Fehlerkorrekturen und Erweiterungen

lib/tools/constants.dart

@@ -19,13 +19,17 @@ class Constants {
   static const String messageTopicPopContentView = "pop_content_view";
   static const String messageTopicReplaceContentView = "replace_content_view";
   static const String messageTopicPublishPayload = "publish_payload_message";
+  static const String messageTopicRemoveState0JobFromMap = "remove_state_0_job_from_map";
   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 messageTopicChatMessagesReceived = "mssg_received";
+
   static const String messageShowPopup = "show_popup";
 
   static const String prefsPreviousMqttSubscription = "previous_mqtt_subscription";
+  static const String prefsDesiredZip = "desired_zip";
 
   static const int stateAssigned = 0;
   static const int stateAccepted = 1;

lib/tools/message_header.dart

@@ -0,0 +1,19 @@
+import 'package:json_annotation/json_annotation.dart';
+
+part 'message_header.g.dart';
+
+@JsonSerializable()
+class MessageHeader {
+  late String messageId;
+
+  late bool requiresAck;
+
+  late Map<String, dynamic> payload;
+
+  MessageHeader();
+
+  factory MessageHeader.fromJson(Map<String, dynamic> json) => _$MessageHeaderFromJson(json);
+
+  Map<String, dynamic> toJson() => _$MessageHeaderToJson(this);
+}
+

lib/tools/message_header.g.dart

@@ -0,0 +1,20 @@
+// GENERATED CODE - DO NOT MODIFY BY HAND
+
+part of 'message_header.dart';
+
+// **************************************************************************
+// JsonSerializableGenerator
+// **************************************************************************
+
+MessageHeader _$MessageHeaderFromJson(Map<String, dynamic> json) =>
+    MessageHeader()
+      ..messageId = json['messageId'] as String
+      ..requiresAck = json['requiresAck'] as bool
+      ..payload = json['payload'] as Map<String, dynamic>;
+
+Map<String, dynamic> _$MessageHeaderToJson(MessageHeader instance) =>
+    <String, dynamic>{
+      'messageId': instance.messageId,
+      'requiresAck': instance.requiresAck,
+      'payload': instance.payload,
+    };

lib/tools/messages/publish_payload_message.dart

@@ -1,11 +1,9 @@
-import 'package:mqtt5_client/mqtt5_client.dart';
-
 class PublishPayloadMessage {
   final String topic;
 
   final Map<String, dynamic> message;
 
-  final MqttQos qos;
+  final bool acknowledgeRequired;
 
-  PublishPayloadMessage({required this.topic, required this.message, required this.qos});
+  PublishPayloadMessage({required this.topic, required this.message, required this.acknowledgeRequired});
 }

lib/tools/messages/remove_state_0_job_from_map.dart

@@ -0,0 +1,7 @@
+import 'package:stadtbote/model/job.dart';
+
+class RemoveState0JobFromMapMessage {
+  final int jobId;
+
+  RemoveState0JobFromMapMessage({required this.jobId});
+}

lib/tools/mqtt_core.dart

@@ -23,11 +23,13 @@ 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';
 import 'package:stadtbote/protocoll_messages/job_del_message.dart';
+import 'package:stadtbote/protocoll_messages/acknowledge_message.dart';
 import 'package:stadtbote/protocoll_messages/start_session_message.dart';
 import 'package:stadtbote/protocoll_messages/status_info_update.dart';
 import 'package:stadtbote/protocoll_messages/sync_jobs_message.dart';
 import 'package:stadtbote/protocoll_messages/track_message.dart';
 import 'package:stadtbote/tools/connection_status.dart';
+import 'package:stadtbote/tools/message_header.dart';
 import 'package:stadtbote/tools/messages/publish_payload_message.dart';
 import 'package:stadtbote/tools/messages/current_connection_status_message.dart';
 import 'package:stadtbote/tools/messages/job_deleted_message.dart';
@@ -39,6 +41,7 @@ import 'package:stadtbote/tools/messages/status_info_update.dart';
 
 // ignore: depend_on_referenced_packages
 import 'package:typed_data/typed_buffers.dart';
+
 // ignore: unused_import
 import 'package:uuid/uuid.dart';
 
@@ -126,7 +129,21 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         _mqttClient!.onDisconnected = _clientDisconnected;
         _mqttClient!.onAutoReconnect = _clientDisconnected;
         _mqttClient!.onAutoReconnected = _clientConnected;
-        _mqttClient!.connectionMessage = MqttConnectMessage().withClientIdentifier(await Util.getAppUuid()).authenticateAs("app", "apppwd");
+
+        var willProperties = MqttWillProperties();
+        willProperties.messageExpiryInterval = 3600;
+        willProperties.willDelayInterval = 0;
+
+        _mqttClient!.connectionMessage = MqttConnectMessage()
+            .withClientIdentifier(await Util.getAppUuid())
+            .authenticateAs("app", "apppwd")
+            .withSessionExpiryInterval(2700)
+            .withWillProperties(willProperties)
+            .withWillTopic("$_toServerTopic$_clientId/connection_lost")
+            .withWillQos(MqttQos.atMostOnce)
+            .keepAliveFor(60)
+            .withWillPayload(MqttUtf8Encoding().toUtf8('{"connection_lost": true}'))
+            .will();
       }
 
       await _mqttClient!.connect();
