diff --git a/lib/tools/mqtt_core.dart b/lib/tools/mqtt_core.dart index da5f003..7c06da0 100644 --- a/lib/tools/mqtt_core.dart +++ b/lib/tools/mqtt_core.dart @@ -39,6 +39,7 @@ 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'; @@ -48,10 +49,7 @@ import '../model/session_status.dart'; import 'constants.dart'; import 'util.dart'; -enum ConnectionStateEnum { - notConnected, - connected -} +enum ConnectionStateEnum { notConnected, connected } class MqttCore { late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected; @@ -109,9 +107,7 @@ class MqttCore { _mqttClient!.onDisconnected = _clientDisconnected; _mqttClient!.onAutoReconnect = _clientDisconnected; _mqttClient!.onAutoReconnected = _clientConnected; - _mqttClient!.connectionMessage = MqttConnectMessage() - .withClientIdentifier(await Util.getAppUuid()) - .authenticateAs("app", "apppwd"); + _mqttClient!.connectionMessage = MqttConnectMessage().withClientIdentifier(await Util.getAppUuid()).authenticateAs("app", "apppwd"); await _mqttClient!.connect(); @@ -126,13 +122,14 @@ class MqttCore { } else { Util.logInfo('Verbindung zum MQTT-Broker fehlgeschlagen!'); - MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); + 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)); + MessageService().publish(Constants.messageTopicConnectionStatusChanged, + arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } } @@ -149,17 +146,15 @@ class MqttCore { } } - publishText(String topic, String text) { - if (_mqttClient != null) { - var buffer = Uint8Buffer(); - buffer.addAll(utf8.encode(text)); + _publishText(String topic, String text) { + var buffer = Uint8Buffer(); + buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable); - _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer); + var messageTopic = "$_toServerTopic$topic/$_clientId"; - Util.logInfo("message\n$text\nsend to $topic"); - } else { - Util.logException("publishBinary called while offline!"); - } + _mqttClient!.publishMessage(messageTopic, MqttQos.exactlyOnce, buffer); + + Util.logInfo("message\n$text\nsend to $messageTopic"); } publishPayloadMessage(String topic, Map payload) { @@ -172,7 +167,7 @@ class MqttCore { var json = jsonEncode(payloadContainer.toJson()); try { - publishText(_toServerTopic, json); + _publishText(_toServerTopic, json); } catch (e) { _enqueueMessage(_toServerTopic, jsonEncode(payloadContainer.toJson())); } @@ -187,57 +182,58 @@ class MqttCore { final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage; if (message.variableHeader!.topicName.endsWith(_clientId)) { - var messageBytes = message.payload.message; - String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes)); + var messageBytes = message.payload.message; + String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes)); - var topicArray = message.variableHeader?.topicName; - var topicArraySplited = topicArray!.split("/"); - var topic = topicArraySplited[3]; + var topicArray = message.variableHeader?.topicName; + var topicArraySplited = topicArray!.split("/"); + var topic = topicArraySplited[3]; - Util.logInfo('message received\ntopic: $topic\nmessage: $payload'); + Util.logInfo('message received\ntopic: $topic\nmessage: $payload'); - switch (topic) { - case 'session': - _sessionMessageReceived(jsonDecode(payload)); - break; + switch (topic) { + case 'session': + _sessionMessageReceived(jsonDecode(payload)); + break; - case 'status': - _statusMessageReceived(jsonDecode(payload)); - break; + case 'status': + _statusMessageReceived(jsonDecode(payload)); + break; - case 'ranks': - _rankingMessageReceived(jsonDecode(payload)); - break; + case 'ranks': + _rankingMessageReceived(jsonDecode(payload)); + break; - case 'job': - _jobMessageReceived(jsonDecode(payload)); - break; + case 'job': + _jobMessageReceived(jsonDecode(payload)); + break; - case 'job_del': - _jobDeleteReceived(jsonDecode(payload)); - break; + case 'job_del': + _jobDeleteReceived(jsonDecode(payload)); + break; - case 'sync_request': - _syncRequestReceived(jsonDecode(payload)); - break; + case 'sync_request': + _syncRequestReceived(jsonDecode(payload)); + break; - case "mssg": - _messageMessageReceived(jsonDecode(payload)); - break; + case "mssg": + _messageMessageReceived(jsonDecode(payload)); + break; - case "upload_device_log": - _messageUploadDeviceLogReceived(jsonDecode(payload)); - break; + case "upload_device_log": + _messageUploadDeviceLogReceived(jsonDecode(payload)); + break; - default: - Util.logInfo('Unknown topic: $topic'); - break; + default: + Util.logInfo('Unknown topic: $topic'); + break; + } } - } - } catch(ex) { + } catch (ex) { Util.logException(ex.toString()); } - }}); + } + }); if (await Util.preferencesContainsKey(Constants.prefsPreviousMqttSubscription)) { String previousMqttSubscription = await Util.preferencesGetString(Constants.prefsPreviousMqttSubscription, "") as String; @@ -253,9 +249,11 @@ class MqttCore { _mqttClient!.subscribe(_fromServerTopic, MqttQos.exactlyOnce); Util.preferencesSaveString(Constants.prefsPreviousMqttSubscription, _fromServerTopic); - Util.logInfo('subscribed to topic: $_fromServerTopic'); + var fromServerTopic = "$_fromServerTopic#"; + Util.logInfo('subscribed to topic: $fromServerTopic'); - MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); + MessageService().publish(Constants.messageTopicConnectionStatusChanged, + arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); } void _sessionMessageReceived(payloadData) async { @@ -272,20 +270,22 @@ class MqttCore { await MyApp.isarInstance.sessionStatus.put(sessionStatusMessage); }); - MessageService().publish(Constants.messageTopicSessionStatusInfoUpdate, arguments: SessionStatusInfoUpdate(status: 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); - }); + void _statusMessageReceived(payloadData) async { + var statusMessage = Status.fromJson(payloadData); - MessageService().publish(Constants.messageTopicStatusInfoUpdated, arguments: StatusInfoUpdate(status: statusMessage)); - } + 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 _locationStatusMessageReceived(payloadData) async { var locationStatus = LocationStatus.fromJson(payloadData); @@ -295,7 +295,8 @@ class MqttCore { await MyApp.isarInstance.locationStatus.put(locationStatus); }); - MessageService().publish(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus)); + MessageService() + .publish(Constants.messageTopicLocationStatusUpdated, arguments: LocationStatusUpdate(status: locationStatus)); } void _rankingMessageReceived(payloadData) async { @@ -334,7 +335,7 @@ class MqttCore { } void _jobDeleteReceived(payloadData) async { - for(var jobId in payloadData) { + for (var jobId in payloadData) { await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.delete(jobId); }); @@ -364,7 +365,11 @@ class MqttCore { void _messageUploadDeviceLogReceived(payloadData) async { var deviceLogRequest = DeviceLogRequest.fromJson(payloadData); - var logEintries = await MyApp.isarInstance.logEntrys.filter().timeGreaterThan(deviceLogRequest.startDateTime).timeLessThan(deviceLogRequest.endDateTime).findAll(); + var logEintries = await MyApp.isarInstance.logEntrys + .filter() + .timeGreaterThan(deviceLogRequest.startDateTime) + .timeLessThan(deviceLogRequest.endDateTime) + .findAll(); var deviceLog = DeviceLogResponse(); deviceLog.entries = logEintries; @@ -404,7 +409,8 @@ class MqttCore { _startTrackTimer(); - MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); + MessageService().publish(Constants.messageTopicConnectionStatusChanged, + arguments: ConnectionStatus(status: ConnectionStatusEnum.notConnected)); } void _clientConnected() { @@ -414,7 +420,8 @@ class MqttCore { _initSession(); - MessageService().publish(Constants.messageTopicConnectionStatusChanged, arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); + MessageService().publish(Constants.messageTopicConnectionStatusChanged, + arguments: ConnectionStatus(status: ConnectionStatusEnum.connected)); } void _processBootstrap() async { @@ -443,6 +450,7 @@ class MqttCore { var response = await http.get(Uri.parse(uri)); var responseText = response.body; + Util.logInfo("Bootstrap response: $responseText"); var bootstrapResponsePayload = BootstrapResponse.fromJson(jsonDecode(responseText)); @@ -465,11 +473,10 @@ class MqttCore { var startSessionMessage = StartSessionMessage(); var jsonMap = startSessionMessage.toJson(); var json = jsonEncode(jsonMap); - publishText("session", json); + _publishText("session", json); } - _sendSyncMessage() async - { + _sendSyncMessage() async { var jobs = await MyApp.isarInstance.jobs.filter().stateGreaterThan(-1).findAll(); var answer = SyncJobsMessage(); @@ -525,7 +532,7 @@ class MqttCore { if (enqueuedMessages.isNotEmpty) { for (var enqueuedMessage in enqueuedMessages) { try { - publishText(enqueuedMessage.topic, enqueuedMessage.payload); + _publishText(enqueuedMessage.topic, enqueuedMessage.payload); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.enqueuedMessages.delete(enqueuedMessage.id); @@ -553,7 +560,8 @@ class MqttCore { publishPayloadMessage(payloadMessage.topic, payloadMessage.message); }); - MessageService().subscribe(Constants.messageTopicDisconnectClient, this, (DisconnectClientMessage disconnectClientMessage) { + MessageService().subscribe(Constants.messageTopicDisconnectClient, this, + (DisconnectClientMessage disconnectClientMessage) { _disconnect(); }); @@ -561,7 +569,8 @@ class MqttCore { _connectToBroker(); }); - MessageService().subscribe(Constants.messageTopicRequestConnectionStatus, this, (RequestConnectionStatusMessage requestConnectionStatusMessage) { + MessageService().subscribe(Constants.messageTopicRequestConnectionStatus, this, + (RequestConnectionStatusMessage requestConnectionStatusMessage) { bool connected = false; if (_mqttClient != null && _mqttClient!.connectionStatus != null) { if (_mqttClient!.connectionStatus!.state == MqttConnectionState.connected) { @@ -569,7 +578,8 @@ class MqttCore { } } - MessageService().publish(Constants.messageTopicCurrentConnectionStatus, arguments: CurrentConnectionStatusMessage(state: connected)); + MessageService().publish(Constants.messageTopicCurrentConnectionStatus, + arguments: CurrentConnectionStatusMessage(state: connected)); }); } }