android - 用于发布的Paho MQTT服务

Question

我是Android和服务的新手.我的目标是能够设置订阅并在主题字符串上做出版物.解析文本字段的输入后,将设置主题字符串和客户端ID.我正在使用Paho MQTT服务(下载源并构建JAR).

以下导致Null指针异常c.publish().在logcat示出了异常的IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, IMqttActionListener callback)在方法MqttAndroidClient被截取的输送令牌其中.

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)的布局似乎也用于发布.

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

        }
    });
}

Answer 2

了解您需要调用 client.setCallback 并实现 MqttCallbackHandler 以接收有关您订阅的主题的消息非常重要。

这是一个可以发布和订阅等的代码示例。

以下代码最初发布 mqtt 主题和有效负载：

• 话题：安卓手机
• 有效负载：您好，我是 Android Mqtt 客户端。

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

• 主题： Fitlet
• Payload：您好，Mosquitto Broker 收到您的消息，说警报已激活。

如果您使用的是 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) {

    }
}