安卓 - 用于发布的Paho MQTT服务

Question

我是Android和服务的新手。我的目标是能够在主题字符串上设置订阅并进行发布。主题字符串和客户端ID是在解析文本字段输入后设置的。我正在使用Paho MQTT服务（下载了源码并构建了JAR）。

以下代码会在c.publish()处引发空指针异常。 logcat显示在MqttAndroidClient中的IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, IMqttActionListener callback)方法处出现了异常，其中一个传递令牌被获取。

public class MainActivity extends Activity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        // Set locale;
        l = getResources().getConfiguration().locale;
    }

    @Override
    protected void onResume() {
        super.onResume();
        addButtonListener();        
    }

    private void addButtonListener() {
        Button submitButton = (Button) findViewById(R.id.buttonSubmit);

        submitButton.setOnClickListener(new OnClickListener() {
// ...
// validation code for fields in layout
// ...
// Finally, this.

                    MemoryPersistence mPer = new MemoryPersistence();
                    String clientId = UUID.randomUUID().toString();
                    String brokerUrl = "tcp://m2m.eclipse.org:1883";
                    MqttAndroidClient c = new MqttAndroidClient(getApplicationContext(), brokerUrl, clientId, mPer);
                    try {
                        c.connect(); 
                        String topic = "transfers/topic";
                        String msg = "topic payload"
                        MqttMessage m = new MqttMessage();
                        m.setPayload(msg.getBytes());
                        m.setQos(2);
                        m.setRetained(false);
                        c.publish(topic, m); 
                    } catch (MqttException e) {
                        Toast.makeText(getApplicationContext(), e.getMessage(), Toast.LENGTH_SHORT).show();
                    }

你能告诉我如何使用服务进行发布和订阅吗？我已经浏览了示例项目（来自Paho Android）。似乎LWT和发布已经合并，因为LWT的布局（activity_publish.xml）也用于发布。

- cogitoergosum

Answer 1

NullPointerException是由于connect()调用了一个异步方法，您需要实现一个ActionListener。

如果成功，您可以发送消息。

Log.i(LOGTAG, "MQTT Start");

MemoryPersistence memPer = new MemoryPersistence();

final MqttAndroidClient client = new MqttAndroidClient(
    context, "tcp://192.168.0.13:1883", username, memPer);

try {
    client.connect(null, new IMqttActionListener() {

        @Override
        public void onSuccess(IMqttToken mqttToken) {
            Log.i(LOGTAG, "Client connected");
            Log.i(LOGTAG, "Topics="+mqttToken.getTopics());

            MqttMessage message = new MqttMessage("Hello, I am Android Mqtt Client.".getBytes());
            message.setQos(2);
            message.setRetained(false);

            try {
                client.publish("messages", message);
                Log.i(LOGTAG, "Message published");

                client.disconnect();
                Log.i(LOGTAG, "client disconnected");

            } catch (MqttPersistenceException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();

            } catch (MqttException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

        @Override
        public void onFailure(IMqttToken arg0, Throwable arg1) {
            // TODO Auto-generated method stub
            Log.i(LOGTAG, "Client connection failed: "+arg1.getMessage());

        }
    });
}

- J3ernhard

---

这真的很有趣，但是怎么知道这个方法会执行异步操作呢？在C#中，如果不将connect()方法标记为async，就无法实现这样的实现，以避免出现此类错误。 - Joao Costa

Answer 2

非常重要的是，为了接收订阅主题的消息，您需要调用client.setCallback并实现MqttCallbackHandler。以下是可以发布和订阅等操作的代码示例。

以下代码最初发布mqtt主题和载荷：

- 主题：AndroidPhone - 载荷：Hello, I am an Android Mqtt Client.

该代码订阅了主题“tester”。如果它收到一个主题为“tester”，载荷为“Alarm Activated”的消息，则通过上述回调发布以下主题和载荷：

- 主题：Fitlet - 载荷：Hello, the Mosquitto Broker got your message saying that the Alarm is Activated.

如果您正在使用Mosquitto，则在终端中运行以下命令将触发此消息。

mosquitto_pub -h 192.168.9.100 -t tester -m "Alarm Activated" -u fred -P 1234

我的Mosquitto用户名为fred，密码为1234

代码：

package colin.android.mqtt;    
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.widget.Toast;
import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import java.io.UnsupportedEncodingException;

public class MainActivity extends AppCompatActivity {
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        String clientId = MqttClient.generateClientId();

        //The URL of the Mosquitto Broker is 192.168.9.100:1883
        final  MqttAndroidClient client = new MqttAndroidClient(this.getApplicationContext(), "tcp://192.168.9.100:1883", clientId);

        client.setCallback(new MqttCallbackHandler(client));//This is here for when a message is received

        MqttConnectOptions options = new MqttConnectOptions();

        try {
            options.setUserName("fred");
            options.setPassword("1234".toCharArray());
            IMqttToken token = client.connect(options);

            token.setActionCallback(new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    // We are connected
                    Log.d("mqtt", "onSuccess");
//-----------------------------------------------------------------------------------------------
                    //PUBLISH THE MESSAGE                    
                    MqttMessage message = new MqttMessage("Hello, I am an Android Mqtt Client.".getBytes());
                    message.setQos(2);
                    message.setRetained(false);

                    String topic = "AndroidPhone";

                    try {
                        client.publish(topic, message);
                        Log.i("mqtt", "Message published");

                        // client.disconnect();
                        //Log.i("mqtt", "client disconnected");

                    } catch (MqttPersistenceException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();

                    } catch (MqttException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
//-----------------------------------------------------------------------------------------------

                    String subtopic = "tester";
                    int qos = 1;
                    try {
                        IMqttToken subToken = client.subscribe(subtopic, qos);
                        subToken.setActionCallback(new IMqttActionListener() {
                            @Override
                            public void onSuccess(IMqttToken asyncActionToken) {
                                // The message was published
                                Log.i("mqtt", "subscription success");
                            }

                            @Override
                            public void onFailure(IMqttToken asyncActionToken,
                                                  Throwable exception) {
                                // The subscription could not be performed, maybe the user was not
                                // authorized to subscribe on the specified topic e.g. using wildcards
                                Log.i("mqtt", "subscription failed");

                            }
                        });



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

//---------------------------------------------------------------------------

                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    // Something went wrong e.g. connection timeout or firewall problems
                    Log.d("mqtt", "onFailure");

                }

            });


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

    }


}//End of Activity class

//-----------------------------------------------------------------------------

class MqttCallbackHandler implements MqttCallbackExtended {

    private final MqttAndroidClient client;

    public MqttCallbackHandler (MqttAndroidClient client)
    {
        this.client=client;
    }

    @Override
    public void connectComplete(boolean b, String s) {
        Log.w("mqtt", s);
    }

    @Override
    public void connectionLost(Throwable throwable) {

    }

    public void AlarmActivatedMessageReceived()
    {
        MqttMessage msg= new MqttMessage("Hello, the Mosquitto Broker got your message saying that the Alarm is Activated.".getBytes());
        try {
            this.client.publish("Fitlet", msg);
            Log.i("mqtt", "Message published");

        } catch (MqttPersistenceException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();

        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        Log.w("mqtt", mqttMessage.toString());

        if (mqttMessage.toString().contains("Alarm Activated"))
        {
            AlarmActivatedMessageReceived();
        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }
}

- CMP

---

如果您使用“Processing”，那么实现MQTT会更容易：http://processing-mqtt.blogspot.com/。Processing代码可在Android手机上运行... - CMP