stadtbote-app/lib/tools/mqtt_core.dart

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/model/start_session_message.dart';
import 'package:stadtbote/tools/connection_status.dart';
import 'package:stadtbote/tools/messages/job_received_message.dart';
import 'package:stadtbote/tools/messages/session_info_update.dart';
import 'package:typed_data/typed_buffers.dart';
import 'package:http/http.dart' as http;
import 'package:uuid/uuid.dart';

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");

      Util.logInfo("broker uri: $_brokerUri");
      Util.logInfo("broker port: $_brokerPort");

      _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();

        _initSession();
      } 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);
    });

    BroadcastReceiver().publish<StatusInfoUpdate>(Constants.messageTopicStatusInfoUpdate, arguments: StatusInfoUpdate(status: statusMessage));
  }

  void _jobMessageReceived(payloadData) async {
    for (var foundJobData in payloadData) {
      Job newJob = Job.fromJson(foundJobData);

      await MyApp.isarInstance.writeTxn(() async {
        await MyApp.isarInstance.jobs.put(newJob);
      });

      BroadcastReceiver().publish<JobReceivedMessage>(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: 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";
      Util.logInfo("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();
    }
  }

  _initSession() {
    var playload = Payload();
    playload.startSession = "true";

    var startSessionMessage = StartSessionMessage();
    startSessionMessage.recv = "server";
    startSessionMessage.store = true;
    startSessionMessage.messageId = const Uuid().v4();
    startSessionMessage.payload = playload;

    var jsonMap = startSessionMessage.toJson();
    var json = jsonEncode(jsonMap);
    Util.logInfo("json: $json");

    final builder = MqttClientPayloadBuilder();
    builder.addString(json);
    var payload = builder.payload;

    var topic = "$_toServerTopic/init";
    Util.logInfo("topic: $topic");

    var messageId = _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, payload!);
    Util.logInfo("messageId: $messageId");
  }

  /*
  private async void StartSession()
    {
        App.Logger.Debug("StartSession called");

        App.Logger.Debug("Setting state to start session");
        _mqttState = MqttStateEnum.StartSession;

        var startSession = new JObject
        {
            new JProperty("recv", "server"),
            new JProperty("store", true),
            new JProperty("id", Guid.NewGuid().ToString()),
            new JProperty("payload", new JObject(
                new JProperty("start_session", "true")))
        };

        var json = JsonConvert.SerializeObject(startSession);

        var topic = _toServerTopic + "/init";
        var message = new MqttApplicationMessage()
        {
            Topic = topic,
            Payload = Encoding.UTF8.GetBytes(json),
            QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce,
            Retain = true,
            MessageExpiryInterval = 3600,
        };
        await _mqttClient.PublishAsync(message);
        App.Logger.Debug("StartSession message " + json + " published to " + topic);

        SendEnqueuedMessages();
    }
   */
}