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:mqtt_client/mqtt_client.dart'; import 'package:mqtt_client/mqtt_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/payload_container.dart'; import 'package:stadtbote/protocoll_messages/start_session_message.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'; 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(); } _disconnect() { _disconnectInProgress = true; _cancelReconnectTimer(); _cancelTrackTimer(); if (_mqttClient != null) { _mqttClient!.disconnect(); } connectionState = ConnectionStateEnum.notConnected; } _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 = 5; _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; _setupMqtt(); _initSession(); } else { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!'); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } catch (ex) { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!: $ex'); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } publishBinary(String topic, List data) { if (_mqttClient != null) { var buffer = Uint8Buffer(); buffer.addAll(data); _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer); Util.logInfo("binary message send to $topic"); } else { Util.logException("publishBinary called while offline!"); } } publishText(String topic, String text) { if (_mqttClient != null) { var buffer = Uint8Buffer(); buffer.addAll(utf8.encode(text)); _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer); Util.logInfo("message\n$text\nsend to $topic"); } else { Util.logException("publishBinary called while offline!"); } } publishPayloadMessage(String topic, Map payload) { var payloadContainer = PayloadContainer(); payloadContainer.id = const Uuid().v4().toString(); payloadContainer.recv = topic; payloadContainer.store = true; payloadContainer.payload = payload; var json = jsonEncode(payloadContainer.toJson()); try { publishText(_toServerTopic, json); } catch (e) { _enqueueMessage(_toServerTopic, jsonEncode(payloadContainer.toJson())); } } void _setupMqtt() async { Util.logInfo('_setupMqtt called'); _mqttClient!.updates?.listen((List> c) { for (var message in c) { try { if (message.topic.endsWith(_clientId)) { var messageBytes = message.payload.; String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes)); var topicArray = message.variableHeader?.topicName; var topicArraySplited = topicArray!.split("/"); var topic = topicArraySplited[3]; Util.logInfo('message received\ntopic: $topic\nmessage: $payload'); switch(topic) { case 'session': _sessionMessageReceived(jsonDecode(payload)); break; case 'status': _statusMessageReceived(jsonDecode(payload)); break; case 'ranks': _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 "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()); } }); 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!.unsubscribe(previousMqttSubscription); } _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce); } _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce); Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, _fromServerTopic); Util.logInfo('subscribed to topic: $_fromServerTopic'); 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); }); 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); }); MessageService().publish(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage)); } void _locationSatusMessageReceived(payloadData) async { var locationStatus = LocationStatus.fromJson(payloadData); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.locationStatus.clear(); await MyApp.isarInstance.locationStatus.put(locationStatus); }); 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); }); 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 { MessageService().publish(Constants.messageTopicJobReceived, arguments: JobReceivedMessage(job: newJob)); } catch (ex) { Util.logException("publish messageTopicJobReceived failed!"); } } void _jobDeleteReceived(payloadData) async { for(var jobId in payloadData) { await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.delete(jobId); }); MessageService().publish(Constants.messageTopicJobDeleted, arguments: JobDeletedMessage(id: jobId)); } } 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'); 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()); var stringBytes = utf8.encode(json); var gzipBytes = GZipEncoder().encode(stringBytes); var topic = "/log/${_toServerTopic.replaceAll("/app/", "").replaceAll("/to_server", "")}"; publishBinary(topic, gzipBytes!); } 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 (_reconnectTimer != null && _reconnectTimer!.isActive) { _reconnectTimer!.cancel(); } } void _clientDisconnected() { Util.logInfo("_clientDisconnected called"); connectionState = ConnectionStateEnum.notConnected; _startTrackTimer(); MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } void _clientConnected() { Util.logInfo("_clientDisconnected called"); connectionState = ConnectionStateEnum.connected; _initSession(); 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://test-app-server.assecutor.org/bootstrap?json=$json"; Util.logInfo("Bootstrap request: $uri"); var response = await http.get(Uri.parse(uri)); var responseText = response.body; var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText)); 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"); _startBootstrapReconnectTimer(); } } _initSession() { var startSessionMessage = StartSessionMessage(); var jsonMap = startSessionMessage.toJson(); var json = jsonEncode(jsonMap); publishText("session", json); } _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); } publishPayloadMessage("sync_request_response", answer.toJson()); } void _startTrackTimer() { _trackTimer ??= Timer.periodic(const Duration(seconds: 60), (Timer t) { var position = MyApp.currentPosition; if (position != null) { var trackMessage = TrackMessage(); trackMessage.pos = TrackPosition(); trackMessage.pos!.time = position.timestamp.toIso8601String(); trackMessage.pos!.lat = position.latitude; trackMessage.pos!.long = position.longitude; trackMessage.pos!.acc = position.accuracy.toInt(); publishPayloadMessage("track", trackMessage.toJson()); } }); } 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 { publishText(enqueuedMessage.topic, enqueuedMessage.payload); 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 = AcknowledgeJob(); message.received = AcknowledgeJobReceived(); message.received.id = job.id; message.received.state = job.state; message.received.ver = job.ver; message.received.utc = DateTime.now().millisecondsSinceEpoch; publishPayloadMessage('job', message.toJson()); } void initMessages() { MessageService().subscribe(Constants.messageTopicPublishPayload, this, (PublishPayloadMessage payloadMessage) { publishPayloadMessage(payloadMessage.topic, payloadMessage.message); }); MessageService().subscribe(Constants.messageTopicDisconnectClient, this, (DisconnectClientMessage disconnectClientMessage) { _disconnect(); }); MessageService().subscribe(Constants.messageTopicConnectClient, this, (ConnectClientMessage connectClientMessage) { _connectToBroker(); }); MessageService().subscribe(Constants.messageTopicRequestConnectionStatus, this, (RequestConnectionStatusMessage requestConnectionStatusMessage) { bool connected = false; if (_mqttClient != null && _mqttClient!.connectionStatus != null) { if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) { connected = true; } } MessageService().publish(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected)); }); } }