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/message.dart'; import 'package:stadtbote/model/payload_container.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/message_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(await Util.getAppUuid()) .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(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); Util.logInfo("message $text send to $topic"); } catch (e) { Util.logException(e.toString()); } } publishPayloadMessage(String receiver, Map 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'); _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; case "mssg": _messageMessageReceived(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); }); BroadcastReceiver().publish(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(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob)); } } void _messageMessageReceived(Map payloadData) async { var newMessage = Message.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.messages.put(newMessage); }); BroadcastReceiver().publish(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage)); } 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 = await Util.getAppUuid(); 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("send start session json: $json"); final builder = MqttClientPayloadBuilder(); builder.addString(json); var payload = builder.payload; var topic = "$_toServerTopic/init"; Util.logInfo("send start session json to topic: $topic"); var messageId = _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, payload!); Util.logInfo("send start session json with messageId: $messageId"); } }