Refactoring

lib/content_views/approach_area_page.dart

@@ -10,6 +10,7 @@ import 'package:stadtbote/popups/tasks_popups/zip_popup.dart';
 import 'package:stadtbote/protocoll_messages/zip_message.dart';
 import 'package:stadtbote/tools/constants.dart';
 import 'package:stadtbote/tools/message_service.dart';
+import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart';
 import 'package:stadtbote/tools/messages/ranking_updated.dart';
 import 'package:stadtbote/tools/mqtt_core.dart';
 
@@ -190,10 +191,7 @@ class _ApproachAreaPageState extends State<ApproachAreaPage> {
       setPosMessage.track!.pos!.acc = position.accuracy.toInt();
     }
 
-    var json = jsonEncode(setPosMessage.toJson());
+    MessageService().publish<PublishPayloadMessage>(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "rank", message: setPosMessage.toJson()));
-    Util.logInfo(json);
-
-    MqttCore.instance.publishMessage("set_zip", jsonEncode(setPosMessage.toJson()));
   }
 
   _initMessages() {

lib/content_views/assigned_jobs_page.dart

@@ -10,6 +10,7 @@ import 'package:badges/badges.dart' as badges;
 import 'package:stadtbote/protocoll_messages/accept_job_message.dart';
 import 'package:stadtbote/protocoll_messages/fin_task_message.dart';
 import 'package:stadtbote/tools/message_service.dart';
+import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart';
 import 'package:stadtbote/tools/messages/refresh_badges_message.dart';
 import 'package:stadtbote/tools/mqtt_core.dart';
 
@@ -200,13 +201,13 @@ class _AssignedJobsPageState extends State<AssignedJobsPage> {
     message.track.pos!.long = currentPosition != null ? currentPosition.longitude : 0.0;
     message.track.pos!.acc = currentPosition != null ? currentPosition.accuracy.toInt() : 0;
 
-    MqttCore.instance.publishMessage("reaction", jsonEncode(message.toJson()));
+    MessageService().publish<PublishPayloadMessage>("reaction", arguments: PublishPayloadMessage(topic: "reaction", message: message.toJson()));
 
     await MyApp.isarInstance.writeTxn(() async {
       await MyApp.isarInstance.jobs.delete(job.id);
     });
 
-    MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage());
+    MessageService().publish<RefreshBadgesMessage>(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage());
 
     _entries = await _createListEntries();
 

lib/content_views/main_page.dart

@@ -16,6 +16,7 @@ import 'package:stadtbote/model/location_status.dart';
 import 'package:stadtbote/popups/infos_popup.dart';
 import 'package:stadtbote/protocoll_messages/logout_message.dart';
 import 'package:stadtbote/tools/message_service.dart';
+import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart';
 import 'package:stadtbote/tools/messages/job_deleted_message.dart';
 import 'package:stadtbote/tools/messages/job_received_message.dart';
 import 'package:stadtbote/tools/messages/message_message.dart';
@@ -280,7 +281,7 @@ class _MainPageState extends State<MainPage> {
       _refreshBadges();
     });
 
-    MessageService().subscribe<RefreshBadgesMessage>(Constants.messageTopicRefreshBadges, this, (RefreshBadgesMessage message) {
+    MessageService().subscribe<RefreshBadgesMessage>(Constants.messageTopicUpdateBadges, this, (RefreshBadgesMessage message) {
       _refreshBadges();
     });
   }
@@ -318,7 +319,7 @@ class _MainPageState extends State<MainPage> {
     if (!_loggedOut) {
       showLogoutAlertDialog(context);
     } else {
-      MqttCore.setup();
+      MessageService().publish<ConnectClientMessage>(Constants.messageTopicConnectClient, arguments: ConnectClientMessage());
 
       _loggedOut = false;
 
@@ -334,7 +335,7 @@ class _MainPageState extends State<MainPage> {
         _sendLogoutMessage();
 
         Future.delayed(const Duration(seconds: 1), () async {
-          MqttCore.instance.diconnect();
+          MessageService().publish<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, arguments: DisconnectClientMessage());
         });
 
         _loggedOut = true;
@@ -380,6 +381,6 @@ class _MainPageState extends State<MainPage> {
       logoutMessage.pos!.acc = position.accuracy.toInt();
     }
 
-    MqttCore.instance.publishMessage("logout", jsonEncode(logoutMessage.toJson()));
+    MessageService().publish<PublishPayloadMessage>(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "logout", message: logoutMessage.toJson()));
   }
 }

