Erweiterungen

lib/tools/mqtt_core.dart

@@ -119,31 +119,29 @@ class MqttCore {
 
         _reconnectDelay = 0;
 
-        _setupMqtt();
+        await _setupMqtt();
 
         _initSession();
       } 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));
     }
   }
 
-  bool _publishText(String topic, String text, {bool enqueueOnError = true}) {
+  bool _publishText(String topic, String text, MqttQos qos, {bool enqueueOnError = true}) {
     try {
       var buffer = Uint8Buffer();
       buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
 
       var messageTopic = "$_toServerTopic$_clientId/$topic";
 
-      _mqttClient!.publishMessage(messageTopic, MqttQos.exactlyOnce, buffer);
+      _mqttClient!.publishMessage(messageTopic, qos, buffer);
 
       Util.logInfo("message\n$text\nsend to $messageTopic");
       Util.logInfo("size in bytes: ${buffer.length}");
@@ -158,7 +156,7 @@ class MqttCore {
     return false;
   }
 
-  void _setupMqtt() async {
+  Future<void> _setupMqtt() async {
     Util.logInfo('_setupMqtt called');
 
     _mqttClient!.updates?.listen((List<MqttReceivedMessage<MqttMessage>> messages) {
@@ -368,7 +366,7 @@ class MqttCore {
 
     var json = jsonEncode(deviceLog.toJson());
 
-    _publishText("device_log", json);
+    _publishText("device_log", json, MqttQos.atMostOnce);
   }
 
   void _startBootstrapReconnectTimer() {
@@ -398,19 +396,17 @@ 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() {
-    Util.logInfo("_clientDisconnected called");
+    Util.logInfo("_clientConnected called");
 
     connectionState = ConnectionStateEnum.connected;
 
     _initSession();
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged,
-        arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _processBootstrap() async {
@@ -434,7 +430,6 @@ class MqttCore {
 
       var jsonMap = handshakeMessage.toJson();
       var json = Uri.encodeFull(jsonEncode(jsonMap));
-      //var uri = "https://test-app-server.assecutor.org/bootstrap?json=$json";
       var uri = "https://bootstrap.assecutor.de/bootstrap?json=$json";
 
       Util.logInfo("Bootstrap request: $uri");
@@ -454,7 +449,7 @@ class MqttCore {
 
       _connectToBroker();
     } catch (ex) {
-      Util.logException("bootstrap failed");
+      Util.logException("bootstrap failed: {$ex}");
 
       _startBootstrapReconnectTimer();
     }
@@ -464,7 +459,7 @@ class MqttCore {
     var startSessionMessage = StartSessionMessage();
     var jsonMap = startSessionMessage.toJson();
     var json = jsonEncode(jsonMap);
-    _publishText("session", json);
+    _publishText("session", json, MqttQos.exactlyOnce);
   }
 
   _sendSyncMessage() async {
@@ -482,7 +477,7 @@ class MqttCore {
       answer.sync.jobs.add(syncJob);
     }
 
-    _publishText("sync", jsonEncode(answer.toJson()));
+    _publishText("sync", jsonEncode(answer.toJson()), MqttQos.exactlyOnce);
   }
 
   void _startTrackTimer() {
@@ -495,7 +490,7 @@ class MqttCore {
         trackMessage.long = position.longitude;
         trackMessage.acc = position.accuracy.toInt();
 
-        _publishText("track", jsonEncode(trackMessage.toJson()));
+        _publishText("track", jsonEncode(trackMessage.toJson()), MqttQos.atMostOnce);
       }
     });
   }
@@ -522,7 +517,7 @@ class MqttCore {
     if (enqueuedMessages.isNotEmpty) {
       for (var enqueuedMessage in enqueuedMessages) {
         try {
-          if (_publishText(enqueuedMessage.topic, enqueuedMessage.payload, enqueueOnError: false)) {
+          if (_publishText(enqueuedMessage.topic, enqueuedMessage.payload, MqttQos.exactlyOnce, enqueueOnError: false)) {
             await MyApp.isarInstance.writeTxn(() async {
               await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id);
             });
@@ -541,12 +536,12 @@ class MqttCore {
     message.ver = job.ver;
     message.utc = DateTime.now().millisecondsSinceEpoch;
 
-    _publishText('job_received', jsonEncode(message.toJson()));
+    _publishText('job_received', jsonEncode(message.toJson()), MqttQos.exactlyOnce);
   }
 
   void initMessages() {
     MessageService().subscribe<PublishPayloadMessage>(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) {
-      _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message));
+      _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message), payloadMessage.qos);
     });
 
     MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this,