import 'dart:async'; import 'dart:convert'; import 'dart:math'; import 'package:archive/archive.dart'; import 'package:device_info_plus/device_info_plus.dart'; import 'package:http/http.dart' as http; import 'package:isar/isar.dart'; import 'package:mqtt5_client/mqtt5_client.dart'; import 'package:mqtt5_client/mqtt5_server_client.dart'; import 'package:stadtbote/model/enqueued_message.dart'; import 'package:stadtbote/model/job.dart'; import 'package:stadtbote/model/log_entry.dart'; import 'package:stadtbote/model/message.dart'; import 'package:stadtbote/model/location_status.dart'; import 'package:stadtbote/model/rank.dart'; import 'package:stadtbote/model/status.dart'; import 'package:stadtbote/protocoll_messages/acknowledge_job.dart'; import 'package:stadtbote/protocoll_messages/bootstrap_request.dart'; import 'package:stadtbote/protocoll_messages/bootstrap_response_payload.dart'; import 'package:stadtbote/protocoll_messages/device_log_request.dart'; import 'package:stadtbote/protocoll_messages/device_log_response.dart'; import 'package:stadtbote/protocoll_messages/device_object.dart'; import 'package:stadtbote/protocoll_messages/job_del_message.dart'; import 'package:stadtbote/protocoll_messages/start_session_message.dart'; import 'package:stadtbote/protocoll_messages/status_info_update.dart'; import 'package:stadtbote/protocoll_messages/sync_jobs_message.dart'; import 'package:stadtbote/protocoll_messages/track_message.dart'; import 'package:stadtbote/tools/connection_status.dart'; import 'package:stadtbote/tools/message_service.dart'; import 'package:stadtbote/tools/messages/publish_payload_message.dart'; import 'package:stadtbote/tools/messages/connect_client_message.dart'; import 'package:stadtbote/tools/messages/current_connection_status_message.dart'; import 'package:stadtbote/tools/messages/disconnect_client_message.dart'; import 'package:stadtbote/tools/messages/job_deleted_message.dart'; import 'package:stadtbote/tools/messages/job_received_message.dart'; import 'package:stadtbote/tools/messages/message_message.dart'; import 'package:stadtbote/tools/messages/ranking_updated.dart'; import 'package:stadtbote/tools/messages/request_connection_status_message.dart'; import 'package:stadtbote/tools/messages/session_info_update.dart'; import 'package:stadtbote/tools/messages/status_info_update.dart'; // ignore: depend_on_referenced_packages import 'package:typed_data/typed_buffers.dart'; // ignore: unused_import import 'package:uuid/uuid.dart'; import '../main.dart'; import '../model/session_status.dart'; import 'constants.dart'; import 'util.dart'; enum ConnectionStateEnum { notConnected, connected } class MqttCore { late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected; late bool _disconnectInProgress = false; late String _brokerUri; late String _clientId; late int _brokerPort; late String _fromServerTopic; late String _toServerTopic; MqttServerClient? _mqttClient; late int _reconnectDelay = 0; Timer? _reconnectTimer; Timer? _trackTimer; MqttCore() { initMessages(); _disconnectInProgress = false; _processBootstrap(); } _disconnectMqttClient() async { await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); _disconnectInProgress = true; _cancelReconnectTimer(); _cancelTrackTimer(); if (_mqttClient != null && _mqttClient!.connectionStatus == ConnectionStateEnum.connected) { _mqttClient!.disconnect(); } connectionState = ConnectionStateEnum.notConnected; _disconnectInProgress = false; } _connectToBroker() async { try { String clientId = await Util.getAppUuid(); Util.logInfo("client id: $clientId"); Util.logInfo("broker uri: $_brokerUri"); Util.logInfo("broker port: $_brokerPort"); if (_mqttClient == null) { _mqttClient = MqttServerClient.withPort(_brokerUri, clientId, _brokerPort); _mqttClient!.logging(on: false); _mqttClient!.keepAlivePeriod = 60; _mqttClient!.disconnectOnNoResponsePeriod = 10; _mqttClient!.autoReconnect = true; _mqttClient!.onDisconnected = _clientDisconnected; _mqttClient!.onAutoReconnect = _clientDisconnected; _mqttClient!.onAutoReconnected = _clientConnected; _mqttClient!.connectionMessage = MqttConnectMessage().withClientIdentifier(await Util.getAppUuid()).authenticateAs("app", "apppwd"); } await _mqttClient!.connect(); if (_mqttClient!.connectionStatus?.state == MqttConnectionState.connected) { Util.logInfo('Verbunden mit dem MQTT-Broker!'); _reconnectDelay = 0; await _setupMqtt(); _initSession(); } else { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!'); await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } catch (ex) { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex'); await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } bool _publishText(String topic, String text, MqttQos qos, {bool enqueueOnError = true}) { try { var buffer = Uint8Buffer(); buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable); var messageTopic = "$_toServerTopic$_clientId/$topic"; _mqttClient!.publishMessage(messageTopic, qos, buffer); Util.logInfo("message\n$text\nsend to $messageTopic"); Util.logInfo("size in bytes: ${buffer.length}"); return true; } catch (ex) { Util.logInfo("send failed!"); if (enqueueOnError) { _enqueueMessage(topic, text); Util.logInfo("message enqueued"); } } return false; } Future _setupMqtt() async { Util.logInfo('_setupMqtt called'); _mqttClient!.updates.listen((List> messages) { for (var mqttMessage in messages) { try { final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage; var topicArray = message.variableHeader?.topicName; var topicArraySplited = topicArray!.split("/"); var topic = topicArraySplited[6]; Util.logInfo('message received\ntopic: $topic\n'); Util.logInfo("size in bytes: ${message.payload.message!.length.toString()}"); var payload = utf8.decode(GZipDecoder().decodeBytes(message.payload.message!)); Util.logInfo('$topic\nmessage: $payload'); switch (topic) { case 'session': _sessionMessageReceived(jsonDecode(payload)); break; case 'status': _statusMessageReceived(jsonDecode(payload)); break; case 'ranking': _rankingMessageReceived(jsonDecode(payload)); break; case 'job': _jobMessageReceived(jsonDecode(payload)); break; case 'job_del': _jobDeleteReceived(jsonDecode(payload)); break; case 'sync_request': _syncRequestReceived(jsonDecode(payload)); break; case 'location_status': _locationStatusMessageReceived(jsonDecode(payload)); break; case "mssg": _messageMessageReceived(jsonDecode(payload)); break; case "upload_device_log": _messageUploadDeviceLogReceived(jsonDecode(payload)); break; default: Util.logInfo('Unknown topic: $topic'); break; } } catch (ex) { Util.logException(ex.toString()); } } }); var fromServerTopic = "$_fromServerTopic$_clientId/#"; if (await Util.preferencesContainsKey(Constants.prefsPreviousMqttSubscription)) { String previousMqttSubscription = await Util.preferencesGetString(Constants.prefsPreviousMqttSubscription, "") as String; if (previousMqttSubscription != fromServerTopic) { Util.logInfo('topic changed -> unsubscribe from $previousMqttSubscription'); _mqttClient!.unsubscribeStringTopic(previousMqttSubscription); } _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce); } _mqttClient!.subscribe(fromServerTopic, MqttQos.exactlyOnce); Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, fromServerTopic); Util.logInfo('subscribed to topic: $fromServerTopic'); await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); } void _sessionMessageReceived(payloadData) async { connectionState = ConnectionStateEnum.connected; await _sendSyncMessage(); await _sendEnqueuedMessages(); var sessionStatusMessage = SessionStatus.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.sessionStatus.clear(); await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage); }); await MessageService().publish(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: sessionStatusMessage)); _startTrackTimer(); } void _statusMessageReceived(payloadData) async { var statusMessage = Status.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.status.clear(); await MyApp.isarInstance.status.put(statusMessage); }); await MessageService().publish(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage)); } void _locationStatusMessageReceived(payloadData) async { var locationStatus = LocationStatus.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.locationStatus.clear(); await MyApp.isarInstance.locationStatus.put(locationStatus); }); await MessageService().publish(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus)); } void _rankingMessageReceived(payloadData) async { var ranking = Ranking.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.rankings.clear(); await MyApp.isarInstance.rankings.put(ranking); }); await MessageService().publish(Constants.messageTopicRankingUpdated, arguments: RankingUpdated()); } void _jobMessageReceived(payloadData) async { Job newJob = Job.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.put(newJob); }); if (newJob.state == Constants.stateAssigned) { if (newJob.global) { await Util.playMp3('aaa.mp3'); } else { await Util.playMp3('job_received.mp3'); } } _acknowlegdeJob(newJob); try { await MessageService().publish(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob)); } catch (ex) { Util.logException("publish messageTopicJobReceived failed!"); } } void _jobDeleteReceived(payloadData) async { var jobDelMessage = JobDelMessage.fromJson(payloadData); var foundJobs = await MyApp.isarInstance.jobs.where().filter().idEqualTo(jobDelMessage.id).findAll(); if (foundJobs.isNotEmpty) { await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.delete(jobDelMessage.id); }); await MessageService().publish(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobDelMessage.id)); } } void _syncRequestReceived(payloadData) async { Util.logInfo('_syncRequestReceived'); _sendSyncMessage(); } void _messageMessageReceived(Map payloadData) async { var newMessage = await Message.mergeFromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.messages.put(newMessage); }); await Util.playMp3('asbeep.mp3'); await MessageService().publish(Constants.messageTopicMessageReceived, arguments: MessageReceivedMessage(message: newMessage)); } void _messageUploadDeviceLogReceived(payloadData) async { var deviceLogRequest = DeviceLogRequest.fromJson(payloadData); var logEintries = await MyApp.isarInstance.logEntrys .filter() .timeGreaterThan(deviceLogRequest.startDateTime) .timeLessThan(deviceLogRequest.endDateTime) .findAll(); var deviceLog = DeviceLogResponse(); deviceLog.entries = logEintries; var json = jsonEncode(deviceLog.toJson()); _publishText("device_log", json, MqttQos.atMostOnce); } void _startBootstrapReconnectTimer() { connectionState = ConnectionStateEnum.notConnected; if (!_disconnectInProgress) { Util.logInfo("_startReconnectTimer called"); _reconnectDelay = min(_reconnectDelay + 1, 30); _reconnectTimer = Timer(Duration(seconds: _reconnectDelay), () { _processBootstrap(); }); } } void _cancelReconnectTimer() { if (!_disconnectInProgress) { if (_reconnectTimer != null && _reconnectTimer!.isActive) { _reconnectTimer!.cancel(); } } } Future _clientDisconnected() async { Util.logInfo("_clientDisconnected called"); connectionState = ConnectionStateEnum.notConnected; _cancelTrackTimer(); await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } Future _clientConnected() async { Util.logInfo("_clientConnected called"); connectionState = ConnectionStateEnum.connected; _initSession(); await MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); } void _processBootstrap() async { try { var deviceInfoPlugin = DeviceInfoPlugin(); AndroidDeviceInfo androidInfo = await deviceInfoPlugin.androidInfo; BaseDeviceInfo deviceInfo = await deviceInfoPlugin.deviceInfo; var deviceObject = DeviceObject(); deviceObject.manufacturer = deviceInfo.data['manufacturer']; deviceObject.model = deviceInfo.data['model']; deviceObject.os = androidInfo.version.release; deviceObject.osVer = androidInfo.version.sdkInt.toString(); var handshakeMessage = BootstrapRequest(); handshakeMessage.app = "sb"; handshakeMessage.mqttClientId = await Util.getAppUuid(); handshakeMessage.deviceObject = deviceObject; handshakeMessage.ver = await Util.getAppVersion(); var jsonMap = handshakeMessage.toJson(); var json = Uri.encodeFull(jsonEncode(jsonMap)); var uri = "https://bootstrap.assecutor.de/bootstrap?json=$json"; Util.logInfo("Bootstrap request: $uri"); var response = await http.get(Uri.parse(uri)); var responseText = response.body; Util.logInfo("Bootstrap response: $responseText"); var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText)); var status = bootstrapResponsePayload.status; var message = bootstrapResponsePayload.message; if (status > 0) { showAlertDialog(status == 1 ? "Information" : "Fehler", message); } if (status < 2) { var brokerUriAndPortArray = bootstrapResponsePayload.broker.replaceFirst("mqtt://", "").split(":"); _brokerUri = brokerUriAndPortArray[0]; _brokerPort = int.parse(brokerUriAndPortArray[1]); _clientId = bootstrapResponsePayload.id; _fromServerTopic = bootstrapResponsePayload.fromServer; _toServerTopic = bootstrapResponsePayload.toServer; _connectToBroker(); } } catch (ex) { Util.logException("bootstrap failed: {$ex}"); _startBootstrapReconnectTimer(); } } _initSession() { var startSessionMessage = StartSessionMessage(); var jsonMap = startSessionMessage.toJson(); var json = jsonEncode(jsonMap); _publishText("session", json, MqttQos.exactlyOnce); } _sendSyncMessage() async { var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll(); var answer = SyncJobsMessage(); answer.utcUnixTime = DateTime.now().toUtc().millisecondsSinceEpoch ~/ 1000; answer.sync = SyncJobMessageSync(); answer.sync.jobs = []; for (var job in jobs) { SyncJobMessageJob syncJob = SyncJobMessageJob(); syncJob.state = job.state; syncJob.id = job.id; syncJob.ver = job.ver.toString(); answer.sync.jobs.add(syncJob); } _publishText("sync", jsonEncode(answer.toJson()), MqttQos.exactlyOnce); } void _startTrackTimer() { _trackTimer ??= Timer.periodic(const Duration(seconds: 120), (Timer t) { var position = MyApp.currentPosition; if (position != null) { var trackMessage = TrackPosition(); trackMessage.time = position.timestamp.toIso8601String(); trackMessage.lat = position.latitude; trackMessage.long = position.longitude; trackMessage.acc = position.accuracy.toInt(); _publishText("track", jsonEncode(trackMessage.toJson()), MqttQos.atMostOnce); } }); } void _cancelTrackTimer() { if (_trackTimer != null && _trackTimer!.isActive) { _trackTimer!.cancel(); _trackTimer = null; } } void _enqueueMessage(String receiver, String payload) async { var enqueuedMessage = EnqueuedMessage(); enqueuedMessage.topic = receiver; enqueuedMessage.payload = payload; await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.enqueuedMessages.put(enqueuedMessage); }); } _sendEnqueuedMessages() async { var enqueuedMessages = await MyApp.isarInstance.enqueuedMessages.where().findAll(); if (enqueuedMessages.isNotEmpty) { for (var enqueuedMessage in enqueuedMessages) { try { if (_publishText(enqueuedMessage.topic, enqueuedMessage.payload, MqttQos.exactlyOnce, enqueueOnError: false)) { await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id); }); } } catch (ex) { Util.logInfo("resend enqueued message failed: $ex"); } } } } void _acknowlegdeJob(Job job) { var message = AcknowledgeJobReceived(); message.id = job.id; message.state = job.state; message.ver = job.ver; message.utc = DateTime.now().millisecondsSinceEpoch; _publishText('job_received', jsonEncode(message.toJson()), MqttQos.exactlyOnce); } void initMessages() { MessageService().subscribe(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) { _publishText(payloadMessage.topic, jsonEncode(payloadMessage.message), payloadMessage.qos); }); MessageService().subscribe(Constants.messageTopicDisconnectClient, this, (DisconnectClientMessage disconnectClientMessage) { _disconnectMqttClient(); }); MessageService().subscribe(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) { _connectToBroker(); }); MessageService().subscribe(Constants.messageTopicRequestConnectionStatus, this, (RequestConnectionStatusMessage requestConnectionStatusMessage) async { bool connected = false; if (_mqttClient != null && _mqttClient!.connectionStatus != null) { if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) { connected = true; } } await MessageService().publish(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected)); }); } }