Android - Paho Mqtt client not receiving messages after network connection change (mobile data is disabled and re-enabled)

The Java client library depends to some extent on the underlying networking API. When publish is called, it will write the MQTT packet to the socket. If this entry fails, then the connection will be ignored, if this entry works, then the client library will continue. The difference in behavior you see is that network libraries behave differently under these circumstances.

The MQTT keepalive interval is intended to help with this. Under certain circumstances, a TCP connection may appear to be alive if it is not. This is especially possible on mobile or satellite connected devices - you cannot expect network APIs to work the same under all circumstances. Keepalive sends a ping packet to the server and waits for a response - if this response is not received, the session is considered closed.

If you set the keepalive interval to 10 seconds, then the connection should be recognized as broken within 15-20 seconds.

In your service: -

 //Receiver that notifies the Service when the phone gets data connection
  private NetworkConnectionIntentReceiver netConnReceiver;

      

        
        
        
      

    

Create the following class: -

/*
* Called in response to a change in network connection - after losing a
*  connection to the server, this allows us to wait until we have a usable
*  data connection again
*/
class NetworkConnectionIntentReceiver extends BroadcastReceiver
{
  private static  String TAG ="NetworkConnectionIntentReceiver";
  @Override
  public void onReceive(Context ctx, Intent intent)
  {
    // we protect against the phone switching off while we're doing this
    //  by requesting a wake lock - we request the minimum possible wake
    //  lock - just enough to keep the CPU running until we've finished

    PowerManager pm = (PowerManager) ctx.getSystemService(ctx.POWER_SERVICE);
    PowerManager.WakeLock wl = pm.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "MQTT");
    wl.acquire();

    Connection c = Connections.getInstance(ctx).getConnection(clientHandle);
    final ActionListener callback = new ActionListener(ctx,
                ActionListener.Action.CONNECT, clientHandle,null);
    c.getClient().setCallback(new MqttCallbackHandler(ctx, clientHandle,messenger_where_incoming_messages_tobe_sent));
    c.getClient().connect(c.getConnectionOptions(), null, callback);

    /*    The Above Reconnect Logic can be put up in a Reconnect() function.
     *    OR WRITE Any Other LOGIC TO RECONNECT TO MQTT
     */       

    // we're finished - if the phone is switched off, it okay for the CPU
    //  to sleep now
    wl.release();
}

      

        
        
        
      

    

Now call the following method Somewhere appropriate in OnResume () or onCreate to register the BroadcastReceiver.

synchronized void handleNetworkChange()
{

    // changes to the phone network - such as bouncing between WiFi
    //  and mobile data networks - can break the MQTT connection
    // the MQTT connectionLost can be a bit slow to notice, so we use
    //  Android inbuilt notification system to be informed of
    //  network changes - so we can reconnect immediately, without
    //  haing to wait for the MQTT timeout
    if (netConnReceiver == null)
    {
        netConnReceiver = new NetworkConnectionIntentReceiver();
        registerReceiver(netConnReceiver,
                new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));

    }
}

      

        
        
        
      

    

I fixed the recconect error as follows (using rxJava2, but not required):

    public void reconnect() {
        Completable.create(emitter -> {
            while (!mqttClient.isConnected()) {
                mqttClient.connect(options, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        emitter.onComplete();
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        LogHelper.d(TAG,"try to connect failed");
                    }
                });

                Thread.sleep(2000);
            }
            emitter.onComplete();
        })
        .subscribeOn(Schedulers.io())
        .subscribe();
    }

      

        
        
        
      

    

and call example

private BroadcastReceiver changeNetworkStateReceiver = new BroadcastReceiver() {
    @Override
    public void onReceive(Context context, Intent intent) {
        if (Objects.equals(intent.getAction(), NetworkStateReceiver.EVENT_CHANGE_NETWORK_STATE)) {
            if(Utils.isOnline(context)) {
                mqttClient.reconnect();
            }
        }
    }
};