MQTT wysyła wiadomość z głównego wątku

Paho nie działa w wątku interfejsu użytkownika, ale może asynchronicznie odwoływać się do wątku interfejsu użytkownika.

Po prostu pozwól Activity lub Fragment zaimplementować interfejs MqttCallbackExtended:

public class SomeActivity extends AppCompatActivity implements MqttCallbackExtended { 

    ...

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        Log.d("Mqtt", "Connect complete > " + serverURI);
    }

    @Override
    public void connectionLost(Throwable cause) {
        Log.d("Mqtt", "Connection lost");
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        Log.d("Mqtt", "Received > " + topic + " > " + message.toString());
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        Log.d("Mqtt", "Delivery complete");
    }
}

I skonstruuj MqttHelper z SomeActivity tak, jak jest MqttCallbackExtended listener:

public MqttHelper(Context context, MqttCallbackExtended listener) {
    this.mqttAndroidClient = new MqttAndroidClient(context, serverUri, clientId);
    this.mqttAndroidClient.setCallback(listener);
}

Na przykład:

this.mqttHelper = new MqttHelper(this);
this.mqttHelper.setCallback(this);

this.mqttHelper.publish("Java", "SomeActivity will handle the callbacks.");

Obsługa ich w Application jest problematyczna, ponieważ Application nie ma interfejsu użytkownika, a Context nie ma Theme. Ale dla klas rozciągających się na Activity, Fragment, DialogFragment, RecyclerView.Adapter itd. sensowne jest zaimplementowanie wywołania zwrotnego interface, gdy chcesz wchodzić w interakcję z ich interfejs użytkownika.

Dla porównania, MqttCallbackExtended extends MqttCallback.

Inne rozwiązanie:

1. Utwórz klasę MQTTService, która rozszerza android.app.Service.
   Klasa Android Service działa w wątku głównym. Więc jeśli chcesz użyć innego wątku, możesz po prostu użyć MqttAsyncClient.

2. Będziesz otrzymywać wiadomości od brokera w innym wątku automatycznie (nie głównym wątku) w messageArrived() przy użyciu metody wywołania zwrotnego.
   

3. Przekaż dane/polecenie z interfejsu użytkownika aplikacji (fragment aktywności,...) do MQTTService przez < strong>EventBus biblioteka po prostu.

4. Ponownie użyj EventBus w metodzie wywołania zwrotnego messageArrived(), aby przekazać odebrane dane od brokera do żądanej sekcji aplikacji.
   Pamiętaj, że w tym kroku, jeśli miejscem docelowym jest interfejs aplikacji, musisz użyć @Subscribe(threadMode = ThreadMode.MAIN) w miejscu docelowym, aby uzyskać dane w głównym wątku.
   

Przykładowy kod:

public class MQTTService extends Service {

    private MqttAsyncClient mqttClient;
    private String serverURI;

     @Override
    public void onCreate() { 
        //do your initialization here
        serverURI = "tcp://yourBrokerUrlOrIP:yourBrokerPort";
        EventBus.getDefault().register(this);
    }

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
      init();
      connect();
    }



    private void init() {
         mqttClient = new MqttAsyncClient(serverURI, yourClientId, new MemoryPersistence())
         mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
               //now you will receive messages from the broker in another thread automatically (not UI Thread).
               //You can do your logic here. for example pass the received data to the different sections of the application:
               EventBus.getDefault().post(new YourPOJO(topic, message, ...));
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
            }
        });
    }

    private MqttConnectOptions getOptions(){
        MqttConnectOptions options = new MqttConnectOptions();
        options.setKeepAliveInterval(...);
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
        options.setAutomaticReconnect(true);
        options.setCleanSession(false);
        options.setUserName(...);
        options.setPassword(...);
        //options.setWill(...);
        //your other configurations
        return options;
    }

    private void connect() {
        try {
            IMqttToken token = mqttClient.connect(getOptions(), null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    //do works after successful connection
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                }
            });

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onDestroy() {
        EventBus.getDefault().unregister(this);
        mqttClient.close();
        mqttClient.disconnect();
    }

    //this method receives your command from the different application sections
    //you can simply create different "MqttCommandPOJO" classes for different purposes
    @Subscribe
    public void receiveFromApp1(MqttCommandPOJO1 pojo1) {
        //do your logic(1). For example:
        //publish or subscribe something to the broker (QOS=1 is a good choice).
    }

    @Subscribe
    public void receiveFromApp2(MqttCommandPOJO2 pojo2) {
        //do your logic(2). For example:
        //publish or subscribe something to the broker (QOS=1 is a good choice).
    }

}

Teraz możesz po prostu otrzymywać dane przekazane z MQTTService w każdej sekcji swojej aplikacji.

Na przykład:

public class MainActivity extends AppCompatActivity {
    ...

    @Subscribe(threadMode = ThreadMode.MAIN)
    public void receiveFromMQTTService(YourPojo pojo){
       //Do your logic. For example update the UI.
    }
}

Inne linki:
Instrukcje ogólne

Wszystkiego najlepszego