控制器研发日志

MQTT

Message Queuing Telemetry Transport 是物联网通讯互联协议中被广泛使用的一种。
International Business Machines 公司的 Andy Stanford-Clark 和 Arcom ( Eurotech ) 公司的 Arlen Nipper 在1999年开发了 MQTT 协议，其3.1.1和5.0版本已经成为 Organization for the Advancement of Structured Information Standards 标准。
MQTT 协议也一种是建立连接后的通讯协议，每个用户首先需要与服务器建立连接，服务器是一个中间环节。当与服务器建立连接后，还需要一个额外的步骤，即订阅 Subscribe 主题，每个主题可以被多个用户订阅，每个用户也可以订阅多个主题。
订阅了同一个主题的用户，无论谁向这个主题发送消息 publish，其他所有已经订阅该主题的用户都会收到该消息，这就好似大家建立了一个小社团，而服务器就是这个社团的秘书或称作中间代理人 broker。

如果要实现点对点的通讯：

• 可以两个人订阅一个只有这两人知道的唯一主题。
• 或是在主题中加入对方的ID信息，服务器辨识发来信息中的主题中如果含有ID信息，则将仅把消息转发给该ID用户。

• Android： Eclipse Paho
• Windows： C# M2MQTT.NET
• iOS and iPadOS： CocoaMQTT

Android with Java to control Camellia Mini Controller which are all connecting to a MQTT Broker (Server) using Eclipse Paho

• 知识短片

1. 检查 Wi-Fi 硬件

Code in Activity:

MainActivity activity = (MainActivity) getActivity();
Context mcontext = activity.getApplicationContext();
ConnectivityManager mConnectivityManager = (ConnectivityManager) mcontext.getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo mWiFiNetworkInfo = mConnectivityManager.getNetworkInfo(ConnectivityManager.TYPE_WIFI);

if( mWiFiNetworkInfo != null && mWiFiNetworkInfo.isConnected() ) {
   WifiManager mWifiManager = (WifiManager) mcontext.getSystemService(Context.WIFI_SERVICE);
   WifiInfo wifiInfo = mWifiManager.getConnectionInfo();
   // ...
}
else{
   // ...
}

2. 定义变量

Code in Activity:

private MqttAndroidClient mqttAndroidClient;
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();

public static final String serverUri = "tcp://mqtt-xxxxxx.mqtt.xxxxxx.com:1883";
public static String clientId = "GID_Android@@@xxxxxx";
public static final String instanceId = "xxxxxx";
public static final String accessKey = "xxxxxx";
public static final String secretKey = "xxxxxx";
public String topic = "CamelliaIoT"; 

3. 连接到服务器

Code in Activity:

mqttConnectOptions.setConnectionTimeout(3000);
mqttConnectOptions.setKeepAliveInterval(90);
mqttConnectOptions.setAutomaticReconnect(true);
mqttConnectOptions.setCleanSession(true);
mqttConnectOptions.setUserName(accessKey);
mqttConnectOptions.setPassword(secretKey);
try {
     mqttAndroidClient.connect(mqttConnectOptions, null, new IMqttActionListener() {
          @Override
          public void onSuccess(IMqttToken asyncActionToken) {
               Log.w("connect", "Connect on Success");
              // subscribeToTopic();
         }
   
         @Override
         public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
              Log.e("connect", "Connect on Failure", exception);
         }
     });
}
catch (MqttException e) {
     Log.e("connect", "exception", e);
}

4. 订阅主题

Code in Main Class:

try {
  final String topicFilter[] = topic;
  final int[] qos = {1};
  mqttAndroidClient.subscribe(topicFilter, qos, null, new IMqttActionListener() {
    @Override
    public void onSuccess(IMqttToken asyncActionToken) {
      Log.w("subscribe", "Subscribe on Success");
    }
    @Overrid
    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
      Log.e("subscribe", "Subsribe on Failure", exception);
    }
  });
}
catch (MqttException ex) {
  Log.e("subscribe", "exception", ex);
}

5. 发送数据

Code in Activity:

