Erweiterungen

lib/tools/mqtt_core.dart

@@ -3,6 +3,7 @@ import 'dart:convert';
 import 'dart:math';
 
 import 'package:archive/archive.dart';
+import 'package:dart_mq/dart_mq.dart';
 import 'package:device_info_plus/device_info_plus.dart';
 import 'package:http/http.dart' as http;
 import 'package:isar/isar.dart';
@@ -11,7 +12,7 @@ import 'package:mqtt5_client/mqtt5_server_client.dart';
 import 'package:stadtbote/model/enqueued_message.dart';
 import 'package:stadtbote/model/job.dart';
 import 'package:stadtbote/model/log_entry.dart';
-import 'package:stadtbote/model/message.dart';
+import 'package:stadtbote/model/chat_message.dart';
 import 'package:stadtbote/model/location_status.dart';
 import 'package:stadtbote/model/rank.dart';
 import 'package:stadtbote/model/status.dart';
@@ -52,7 +53,7 @@ import 'util.dart';
 
 enum ConnectionStateEnum { notConnected, connected }
 
-class MqttCore {
+class MqttCore with ProducerMixin, ConsumerMixin {
   late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected;
 
   late bool _disconnectInProgress = false;
@@ -80,7 +81,13 @@ class MqttCore {
   }
 
   _disconnectMqttClient() async {
-    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicConnectionStatusChanged,
+      payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
+    );
+
+    //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
 
     _disconnectInProgress = true;
 
@@ -131,12 +138,24 @@ class MqttCore {
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
-        await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+        // 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');
 
-      await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+      // TODO MESSAGING
+      sendMessage(
+        routingKey: Constants.messageTopicConnectionStatusChanged,
+        payload: ConnectionStatus(status: ConnectionStatusEnum.notConnected),
+      );
+
+      //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
     }
   }
 
@@ -250,7 +269,13 @@ class MqttCore {
 
     Util.logInfo('subscribed to topic: $fromServerTopic');
 
-    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicConnectionStatusChanged,
+      payload: ConnectionStatus(status: ConnectionStatusEnum.connected),
+    );
+
+    //await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _sessionMessageReceived(payloadData) async {
@@ -267,7 +292,13 @@ class MqttCore {
       await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage);
     });
 
-    await MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicSessionStatusInfoUpdate,
+      payload: SessionStatusInfoUpdate(status: sessionStatusMessage),
+    );
+
+    //await MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
 
     _startTrackTimer();
   }
@@ -280,7 +311,13 @@ class MqttCore {
       await MyApp.isarInstance.status.put(statusMessage);
     });
 
-    await MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicStatusInfoUpdated,
+      payload: StatusInfoUpdate(status: statusMessage),
+    );
+
+    //await MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
   }
 
   void _locationStatusMessageReceived(payloadData) async {
@@ -291,7 +328,13 @@ class MqttCore {
       await MyApp.isarInstance.locationStatus.put(locationStatus);
     });
 
-    await MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicLocationStatusUpdated,
+      payload: LocationStatusUpdate(status: locationStatus),
+    );
+
+    // await MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
   }
 
   void _rankingMessageReceived(payloadData) async {
@@ -302,7 +345,13 @@ class MqttCore {
       await MyApp.isarInstance.rankings.put(ranking);
     });
 
-    await MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicRankingUpdated,
+      payload: RankingUpdated(),
+    );
+
+    //await MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
   }
 
   void _jobMessageReceived(payloadData) async {
@@ -322,11 +371,13 @@ class MqttCore {
 
     _acknowlegdeJob(newJob);
 
-    try {
-      await MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
-    } catch (ex) {
-      Util.logException("publish messageTopicJobReceived failed!");
-    }
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicJobReceived,
+      payload: JobReceivedMessage(job: newJob),
+    );
+
+    // await MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
   }
 
   void _jobDeleteReceived(payloadData) async {
@@ -336,7 +387,13 @@ class MqttCore {
       await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).deleteAll();
     });
 
