Как правильно заполнить данные MQTT из MqttAsyncClient в пользовательский интерфейс приложения? (RecyclerView, проблемы ⇐ Android

[Гость]

Я хотел бы начать с информации о том, что мое решение MQTT очень хорошо работало с предыдущим org.eclipse.paho.android.service.MqttAndroidClient, но недавно я решил переключиться на ->> < code>org.eclipse.paho.client.mqttv3.MqttAsyncClient.

С тех пор начались следующие проблемы:
[*]RecyclerView, который использует данные MQTT, собранные в messageArrived(), не обновляется после вызова notifyDataSetChanged() в настройке моего адаптера и когда notifyDataSetChanged() вызывается, получена ошибка:
android.view.ViewRootImpl$CalledFromWrongThreadException: только исходный поток, создавший иерархию представлений, может касаться ее представлений.

Вызов notifyDataSetChanged() из Runnable:

getActivity().runOnUiThread(new Runnable() { @Override общественный недействительный запуск () { //... sonoff_devices_arrayList.add(sonoff_obj); sonoff_adapter_rv.notifyDataSetChanged(); } }); вызывает сбой приложения:

Попытка чтения из поля androidx.recyclerview.widget.ViewInfoStore androidx.recyclerview.widget.RecyclerView.mViewInfoStore по нулевой ссылке на объект
[*]SnackBar, вызванный сообщением MQTT, не отображается в пользовательском интерфейсе (но после того, как приложение помещено в фоновый и передний план, он отображается) [*]Некоторые, казалось бы, несвязанные незначительные изменения пользовательского интерфейса не распространяются, поскольку один навигационный ящик неожиданно остается выделенным.
Как уже говорилось, когда приложение переводится в фоновый режим и возвращается на передний план, пользовательский интерфейс неожиданно обновляется сразу после возобновления работы приложения. Понятия не имею, почему.

Я уверен, что делаю что-то не так, пытаясь обновить пользовательский интерфейс так же, как я использовал это с MqttAndroidClient

Мой класс mqtt_helper, обеспечивающий базовое подключение MqttAsyncClient:

пакет cz.removed; импортировать android.content.Context; импортировать android.util.Log; импортировать org.eclipse.paho.client.mqttv3.IMqttActionListener; импортировать org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; импортировать org.eclipse.paho.client.mqttv3.MqttAsyncClient; импортировать org.eclipse.paho.client.mqttv3.IMqttToken; импортировать org.eclipse.paho.client.mqttv3.MqttConnectOptions; импортировать org.eclipse.paho.client.mqttv3.MqttException; импортировать org.eclipse.paho.client.mqttv3.MqttMessage; импортировать org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; импортировать org.json.JSONObject; импортировать java.io.IOException; импортировать java.security.KeyManagementException; импортировать java.security.KeyStoreException; импортировать java.security.NoSuchAlgorithmException; импортировать java.security.UnrecoverableKeyException; импортировать java.security.cert.CertificateException; общественный класс MQTThelper { MqttConnectOptions MQTT_CONNECTION_OPTIONS; PreferenceData; PreferenceData; Контекстный контекст; public MQTThelper (Контекстный контекст) { this.context = контекст; PreferenceData = новые PreferenceData (контекст); } общественный MqttAsyncClient getMqttCLIENT(){ //материалы для подключения MQTT MqttAsyncClient КЛИЕНТ; String BROKER_HOST = PreferenceData.getUsersStringData("key_mqtt_server_host"); //Собранные данные о предпочтениях Строка BROKER = "ssl://" + BROKER_HOST; пытаться { КЛИЕНТ = новый MqttAsyncClient(BROKER, MqttAsyncClient.generateClientId(), новый MemoryPersistence()); } catch (MqttException e) { выбросить новое RuntimeException(e); } MQTT_CONNECTION_OPTIONS = новый MqttConnectOptions(); MQTT_CONNECTION_OPTIONS.setCleanSession (ложь); MQTT_CONNECTION_OPTIONS.setKeepAliveInterval(30); //Необязательный вход String USERNAME = PreferenceData.getUsersStringData("key_mqtt_server_username"); //Собранные данные о предпочтениях Строка PASSWORD = PreferenceData.getUsersStringData("key_mqtt_server_password"); //Собранные данные о предпочтениях MQTT_CONNECTION_OPTIONS.setUserName(ИМЯ ПОЛЬЗОВАТЕЛЯ); MQTT_CONNECTION_OPTIONS.setPassword(PASSWORD.toCharArray()); MQTT_CONNECTION_OPTIONS.setCleanSession (ложь); если (BROKER.contains("ssl")) { SocketFactory.SocketFactoryOptions socketFactoryOptions = новый SocketFactory.SocketFactoryOptions(); пытаться { socketFactoryOptions.withCaInputStream(context.getResources().openRawResource(R.raw.ca_root)); MQTT_CONNECTION_OPTIONS.setSocketFactory(новый SocketFactory(socketFactoryOptions)); } catch (IOException | NoSuchAlgorithmException | KeyStoreException | CertificateException | KeyManagementException | UnrecoverableKeyException e) { е.printStackTrace(); } } System.out.print("Выделенный CLIENT в помощниках -> getMqttCLIENT: " + CLIENT.getClientId()); вернуть КЛИЕНТА; } public void MqttConnect (MqttAsyncClient CLIENT) { пытаться { System.out.println("Идентификатор клиента MQTT в помощниках: " + CLIENT.getClientId()); окончательный токен IMqttToken = CLIENT.connect(MQTT_CONNECTION_OPTIONS); токен.waitForCompletion(3500); token.setActionCallback(новый IMqttActionListener() { @Override public void onSuccess (IMqttToken asyncActionToken) { //связанный Log.d("MQTT:", "подключено, токен: " + asyncActionToken.toString()); } @Override public void onFailure (IMqttToken asyncActionToken, исключение Throwable) { // Что-то пошло не так Log.d("MQTT:", "не подключено" + asyncActionToken.toString()); } }); } catch (MqttException e) { е.printStackTrace(); } } void MqttDisconnect (MqttAsyncClient CLIENT) { if(CLIENT != null && CLIENT.isConnected() ) { пытаться { IMqttToken disconToken = CLIENT.disconnect(); disconToken.setActionCallback(новый IMqttActionListener() { @Override public void onSuccess (IMqttToken asyncActionToken) { Log.d("mqtt:", "отключено"); } @Override public void onFailure (IMqttToken asyncActionToken, Выбрасываемое исключение) { Log.d( "mqtt:", "не удалось отключиться" ); } }); } catch (MqttException e) { е.printStackTrace(); } } } public void subscribe (MqttAsyncClient CLIENT, строковая тема, байт qos) { пытаться { IMqttToken subToken = CLIENT.subscribe(тема, qos); subToken.setActionCallback(новый IMqttActionListener() { @Override public void onSuccess (IMqttToken asyncActionToken) { Log.d("MQTT", "подписка успешна: " + asyncActionToken.toString() + "тема: " + тема + ", с QOS:" + qos); } @Override public void onFailure (IMqttToken asyncActionToken, исключение Throwable) { Log.d("MQTT", "ошибка подписки на: " + тема); } }); } catch (MqttException e) { е.printStackTrace(); } } public void unsubscribe (MqttAsyncClient CLIENT, строковая тема) { //отписываемся от MQTT if(CLIENT != null && CLIENT.isConnected()) { пытаться { IMqttToken unsubToken = CLIENT.unsubscribe(тема); unsubToken.setActionCallback(новый IMqttActionListener() { @Override public void onSuccess (IMqttToken asyncActionToken) { Log.d( "mqtt:", "отписался от " + тема ); } @Override public void onFailure (IMqttToken asyncActionToken, Выбрасываемое исключение) { Log.d( "mqtt:", "не удалось отменить регистрацию"); } }); } catch (MqttException e) { е.printStackTrace(); } } } public voidpublish(MqttAsyncClient CLIENT,String theme,JSONObject jmsg,String msg,boolean isRetained,int qos) { пытаться { Сообщение MqttMessage = новое MqttMessage(); message.setQos(qos); message.setRetained(isRetained); если (сообщение! = ноль) { message.setPayload(msg.getBytes()); } иначе, если (jmsg != null ){ message.setPayload(jmsg.toString().getBytes()); } Токен IMqttToken = CLIENT.publish(тема, сообщение); токен.waitForCompletion(); token.setActionCallback(новый IMqttActionListener() { @Override public void onSuccess (IMqttToken asyncActionToken) { Log.d("mqtt:", "публикация выполнена -> " + asyncActionToken.toString()); } @Override public void onFailure (IMqttToken asyncActionToken, исключение Throwable) { Log.d("mqtt:", "ошибка публикации -> " + asyncActionToken.toString()); } }); } catch (Исключение е) { е.printStackTrace(); } } } Класс, использующий указанный выше помощник MQTT:

пакет cz.removed; импортировать android.widget.Toast; импортировать androidx.annotation.NonNull; импортировать androidx.annotation.Nullable; импортировать androidx.appcompat.app.AlertDialog; импортировать androidx.core.content.ContextCompat; импортировать androidx.fragment.app.Fragment; импортировать androidx.recyclerview.widget.GridLayoutManager; импортировать androidx.recyclerview.widget.LinearLayoutManager; импортировать androidx.recyclerview.widget.RecyclerView; импортировать androidx.swiperefreshlayout.widget.SwipeRefreshLayout; импортировать androidx.viewpager2.widget.ViewPager2; импортировать com.google.android.material.snackbar.Snackbar; импортировать org.eclipse.paho.android.service.MqttAndroidClient; импортировать org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; импортировать org.eclipse.paho.client.mqttv3.MqttAsyncClient; импортировать org.eclipse.paho.client.mqttv3.MqttCallbackExtended; импортировать org.eclipse.paho.client.mqttv3.MqttException; импортировать org.eclipse.paho.client.mqttv3.MqttMessage; импортировать org.json.JSONArray; импортировать org.json.JSONException; импортировать org.json.JSONObject; общедоступный класс EFragment расширяет Fragment реализует MqttCallbackExtended { MqttAsyncClient КЛИЕНТ; isMQTTEnabled = Истина; @Override public void onCreate (Bundle saveInstanceState) { super.onCreate(saveInstanceState); КЛИЕНТ = mqtt_helper.getMqttCLIENT(); CLIENT.setCallback(это); } @Override public void onViewCreated (Просмотр, @Nullable Bundle saveInstanceState) { super.onViewCreated(просмотр, saveInstanceState); //# Означает подписку на все ТЕМА = новый ArrayList(); TOPIC.add("a/discovery/#"); TOPIC.add( "б/#" ); @Override общественная недействительность onPause () { супер.онПауза(); //Отписываемся от MQTT если (isMQTTEnabled) { отписатьсяMQTT(); } } @Override общественная недействительность onDetach () { супер.onDetach(); } @Override общественный недействительный onResume () { супер.онрезюме(); если (isMQTTEnabled) { пытаться{ if ( CLIENT !=null && ! CLIENT.isConnected()){ mqtt_helper.MqttConnect(КЛИЕНТ); } еще{ подписатьсяMQTT(); } } } } @Override общественная недействительность onStop() { супер.онСтоп(); если (isMQTTEnabled) { mqtt_helper.MqttDisconnect(КЛИЕНТ); } } общественная недействительность отпискиMQTT() { если (isMQTTEnabled) { for (int i = 0; i