1. Import

lib/tools/mqtt_core.dart

@@ -0,0 +1,214 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:math';
+
+import 'package:flutter_broadcast_receiver/flutter_broadcast_receiver.dart';
+import 'package:mqtt_client/mqtt_client.dart';
+import 'package:mqtt_client/mqtt_server_client.dart';
+import 'package:stadtbote/model/bootstrap_request.dart';
+import 'package:stadtbote/model/bootstrap_response_payload.dart';
+import 'package:stadtbote/model/device_object.dart';
+import 'package:stadtbote/model/job.dart';
+import 'package:stadtbote/tools/connection_status.dart';
+import 'package:typed_data/typed_buffers.dart';
+import 'package:http/http.dart' as http;
+
+import '../main.dart';
+import '../model/status.dart';
+import 'constants.dart';
+import 'util.dart';
+
+class MqttCore {
+  static late MqttCore instance;
+
+  late String _brokerUri;
+  late int _brokerPort;
+  late String _fromServerTopic;
+  late String _toServerTopic;
+
+  late MqttServerClient _mqttClient;
+
+  late int _reconnectDelay = 0;
+
+  MqttCore() {
+    _processBootstrap();
+  }
+
+  static setup() {
+    instance = MqttCore();
+  }
+
+  _connectToBroker() async {
+    try {
+      String clientId = await Util.getAppUuid();
+      Util.logInfo("client id: $clientId");
+
+      _mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort);
+      _mqttClient.logging(on: false);
+      _mqttClient.keepAlivePeriod = 20;
+      _mqttClient.autoReconnect = false;
+      _mqttClient.onDisconnected = _clientDisconnected;
+
+      _mqttClient.connectionMessage = MqttConnectMessage()
+          .withClientIdentifier('clientId')
+          .withWillQos(MqttQos.atMostOnce)
+          .withWillTopic('willTopic')
+          .withWillMessage('willMessage')
+          .authenticateAs("app", "apppwd");
+
+      await _mqttClient.connect();
+
+      if (_mqttClient.connectionStatus?.state == MqttConnectionState.connected) {
+        Util.logInfo('Verbunden mit dem MQTT-Broker!');
+
+        _reconnectDelay = 0;
+
+        _setupSubscriptions();
+      } else {
+        Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!');
+
+        BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+
+        _startReconnectTimer();
+      }
+
+    } catch (ex) {
+      _startReconnectTimer();
+
+      Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex');
+
+      BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+    }
+  }
+
+  publishText(String topic, String text) {
+    try {
+      var buffer = Uint8Buffer();
+      buffer.addAll(text.codeUnits);
+
+      _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer);
+    } catch (e) {
+      Util.logException(e.toString());
+    }
+  }
+
+  void _setupSubscriptions() async {
+    Util.logInfo('_setupSubscriptions called');
+
+    _mqttClient.updates?.listen((List<MqttReceivedMessage<MqttMessage>> c) {
+      try {
+        final MqttPublishMessage message = c[0].payload as MqttPublishMessage;
+        final payload = MqttPublishPayload.bytesToStringAsString(message.payload.message);
+        var topic = message.variableHeader?.topicName;
+
+        Util.logInfo('message received');
+        Util.logInfo('topic: $topic');
+        Util.logInfo('payload: $payload');
+
+        Map<String, dynamic> messageData = jsonDecode(payload);
+
+        for (var entry in messageData.keys) {
+          if (entry == 'payload') {
+            var payload = messageData[entry];
+            for (var payloadEntry in payload.keys) {
+              var payloadData = payload[payloadEntry];
+
+              switch(payloadEntry) {
+                case 'session':
+                  _sessionMessageReceived(payloadData);
+                  break;
+
+                case 'jobs':
+                  _jobMessageReceived(payloadData);
+                  break;
+
+                default:
+                  Util.logInfo('Unknown topic: $payloadEntry');
+                  break;
+              }
+            }
+          }
+        }
+      } catch(ex) {
+        Util.logException(ex.toString());
+      }
+    });
+
+    _mqttClient.subscribe(_fromServerTopic, MqttQos.exactlyOnce);
+    Util.logInfo('subscribed to topic: $_fromServerTopic');
+
+    BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected));
+  }
+
+  void _sessionMessageReceived(payloadData) async {
+    var statusMessage = Status.fromJson(payloadData['status']);
+
+    await MyApp.isarInstance.writeTxn(() async {
+      await MyApp.isarInstance.status.put(statusMessage);
+    });
+  }
+
+  void _jobMessageReceived(payloadData) async {
+    for (var foundJobData in payloadData) {
+      var newJob = Job.fromJson(foundJobData);
+
+      await MyApp.isarInstance.writeTxn(() async {
+        await MyApp.isarInstance.jobs.put(newJob);
+      });
+    }
+  }
+
+  void _startReconnectTimer() {
+    Util.logInfo("_startReconnectTimer called");
+
+    _reconnectDelay = min(_reconnectDelay + 1, 30);
+
+    Timer(Duration(seconds: _reconnectDelay), () {
+      _processBootstrap();
+    });
+  }
+
+  void _clientDisconnected() {
+    Util.logInfo("_clientDisconnected called");
+
+    BroadcastReceiver().publish<ConnectionStatus>(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected));
+
+    _startReconnectTimer();
+  }
+
+  void _processBootstrap() async {
+    try {
+      var deviceObject = DeviceObject();
+      deviceObject.manufacturer = "sven";
+      deviceObject.model = "emulator";
+      deviceObject.os = "hat es";
+      deviceObject.osVer = "0.0";
+
+      var handshakeMessage = BootstrapRequest();
+      handshakeMessage.app = "sb";
+      handshakeMessage.mqttClientId = "12345";
+      handshakeMessage.deviceObject = deviceObject;
+      handshakeMessage.ver = 200;
+
+      var jsonMap = handshakeMessage.toJson();
+      var json = jsonEncode(jsonMap);
+      var uri = "https://init-x.assecutor.de/bootstrap?get_mqtt_broker=$json";
+      print("Bootstrap request: $uri");
+
+      var response = await http.get(Uri.parse(uri));
+      var responseText = response.body;
+
+      var bootstrapResponsePayload = BootstrapResponsePayload.fromJson(jsonDecode(responseText)["payload"]);
+
+      var brokerUriAndPortArray = bootstrapResponsePayload.config.broker.replaceFirst("mqtt://", "").split(":");
+      _brokerUri = brokerUriAndPortArray[0];
+      _brokerPort = int.parse(brokerUriAndPortArray[1]);
+      _fromServerTopic = bootstrapResponsePayload.config.topics.fromServer;
+      _toServerTopic = bootstrapResponsePayload.config.topics.toServer;
+
+      _connectToBroker();
+    } catch (ex) {
+      _startReconnectTimer();
+    }
+  }
+}