-    await MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id));
+    // 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 {
@@ -346,15 +403,21 @@ class MqttCore {
   }
 
   void _messageMessageReceived(Map<String, dynamic> payloadData) async {
-    var newMessage = await Message.mergeFromJson(payloadData);
+    var newMessage = await ChatMessage.mergeFromJson(payloadData);
 
     await MyApp.isarInstance.writeTxn(() async {
-      await MyApp.isarInstance.messages.put(newMessage);
+      await MyApp.isarInstance.chatMessages.put(newMessage);
     });
 
     await Util.playMp3('asbeep.mp3');
 
-    await MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
+    // TODO MESSAGING
+    sendMessage(
+      routingKey: Constants.messageTopicMessageReceived,
+      payload: MessageReceivedMessage(message: newMessage),
+    );
+
+    //await MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
   }
 
   void _messageUploadDeviceLogReceived(payloadData) async {
@@ -403,7 +466,13 @@ class MqttCore {
 
     _cancelTrackTimer();
 
-    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    // 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 {
@@ -413,7 +482,13 @@ class MqttCore {
 
     _initSession();
 
-    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    // 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 {
@@ -458,8 +533,6 @@ class MqttCore {
         var brokerUriAndPortArray = bootstrapResponsePayload.broker.replaceFirst("mqtt://", "").split(":");
         _brokerUri = brokerUriAndPortArray[0];
         _brokerPort = int.parse(brokerUriAndPortArray[1]);
-        //_brokerUri = "192.168.180.27";
-        //_brokerPort = 1883;
         _clientId = bootstrapResponsePayload.id;
         _fromServerTopic = bootstrapResponsePayload.fromServer;
         _toServerTopic = bootstrapResponsePayload.toServer;
@@ -559,20 +632,60 @@ class MqttCore {
   }
 
   void initMessages() {
-    MessageService().subscribe<PublishPayloadMessage>(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) {
-      _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message), payloadMessage.qos);
-    });
+    subscribe(
+      queueId: Constants.messageTopicPublishPayload,
+      callback: (message) {
+        var publisPayloadMessage = message.payload as PublishPayloadMessage;
 
-    MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this,
+        _publishText(publisPayloadMessage.topic, jsonEncode(publisPayloadMessage.message), publisPayloadMessage.qos);
+      },
+    );
+    /*MessageService().subscribe<PublishPayloadMessage>(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) {
+      _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message), payloadMessage.qos);
+    });*/
+
+    subscribe(
+      queueId: Constants.messageTopicDisconnectClient,
+      callback: (message) {
+        _disconnectMqttClient();
+      },
+    );
+    /*MessageService().subscribe<DisconnectClientMessage>(Constants.messageTopicDisconnectClient, this,
         (DisconnectClientMessage disconnectClientMessage) {
       _disconnectMqttClient();
-    });
+    });*/
 
-    MessageService().subscribe<ConnectClientMessage>(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) {
+    subscribe(
+      queueId: Constants.messageTopicConnectClient,
+      callback: (message) {
+        _connectToBroker();
+      },
+    );
+    /*MessageService().subscribe<ConnectClientMessage>(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) {
       _connectToBroker();
-    });
+    });*/
 
-    MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this,
+    subscribe(
+      queueId: Constants.messageTopicRequestConnectionStatus,
+      callback: (message) {
+        bool connected = false;
+        if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
+          if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
+            connected = true;
+          }
+        }
+
+        // TODO MESSAGING
+        sendMessage(
+          routingKey: Constants.messageTopicCurrentConnectionStatus,
+          payload: CurrentConnectionStatusMessage(state: connected),
+        );
+
+        //await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
+      },
+    );
+
+    /*MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this,
         (RequestConnectionStatusMessage requestConnectionStatusMessage) async {
       bool connected = false;
       if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
@@ -582,6 +695,6 @@ class MqttCore {
       }
 
       await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
-    });
+    });*/
   }
 }