try {
  if(mqttAndroidClient.isConnected()) {
    MqttMessage message = new MqttMessage();
    final String msg = "Hello, Camellia!";
    message.setPayload(msg.getBytes());
    mqttAndroidClient.publish(topic, message, null, new IMqttActionListener() {
      @Override
      public void onSuccess(IMqttToken asyncActionToken) {
       Log.w("publish", "Publish successfully:" + msg); 
      }
      @Override
      public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
        Log.w("publish", "Publish failed!");
      }
    });
  }
}
catch (MqttException e) {
  Log.e("publish", "exception", e);
}

6. 接收数据

Code in Activity:

mqttAndroidClient.setCallback(new MqttCallbackExtended() {
  // Fire Automatically
  @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        byte[] buf = new byte[1024];
        buf = message.getPayload();
        String indicateChinese = new String(buf,"GB18030");
        String indicateHEX = Integer.toHexString(buf[n]);
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //when connect success, need resub
    }
    @Override
    public void connectionLost(Throwable cause) {
        Log.e("close", "Connection Lost", cause);
    }
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    }
});

7. 线程间数据的通讯

Code in Activity:

private Handler handler = new Handler() {
   @Override
   public void handleMessage(Message msg) {
      // ...
   }
}

Code in Thread:

Message msg = new Message();
handler.sendMessage(msg);

8. 断开连接

try {
  if (mqttAndroidClient.isConnected())
  {
    mqttAndroidClient.disconnect();
  }
}
catch (MqttException e) {
  Log.e("Disconnect", "exception", e);
}

Windows with C# to control Camellia Mini Controller which are all connecting to a MQTT Broker (Server) using C# M2MQTT.NET

• 知识短片

1. 检查 Wi-Fi 硬件（如果采用有线连接，可以忽略此步骤）

Code in Main Class:

NetworkInterface[] nics = NetworkInterface.GetAllNetworkInterfaces();
string ipAddress = null;
foreach (NetworkInterface adapter in nics)
{
   /// Wi-Fi
   if (adapter.NetworkInterfaceType == NetworkInterfaceType.Wireless80211)
   {
      IPInterfaceProperties ip = adapter.GetIPProperties();
      UnicastIPAddressInformationCollection ipCollection = ip.UnicastAddresses;
      foreach (UnicastIPAddressInformation ipadd in ipCollection)
      {
         if (ipadd.Address.AddressFamily == AddressFamily.InterNetwork)
         { 
            ipAddress = ipadd.Address.ToString();
         }
      }
   }
}

2. 定义变量

Code in Main Class:

private MqttClient mqttclient;
public String instanceId = "mqtt-xxxxxx";
public String brokerUrl = "mqtt-xxxxxx.mqtt.xxxxxx.com";
public String accessKey = "xxxxxx";
public String secretKey = "xxxxxx";
public String parentTopic = "CamelliaIoT";
public String[] topic = { "CamelliaIoT" };
public String clientId = "GID_Windows@@@xxxxxx";

3. 连接到服务器

Code in Main Class:

try {
    mqttclient = new MqttClient(brokerUrl);
}
catch (Exception error){
    Console.WriteLine("MQTT initial failed: " + error);
}

if ( mqttclient != null ){
    mqttclient.MqttMsgSubscribed += client_subscribedSuccess;
    mqttclient.MqttMsgPublishReceived += client_recvMsg;
    mqttclient.MqttMsgPublished += client_publishSuccess;
    mqttclient.ConnectionClosed += client_connectLose;
    String userName = accessKey;
    String passWord = secretKey;
    mqttclient.Connect(clientId, userName, passWord, true, 60);
}

4. 订阅主题

Code in Main Class:

if (mqttclient.IsConnected) {
  Console.WriteLine("MQTT Connected!");
  string[] subTopicArray = { parentTopic };
  byte[] qosLevels = { MqttMsgBase.QOS_LEVEL_AT_MOST_ONCE };
  try {
    mqttclient.Subscribe(subTopicArray, qosLevels);
  }
  catch (Exception error) {
    Console.WriteLine("MQTT subscribe failed: " + error);
  }
}

5. 发送数据

Code in Main Class:

if ((mqttclient != null) && (mqttclient.IsConnected)) {
  try {
    byte[] data = new byte[1024];
    mqttclient.Publish(parentTopic, data, MqttMsgBase.QOS_LEVEL_AT_LEAST_ONCE, false); 
  } 
  catch (Exception error) {
    Console.WriteLine("MQTT Send Error: " + error.Message); 
  }
}