lib/content_views/messages_page.dart

@@ -10,6 +10,9 @@ import 'package:stadtbote/main.dart';
 import 'package:stadtbote/model/message.dart' as sbmessage;
 import 'package:stadtbote/model/message.dart';
 import 'package:stadtbote/model/session_status.dart';
+import 'package:stadtbote/tools/constants.dart';
+import 'package:stadtbote/tools/message_service.dart';
+import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart';
 import 'package:stadtbote/tools/mqtt_core.dart';
 
 import '../tools/util.dart';
@@ -108,7 +111,7 @@ class _MessagesPageState extends State<MessagesPage> {
       await MyApp.isarInstance.messages.put(message);
     });
 
-    MqttCore.instance.publishMessage('message', jsonEncode(message.toJson()));
+    MessageService().publish<PublishPayloadMessage>(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "message", message: message.toJson()));
 
     _messages.insert(0, types.TextMessage(
       author: _user,

lib/content_views/task_list_page.dart

@@ -23,6 +23,7 @@ import 'package:stadtbote/protocoll_messages/fin_tour_message.dart';
 import 'package:stadtbote/protocoll_messages/pic_message.dart';
 import 'package:stadtbote/tools/constants.dart';
 import 'package:stadtbote/tools/message_service.dart';
+import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart';
 
 import '../main.dart';
 import '../model/task_info.dart';

lib/main.dart

@@ -458,7 +458,7 @@ class _MainState extends State<Main> with SingleTickerProviderStateMixin {
           await showAcceptJobsPopup(context, await MyApp.isarInstance.jobs.filter().stateEqualTo(0).findAll(), () {
             _dialogOpened = false;
 
-            MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage());
+            MessageService().publish(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage());
           });
         }
       }

lib/popups/accept_jobs_popup.dart

@@ -16,6 +16,8 @@ import 'package:stadtbote/tools/mqtt_core.dart';
 import 'package:stadtbote/tools/util.dart';
 import 'package:badges/badges.dart' as badges;
 
+import '../tools/messages/PublishPayloadMessage.dart';
+
 Future<void> showAcceptJobsPopup(BuildContext context, List<Job> jobs, Function() onFinished) async {
   var jobsSet = jobs.toSet();
 
@@ -217,11 +219,11 @@ _sendAnswer(Job job, int item) async {
   message.track.pos!.long = currentPosition != null ? currentPosition.longitude : 0.0;
   message.track.pos!.acc = currentPosition != null ? currentPosition.accuracy.toInt() : 0;
 
-  MqttCore.instance.publishMessage("reaction", jsonEncode(message.toJson()));
+  MessageService().publish<PublishPayloadMessage>(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "reaction", message: message.toJson()));
 
   await MyApp.isarInstance.writeTxn(() async{
     await MyApp.isarInstance.jobs.delete(job.id);
   });
 
-  MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage());
+  MessageService().publish(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage());
 }

lib/popups/infos_popup.dart

@@ -1,14 +1,43 @@
 import 'package:flutter/material.dart';
 import 'package:stadtbote/main.dart';
 import 'package:stadtbote/model/session_status.dart';
