d11-react-native-mqtt

消息队列遥测传输（MQTT）是一种轻量级、高效且可靠的消息协议，专为低带宽、高延迟或不可靠的网络而设计。

MQTT 基于发布/订阅模式运行，实现了消息发送者（发布者）与消息接收者（订阅者）之间的解耦。这是通过利用一个称为消息代理（message broker）的核心组件来实现的，该组件负责协调发布者与订阅者之间的通信。

此包为 React Native 应用提供 MQTT 功能。

内部实现上，此包在 Android 平台上利用 HiveMQTT，在 iOS 平台上利用 CocoaMQTT，确保在两个平台上都能实现无缝集成和最佳性能。

贡献指南：CONTRIBUTING.md
许可证：MIT
MQTT 资源：
- MQTT Essentials
- MQTT 5.0

功能特性

API 风格：
- 异步 API
背压支持：
- QoS 1 和 2
- QoS 0（必要时丢弃传入消息）
- 将 MQTT 流量控制与响应式拉取背压相结合
传输协议：
- TCP
- SSL/TLS
自动且可配置的线程管理
自动且可配置的重连处理
生命周期监听器
- 连接时
- 断开连接或连接失败时

最低支持版本

此项目需要以下最低版本才能正常运行：

Java：JDK 8 或更高版本
iOS：iOS 12.0 或更高版本
Android：API 级别 24（Android 7.0）或更高版本

安装

using NPM
npm install @d11/react-native-mqtt

using Yarn
yarn add @d11/react-native-mqtt

通用配置

变量名	描述	默认值
topic	主题是一个 UTF-8 编码的字符串，是 MQTT 协议中消息路由的基础	无
clientId	MQTT 客户端的唯一标识符	无
host	MQTT 服务器的主机名或 IP 地址	无
port	MQTT 服务器的端口	无
keepAlive	保持连接存活的时间（以秒为单位）。	60 秒
cleanSession	连接时创建新会话或使用旧会话的标志	true
backoffTime	重试连接前的等待时间	2 秒
maxBackoffTime	重试连接前的最大等待时间	60 秒
jitter	抖动用于在退避时间中添加随机性	1
enableSslConfig	一个布尔值，指示是否应启用 SSL/TLS 配置	false
autoReconnect	一个布尔值，用于确定自动重连	None
retryCount	一个整数，用于确定重试次数	None

服务质量 (QoS)

枚举名称	描述
AT_MOST_ONCE	根据网络能力，实现至多一次投递的 QoS。
AT_LEAST_ONCE	确保至少一次投递的 QoS。
EXACTLY_ONCE	确保恰好一次投递的 QoS。

客户端的创建

import { createMqttClient, MqttConfig } from "@d11/react-native-mqtt";
import { MqttClient } from "@d11/react-native-mqtt/dist/Mqtt/MqttClient";

export const createMqtt = (mqttConfig: MqttConfig): MqttClient => {
  const client = createMqttClient({
      clientId: mqttConfig.clientId,
      host: mqttConfig.host,
      port: mqttConfig.port,
      options: {
        password: '',
        enableSslConfig: false,
        autoReconnect: true,
        maxBackoffTime: mqttConfig.maxBackoffTime,
        retryCount: mqttConfig.connectRetryCount,
        cleanSession: mqttConfig.cleanSession,
        keepAlive: mqttConfig.keepAlive,
        jitter: mqttConfig.jitter,
        username: '',
      },
  });
  return client;
}

示例

import * as React from 'react';
import { StyleSheet, View, Text, Button } from 'react-native';
import { createMqtt } from './components/create-mqtt';
import { MqttClient } from '@d11/react-native-mqtt/dist/Mqtt/MqttClient';
import { MqttConfig } from "@d11/react-native-mqtt";

const mqttConfig: MqttConfig =  {
  "clientId": "mqttx_0121",
  "cleanSession": true,
  "host": "dummy.mqttx.com",
  "jitter": 1,
  "keepAlive": 60,
  "maxBackoffTime": 60,
  "port": 8883,
  "retryCount": 3
}

enum MqttQos {
  AT_MOST_ONCE = 0,
  AT_LEAST_ONCE = 1,
  EXACTLY_ONCE = 2,
}

export default function App() {
  const [client, setClient] = React.useState<MqttClient | undefined>(undefined);

  const createMqttClient = () => {
    const newClient = createMqtt(MqttConfig);
    setClient(newClient);
  };

  const connectMqtt = () => {
    if (client) {
      client.connect();
    }
  };

  const disconnectMqtt = () => {
    if (client) {
      client.disconnect();
    }
  };

  const subscribeMQTT = () => {
    if(client){
      client.subscribe({
        topic: "sample_topic",
        qos: MqttQos.AT_LEAST_ONCE,
        onSuccess: (ack) => {
          console.log(`MQTT Subscription Success: ${ack}`);
        },
        onError: (error) => {
          console.log(`MQTT Subscription Failed: ${error}}`);
        },
        onEvent: ({ payload }) => {
          console.log(`MQTT Subscription data: ${payload}`);
        }
      })
    }
  }

  return (
    <View style={styles.container}>
      <Button
        title="Create Mqtt"
        color={'green'}
        onPress={createMqttClient}
      />
      <Button
        title="Connect Mqtt"
        color={'green'}
        onPress={connectMqtt}
      />

      <Button
        title="Subscribe MQTT"
        color={'green'}
        onPress={subscribeMQTT}
      />
      <Button
        title="Disconnect Mqtt"
        color={'red'}
        onPress={disconnectMqtt}
      />
    </View>
  );
}

const styles = StyleSheet.create({
  container: {
    flex: 1,
    alignItems: 'center',
    justifyContent: 'center',
  }
});

