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(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); _startReconnectTimer(); } } catch (ex) { _startReconnectTimer(); Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex'); BroadcastReceiver().publish(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> 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 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(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(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(); } } }