@@ -138,7 +155,7 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
         await _setupMqtt();
 
-        _initSession();
+        await _initSession();
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
@@ -153,37 +170,37 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     } catch (ex) {
       Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
 
-      // TODO MESSAGING
       sendMessage(
         routingKey: Constants.messageTopicConnectionStatusChanged,
         payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
       );
-
-      //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
     }
   }
 
-  bool _publishText(String topic, String text, MqttQos qos, {bool enqueueOnError = true}) {
+  bool _publishMessage(String topic, Map<String, dynamic> message, bool acknowledgeRequired, {bool enqueueOnError = true}) {
     try {
+      String messageId = const Uuid().v4();
+      String messageText = jsonEncode(message);
+
+      if (acknowledgeRequired) {
+        _enqueueMessage(messageId, topic, messageText);
+      }
+
+      var encapsulatedMessageText = _encapsulateMessage(messageId, acknowledgeRequired, message);
+
       var buffer = Uint8Buffer();
-      buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
+      buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(encapsulatedMessageText)) as Iterable<int>);
 
       var messageTopic = "$_toServerTopic$_clientId/$topic";
 
-      _mqttClient!.publishMessage(messageTopic, qos, buffer);
+      _mqttClient!.publishMessage(messageTopic, MqttQos.atMostOnce, buffer);
 
-      Util.logInfo("message\n$text\nsend to $messageTopic");
+      Util.logInfo("message\n$encapsulatedMessageText\nsend to $messageTopic");
       Util.logInfo("size in bytes: ${buffer.length}");
 
       return true;
     } catch (ex) {
       Util.logInfo("send failed!");
-
-      if (enqueueOnError) {
-        _enqueueMessage(topic, text);
-
-        Util.logInfo("message enqueued");
-      }
     }
 
     return false;
@@ -199,50 +216,58 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
           var topicArray = message.variableHeader?.topicName;
           var topicArraySplited = topicArray!.split("/");
-          var topic = topicArraySplited[6];
+          var topic = topicArraySplited[7];
 
           Util.logInfo('message received\ntopic: $topic\n');
           Util.logInfo("size in bytes: ${message.payload.message!.length.toString()}");
 
-          var payload = utf8.decode(GZipDecoder().decodeBytes(message.payload.message!));
+          var decompressedPayload = utf8.decode(GZipDecoder().decodeBytes(message.payload.message!));
 
-          Util.logInfo('$topic\nmessage: $payload');
+          var headerMessage = MessageHeader.fromJson(jsonDecode(decompressedPayload));
+          Util.logInfo('$topic\nmessageId: ${headerMessage.messageId}');
+
+          var payloadString = jsonEncode(headerMessage.payload);
+          Util.logInfo('$topic\nplayloadString: $payloadString');
 
           switch (topic) {
             case 'session':
-              _sessionMessageReceived(jsonDecode(payload));
+              _sessionMessageReceived(jsonDecode(payloadString));
               break;
 
             case 'status':
-              _statusMessageReceived(jsonDecode(payload));
+              _statusMessageReceived(jsonDecode(payloadString));
               break;
 
             case 'ranking':
-              _rankingMessageReceived(jsonDecode(payload));
+              _rankingMessageReceived(jsonDecode(payloadString));
               break;
 
             case 'job':
-              _jobMessageReceived(jsonDecode(payload));
+              _jobMessageReceived(jsonDecode(payloadString));
               break;
 
             case 'job_del':
-              _jobDeleteReceived(jsonDecode(payload));
+              _jobDeleteReceived(jsonDecode(payloadString));
               break;
 
             case 'sync_request':
-              _syncRequestReceived(jsonDecode(payload));
+              _syncRequestReceived(jsonDecode(payloadString));
               break;
 
             case 'location_status':
-              _locationStatusMessageReceived(jsonDecode(payload));
+              _locationStatusMessageReceived(jsonDecode(payloadString));
               break;
 
             case "mssg":
-              _messageMessageReceived(jsonDecode(payload));
+              _messageMessageReceived(jsonDecode(payloadString));
               break;
 
             case "upload_device_log":
-              _messageUploadDeviceLogReceived(jsonDecode(payload));
+              _messageUploadDeviceLogReceived(jsonDecode(payloadString));
+              break;
+
+            case "acknowledge":
+              _messageAcknowledgeReceived(jsonDecode(payloadString));
               break;
 
             default:
@@ -391,13 +416,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
       await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).deleteAll();
     });
 
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicJobDeleted,
       payload: JobDeletedMessage(id: jobDelMessage.id),
     );
