Odtnhj

I've a Service to manage my MQTT Client connection, the MQTT works fine, but the problem is when I restart Broker Server, the Android client not reconnect. A exception is triggered on onConnectionLost() callback.

Notes

1. I'm using Moquette Broker at same computer -> Moquette
   


2. I've two Android clients app, a using Service (the problematic) and other working on a Thread, without Service (this works fine, reconnect is ok).


3. I can't run the Android Client MQTT lib, because this I'm using the Eclipse Paho MQTT.


4. Yes, I make setAutomaticReconnect(true);
   

Problem

The Android app that use Service, to works forever, not reconnect to MQTT Broker.

Code

MQTTService.java

public class MQTTService extends Service implements MqttCallbackExtended 

 boolean running;
private static final String TAG = "MQTTService";

public static final String ACTION_MQTT_CONNECTED = "ACTION_MQTT_CONNECTED";
public static final String ACTION_MQTT_DISCONNECTED = "ACTION_MQTT_DISCONNECTED";
public static final String ACTION_DATA_ARRIVED = "ACTION_DATA_ARRIVED";

// MQTT
MqttClient mqttClient;
final String serverURI = "tcp://"+ServidorServices.IP+":1883";
final String clientId = "Responsavel";
String topicoId;
Thread mqttStartThread;

public boolean subscribe(String topic) 
 try 
 Log.i(TAG,"Subscripe: " + topic);
 mqttClient.subscribe(topic);
 mqttClient.subscribe("LOCATION_REAL");
 return true;
 catch (Exception e) 
 e.printStackTrace();
 
 return false;


// Life Cycle
@Override
public IBinder onBind(Intent intent) 
 Log.d(TAG,"onBind()");
 return null;


@Override
public void onCreate() 
 Log.d(TAG,"onCreate()");
 running = true;
 topicoId = getSharedPreferences("myprefs",MODE_PRIVATE).getString("tag_id_aluno","0");

 mqttStartThread = new MQTTStartThread(this);

 if(topicoId.equals("0")) 
 Log.i(TAG,"Error to subscribe");
 return;
 

 mqttStartThread.start();


@Override
public int onStartCommand(Intent intent, int flags, int startId) 
 Log.d(TAG,"onStartCommand()");
 return super.onStartCommand(intent, flags, startId);


class MQTTStartThread extends Thread 

 MqttCallbackExtended mqttCallbackExtended;

 public MQTTStartThread(MqttCallbackExtended callbackExtended) 
 this.mqttCallbackExtended = callbackExtended;
 

 @Override
 public void run() 
 try 
 mqttClient = new MqttClient(serverURI,clientId,new MemoryPersistence());
 MqttConnectOptions options = new MqttConnectOptions();
 options.setAutomaticReconnect(true);
 options.setCleanSession(true);
 mqttClient.setCallback(mqttCallbackExtended);
 mqttClient.connect();
 catch (Exception e) 
 Log.i(TAG,"Exception MQTT CONNECT: " + e.getMessage());
 e.printStackTrace();
 
 


@Override
public void onDestroy() 
 Log.d(TAG,"onDestroy()");
 running = false;
 if (mqttClient != null) 
 try 
 if (mqttClient.isConnected()) mqttClient.disconnect();
 catch (Exception e) 
 e.printStackTrace();
 
 


@Override
public boolean onUnbind(Intent intent) 
 Log.i(TAG,"onUnbind()");
 return super.onUnbind(intent);


// Callbacks MQTT
@Override
public void connectComplete(boolean reconnect, String serverURI) 
 Log.i(TAG,"connectComplete()");
 if (topicoId == null) 
 Log.i(TAG,"Erro ao ler ID da Tag");
 return;
 
 sendBroadcast(new Intent(ACTION_MQTT_CONNECTED));
 subscribe(topicoId);


@Override
public void connectionLost(Throwable cause) 
 Log.i(TAG,"connectionLost(): " + cause.getMessage());
 cause.printStackTrace();
 sendBroadcast(new Intent(ACTION_MQTT_DISCONNECTED));


@Override
public void messageArrived(String topic, MqttMessage message) throws Exception 
 Log.i(TAG,"messageArrived() topic: " + topic);

 if (topic.equals("LOCATION_REAL")) 
 Log.i(TAG,"Data: " + new String(message.getPayload()));
 else 
 Context context = MQTTService.this;
 String data = new String(message.getPayload());
 Intent intent = new Intent(context,MapsActivity.class);
 intent.putExtra("location",data);
 LatLng latLng = new LatLng(Double.valueOf(data.split("_")[0]),Double.valueOf(data.split("_")[1]));
 String lugar = Utils.getAddressFromLatLng(latLng,getApplicationContext());
 NotificationUtil.create(context,intent,"Embarque",lugar,1);

 if (data.split("_").length < 3) 
 return;
 

 double latitude = Double.valueOf(data.split("_")[0]);
 double longitude = Double.valueOf(data.split("_")[1]);
 String horario = data.split(" ")[2];

 Intent iMqttBroadcast = new Intent(ACTION_DATA_ARRIVED);
 iMqttBroadcast.putExtra("topico",String.valueOf(topic));
 iMqttBroadcast.putExtra("latitude",latitude);
 iMqttBroadcast.putExtra("longitude",longitude);
 iMqttBroadcast.putExtra("evento","Embarcou");
 iMqttBroadcast.putExtra("horario",horario);

 sendBroadcast(iMqttBroadcast);
 


@Override
public void deliveryComplete(IMqttDeliveryToken token) 
 Log.i(TAG,"deliveryComplete()");

Exception Stacktrace

I/MQTTService: connectionLost(): Connection lost
W/System.err: Connection lost (32109) - java.io.EOFException
W/System.err: at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:146)
W/System.err: at java.lang.Thread.run(Thread.java:818)
W/System.err: Caused by: java.io.EOFException
W/System.err: at java.io.DataInputStream.readByte(DataInputStream.java:77)
W/System.err: at org.eclipse.paho.client.mqttv3.internal.wire.MqttInputStream.readMqttWireMessage(MqttInputStream.java:65)
W/System.err: at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:107)
W/System.err: ... 1 more

I think you forgot to include MqttConnectOptions with MqttClient object.

Please try like following

mqttClient.connect(options);

instead of

mqttClient.connect();

Hope it may help to resolve your re-connect issue.

As method description says.

 options.setAutomaticReconnect(true);

The client will attempt to reconnect to the server. It will initially wait 1 second before it attempts to reconnect, for every failed reconnect attempt, the delay will doubleuntil it is at 2 minutes at which point the delay will stay at 2 minutes.

Another option would be you could manage retry interval in case of connection lost events.