MQTT callback client Reconnection logic

Question

I am unable to find a logic for reconnection of mqtt call back client. There is method onDisconnected() but I am unable to find documentation or any sample example on internet.

My Listener

public class MyListener implements Listener {

    public MyListener()
    {

    }

    @Override
    public void onConnected()
    {
        System.out.println("Connected ....");
    }

    @Override
    public void onDisconnected()
    {
        System.out.println("Disconnected");
    }

    @Override
    public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack)
    {
        System.out.println("Entered Onpublish");

        try
        {
         System.out.println("received msg:" + msg);
        }
        catch (HikeException e)
        {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        finally{
            ack.run();
        }

    }


    @Override
    public void onFailure(Throwable value)
    {
        value.printStackTrace();
    }

}

Create Connection

private void createConnection(String host, int port,String id, String token) throws Exception
{

    this.disconnect();
    MQTT mqtt = new MQTT();
    mqtt.setHost(host, port);
    mqtt.setUserName(id);
    mqtt.setPassword(token);
    CallbackConnection callbackConnection = null;
    callbackConnection = mqtt.callbackConnection();
    callbackConnection.listener(new MyListener());
    callbackConnection.connect(new MyCallback<Void>("CONNECT"));
    callbackConnection.subscribe(new Topic[] { new Topic(uid + "/u", QoS.AT_MOST_ONCE) }, new MyCallback<byte[]>("EVENT SUBSCRIBE"));
    callbackConnection.subscribe(new Topic[] { new Topic(uid + "/s", QoS.AT_LEAST_ONCE), new Topic(uid + "/a", QoS.AT_LEAST_ONCE) }, new MyCallback<byte[]>("MSG SUBSCRIBE"));

    this.callbackConnection = callbackConnection;
}

MyCallback

class MyCallback<T> implements Callback<T>
{
    public MyCallback(String tag)
    {
        super();
        this.tag = tag;
    }

    String tag;

    @Override
    public void onSuccess(T value)
    {
        System.out.println("TAG:" + tag + " =SUCCESS value=" + value);
    }

    @Override
    public void onFailure(Throwable value)
    {
        System.out.println("TAG:" + tag + "Fail");
        value.printStackTrace();
    }

}

My question is how to implement mqtt reconnection to server logic ? If I should use onDisconnect() method, then how I can use it ?

Answer 1

I have implemented this way

        //check when network connectivity is back and implement the connection logic again
        System.out.println("Connection Lost\n trying to re-connect");
        int tries=0;
        while(true){
            Thread.sleep(MQTT_RETRY_INTERVAL);
            if(checkIfNetworkAvailable()&& !MQTTClient.getInstance().mqttClient.isConnected()){
                try{
                    tries++;
                MQTTClient.getInstance().mqttClient.connect(MachineDetails.getInstance().getMACDetails(), true, (short) 1000);
                //register handler
                MQTTClient.getInstance().mqttClient.registerAdvancedHandler(ApplicationPublishHandler.getInstance());
                String[] topics={Constants.PUBLIC_BROADCAST_TOPIC};
                int[] qos={1};
                MQTTClient.getInstance().mqttClient.subscribe(topics, qos);
                }catch(Exception e){
                    //Service down  and give an alert
//                  break;
                }
                if(tries>No of retries on network available)
                break;
            }
        }


    private boolean checkIfNetworkAvailable() {
        try {
            InetAddress.getByName("<<your host name>>");
            return true;
        } catch (UnknownHostException e) {
            return false;
        }


    }

Answer 2

Here is how i implemented Mqtt reconnect on connection lost, starts a thread to try to connect to MqttServer which will be destroyed on successful connection.

  boolean retrying = false;
   public void reConnect(){
        if (!retrying) {
            retrying = true;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    for (;;) {
                        try {
                            if (isInetAvailable() && !mqttClient.isConnected()) {
                                if(isPasswdProtected) {
                                     //connect with MqttConnectionOptions
                                    Connect_with_passwd();
                                } else {
                                    Connect();
                                }
                                Thread.sleep(MQTT_RETRY_INTERVAL);
                            } else if (isConnected()) {
                                List<String> topics = topicsSubscribed;
                                topicsSubscribed.clear();
                                for (String topic : topics) {
                                    try {
                                        subscribeToTopic(topic);
                                    } catch (MqttException e) {
                                    }
                                }
                                retrying = false;
                                break;
                            } else if (!Internet.isAvailable()) {
                                Thread.sleep(INET_RETRY_INTERVEL);
                            }
                        } catch (MqttException | InterruptedException e) {
                            try {
                                Thread.sleep(MQTT_RETRY_INTERVAL);
                            } catch (InterruptedException ex) {
                            }
                        }
                    }
                }
            }).start();
        }
}
 /*Check internet connection*/

 public static boolean isInetAvailable() {
    boolean connectivity;
    try {
        URL url = new URL(GOOGLE);
        URLConnection conn = url.openConnection();
        conn.connect();
        connectivity = true;
    } catch (IOException e) {
        connectivity = false;
    }
    return connectivity;
}