-
-    //await MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id));
   }
 
   void _syncRequestReceived(payloadData) async {
@@ -415,13 +437,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
     await Util.playMp3('asbeep.mp3');
 
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicMessageReceived,
       payload: MessageReceivedMessage(message: newMessage),
     );
-
-    //await MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
   }
 
   void _messageUploadDeviceLogReceived(payloadData) async {
@@ -436,12 +455,22 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     var deviceLog = DeviceLogResponse();
     deviceLog.entries = logEintries;
 
-    var json = jsonEncode(deviceLog.toJson());
-
-    _publishText("device_log", json, MqttQos.atMostOnce);
+    _publishMessage("device_log", deviceLog.toJson(), false);
   }
 
-  void _startBootstrapReconnectTimer() {
+  void _messageAcknowledgeReceived(payloadData) async {
+    var acknowledgeMessage = AcknowledgeMessage.fromJson(payloadData);
+
+    await MyApp.isarInstance.writeTxn(() async {
+      await MyApp.isarInstance.enqueuedMessages
+          .filter()
+          .messageIdEqualTo(acknowledgeMessage.messageId)
+          .deleteAll();
+    });
+  }
+
+
+  void _startReconnectTimer() {
     connectionState = ConnectionStateEnum.notConnected;
 
     if (!_disconnectInProgress) {
@@ -470,13 +499,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
     _cancelTrackTimer();
 
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicConnectionStatusChanged,
       payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
     );
-
-    //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
   }
 
   Future<void> _clientConnected() async {
@@ -486,13 +512,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
 
     _initSession();
 
-    // TODO MESSAGING
     sendMessage(
       routingKey: Constants.messageTopicConnectionStatusChanged,
       payload: ConnectionStatus(status: ConnectionStatusEnum.connected),
     );
-
-    //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _processBootstrap() async {
@@ -546,16 +569,17 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     } catch (ex) {
       Util.logException("bootstrap failed: {$ex}");
 
-      _startBootstrapReconnectTimer();
+      _startReconnectTimer();
     }
   }
 
-  _initSession() {
+  _initSession() async {
     var startSessionMessage = StartSessionMessage();
-    var jsonMap = startSessionMessage.toJson();
-    var json = jsonEncode(jsonMap);
+    startSessionMessage.desiredZip = (await Util.preferencesGetString(Constants.prefsDesiredZip, ""))!;
 
-    _publishText("session", json, MqttQos.exactlyOnce);
+    var jsonMap = startSessionMessage.toJson();
+
+    _publishMessage("session", jsonMap, false);
   }
 
   _sendSyncMessage() async {
@@ -573,7 +597,7 @@ class MqttCore with ProducerMixin, ConsumerMixin {
       answer.sync.jobs.add(syncJob);
     }
 
-    _publishText("sync", jsonEncode(answer.toJson()), MqttQos.exactlyOnce);
+    _publishMessage("sync", answer.toJson(), false);
   }
 
   void _startTrackTimer() {
@@ -586,7 +610,7 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         trackMessage.long = position.longitude;
         trackMessage.acc = position.accuracy.toInt();
 
-        _publishText("track", jsonEncode(trackMessage.toJson()), MqttQos.atMostOnce);
+        _publishMessage("track", trackMessage.toJson(), false);
       }
     });
   }
@@ -598,8 +622,8 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     }
   }
 