6. 接收数据

Code in Main Class:

mqttclient.MqttMsgPublishReceived += client_recvMsg;
// 自动触发
public void client_recvMsg(object sender, MqttMsgPublishEventArgs e)
{
    string indicateHEX = "";
    string indicate = "";
    indicateHEX = ToHexString(e.Message);
    indicate = Encoding.UTF8.GetString(e.Message);
}

7. 断开连接

if ((mqttclient != null) && (mqttclient.IsConnected)) {
  try {    
    mqttclient.Disconnect();    
    mqttclient = null;  
  }  
  catch (Exception error) {
  }
}

iOS and iPadOS with Swift to control Camellia Mini Controller which are all connecting to a MQTT Broker (Server) using CocoaMQTT

• 知识短片

1. 检查 Wi-Fi 硬件

Code in Main Class:

let reachability = try? Reachability() 
if(reachability != nil){
   if(reachability?.connection == .wifi){
       print("Reachable via WiFi!")
   }
}

let mresult = getMAC()
if((mresult.success)){
   mdevice.text = mresult.ssid
   print("Wi-Fi is OK!")
}

2. 定义变量

Code in Main Class:

var mmqtt = CocoaMQTT(clientID: "GID_iOS@@@", host: "localhost", port: 1883)
var instanceId: String = "mqtt-xxxxxx"
var brokerUrl: String = "mqtt-xxxxxx.mqtt.axxxxxx.com"
var accessKey: String = "xxxxxx"
var secretKey: String = "xxxxxx"
var parentTopic: String = "CamelliaIoT"
var clientId: String = "GID_iOS@@@xxxxxx"

3. 连接到服务器

Code in Main Class:

mmqtt = CocoaMQTT(clientID: clientId, host: brokerUrl, port: 1883)
mmqtt.username = accessKey
mmqtt.password = secretKey
mmqtt.keepAlive = 60
mmqtt.delegate = self
mmqtt.connect()

4. 订阅主题

Code in Main Class:

mmqtt.subscribe(parentTopic, qos: CocoaMQTTQoS.qos0)

5. 发送数据

Code in Main Class:

print("Publish MQTT message: ")
var senddata: [UInt8] = [ 2 , 0 , 1 , 4 , 0 , 6 , 1 , 5 ]
let sendmessage = CocoaMQTTMessage(topic: parentTopic, payload: senddata,qos: .qos0,retained: false)
mmqtt.publish(sendmessage)

6. 接收数据

Code in Main Class:

extension PanelViewController: CocoaMQTTDelegate{

    func mqtt(_ mqtt: CocoaMQTT, didReceiveMessage message: CocoaMQTTMessage, id: UInt16) {
        let receivemessage: String = message.payload;
        print("MQTT receive Message: in topic -  \(message.topic) with payload - \(message.payload)")  
    }

    // 回调函数   
    func mqtt(_ mqtt: CocoaMQTT, didConnectAck ack: CocoaMQTTConnAck) {
        print("MQTT connect successfully!")
        mmqtt.subscribe(parentTopic, qos: CocoaMQTTQoS.qos0)
    }

    func mqtt(_ mqtt: CocoaMQTT, didPublishMessage message: CocoaMQTTMessage, id: UInt16) {
        print("MQTT publish successfully!")   
    }

    func mqtt(_ mqtt: CocoaMQTT, didPublishAck id: UInt16) {  
    }

    func mqtt(_ mqtt: CocoaMQTT, didReceiveMessage message: CocoaMQTTMessage, id: UInt16) {
        print("MQTT receive Message: in topic -  \(message.topic) with payload - \(message.payload)")  
    }

    func mqtt(_ mqtt: CocoaMQTT, didSubscribeTopics topics: [String]) {      
        print("MQTT subscribe successfully!")    
    }

    func mqtt(_ mqtt: CocoaMQTT, didUnsubscribeTopics topics: [String]) {     
    }

    func mqttDidPing(_ mqtt: CocoaMQTT) {    
    }

    func mqttDidReceivePong(_ mqtt: CocoaMQTT) {    
    }

    func mqttDidDisconnect(_ mqtt: CocoaMQTT, withError err: Error?) {     
    }
}

7. 断开连接

iOS 和 iPadOS 不需要断开连接。

所有核心代码