API 支持

createMqttClient: 一个返回给定输入配置的 MqttClient 实例的函数。

createMqttClient: (config: MqttConfig) => MqttClient
const client = createMqttClient(config)

type MqttConfig = {
  clientId: string;
  host: string;
  port: number;
  options?: MqttOptions;
}

type MqttOptions = {
  keepAlive?: number;
  cleanSession?: boolean;
  username?: string;
  password?: string;
  maxBackoffTime?: number;
  backoffTime?: number;
  jitter?: number;
  enableSslConfig?: boolean;
  autoReconnect?: boolean;
  retryCount?: number;
}

connect: 支持 auto-reconnect 和 retry 连接到 MQTT 服务器。

connect(options?: MqttOptions) => void

client.connect()

setOnConnectCallback: 在连接建立后执行操作。

type onConnectCallback = (ack: MqttEventsInterface[MQTT_EVENTS.CONNECTED_EVENT]) => void

client.setOnConnectCallback(onConnectCallback)

MQTT_EVENTS {
  CONNECTED_EVENT = 'connected',
  DISCONNECTED_EVENT = 'disconnected',
  SUBSCRIPTION_EVENT = 'subscription_event',
  SUBSCRIPTION_SUCCESS_EVENT = 'subscribe_success',
  SUBSCRIPTION_FAILED_EVENT = 'subscribe_failed',
}

setOnConnectFailureCallback: 设置用于处理连接失败事件的回调。

type onConnectFailureCallback = (ack: MqttEventsInterface[MQTT_EVENTS.CONNECTED_EVENT]) => void

client.setOnConnectFailureCallback(onConnectFailureCallback)

Mqtt5ReasonCode {
  BAD_USER_NAME_OR_PASSWORD = 134,
  NOT_AUTHORIZED = 135,
  BAD_AUTHENTICATION_METHOD = 140,
  NORMAL_DISCONNECTION = 0,
  DEFAULT = -1,
  CONNECTION_ERROR = -2,
  DISCONNECTION_ERROR = -3,
  SUBSCRIPTION_ERROR = -4,
  UNSUBSCRIPTION_ERROR = -5,
  INITIALIZATION_ERROR = -6,
  RX_CHAIN_ERROR = -7
}

setOnErrorCallback: 设置用于处理错误事件的回调。

type onErrorCallback = (ack: MqttEventsInterface[MQTT_EVENTS.ERROR_EVENT]) => void

client.setOnErrorCallback(onErrorCallback)

export enum MqttErrorType {
  INITIALIZATION = 'INITIALIZATION',
  CONNECTION = 'CONNECTION',
  SUBSCRIPTION = 'SUBSCRIPTION',
  UNSUBSCRIPTION = 'UNSUBSCRIPTION',
  DISCONNECTION = 'DISCONNECTION',
  GENERAL = 'GENERAL',
}

setOnDisconnectCallback: 设置用于处理客户端断开连接事件的回调。

type onDisconnectCallback = (ack: MqttEventsInterface[MQTT_EVENTS.DISCONNECTED_EVENT],
options: DisconnectCallback['options']) => void

client.setOnDisconnectCallback(onDisconnectCallback)

connectInterceptor: 用于拦截连接选项的方法，可能会修改或覆盖这些选项。

type connectInterceptor?: (options?: MqttOptions) => Promise<MqttConnect | undefined>

client.connectInterceptor()

reconnectInterceptor: 用于拦截连接选项的方法，可能会修改或覆盖这些选项。

type reconnectInterceptor?: (mqtt5ReasonCode?: Mqtt5ReasonCode) => Promise<MqttConnect | undefined>

client.reconnectInterceptor()

setOnReconnectInterceptor: 在首次连接失败后，每次重连前执行一次操作。

type reconnectInterceptor = (
  mqtt5ReasonCode?: Mqtt5ReasonCode
) => Promise<MqttConnect | undefined>

client.setOnReconnectInterceptor(reconnectInterceptor)

setOnConnectFailureCallback: 设置用于处理连接失败事件的回调。

type onConnectFailureCallback = (ack: MqttEventsInterface[MQTT_EVENTS.CONNECTED_EVENT]) => void

client.setOnConnectFailureCallback(onConnectFailureCallback)

getConnectionStatus: 获取当前连接状态。

getConnectionStatus: () => string
const status = client.getConnectionStatus();

getCurrentRetryCount: 获取当前重试次数。

getConnectionStatus: () => number
const status = client.getCurrentRetryCount();

subscribe: 订阅主题，支持 onEvent、onSuccess、onError 回调。

subscribe: ({ topic, qos, onEvent, onSuccess, onError}: Subscribe) => void

client.subscribe({
  topic,
  qos,
  onEvent,
  onSuccess,
  onError,
})

type Subscribe = {
  topic: string;
  qos?: MqttQos;
  onEvent: (
    payload: MqttEventsInterface[MQTT_EVENTS.SUBSCRIPTION_EVENT]
  ) => void;
  onSuccess?: (
    ack: MqttEventsInterface[MQTT_EVENTS.SUBSCRIPTION_SUCCESS_EVENT]
  ) => void;
  onError?: (
    error: MqttEventsInterface[MQTT_EVENTS.SUBSCRIPTION_FAILED_EVENT]
  ) => void;
}

disconnect: 与 MQTT 服务器断开连接。

disconnect: () => void

client.disconnect();

remove: 移除特定客户端。

remove: () => void

client.remove();

它是如何工作的？

Alt text

运行示例应用

# install all packages
yarn

# for android
yarn example android

# for ios
yarn example ios

贡献

请参阅贡献指南，了解如何为仓库做出贡献以及开发工作流程。

许可证

MIT