Erweiterungen

lib/tools/mqtt_core.dart

@@ -4,6 +4,7 @@ import 'dart:math';
 
 import 'package:archive/archive.dart';
 import 'package:device_info_plus/device_info_plus.dart';
+import 'package:flutter/foundation.dart';
 import 'package:http/http.dart' as http;
 import 'package:isar/isar.dart';
 import 'package:mqtt_client/mqtt_client.dart';
@@ -79,8 +80,8 @@ class MqttCore {
     _processBootstrap();
   }
 
-  _disconnectMqttClient() {
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+  _disconnectMqttClient() async {
+    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
 
     _disconnectInProgress = true;
 
@@ -93,7 +94,6 @@ class MqttCore {
     }
 
     connectionState = ConnectionStateEnum.notConnected;
-
   }
 
   _connectToBroker() async {
@@ -128,19 +128,23 @@ class MqttCore {
       } else {
         Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
 
-        MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: 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');
 
-      MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: 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}) {
     try {
       var buffer = Uint8Buffer();
-      buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
+      if (kDebugMode) {
+        buffer.addAll(const Utf8Encoder().convert(text));
+      } else {
+        buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable<int>);
+      }
 
       var messageTopic = "$_toServerTopic$_clientId/$topic";
 
@@ -174,8 +178,12 @@ class MqttCore {
           Util.logInfo('message received\ntopic: $topic\n');
           Util.logInfo("size in bytes: ${message.payload.message.length.toString()}");
 
-          var messageBytes = message.payload.message;
-          String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes));
+          String payload;
+          if (kDebugMode) {
+            payload = utf8.decode(message.payload.message);
+          } else {
+            payload = utf8.decode(GZipDecoder().decodeBytes(message.payload.message));
+          }
 
           Util.logInfo('$topic\nmessage: $payload');
 
@@ -244,7 +252,7 @@ class MqttCore {
 
     Util.logInfo('subscribed to topic: $fromServerTopic');
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _sessionMessageReceived(payloadData) async {
@@ -261,8 +269,7 @@ class MqttCore {
       await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage);
     });
 
-    MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate,
-        arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
+    await MessageService().publish<SessionStatusInfoUpdate>(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage));
 
     _startTrackTimer();
   }
@@ -275,7 +282,7 @@ class MqttCore {
       await MyApp.isarInstance.status.put(statusMessage);
     });
 
-    MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
+    await MessageService().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage));
   }
 
   void _locationStatusMessageReceived(payloadData) async {
@@ -286,7 +293,7 @@ class MqttCore {
       await MyApp.isarInstance.locationStatus.put(locationStatus);
     });
 
-    MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
+    await MessageService().publish<LocationStatusUpdate>(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus));
   }
 
   void _rankingMessageReceived(payloadData) async {
@@ -297,7 +304,7 @@ class MqttCore {
       await MyApp.isarInstance.rankings.put(ranking);
     });
 
-    MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
+    await MessageService().publish<RankingUpdated>(Constants.messageTopicRankingUpdated, arguments: RankingUpdated());
   }
 
   void _jobMessageReceived(payloadData) async {
@@ -318,7 +325,7 @@ class MqttCore {
     _acknowlegdeJob(newJob);
 
     try {
-      MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
+      await MessageService().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob));
     } catch (ex) {
       Util.logException("publish messageTopicJobReceived failed!");
     }
@@ -333,7 +340,7 @@ class MqttCore {
         await MyApp.isarInstance.jobs.delete(jobDelMessage.id);
       });
 
-      MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id));
+      await MessageService().publish<JobDeletedMessage>(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id));
     }
   }
 
@@ -352,7 +359,7 @@ class MqttCore {
 
     await Util.playMp3('asbeep.mp3');
 
-    MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
+    await MessageService().publish<MessageReceivedMessage>(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage));
   }
 
   void _messageUploadDeviceLogReceived(payloadData) async {
@@ -394,24 +401,24 @@ class MqttCore {
     }
   }
 
-  void _clientDisconnected() {
+  Future<void> _clientDisconnected() async {
     Util.logInfo("_clientDisconnected called");
 
     connectionState = ConnectionStateEnum.notConnected;
 
     _cancelTrackTimer();
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
   }
 
-  void _clientConnected() {
+  Future<void> _clientConnected() async {
     Util.logInfo("_clientConnected called");
 
     connectionState = ConnectionStateEnum.connected;
 
     _initSession();
 
-    MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+    await MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
 
   void _processBootstrap() async {
@@ -446,8 +453,8 @@ class MqttCore {
       var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText));
 
       var brokerUriAndPortArray = bootstrapResponsePayload.broker.replaceFirst("mqtt://", "").split(":");
-      _brokerUri = brokerUriAndPortArray[0];
-      _brokerPort = int.parse(brokerUriAndPortArray[1]);
+      _brokerUri = "192.168.180.27"; //brokerUriAndPortArray[0];
+      _brokerPort = 1883; //int.parse(brokerUriAndPortArray[1]);
       _clientId = bootstrapResponsePayload.id;
       _fromServerTopic = bootstrapResponsePayload.fromServer;
       _toServerTopic = bootstrapResponsePayload.toServer;
@@ -464,6 +471,7 @@ class MqttCore {
     var startSessionMessage = StartSessionMessage();
     var jsonMap = startSessionMessage.toJson();
     var json = jsonEncode(jsonMap);
+
     _publishText("session", json, MqttQos.exactlyOnce);
   }
 
@@ -559,7 +567,7 @@ class MqttCore {
     });
 
     MessageService().subscribe<RequestConnectionStatusMessage>(Constants.messageTopicRequestConnectionStatus, this,
-        (RequestConnectionStatusMessage requestConnectionStatusMessage) {
+        (RequestConnectionStatusMessage requestConnectionStatusMessage) async {
       bool connected = false;
       if (_mqttClient != null && _mqttClient!.connectionStatus != null) {
         if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) {
@@ -567,8 +575,7 @@ class MqttCore {
         }
       }
 
-      MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus,
-          arguments: CurrentConnectionStatusMessage(state: connected));
+      await MessageService().publish<CurrentConnectionStatusMessage>(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected));
     });
   }
 }