android - Paho MQTT service for publishing

Question

I am new to Android and services. My aim is to be able to set-up subscriptions and do publications on topic strings. The topic strings and client ID are set-up after parsing input of text fields. I am using the Paho MQTT service (downloaded the source and built the JAR).

The following causes a Null Pointer Exception at c.publish(). The logcat shows the exception at the IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, IMqttActionListener callback) method in MqttAndroidClient where a delivery token is being taken.

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

Can you please tell me how to use the service to publish and subscribe ? I did browse through the sample project (from Paho Android). The LWT and publish seems to be merged as the layout for LWT (activity_publish.xml) seems to be used for publication as well.

Answer 1

The NullPointerException is because connect() calls an asynchronous method and you need to implement an ActionListener. In case of success you could send messages.

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

It is very important to understand that you need to call client.setCallback and implement MqttCallbackHandler in order to receive messages on the topics to which you subscribed.

Here is an example of code that can both publish and subscribe etc.

The following code initially publishes the mqtt topic and payload:

• Topic: AndroidPhone
• Payload: Hello, I am an Android Mqtt Client.

The code subscribes to the topic "tester". If it receives a message with topic "tester" and payload of "Alarm Activated" then it publishes the following topic and payload (via the callback mentioned above):

• Topic: Fitlet
• Payload: Hello, the Mosquitto Broker got your message saying that the Alarm is Activated.

If you are using Mosquitto then the following command in the terminal would cause this message to be fired out:

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

Where my Mosquitto username is fred and my password is 1234

The Code:

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) {

    }
}