-  void _enqueueMessage(String receiver, String payload) async {
-    var enqueuedMessage = EnqueuedMessage();
+  void _enqueueMessage(String uuid, String receiver, String payload) async {
+    var enqueuedMessage = EnqueuedMessage(messageId: uuid);
     enqueuedMessage.topic = receiver;
     enqueuedMessage.payload = payload;
 
@@ -613,7 +637,9 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     if (enqueuedMessages.isNotEmpty) {
       for (var enqueuedMessage in enqueuedMessages) {
         try {
-          if (_publishText(enqueuedMessage.topic, enqueuedMessage.payload, MqttQos.exactlyOnce, enqueueOnError: false)) {
+          var json = jsonDecode(enqueuedMessage.payload);
+
+          if (_publishMessage(enqueuedMessage.topic, json, false, enqueueOnError: false)) {
             await MyApp.isarInstance.writeTxn(() async {
               await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id);
             });
@@ -632,21 +658,18 @@ class MqttCore with ProducerMixin, ConsumerMixin {
     message.ver = job.ver;
     message.utc = DateTime.now().millisecondsSinceEpoch;
 
-    _publishText('job_received', jsonEncode(message.toJson()), MqttQos.exactlyOnce);
+    _publishMessage('job_received', message.toJson(), false);
   }
 
   void initMessages() {
     subscribe(
       queueId: Constants.messageTopicPublishPayload,
       callback: (message) {
-        var publisPayloadMessage = message.payload as PublishPayloadMessage;
+        var publishPayloadMessage = message.payload as PublishPayloadMessage;
 
-        _publishText(publisPayloadMessage.topic, jsonEncode(publisPayloadMessage.message), publisPayloadMessage.qos);
+        _publishMessage(publishPayloadMessage.topic, publishPayloadMessage.message, publishPayloadMessage.acknowledgeRequired);
       },
     );
-    /*MessageService().subscribe<PublishPayloadMessage>(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) {
-      _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message), payloadMessage.qos);
-    });*/
 
     subscribe(
       queueId: Constants.messageTopicDisconnectClient,
@@ -654,10 +677,6 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         _disconnectMqttClient();
       },
     );
-    /*MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this,
-        (DisconnectClientMessage disconnectClientMessage) {
-      _disconnectMqttClient();
-    });*/
 
     subscribe(
       queueId: Constants.messageTopicConnectClient,
@@ -665,9 +684,6 @@ class MqttCore with ProducerMixin, ConsumerMixin {
         _connectToBroker();
       },
     );
-    /*MessageService().subscribe<ConnectClientMessage>(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) {
-      _connectToBroker();
-    });*/
 
     subscribe(
       queueId: Constants.messageTopicRequestConnectionStatus,
@@ -679,13 +695,10 @@ class MqttCore with ProducerMixin, ConsumerMixin {
           }
         }
 
-        // TODO MESSAGING
         sendMessage(
           routingKey: Constants.messageTopicCurrentConnectionStatus,
           payload: CurrentConnectionStatusMessage(state: connected),
         );
-
-        //await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
       },
     );
 
@@ -701,4 +714,15 @@ class MqttCore with ProducerMixin, ConsumerMixin {
       await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
     });*/
   }
+
+  String _encapsulateMessage(String messageId, bool requiresAck, Map<String, dynamic> message) {
+    var messageHeader = MessageHeader();
+    messageHeader.messageId = messageId;
+    messageHeader.requiresAck = requiresAck;
+    messageHeader.payload = message;
+
+    var encapsulatedMessageText = jsonEncode(messageHeader.toJson());
+
+    return encapsulatedMessageText;
+  }
 }

lib/tools/util.dart

@@ -452,8 +452,11 @@ class DecimalTextInputFormatter extends TextInputFormatter {
   }
 }
 
-bool isJobInFuture(Job job) {
-  return DateTime.parse(job.ordertime).add(Duration(minutes: job.offsWarn)).toUtc().millisecondsSinceEpoch > DateTime.now().toUtc().millisecondsSinceEpoch;
+bool isJobTooFarInFuture(Job job) {
+  var ordertime = DateTime.parse(job.ordertime).toUtc().millisecondsSinceEpoch;
+  var dateTimeThreshold = DateTime.now().add(Duration(minutes: job.offsWarn)).toUtc().millisecondsSinceEpoch;
+
+  return ordertime > dateTimeThreshold;
 }
 
 Future<bool> taskDialogSecurityException(BuildContext context) async {