Subscribe and Read MQTT Message Using PAHO

Goot picture Goot · Mar 28, 2014 · Viewed 48.6k times · Source

I'm using paho to send and receive mqtt messages. So far it has been no problem to send the messages, I'm receiving them by using mosquitto.

Now I want to read the messages by using a java client and I noticed that there has been less documentation about receiving the messages.

I implemented the MqttCallback interface but I still couldn't figure out how to read a message of a topic I've subscribed to.

This is my sourcecode so far, I can read the messages using mosquitto_sub.

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class PahoDemo implements MqttCallback {
    MqttClient client;
    MqttClient subClient;

    public PahoDemo() {
    }

    public static void main(String[] args) {
        new PahoDemo().doDemo();
    }

    public void doDemo() {
        try {
            client = new MqttClient("tcp://192.168.118.11:1883", "Sending");
            subClient = new MqttClient("tcp://192.168.118.11:1883",
                    "Subscribing");
            client.connect();
            subClient.connect();
            subClient.subscribe("foo");
            MqttMessage message = new MqttMessage();
            message.setPayload("A single message from my computer fff"
                    .getBytes());
            client.publish("foo", message);
            client.disconnect();
            client.close();
            subClient.disconnect();
            subClient.close();
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        // TODO Auto-generated method stub

    }

    @Override
    public void messageArrived(String topic, MqttMessage message)
            throws Exception {
 System.out.println(message);       
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // TODO Auto-generated method stub

    }

}

Answer

hardillb picture hardillb · Mar 28, 2014

You are closing the client down before the broker has time to send the message back.

Also you don't need 2 instance of the client, you can send and receive with just one.

I've edited your code a little, it now will continue to run and receive messages until you kill it.

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class PahoDemo implements MqttCallback {

MqttClient client;

public PahoDemo() {
}

public static void main(String[] args) {
    new PahoDemo().doDemo();
}

public void doDemo() {
    try {
        client = new MqttClient("tcp://192.168.118.11:1883", "Sending");
        client.connect();
        client.setCallback(this);
        client.subscribe("foo");
        MqttMessage message = new MqttMessage();
        message.setPayload("A single message from my computer fff"
                .getBytes());
        client.publish("foo", message);
    } catch (MqttException e) {
        e.printStackTrace();
    }
}

@Override
public void connectionLost(Throwable cause) {
    // TODO Auto-generated method stub

}

@Override
public void messageArrived(String topic, MqttMessage message)
        throws Exception {
 System.out.println(message);   
}

@Override
public void deliveryComplete(IMqttDeliveryToken token) {
    // TODO Auto-generated method stub

}

}

EDIT: added the missing client.setCallback(this)