使用Android应用程序的基于本地经纪人（MQTT）的发布/订阅

Question

在我当前的项目中，我使用基于MQTT本地代理的实施和设置，使用this link。本地基于经纪人的MQTT实施工作正常。

在最初，我使用Node.js的发布者和用户使用此本地代理，以使信息的交流和它的作品没有互联网连接也（因为这些设备都连接到同一网络，本地券商MQTT基于API实施如http://192.168.0.105）。

现在，我想用Android应用程序替换Node.js订户。我的android应用程序使用mqtt paho库进行通信并使用mqtt API mqtt://test.mosquitto.org:1883。如果我正在使用mqtt开放代理，即比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并将其部署到移动设备时，它显示错误不幸的是，Android应用程序在andoid应用程序中停止了。

如何在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的用户（本地代理基于实现）MQTT用户的更多，更清晰

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()); 
    }  
}); 

共享代码：

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代理。

您需要使用tcp:// URI为Android（这可能为工作的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);