+import 'package:stadtbote/tools/constants.dart';
+import 'package:stadtbote/tools/message_service.dart';
 import 'package:stadtbote/tools/mqtt_core.dart';
 import 'package:stadtbote/tools/util.dart';
 
 void showInfosPopup(BuildContext context, SessionStatus? status) async {
-  var version = await Util.getAppVersion();
+  final int version = await Util.getAppVersion();
 
+  late String connectedString = "Bitte warten...";
+
+  Future.delayed(const Duration(seconds: 2), () async {
+    MessageService().publish<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, arguments: RequestConnectionStatusMessage());
+  });
+
+  void unsubscribeMessages() {
+    MessageService().unsubscribe(Constants.messageTopicCurrentConnectionStatus, version);
+    MessageService().unsubscribe(Constants.messageTopicConnectionStatusChanged, version);
+  }
+
+  await showDialog(
+      // ignore: use_build_context_synchronously
+      context: context,
+      builder: (BuildContext context) {
+        // ignore: deprecated_member_use
+        return WillPopScope(
+            onWillPop: () async {
+              unsubscribeMessages();
+
+              return true;
+            },
+            child: AlertDialog(
+                contentPadding: EdgeInsets.zero,
+                shadowColor: Colors.transparent,
+                backgroundColor: Colors.white,
+                content: StatefulBuilder(builder: (BuildContext context, StateSetter setState) {
                   var text = "Versionsnummer: a$version\n";
-  text += "Verbindungsstatus: ${MqttCore.instance.connectionState == ConnectionStateEnum.connected ? "Verbunden" : "NICHT verbunden"}\n";
+                  text += "Verbindungsstatus: $connectedString\n";
                   text += "GPS-Status: ${MyApp.currentPosition != null ? MyApp.currentPosition!.timestamp.toString() : "Unbekannt"}\n";
                   text += "\n";
                   text += "User Status:\n";

lib/tools/constants.dart

@@ -7,18 +7,25 @@ class Constants {
   static const String messageTopicConnectionStatusChanged = "connection_status_changed";
   static const String messageTopicSessionInfoUpdate = "status_info_updated";
   static const String messageTopicSessionStatusInfoUpdate = "session_status_info_updated";
-  static const String messageTopicLocationStatusUpdated = "status_info_updated";
+  static const String messageTopicLocationStatusUpdated = "location_status_updated";
   static const String messageTopicRankingUpdated = "ranking_updated";
   static const String messageTopicJobReceived = "job_received";
   static const String messageTopicJobDeleted = "job_deleted";
-  static const String messageTopicRefreshBadges = "refresh_badges";
+  static const String messageTopicUpdateBadges = "update_badges";
   static const String messageTopicMessageReceived = "message_received";
   static const String messageTopicPushContentView = "push_content_view";
   static const String messageTopicPopContentView = "pop_content_view";
   static const String messageTopicReplaceContentView = "replace_content_view";
   static const String messageTopicHideRightButton = "hide_right_button";
+  static const String messageTopicPublishPayload = "publish_payload_message";
+  static const String messageTopicConnectClient = "connect_client";
+  static const String messageTopicDisconnectClient = "disconnect_client";
+  static const String messageTopicRequestConnectionStatus = "request_connection_status";
+  static const String messageTopicCurrentConnectionStatus = "current_connection_status";
   static const String messageShowPopup = "show_popup";
 
+  static const String prefsPreviousMqttSubscription = "previous_mqtt_subscription";
+
   static const int stateAssigned = 0;
   static const int stateAccepted = 1;
   static const int stateFinished = 2;

lib/tools/messages/PublishPayloadMessage.dart

@@ -0,0 +1,7 @@
+class PublishPayloadMessage {
+  final String topic;
+
+  final Map<String, dynamic> message;
+
+  PublishPayloadMessage({required this.topic, required this.message});
+}

lib/tools/mqtt_core.dart

@@ -45,8 +45,6 @@ enum ConnectionStateEnum {
 }
 
 class MqttCore {
-  static late MqttCore instance;
-
   late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected;
 
   late bool _disconnectInProgress = false;
@@ -57,7 +55,7 @@ class MqttCore {
   late String _fromServerTopic;
   late String _toServerTopic;
 
-  late MqttServerClient _mqttClient;
+  MqttServerClient? _mqttClient;
 
   late int _reconnectDelay = 0;
 
@@ -66,23 +64,23 @@ class MqttCore {
   Timer? _trackTimer;
 
   MqttCore() {
+    initMessages();
+
     _disconnectInProgress = false;
 
     _processBootstrap();
   }
 
-  static setup() {
+  _disconnect() {
-    instance = MqttCore();
-  }
-
-  diconnect() {
     _disconnectInProgress = true;
 
     _cancelReconnectTimer();
 
     _cancelTrackTimer();
 
-    _mqttClient.disconnect();
+    if (_mqttClient != null) {
+      _mqttClient!.disconnect();
+    }
 
     connectionState = ConnectionStateEnum.notConnected;
   }
@@ -95,25 +93,25 @@ class MqttCore {
       Util.logInfo("broker port: $_brokerPort");
 
       _mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort);
-      _mqttClient.logging(on: false);
+      _mqttClient!.logging(on: false);
-      _mqttClient.keepAlivePeriod = 5;
+      _mqttClient!.keepAlivePeriod = 5;
-      _mqttClient.disconnectOnNoResponsePeriod = 10;
+      _mqttClient!.disconnectOnNoResponsePeriod = 10;
-      _mqttClient.autoReconnect = true;
+      _mqttClient!.autoReconnect = true;
-      _mqttClient.onDisconnected = _clientDisconnected;
+      _mqttClient!.onDisconnected = _clientDisconnected;
-      _mqttClient.onAutoReconnect = _clientDisconnected;
+      _mqttClient!.onAutoReconnect = _clientDisconnected;
-      _mqttClient.onAutoReconnected = _clientConnected;
+      _mqttClient!.onAutoReconnected = _clientConnected;
-      _mqttClient.connectionMessage = MqttConnectMessage()
+      _mqttClient!.connectionMessage = MqttConnectMessage()
           .withClientIdentifier(await Util.getAppUuid())
           .authenticateAs("app", "apppwd");
 
-      await _mqttClient.connect();
+      await _mqttClient!.connect();
 
-      if (_mqttClient.connectionStatus?.state == MqttConnectionState.connected) {
+      if (_mqttClient!.connectionStatus?.state == MqttConnectionState.connected) {
         Util.logInfo('Verbunden mit dem MQTT-Broker!');
 
         _reconnectDelay = 0;
 
-        _setupSubscriptions();
+        _setupMqtt();
 
         _initSession();
       } else {
@@ -130,46 +128,53 @@ class MqttCore {
   }
 
   publishBinary(String topic, List<int> data) {
+    if (_mqttClient != null) {
       var buffer = Uint8Buffer();
       buffer.addAll(data);
 
-    _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
+      _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer);
 
       Util.logInfo("binary message send to $topic");
+    } else {
+      Util.logException("publishBinary called while offline!");
+    }
   }
 
-  publishMessage(String topic, String text) {
+  publishText(String topic, String text) {
+    if (_mqttClient != null) {
       var buffer = Uint8Buffer();
-    buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
+      buffer.addAll(utf8.encode(text));
 
-    var messageTopic = "$_toServerTopic$topic/$_clientId";
+      _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer);
 
-    _mqttClient.publishMessage(messageTopic, MqttQos.exactlyOnce, buffer);
+      Util.logInfo("message\n$text\nsend to $topic");
-
+    } else {
-    Util.logInfo("message\n$text\nsend to $messageTopic");
+      Util.logException("publishBinary called while offline!");
+    }
   }
 
-  void _setupSubscriptions() async {
+  void _setupMqtt() async {
-    Util.logInfo('_setupSubscriptions called');
+    Util.logInfo('_setupMqtt called');
 
-    _mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> messages) {
+    _mqttClient!.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
       try {
-        for (var mqttMessage in messages) {
+        final MqttPublishMessage message = c[0].payload as MqttPublishMessage;
-          final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage;
+        String payload = utf8.decode(MqttPublishPayload.bytesToStringAsString(message.payload.message).codeUnits);
-
+        var topic = message.variableHeader?.topicName;
-          if (message.variableHeader!.topicName.endsWith(_clientId)) {
-            var messageBytes = message.payload.message;
-            String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes));
-
-            var topicArray = message.variableHeader?.topicName;
-            var topicArraySplited = topicArray!.split("/");
-            var topic = topicArraySplited[3];
 
         Util.logInfo('message received\ntopic: $topic\nmessage: $payload');
 
-            switch(topic) {
+        Map<String, dynamic> messageData = jsonDecode(payload);
+
+        for (var entry in messageData.keys) {
+          if (entry == 'payload') {
+            var payload = messageData[entry];
+            for (var payloadEntry in payload.keys) {
+              var payloadData = payload[payloadEntry];
+
+              switch (payloadEntry) {
                 case 'session':
-                _sessionMessageReceived(jsonDecode(payload));
+                  _sessionMessageReceived(payloadData);
                   break;
 
                 case 'location_status':
@@ -206,14 +211,27 @@ class MqttCore {
               }
             }
           }
+        }
       } catch(ex) {
         Util.logException(ex.toString());
       }
     });
 
-    var fromServerTopic = "$_fromServerTopic#";
+    if (await Util.preferencesContainsKey(Constants.prefsPreviousMqttSubscription)) {
-    _mqttClient.subscribe(fromServerTopic, MqttQos.exactlyOnce);
+      String previousMqttSubscription = await Util.preferencesGetString(Constants.prefsPreviousMqttSubscription, "") as String;
-    Util.logInfo('subscribed to topic: $fromServerTopic');
+
+      if (previousMqttSubscription != _fromServerTopic) {
+        Util.logInfo('topic changed -> unsubscribe from $previousMqttSubscription');
+        _mqttClient!.unsubscribe(previousMqttSubscription);
+      }
+
+      _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
+    }
+
+    _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
+    Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, _fromServerTopic);
+
+    Util.logInfo('subscribed to topic: $_fromServerTopic');
 
     MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
@@ -388,7 +406,6 @@ class MqttCore {
 
       var jsonMap = handshakeMessage.toJson();
       var json = Uri.encodeFull(jsonEncode(jsonMap));
-      //var uri = "http://test-app-server.dns.army:8080/bootstrap?json=$json"; // Testserver bei Torben
       var uri = "https://test-app-server.assecutor.org/bootstrap?json=$json";
       Util.logInfo("Bootstrap request: $uri");
 
@@ -496,6 +513,31 @@ class MqttCore {
     message.received.ver = job.ver;
     message.received.utc = DateTime.now().millisecondsSinceEpoch;
 
-    MqttCore.instance.publishMessage('job_received', jsonEncode(message.toJson()));
+    publishPayloadMessage('job', message.toJson());
+  }
+
+  void initMessages() {
+    MessageService().subscribe<PublishPayloadMessage>(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) {
+      publishPayloadMessage(payloadMessage.topic, payloadMessage.message);
+    });
+
+    MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this, (DisconnectClientMessage disconnectClientMessage) {
+      _disconnect();
+    });
+
+    MessageService().subscribe<ConnectClientMessage>(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) {
+      _connectToBroker();
+    });
+
+    MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this, (RequestConnectionStatusMessage requestConnectionStatusMessage) {
+      bool connected = false;
+      if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
+        if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
+          connected = true;
+        }
+      }
+
+      MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
+    });
   }
 }