Erweiterungen

lib/tools/mqtt_core.dart

@@ -48,6 +48,7 @@ class MqttCore {
   late String _brokerUri;
   late int _brokerPort;
   late String _fromServerTopic;
+  late String _clientId;
   late String _toServerTopic;
 
   late MqttServerClient _mqttClient;
@@ -127,12 +128,14 @@ class MqttCore {
     }
   }
 
-  publishText(String topic, String text) {
+  publishText(String subtopic, String text) {
     try {
       var buffer = Uint8Buffer();
       buffer.addAll(text.codeUnits);
 
-      _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
+      var topic = "$_toServerTopic$subtopic/$_clientId";
+
+      _mqttClient.publishMessage( topic, MqttQos.exactlyOnce, buffer);
 
       Util.logInfo("send message $text to $topic");
     } catch (e) {
@@ -140,22 +143,6 @@ class MqttCore {
     }
   }
 
-  publishPayloadMessage(String receiver, Map<String, dynamic> payload) {
-    try {
-      var payloadContainer = PayloadContainer();
-      payloadContainer.id = const Uuid().v4().toString();
-      payloadContainer.recv = receiver;
-      payloadContainer.store = true;
-      payloadContainer.payload = payload;
-
-      var json = jsonEncode(payloadContainer.toJson());
-
-      publishText(_toServerTopic, json);
-    } catch (e) {
-      Util.logException(e.toString());
-    }
-  }
-
   void _setupSubscriptions() async {
     Util.logInfo('_setupSubscriptions called');
 
@@ -170,7 +157,7 @@ class MqttCore {
         Util.logInfo('mesage: $payload');
 
         var topicArray = topic!.split('/');
-        var messageType = topicArray[topicArray.length - 1];
+        var messageType = topicArray[3];
 
         switch(messageType) {
           case 'session':
@@ -210,12 +197,9 @@ class MqttCore {
       }
     });
 
-    _mqttClient.subscribe(_fromServerTopic + "/session", MqttQos.exactlyOnce);
-    _mqttClient.subscribe(_fromServerTopic + "/status", MqttQos.exactlyOnce);
-    _mqttClient.subscribe(_fromServerTopic + "/job", MqttQos.exactlyOnce);
-    _mqttClient.subscribe(_fromServerTopic + "/job_del", MqttQos.exactlyOnce);
-    _mqttClient.subscribe(_fromServerTopic + "/ranking", MqttQos.exactlyOnce);
-    Util.logInfo('subscribed to topic: $_fromServerTopic');
+    var topic = "$_fromServerTopic#";
+    _mqttClient.subscribe(topic, MqttQos.exactlyOnce);
+    Util.logInfo('subscribed to topic: $topic');
 
     MessageService().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
   }
@@ -369,13 +353,14 @@ class MqttCore {
 
       var responseText = await rootBundle.loadString('assets/json/bootstrap_response.json');
 
-      var bootstrapResponsePayload = BootstrapResponsePayload.fromJson(jsonDecode(responseText)["payload"]);
+      var bootstrapResponse = BootstrapResponse.fromJson(jsonDecode(responseText));
 
-      var brokerUriAndPortArray = bootstrapResponsePayload.config.broker.replaceFirst("mqtt://", "").split(":");
+      var brokerUriAndPortArray = bootstrapResponse.broker.replaceFirst("mqtt://", "").split(":");
       _brokerUri = brokerUriAndPortArray[0];
       _brokerPort = int.parse(brokerUriAndPortArray[1]);
-      _fromServerTopic = bootstrapResponsePayload.config.topics.fromServer;
-      _toServerTopic = bootstrapResponsePayload.config.topics.toServer;
+      _clientId = bootstrapResponse.id;
+      _fromServerTopic = bootstrapResponse.fromServer;
+      _toServerTopic = bootstrapResponse.toServer;
 
       _connectToBroker();
     } catch (ex) {
@@ -385,13 +370,10 @@ class MqttCore {
 
   _initSession() {
     var startSessionMessage = StartSessionMessage();
-    startSessionMessage.startSession = true;
-
-    var topic = "$_toServerTopic/session";
 
     var jsonMap = startSessionMessage.toJson();
     var json = jsonEncode(jsonMap);
-    publishText(topic, json);
+    publishText("session", json);
   }
 
   _sendSyncMessage() async
@@ -410,7 +392,7 @@ class MqttCore {
       answer.sync.jobs.add(syncJob);
     }
 
-    MqttCore.instance.publishText("/appserver/to_server/100/1/104400/sync", jsonEncode(answer.toJson()));
+    MqttCore.instance.publishText("sync", jsonEncode(answer.toJson()));
   }
 
   void _startTrackTimer() {
@@ -424,7 +406,7 @@ class MqttCore {
           trackMessage.pos.long = answer.longitude;
           trackMessage.pos.acc = answer.accuracy.toInt();
 
-          publishText("/appserver/to_server/100/1/104400/track", jsonEncode(answer.toJson()));
+          publishText("track", jsonEncode(answer.toJson()));
         }
       });
   }