Android Paho Mqtt не повторно подключается после повторного подключения Wi-Fi

Я использую клиент Android mqtt для потоковой передачи местоположений. Это работает нормально до тех пор, пока не потеряется подключение к Интернету. Клиент mqtt не переподключается должным образом. Что мне делать, чтобы убедиться, что paho mqtt повторно подключается к брокеру. Я хочу заставить клиент mqtt повторно подключиться к брокеру и продолжить публикацию местоположений, когда соединение вернется.

Ниже мой код.

public class MqttClientHelperService2 extends Service {
    private MqttAndroidClient mqttAndroidClient;
    BroadcastReceiver broadcastReceiver;
    private final String SERVER_URL = "tcp://000.102.110.**:1883";
    private final String CLIENT_ID = "client_id";
    private final String MQTT_TOPIC = "livelocations/local";
    LiveLocation liveLocation;

    @Override
    public void onCreate() {
        super.onCreate();
        Thread newThread = new Thread(){
            public void run(){
                init();
            }
        };

        newThread.start();

        if (Build.VERSION.SDK_INT >= 26) {
            String CHANNEL_ID = "my_channel_01";
            NotificationChannel channel = new NotificationChannel(CHANNEL_ID,
                    "Channel human readable title",
                    NotificationManager.IMPORTANCE_LOW);

            ((NotificationManager) Objects.requireNonNull(getSystemService(Context.NOTIFICATION_SERVICE))).
                    createNotificationChannel(channel);
            Notification notification = new NotificationCompat.Builder(this, CHANNEL_ID)
                    .setContentTitle("")
                    .setContentText("")
                    .setSmallIcon(R.mipmap.ic_launcher).build();

            startForeground(1, notification);
        }
    }

    private void init() {
        mqttAndroidClient = new MqttAndroidClient(getApplicationContext(), SERVER_URL, CLIENT_ID);
        mqttAndroidClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {

            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {

            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {

            }


        });
    }


    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
        liveLocation= new LiveLocation(
                intent.getIntExtra("driver_id", 0),
                intent.getDoubleExtra("latitude", 0),
                intent.getDoubleExtra("longitude", 0),
                "");
        connectMqtt();
        return START_STICKY;
    }


    private MqttConnectOptions getMqttOptions(){
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
        mqttConnectOptions.setAutomaticReconnect(true);
        mqttConnectOptions.setCleanSession(false);
        return mqttConnectOptions;
    }

    private void connectMqtt() {
        try {
           IMqttToken iMqttToken =  mqttAndroidClient.connect(getMqttOptions());
           iMqttToken.setActionCallback(new IMqttActionListener() {
               @Override
               public void onSuccess(IMqttToken asyncActionToken) {
                    Log.e("phanuel-log", "connecting ......." + Calendar.getInstance().getTime());
                   publishToServer(liveLocation);
               }

               @Override
               public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.e("phanuel-log-error", "connection error");
               }
           });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onDestroy() {
        if(mqttAndroidClient != null){
            try {
                mqttAndroidClient.unregisterResources();
                mqttAndroidClient.close();
                mqttAndroidClient.disconnect();
            } catch (Exception e){
                e.printStackTrace();
            }
        }
        unregisterReceiver(broadcastReceiver);
        broadcastReceiver = null;
        super.onDestroy();
    }

    private boolean isNetworkAvailable() {
        ConnectivityManager connectivityManager
                = (ConnectivityManager) getSystemService(Context.CONNECTIVITY_SERVICE);
        assert  connectivityManager != null;
        NetworkInfo activeNetworkInfo = connectivityManager.getActiveNetworkInfo();
        return activeNetworkInfo != null && activeNetworkInfo.isConnectedOrConnecting();
    }

    void publishToServer(final LiveLocation location) {
        try {
            if (location.getDriverId() > 0){
                mqttAndroidClient.connect(getMqttOptions(), null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        JSONObject locationJsonObject = new JSONObject();
                        try {
                            locationJsonObject.put("driverId", location.getDriverId());
                            locationJsonObject.put("driverLatitude", location.getLatitude());
                            locationJsonObject.put("driverLongitude", location.getLongitude());
                            locationJsonObject.put("driverTimeStamp", Calendar.getInstance().getTime());
                        } catch (JSONException e) {
                            e.printStackTrace();
                        }

                        MqttMessage mqttMessage = new MqttMessage();
                        mqttMessage.setPayload(locationJsonObject.toString().getBytes());
                        mqttMessage.setQos(0);
                        mqttMessage.setRetained(false);

                        try {
                            mqttAndroidClient.publish(MQTT_TOPIC, mqttMessage);
                        }catch (MqttException e){
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        Log.e("mqtt-client", "Failed to connect");
                        Log.e("mqtt-cleint", exception.toString());
                    }
                });
            }
        } catch (MqttException e){
            e.printStackTrace();
        }
    }
}