Fehlerkorrekturen und Erweiterungen

lib/tools/mqtt_core.dart

@@ -19,6 +19,7 @@ import 'package:stadtbote/model/status.dart';
 import 'package:stadtbote/protocoll_messages/acknowledge_job.dart';
 import 'package:stadtbote/protocoll_messages/bootstrap_request.dart';
 import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.dart';
+import 'package:stadtbote/protocoll_messages/connection_lost_message.dart';
 import 'package:stadtbote/protocoll_messages/device_log_request.dart';
 import 'package:stadtbote/protocoll_messages/device_log_response.dart';
 import 'package:stadtbote/protocoll_messages/device_object.dart';
@@ -88,14 +89,11 @@ class MqttCore with ProducerMixin, ConsumerMixin {
   }
 
   _disconnectMqttClient() async {
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicConnectionStatusChanged,
       payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
     );
 
-    //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
-
     _disconnectInProgress = true;
 
     _cancelReconnectTimer();
@@ -134,6 +132,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         willProperties.messageExpiryInterval = 3600;
         willProperties.willDelayInterval = 0;
 
+        var connectionLostMessage = ConnectionLostMessage();
+        connectionLostMessage.connectionLost = true;
+        var connectionLostEncapsulatedMessage = _encapsulateMessage(const Uuid().v4(), false, connectionLostMessage.toJson());
+
         _mqttClient!.connectionMessage = MqttConnectMessage()
             .withClientIdentifier(await Util.getAppUuid())
             .authenticateAs("app", "apppwd")
@@ -142,7 +144,7 @@ class MqttCore with ProducerMixin, ConsumerMixin {
             .withWillTopic("$_toServerTopic$_clientId/connection_lost")
             .withWillQos(MqttQos.atMostOnce)
             .keepAliveFor(60)
-            .withWillPayload(MqttUtf8Encoding().toUtf8('{"connection_lost": true}'))
+            .withWillPayload(MqttUtf8Encoding().toUtf8(connectionLostEncapsulatedMessage))
             .will();
       }
 
@@ -159,13 +161,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
-        // TODO MESSAGING
         sendMessage(
           routingKey: Constants.messageTopicConnectionStatusChanged,
           payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
         );
-
-        //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
       }
     } catch (ex) {
       Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
@@ -400,32 +399,35 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
     _acknowlegdeJob(newJob);
 
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicJobReceived,
       payload: JobReceivedMessage(job: newJob),
     );
-
-    // await MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
   }
 
   void _jobDeleteReceived(payloadData) async {
     var jobDelMessage = JobDelMessage.fromJson(payloadData);
 
+    var job = await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).findFirst();
+
     await MyApp.isarInstance.writeTxn(() async {
       await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).deleteAll();
     });
 
-    sendMessage(
-      routingKey: Constants.messageTopicJobDeleted,
-      payload: JobDeletedMessage(id: jobDelMessage.id),
-    );
+    if (job != null) {
+      sendMessage(
+        routingKey: Constants.messageTopicJobDeleted,
+        payload: JobDeletedMessage(job: job!),
+      );
+    }
   }
 
   void _syncRequestReceived(payloadData) async {
     Util.logInfo('_syncRequestReceived');
 
-    _sendSyncMessage();
+    await _sendEnqueuedMessages();
+
+    await _sendSyncMessage();
   }
 
   void _messageMessageReceived(Map<String, dynamic> payloadData) async {
@@ -701,18 +703,6 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         );
       },
     );
-
-    /*MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this,
-        (RequestConnectionStatusMessage requestConnectionStatusMessage) async {
-      bool connected = false;
-      if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
-        if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
-          connected = true;
-        }
-      }
-
-      await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
-    });*/
   }
 
   String _encapsulateMessage(String messageId, bool requiresAck, Map<String, dynamic> message) {