Re: [mqtt] Using of Client persistence

[Nicholas O'Leary]

Hi,

Paho includes a reference implementation - MqttDefaultFilePersistence.java.

Have a look at that, along with the documentation for the interface in
MqttClientPersistence.java

That should hopefully get you started.

Regards,
Nick

On 21 February 2013 09:32, dileep <dile...@gmail.com> wrote:
> Hi all,
>
> Can anybody has reference how to implement client persistence concept using
> eclipse paho (for java client).
>
> --
> --
> To learn more about MQTT please visit http://mqtt.org
>
> To post to this group, send email to mq...@googlegroups.com
> To unsubscribe from this group, send email to
> mqtt+uns...@googlegroups.com
>
> For more options, visit this group at
> http://groups.google.com/group/mqtt
>
> ---
> You received this message because you are subscribed to the Google Groups
> "MQ Telemetry Transport" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to mqtt+uns...@googlegroups.com.
> For more options, visit https://groups.google.com/groups/opt_out.
>
>

[dileep]

Hi knolleary ,

 

My aim to retrieve the messages that are published when he subscriber is not connected to the server . I think it can be achieved by using client persistence concept . I have been searching for implementation of this which I didnt found. I am using mosquitto server and paho client(java) . Can you give any reference or implementation how to persists and retrieve of perisisted messages .

 

 

[Roger Light]

Hi,

> My aim to retrieve the messages that are published when he subscriber is not
> connected to the server .

What you need is the "clean session" option. Setting clean session to
false in the client before it connects means that the server will
regard it as a durable client. When the client disconnects, any
subscriptions it have will be kept and if the QoS of the subscription
is greater than 0, then messages that match the subscription will be
queued. When the client reconnects (assuming that clean session is
still set to false), then the server will deliver the messages.

I'm not familiar with the paho client so I can't tell you how to set
clean session, but I assume it's very easy.

Cheers,

Roger

[dileep]

Hi Roger,

I have tried the above but I am not able to get the old messages... Do I need to do any setting in the  broker ... Because I had made clean session equal to false , qos=2 in both the publisher and subscriber I am still not able to get the old messages when I published before the subscriber connects to the server

[Karl Palsson]

> I have tried the above but I am not able to get the old messages... Do I
> need to do any setting in the broker ... Because I had made clean session
> equal to false , qos=2 in both the publisher and subscriber I am still not
> able to get the old messages when I published before the subscriber
> connects to the server

The subscriber has to connect at least once first. Before that, brokers have no way of
knowing that they should be keeping the messages. Once a clientid with
cleansession=false has connected, they can then disconnect whenever they like and receive
messages published while they were away.

Cheers,
Karl P

[dileep]

Tried it , But didnt worked out. Do I need to persist the data so that the broker will read the persisted data and send it the client whenever it connects again. 

 

 

 

[Nicholas O'Leary]

Hi,

What you are trying to do should just work, so lets go back to basics.

Here is the most simple example demonstrating how durable subscriptions work:

1. Connect the subscriber client to the broker, with clean-session set
to false. By default, the Paho java client uses clean-session set to
true, so you must pass in an instance of MqttConnectOptions to the
connect method having set clean-session to false.
2. Subscribe the client to the topic you want at Qos 1 or 2.
3. Disconnect the client.
4. From another client, publish messages to the topic at Qos 1 or 2.
5. Re-connect the subscriber. You must use the same Client ID as
before and clean-session must still be false. Ensure you have
registered a callback handler (MqttClient.setCallback) before you
connect. As soon as you connect, the messages will arrive.

How does that compare to what you are trying?

Cheers,
Nick

[dileep]

Tried the above lines but didnt got the packets..... Plz check the below code

Subscriber :

import java.io.IOException;
import java.sql.Timestamp;

import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;


public class MQTTListener implements MqttCallback {
    private MqttClient client;
    private MqttConnectOptions mConOpt;
    private String url = "";
   

public MQTTListener(String url, String clientId, String topic, int qos) {
   
    this.url=url;
    try {
        mConOpt = new MqttConnectOptions();
        mConOpt.setCleanSession(false);
        client = new MqttClient(url, clientId);
        client.setCallback(this);
        subscribe(topic, qos);
    } catch (MqttException e) {
        e.printStackTrace();
        // System.exit(1);
    }
   
}
    public void subscribe(String topicName, int qos) throws MqttException {
        // Connect to the server
        client.connect();
        System.out.println("Connected to "+url+" with client ID "+client.getClientId());
        System.out.println("Subscribing to topic \""+topicName+"\" qos "+qos);
        client.subscribe(topicName, qos);
        try {
            System.in.read();
        }catch (IOException e) {
            //If we can't read we'll just exit
        }
        // Disconnect the client
        client.disconnect();
        System.out.println("Disconnected");
    }

    @Override
    public void connectionLost(Throwable arg0) {
        // TODO Auto-generated method stub
        System.out.println("Connection to " + url + " lost!" + arg0);
    }

    @Override
    public void deliveryComplete(MqttDeliveryToken arg0) {
        // TODO Auto-generated method stub
        
    }

    @Override
    public void messageArrived(MqttTopic topic, MqttMessage message)
            throws MqttException {
        // TODO Auto-generated method stub
        String time = new Timestamp(System.currentTimeMillis()).toString();

        System.out.println("Time:\t" +time +
                           "  Topic:\t" + topic.getName() +
                           "  Message:\t" + new String(message.getPayload()) +
                           "  QoS:\t" + message.getQos());
    }
   
    public static void main(String[] args) {
        String broker="localhost";
        String port="1883";
        String clientId="Listener1";
        String topic="AAA";
        String url="";
        int qos=2;
        url="tcp://"+broker+":"+port;
        try{
    new MQTTListener(url, clientId, topic, qos);
        }catch(Exception e){
            e.printStackTrace();
        }
    }

}





Publisher :

import java.sql.Timestamp;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;


public class MQTTSender {
    private String url = "";
    private MqttClient client;
    private MqttConnectOptions mConOpt;
    public MQTTSender(String url,String clientId,int qos,String topic,String message) throws MqttException {
        // TODO Auto-generated constructor stub
        mConOpt = new MqttConnectOptions();
        mConOpt.setCleanSession(false);
        client = new MqttClient(url, clientId, null);       
        publish(topic, qos, message.getBytes());
    }
     public void publish(String topicName, int qos, byte[] payload) throws MqttException {
           
            client.connect();
            System.out.println("Connected to "+url + " with client ID "+client.getClientId());   
            MqttTopic topic = client.getTopic(topicName);
            MqttMessage message = new MqttMessage(payload);
            message.setQos(qos);
               String time = new Timestamp(System.currentTimeMillis()).toString();
            System.out.println("Publishing at: "+time+ " to topic \""+topicName+"\" qos "+qos);
            MqttDeliveryToken token = topic.publish(message);
            token.waitForCompletion();
            client.disconnect();
            System.out.println("Disconnected");

         }
    public static void main(String[] args) {
        String broker="localhost";
        String port="1883";
        String url="";
        String topic="AAA";
        String clientId="Sender1";
        String message="Hello World";
        int qos=2;
        url="tcp://"+broker+":"+port;
        try {
            new MQTTSender(url, clientId, qos, topic, message);
        } catch (MqttException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

 

[Nicholas O'Leary]

Hi,

in MQTTListener, you create an instance of MqttConnectOptions:

mConOpt = new MqttConnectOptions();
mConOpt.setCleanSession(false);

but you don't then use it. You need to pass this object in the call to
client.connect() in your subscribe method.

Without this, the listener is still connecting with the default
options - ie CleanSession is true.


Cheers,
Nick

[dileep]

Thanks a lot .. Finally got it..

[Dave Locke]

there is a bug in the paho MQTT Java sample that is the same as this i.e. an MqttConnectOptions object is created but not passed on the connect method.  Will fix this.


All the best
Dave

Unless stated otherwise above:
IBM United Kingdom Limited - Registered in England and Wales with number 741598.
Registered office: PO Box 41, North Harbour, Portsmouth, Hampshire PO6 3AU