android - 使用 Android 应用程序的基于本地代理 (MQTT) 的发布/订阅

Question

在我当前的项目中，我正在使用基于 MQTT 本地代理的实现和设置 this link .基于本地代理的 MQTT 实现工作完美。

最初，我使用 Node.js 发布者和订阅者与这个本地代理来启用消息交换，它也可以在没有互联网连接的情况下工作(因为这些设备连接到同一个网络，mqtt API 用于基于本地代理的实现就像 http://192.168.0.105)。

现在，我想用 Android 应用程序替换 Node.js 订阅者。我的 android 应用程序使用 mqtt paho 库进行通信并使用 mqtt APImqtt://test.mosquitto.org:1883。如果我使用的是 mqtt 开放式代理，即 mosquitto，那么 Node.js 发布者和 Android 应用程序订阅者都可以正常工作(这种情况是使用 API mqtt://test.mosquitto.org:1883< 讨论基于 mosquitto 的 mqtt 实现)。

现在，我想在 Android 应用程序中使用基于本地代理的实现。所以我必须用 http://192.168.0.105 替换 mqtt://test.mosquitto.org:1883 API，因为我想使用基于本地代理的 mqtt 实现。当我在android代码中使用http://192.168.0.105 API并部署到移动设备时，在andoid应用程序中显示Unfortunately, Android Application has stopped错误。

如何在 android 应用程序中使用基于本地代理的实现？

另外分享 Node.js 发布者和订阅者的代码。 Node.js 发布者(基于本地代理的实现)

var mqtt = require('mqtt');
var client = mqtt.connect('http://192.168.0.105');
setInterval(function() {
var data = {
     "tempValue" : Math.random(),
     "unitOfMeasurement" : 'C'
     };
client.publish('tempMeasurement', JSON.stringify(data));
}, 5000);

Node.js 订阅者(基于本地代理的实现)

var mqtt = require('mqtt');
var client = mqtt.connect('http://192.168.0.105');
client.subscribe('tempMeasurement'); 
client.on('message', function(topic, payload) {
       if (topic.toString() == "tempMeasurement") {  
       console.log("Mesage Received "+ payload.toString());
   }       
 });

共享 MQTT Subscriber 代码以便更清晰:

package iotsuite.pubsubmiddleware;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.internal.MemoryPersistence;

public class MQTTSubscriber {

// URI for open MQTT broker
// public static final String BROKER_URL = "tcp://test.mosquitto.org:1883";
public static final String BROKER_URL = "mqtt://192.168.0.105";

private MqttClient client;

public MQTTSubscriber(PubSubMiddleware pubsub) {
    try {
        client = new MqttClient(BROKER_URL, MqttClient.generateClientId(),
                new MemoryPersistence());

        client.setCallback(new PushCallback(pubsub));
        client.connect();

    } catch (MqttException e) {
        e.printStackTrace();
    }
}

public void subscribe(String topicName) throws MqttException {

    client.subscribe(topicName);

    System.out.println("Subscribed. Topic: " + topicName);
}

}

并使用以下代码订阅 tempMeasurement:

package framework;

 import iotsuite.common.Logger;
 import iotsuite.pubsubmiddleware.PubSubMiddleware;
 import iotsuite.pubsubmiddleware.Subscriber;
 import iotsuite.semanticmodel.Device;
 import android.util.Log;
 import com.google.gson.Gson;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;

 public abstract class SmartHomeApp implements Runnable, Subscriber {

protected final PubSubMiddleware myPubSubMiddleware;
protected final Device myDeviceInfo;
Gson gson = new Gson();

public SmartHomeApp(PubSubMiddleware pubSubM, Device deviceInfo) {

    this.myPubSubMiddleware = pubSubM;
    this.myDeviceInfo = deviceInfo;
    postInitialize();

}

protected void postInitialize() {
    subscribeDisplayTemp();

}

@Override
public void notifyReceived(String eventName, Object arg) {
    try {

        if (eventName.equals("tempMeasurement")) {
            Logger.log(myDeviceInfo.getName(), "TempMonitoringApp",
                    "Notification Received tempMeasurement ");
            JsonObject jsonObject = new JsonParser().parse(arg.toString())
                    .getAsJsonObject();
            double tempValue = jsonObject.get("tempValue").getAsDouble();
            // double yahootempValue =
            // jsonObject.get("yahootempValue").getAsDouble();
            TempStruct tempStruct = new TempStruct(tempValue, "C");
            System.out.println("tempValue is " + tempValue
                    + " in framework");
            Log.i("tempValue", "tempValue is received in framework");
            onNewDisplayTempNotify(tempStruct);
        }

    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Override
public void run() {
}

public abstract void onNewDisplayTempNotify(TempStruct arg);

public void subscribeDisplayTemp() {

    this.myPubSubMiddleware.subscribe(this, "tempMeasurement");
}

}

使用上面的代码，我想显示从 Node.js 代码接收到的 tempValue 到 Android 应用程序。来自 Android 的错误如下:

01-04 23:03:26.786: E/AndroidRuntime(21793): FATAL EXCEPTION: main
01-04 23:03:26.786: E/AndroidRuntime(21793): Process: com.example.android,      PID: 21793
01-04 23:03:26.786: E/AndroidRuntime(21793): java.lang.RuntimeException:    Unable to start activity         ComponentInfo{com.example.android/com.example.android.MainActivity}:       java.lang.IllegalArgumentException
01-04 23:03:26.786: E/AndroidRuntime(21793):    at    android.app.ActivityThread.performLaunchActivity(ActivityThread.java:2426)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java:2490)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.ActivityThread.-wrap11(ActivityThread.java)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1354)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.os.Handler.dispatchMessage(Handler.java:102)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.os.Looper.loop(Looper.java:148)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.ActivityThread.main(ActivityThread.java:5443)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at java.lang.reflect.Method.invoke(Native Method)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:728)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:618)
01-04 23:03:26.786: E/AndroidRuntime(21793): Caused by: java.lang.IllegalArgumentException
01-04 23:03:26.786: E/AndroidRuntime(21793):    at  org.eclipse.paho.client.mqttv3.MqttClient.validateURI(MqttClient.java:204)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at org.eclipse.paho.client.mqttv3.MqttClient.<init>(MqttClient.java:175)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at iotsuite.pubsubmiddleware.MQTTPublisher.<init>(MQTTPublisher.java:21)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at iotsuite.pubsubmiddleware.PubSubMiddleware.<init>(PubSubMiddleware.java:58)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at      iotsuite.pubsubmiddleware.IoTSuiteFactory.getInstance(IoTSuiteFactory.java:16)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at  sim.deviceD9.Startup.setUpNode(Startup.java:25)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at sim.deviceD9.Startup.startDevice(Startup.java:63)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at com.example.android.MainActivity.onCreate(MainActivity.java:21)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.Activity.performCreate(Activity.java:6259)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.Instrumentation.callActivityOnCreate(Instrumentation.java:1130)
01-04 23:03:26.786: E/AndroidRuntime(21793):    at android.app.ActivityThread.performLaunchActivity(ActivityThread.java:2379)
01-04 23:03:26.786: E/AndroidRuntime(21793):    ... 9 more

Answer 1

问题是您的 URI 使用的是 http:// 方案。 http:// 用于连接到 HTTP 服务器而不是 MQTT 代理。

您需要为 Android 使用 tcp:// URI(这可能也适用于 NodeJS，但 mqtt:// 绝对适用于 NodeJS

mqtt://192.168.0.105

var mqtt = require('mqtt');
var client = mqtt.connect('mqtt://192.168.0.105');
setInterval(function() {
var data = {
     "tempValue" : Math.random(),
     "unitOfMeasurement" : 'C'
     };
client.publish('tempMeasurement', JSON.stringify(data));
}, 5000);