Fehlerkorrekturen

lib/tools/mqtt_core.dart

@@ -39,6 +39,7 @@ import 'package:stadtbote/tools/messages/ranking_updated.dart';
 import 'package:stadtbote/tools/messages/request_connection_status_message.dart';
 import 'package:stadtbote/tools/messages/session_info_update.dart';
 import 'package:stadtbote/tools/messages/status_info_update.dart';
+
 // ignore: depend_on_referenced_packages
 import 'package:typed_data/typed_buffers.dart';
 import 'package:uuid/uuid.dart';
@@ -48,10 +49,7 @@ import '../model/session_status.dart';
 import 'constants.dart';
 import 'util.dart';
 
-enum ConnectionStateEnum {
+enum ConnectionStateEnum { notConnected, connected }
-  notConnected,
-  connected
-}
 
 class MqttCore {
   late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected;
@@ -109,9 +107,7 @@ class MqttCore {
       _mqttClient!.onDisconnected = _clientDisconnected;
       _mqttClient!.onAutoReconnect = _clientDisconnected;
       _mqttClient!.onAutoReconnected = _clientConnected;
-      _mqttClient!.connectionMessage = MqttConnectMessage()
+      _mqttClient!.connectionMessage = MqttConnectMessage().withClientIdentifier(await Util.getAppUuid()).authenticateAs("app", "apppwd");
-          .withClientIdentifier(await Util.getAppUuid())
-          .authenticateAs("app", "apppwd");
 
       await _mqttClient!.connect();
 
@@ -126,13 +122,14 @@ class MqttCore {
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
-        MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+        MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
+            arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
       }
-
     } catch (ex) {
       Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
 
-      MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+      MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
+          arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
     }
   }
 
@@ -149,17 +146,15 @@ class MqttCore {
     }
   }
 
-  publishText(String topic, String text) {
+  _publishText(String topic, String text) {
-    if (_mqttClient != null) {
     var buffer = Uint8Buffer();
-      buffer.addAll(utf8.encode(text));
+    buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
 
-      _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer);
+    var messageTopic = "$_toServerTopic$topic/$_clientId";
 
-      Util.logInfo("message\n$text\nsend to $topic");
+    _mqttClient!.publishMessage(messageTopic, MqttQos.exactlyOnce, buffer);
-    } else {
+
-      Util.logException("publishBinary called while offline!");
+    Util.logInfo("message\n$text\nsend to $messageTopic");
-    }
   }
 
   publishPayloadMessage(String topic, Map<String, dynamic> payload) {
@@ -172,7 +167,7 @@ class MqttCore {
     var json = jsonEncode(payloadContainer.toJson());
 
     try {
-      publishText(_toServerTopic, json);
+      _publishText(_toServerTopic, json);
     } catch (e) {
       _enqueueMessage(_toServerTopic, jsonEncode(payloadContainer.toJson()));
     }
@@ -237,7 +232,8 @@ class MqttCore {
         } catch (ex) {
           Util.logException(ex.toString());
         }
-    }});
+      }
+    });
 
     if (await Util.preferencesContainsKey(Constants.prefsPreviousMqttSubscription)) {
       String previousMqttSubscription = await Util.preferencesGetString(Constants.prefsPreviousMqttSubscription, "") as String;
@@ -253,9 +249,11 @@ class MqttCore {
     _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
     Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, _fromServerTopic);
 
-    Util.logInfo('subscribed to topic: $_fromServerTopic');
+    var fromServerTopic = "$_fromServerTopic#";
+    Util.logInfo('subscribed to topic: $fromServerTopic');
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
+        arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _sessionMessageReceived(payloadData) async {
@@ -272,10 +270,12 @@ class MqttCore {
       await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage);
     });
 
-    MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
+    MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate,
+        arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
 
     _startTrackTimer();
   }
+
   void _statusMessageReceived(payloadData) async {
     var statusMessage = Status.fromJson(payloadData);
 
@@ -295,7 +295,8 @@ class MqttCore {
       await MyApp.isarInstance.locationStatus.put(locationStatus);
     });
 
-    MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
+    MessageService()
+        .publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
   }
 
   void _rankingMessageReceived(payloadData) async {
@@ -364,7 +365,11 @@ class MqttCore {
   void _messageUploadDeviceLogReceived(payloadData) async {
     var deviceLogRequest = DeviceLogRequest.fromJson(payloadData);
 
-    var logEintries = await MyApp.isarInstance.logEntrys.filter().timeGreaterThan(deviceLogRequest.startDateTime).timeLessThan(deviceLogRequest.endDateTime).findAll();
+    var logEintries = await MyApp.isarInstance.logEntrys
+        .filter()
+        .timeGreaterThan(deviceLogRequest.startDateTime)
+        .timeLessThan(deviceLogRequest.endDateTime)
+        .findAll();
 
     var deviceLog = DeviceLogResponse();
     deviceLog.entries = logEintries;
@@ -404,7 +409,8 @@ class MqttCore {
 
     _startTrackTimer();
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
+        arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
   }
 
   void _clientConnected() {
@@ -414,7 +420,8 @@ class MqttCore {
 
     _initSession();
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
+        arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _processBootstrap() async {
@@ -443,6 +450,7 @@ class MqttCore {
 
       var response = await http.get(Uri.parse(uri));
       var responseText = response.body;
+      Util.logInfo("Bootstrap response: $responseText");
 
       var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText));
 
@@ -465,11 +473,10 @@ class MqttCore {
     var startSessionMessage = StartSessionMessage();
     var jsonMap = startSessionMessage.toJson();
     var json = jsonEncode(jsonMap);
-    publishText("session", json);
+    _publishText("session", json);
   }
 
-  _sendSyncMessage() async
+  _sendSyncMessage() async {
-  {
     var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll();
 
     var answer = SyncJobsMessage();
@@ -525,7 +532,7 @@ class MqttCore {
     if (enqueuedMessages.isNotEmpty) {
       for (var enqueuedMessage in enqueuedMessages) {
         try {
-          publishText(enqueuedMessage.topic, enqueuedMessage.payload);
+          _publishText(enqueuedMessage.topic, enqueuedMessage.payload);
 
           await MyApp.isarInstance.writeTxn(() async {
             await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id);
@@ -553,7 +560,8 @@ class MqttCore {
       publishPayloadMessage(payloadMessage.topic, payloadMessage.message);
     });
 
-    MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this, (DisconnectClientMessage disconnectClientMessage) {
+    MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this,
+        (DisconnectClientMessage disconnectClientMessage) {
       _disconnect();
     });
 
@@ -561,7 +569,8 @@ class MqttCore {
       _connectToBroker();
     });
 
-    MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this, (RequestConnectionStatusMessage requestConnectionStatusMessage) {
+    MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this,
+        (RequestConnectionStatusMessage requestConnectionStatusMessage) {
       bool connected = false;
       if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
         if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
@@ -569,7 +578,8 @@ class MqttCore {
         }
       }
 
-      MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
+      MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus,
+          arguments: CurrentConnectionStatusMessage(state: connected));
     });
   }
 }