From 91d50fe58ce714a2f72ca645f1eee708597cf144 Mon Sep 17 00:00:00 2001 From: Sven Carstensen Date: Tue, 30 Apr 2024 14:47:13 +0200 Subject: [PATCH] Refactoring --- lib/content_views/approach_area_page.dart | 6 +- lib/content_views/assigned_jobs_page.dart | 5 +- lib/content_views/main_page.dart | 9 +- lib/content_views/messages_page.dart | 5 +- lib/content_views/task_list_page.dart | 1 + lib/main.dart | 2 +- lib/popups/accept_jobs_popup.dart | 6 +- lib/popups/infos_popup.dart | 59 ++++-- lib/tools/constants.dart | 11 +- lib/tools/messages/PublishPayloadMessage.dart | 7 + lib/tools/mqtt_core.dart | 194 +++++++++++------- 11 files changed, 198 insertions(+), 107 deletions(-) create mode 100644 lib/tools/messages/PublishPayloadMessage.dart diff --git a/lib/content_views/approach_area_page.dart b/lib/content_views/approach_area_page.dart index a9c7192..183b8ec 100644 --- a/lib/content_views/approach_area_page.dart +++ b/lib/content_views/approach_area_page.dart @@ -10,6 +10,7 @@ import 'package:stadtbote/popups/tasks_popups/zip_popup.dart'; import 'package:stadtbote/protocoll_messages/zip_message.dart'; import 'package:stadtbote/tools/constants.dart'; import 'package:stadtbote/tools/message_service.dart'; +import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart'; import 'package:stadtbote/tools/messages/ranking_updated.dart'; import 'package:stadtbote/tools/mqtt_core.dart'; @@ -190,10 +191,7 @@ class _ApproachAreaPageState extends State { setPosMessage.track!.pos!.acc = position.accuracy.toInt(); } - var json = jsonEncode(setPosMessage.toJson()); - Util.logInfo(json); - - MqttCore.instance.publishMessage("set_zip", jsonEncode(setPosMessage.toJson())); + MessageService().publish(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "rank", message: setPosMessage.toJson())); } _initMessages() { diff --git a/lib/content_views/assigned_jobs_page.dart b/lib/content_views/assigned_jobs_page.dart index b463ad6..63106eb 100644 --- a/lib/content_views/assigned_jobs_page.dart +++ b/lib/content_views/assigned_jobs_page.dart @@ -10,6 +10,7 @@ import 'package:badges/badges.dart' as badges; import 'package:stadtbote/protocoll_messages/accept_job_message.dart'; import 'package:stadtbote/protocoll_messages/fin_task_message.dart'; import 'package:stadtbote/tools/message_service.dart'; +import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart'; import 'package:stadtbote/tools/messages/refresh_badges_message.dart'; import 'package:stadtbote/tools/mqtt_core.dart'; @@ -200,13 +201,13 @@ class _AssignedJobsPageState extends State { message.track.pos!.long = currentPosition != null ? currentPosition.longitude : 0.0; message.track.pos!.acc = currentPosition != null ? currentPosition.accuracy.toInt() : 0; - MqttCore.instance.publishMessage("reaction", jsonEncode(message.toJson())); + MessageService().publish("reaction", arguments: PublishPayloadMessage(topic: "reaction", message: message.toJson())); await MyApp.isarInstance.writeTxn(() async { await MyApp.isarInstance.jobs.delete(job.id); }); - MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage()); + MessageService().publish(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage()); _entries = await _createListEntries(); diff --git a/lib/content_views/main_page.dart b/lib/content_views/main_page.dart index f13c62e..0d27d2f 100644 --- a/lib/content_views/main_page.dart +++ b/lib/content_views/main_page.dart @@ -16,6 +16,7 @@ import 'package:stadtbote/model/location_status.dart'; import 'package:stadtbote/popups/infos_popup.dart'; import 'package:stadtbote/protocoll_messages/logout_message.dart'; import 'package:stadtbote/tools/message_service.dart'; +import 'package:stadtbote/tools/messages/PublishPayloadMessage.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'; @@ -280,7 +281,7 @@ class _MainPageState extends State { _refreshBadges(); }); - MessageService().subscribe(Constants.messageTopicRefreshBadges, this, (RefreshBadgesMessage message) { + MessageService().subscribe(Constants.messageTopicUpdateBadges, this, (RefreshBadgesMessage message) { _refreshBadges(); }); } @@ -318,7 +319,7 @@ class _MainPageState extends State { if (!_loggedOut) { showLogoutAlertDialog(context); } else { - MqttCore.setup(); + MessageService().publish(Constants.messageTopicConnectClient, arguments: ConnectClientMessage()); _loggedOut = false; @@ -334,7 +335,7 @@ class _MainPageState extends State { _sendLogoutMessage(); Future.delayed(const Duration(seconds: 1), () async { - MqttCore.instance.diconnect(); + MessageService().publish(Constants.messageTopicDisconnectClient, arguments: DisconnectClientMessage()); }); _loggedOut = true; @@ -380,6 +381,6 @@ class _MainPageState extends State { logoutMessage.pos!.acc = position.accuracy.toInt(); } - MqttCore.instance.publishMessage("logout", jsonEncode(logoutMessage.toJson())); + MessageService().publish(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "logout", message: logoutMessage.toJson())); } } diff --git a/lib/content_views/messages_page.dart b/lib/content_views/messages_page.dart index 944334e..67ececf 100644 --- a/lib/content_views/messages_page.dart +++ b/lib/content_views/messages_page.dart @@ -10,6 +10,9 @@ import 'package:stadtbote/main.dart'; import 'package:stadtbote/model/message.dart' as sbmessage; import 'package:stadtbote/model/message.dart'; import 'package:stadtbote/model/session_status.dart'; +import 'package:stadtbote/tools/constants.dart'; +import 'package:stadtbote/tools/message_service.dart'; +import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart'; import 'package:stadtbote/tools/mqtt_core.dart'; import '../tools/util.dart'; @@ -108,7 +111,7 @@ class _MessagesPageState extends State { await MyApp.isarInstance.messages.put(message); }); - MqttCore.instance.publishMessage('message', jsonEncode(message.toJson())); + MessageService().publish(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "message", message: message.toJson())); _messages.insert(0, types.TextMessage( author: _user, diff --git a/lib/content_views/task_list_page.dart b/lib/content_views/task_list_page.dart index 402eddc..c79ec03 100644 --- a/lib/content_views/task_list_page.dart +++ b/lib/content_views/task_list_page.dart @@ -23,6 +23,7 @@ import 'package:stadtbote/protocoll_messages/fin_tour_message.dart'; import 'package:stadtbote/protocoll_messages/pic_message.dart'; import 'package:stadtbote/tools/constants.dart'; import 'package:stadtbote/tools/message_service.dart'; +import 'package:stadtbote/tools/messages/PublishPayloadMessage.dart'; import '../main.dart'; import '../model/task_info.dart'; diff --git a/lib/main.dart b/lib/main.dart index c3a1536..3a4c866 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -458,7 +458,7 @@ class _MainState extends State
with SingleTickerProviderStateMixin { await showAcceptJobsPopup(context, await MyApp.isarInstance.jobs.filter().stateEqualTo(0).findAll(), () { _dialogOpened = false; - MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage()); + MessageService().publish(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage()); }); } } diff --git a/lib/popups/accept_jobs_popup.dart b/lib/popups/accept_jobs_popup.dart index 04709fd..fb6b226 100644 --- a/lib/popups/accept_jobs_popup.dart +++ b/lib/popups/accept_jobs_popup.dart @@ -16,6 +16,8 @@ import 'package:stadtbote/tools/mqtt_core.dart'; import 'package:stadtbote/tools/util.dart'; import 'package:badges/badges.dart' as badges; +import '../tools/messages/PublishPayloadMessage.dart'; + Future showAcceptJobsPopup(BuildContext context, List jobs, Function() onFinished) async { var jobsSet = jobs.toSet(); @@ -217,11 +219,11 @@ _sendAnswer(Job job, int item) async { message.track.pos!.long = currentPosition != null ? currentPosition.longitude : 0.0; message.track.pos!.acc = currentPosition != null ? currentPosition.accuracy.toInt() : 0; - MqttCore.instance.publishMessage("reaction", jsonEncode(message.toJson())); + MessageService().publish(Constants.messageTopicPublishPayload, arguments: PublishPayloadMessage(topic: "reaction", message: message.toJson())); await MyApp.isarInstance.writeTxn(() async{ await MyApp.isarInstance.jobs.delete(job.id); }); - MessageService().publish(Constants.messageTopicRefreshBadges, arguments: RefreshBadgesMessage()); + MessageService().publish(Constants.messageTopicUpdateBadges, arguments: RefreshBadgesMessage()); } diff --git a/lib/popups/infos_popup.dart b/lib/popups/infos_popup.dart index ec7c200..43e5c3e 100644 --- a/lib/popups/infos_popup.dart +++ b/lib/popups/infos_popup.dart @@ -1,28 +1,57 @@ import 'package:flutter/material.dart'; import 'package:stadtbote/main.dart'; import 'package:stadtbote/model/session_status.dart'; +import 'package:stadtbote/tools/constants.dart'; +import 'package:stadtbote/tools/message_service.dart'; import 'package:stadtbote/tools/mqtt_core.dart'; import 'package:stadtbote/tools/util.dart'; void showInfosPopup(BuildContext context, SessionStatus? status) async { - var version = await Util.getAppVersion(); + final int version = await Util.getAppVersion(); - var text = "Versionsnummer: a$version\n"; - text += "Verbindungsstatus: ${MqttCore.instance.connectionState == ConnectionStateEnum.connected ? "Verbunden" : "NICHT verbunden"}\n"; - text += "GPS-Status: ${MyApp.currentPosition != null ? MyApp.currentPosition!.timestamp.toString() : "Unbekannt"}\n"; - text += "\n"; - text += "User Status:\n"; - if (status != null) { - text += "cr_id: ${status.crId}\n"; - text += "cust_id: ${status.custId}\n"; - text += "sid: ${status.sid}\n"; - text += "vht_id: ${status.vhtId}\n"; - text += "server_version: ${status.serverVersion}\n"; - text += "server_instance: ${status.serverInstance}\n"; - } else { - text += "Keine Informationen gefunden"; + late String connectedString = "Bitte warten..."; + + Future.delayed(const Duration(seconds: 2), () async { + MessageService().publish(Constants.messageTopicRequestConnectionStatus, arguments: RequestConnectionStatusMessage()); + }); + + void unsubscribeMessages() { + MessageService().unsubscribe(Constants.messageTopicCurrentConnectionStatus, version); + MessageService().unsubscribe(Constants.messageTopicConnectionStatusChanged, version); } + await showDialog( + // ignore: use_build_context_synchronously + context: context, + builder: (BuildContext context) { + // ignore: deprecated_member_use + return WillPopScope( + onWillPop: () async { + unsubscribeMessages(); + + return true; + }, + child: AlertDialog( + contentPadding: EdgeInsets.zero, + shadowColor: Colors.transparent, + backgroundColor: Colors.white, + content: StatefulBuilder(builder: (BuildContext context, StateSetter setState) { + var text = "Versionsnummer: a$version\n"; + text += "Verbindungsstatus: $connectedString\n"; + text += "GPS-Status: ${MyApp.currentPosition != null ? MyApp.currentPosition!.timestamp.toString() : "Unbekannt"}\n"; + text += "\n"; + text += "User Status:\n"; + if (status != null) { + text += "cr_id: ${status.crId}\n"; + text += "cust_id: ${status.custId}\n"; + text += "sid: ${status.sid}\n"; + text += "vht_id: ${status.vhtId}\n"; + text += "server_version: ${status.serverVersion}\n"; + text += "server_instance: ${status.serverInstance}\n"; + } else { + text += "Keine Informationen gefunden"; + } + await showDialog( context: context, builder: (BuildContext context) { diff --git a/lib/tools/constants.dart b/lib/tools/constants.dart index f115920..ce08f75 100644 --- a/lib/tools/constants.dart +++ b/lib/tools/constants.dart @@ -7,18 +7,25 @@ class Constants { static const String messageTopicConnectionStatusChanged = "connection_status_changed"; static const String messageTopicSessionInfoUpdate = "status_info_updated"; static const String messageTopicSessionStatusInfoUpdate = "session_status_info_updated"; - static const String messageTopicLocationStatusUpdated = "status_info_updated"; + static const String messageTopicLocationStatusUpdated = "location_status_updated"; static const String messageTopicRankingUpdated = "ranking_updated"; static const String messageTopicJobReceived = "job_received"; static const String messageTopicJobDeleted = "job_deleted"; - static const String messageTopicRefreshBadges = "refresh_badges"; + static const String messageTopicUpdateBadges = "update_badges"; static const String messageTopicMessageReceived = "message_received"; static const String messageTopicPushContentView = "push_content_view"; static const String messageTopicPopContentView = "pop_content_view"; static const String messageTopicReplaceContentView = "replace_content_view"; static const String messageTopicHideRightButton = "hide_right_button"; + static const String messageTopicPublishPayload = "publish_payload_message"; + static const String messageTopicConnectClient = "connect_client"; + static const String messageTopicDisconnectClient = "disconnect_client"; + static const String messageTopicRequestConnectionStatus = "request_connection_status"; + static const String messageTopicCurrentConnectionStatus = "current_connection_status"; static const String messageShowPopup = "show_popup"; + static const String prefsPreviousMqttSubscription = "previous_mqtt_subscription"; + static const int stateAssigned = 0; static const int stateAccepted = 1; static const int stateFinished = 2; diff --git a/lib/tools/messages/PublishPayloadMessage.dart b/lib/tools/messages/PublishPayloadMessage.dart new file mode 100644 index 0000000..f81c93b --- /dev/null +++ b/lib/tools/messages/PublishPayloadMessage.dart @@ -0,0 +1,7 @@ +class PublishPayloadMessage { + final String topic; + + final Map message; + + PublishPayloadMessage({required this.topic, required this.message}); +} \ No newline at end of file diff --git a/lib/tools/mqtt_core.dart b/lib/tools/mqtt_core.dart index 290ab19..c63e588 100644 --- a/lib/tools/mqtt_core.dart +++ b/lib/tools/mqtt_core.dart @@ -45,8 +45,6 @@ enum ConnectionStateEnum { } class MqttCore { - static late MqttCore instance; - late ConnectionStateEnum connectionState = ConnectionStateEnum.notConnected; late bool _disconnectInProgress = false; @@ -57,7 +55,7 @@ class MqttCore { late String _fromServerTopic; late String _toServerTopic; - late MqttServerClient _mqttClient; + MqttServerClient? _mqttClient; late int _reconnectDelay = 0; @@ -66,23 +64,23 @@ class MqttCore { Timer? _trackTimer; MqttCore() { + initMessages(); + _disconnectInProgress = false; _processBootstrap(); } - static setup() { - instance = MqttCore(); - } - - diconnect() { + _disconnect() { _disconnectInProgress = true; _cancelReconnectTimer(); _cancelTrackTimer(); - _mqttClient.disconnect(); + if (_mqttClient != null) { + _mqttClient!.disconnect(); + } connectionState = ConnectionStateEnum.notConnected; } @@ -95,25 +93,25 @@ class MqttCore { 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() + _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(); + await _mqttClient!.connect(); - if (_mqttClient.connectionStatus?.state == MqttConnectionState.connected) { + if (_mqttClient!.connectionStatus?.state == MqttConnectionState.connected) { Util.logInfo('Verbunden mit dem MQTT-Broker!'); _reconnectDelay = 0; - _setupSubscriptions(); + _setupMqtt(); _initSession(); } else { @@ -130,79 +128,87 @@ class MqttCore { } publishBinary(String topic, List data) { - var buffer = Uint8Buffer(); - buffer.addAll(data); + if (_mqttClient != null) { + var buffer = Uint8Buffer(); + buffer.addAll(data); - _mqttClient.publishMessage(topic, MqttQos.exactlyOnce, buffer); + _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer); - Util.logInfo("binary message send to $topic"); + Util.logInfo("binary message send to $topic"); + } else { + Util.logException("publishBinary called while offline!"); + } } - publishMessage(String topic, String text) { - var buffer = Uint8Buffer(); - buffer.addAll(GZipEncoder().encode(const Utf8Encoder().convert(text)) as Iterable); + publishText(String topic, String text) { + if (_mqttClient != null) { + var buffer = Uint8Buffer(); + buffer.addAll(utf8.encode(text)); - var messageTopic = "$_toServerTopic$topic/$_clientId"; + _mqttClient!.publishMessage(topic, MqttQos.exactlyOnce, buffer); - _mqttClient.publishMessage(messageTopic, MqttQos.exactlyOnce, buffer); - - Util.logInfo("message\n$text\nsend to $messageTopic"); + Util.logInfo("message\n$text\nsend to $topic"); + } else { + Util.logException("publishBinary called while offline!"); + } } - void _setupSubscriptions() async { - Util.logInfo('_setupSubscriptions called'); + void _setupMqtt() async { + Util.logInfo('_setupMqtt called'); - _mqttClient.updates?.listen((List> messages) { + _mqttClient!.updates?.listen((List> c) { try { - for (var mqttMessage in messages) { - final MqttPublishMessage message = mqttMessage.payload as MqttPublishMessage; + final MqttPublishMessage message = c[0].payload as MqttPublishMessage; + String payload = utf8.decode(MqttPublishPayload.bytesToStringAsString(message.payload.message).codeUnits); + var topic = message.variableHeader?.topicName; - if (message.variableHeader!.topicName.endsWith(_clientId)) { - var messageBytes = message.payload.message; - String payload = utf8.decode(GZipDecoder().decodeBytes(messageBytes)); + Util.logInfo('message received\ntopic: $topic\nmessage: $payload'); - var topicArray = message.variableHeader?.topicName; - var topicArraySplited = topicArray!.split("/"); - var topic = topicArraySplited[3]; + Map messageData = jsonDecode(payload); - Util.logInfo('message received\ntopic: $topic\nmessage: $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(topic) { - case 'session': - _sessionMessageReceived(jsonDecode(payload)); - break; + switch (payloadEntry) { + case 'session': + _sessionMessageReceived(payloadData); + break; - case 'location_status': - _locationSatusMessageReceived(jsonDecode(payload)); - break; + case 'location_status': + _locationSatusMessageReceived(jsonDecode(payload)); + break; - case 'ranking': - _rankingMessageReceived(jsonDecode(payload)); - break; + case 'ranking': + _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; + } } } } @@ -211,9 +217,21 @@ class MqttCore { } }); - var fromServerTopic = "$_fromServerTopic#"; - _mqttClient.subscribe(fromServerTopic, MqttQos.exactlyOnce); - Util.logInfo('subscribed to topic: $fromServerTopic'); + 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)); } @@ -388,7 +406,6 @@ class MqttCore { var jsonMap = handshakeMessage.toJson(); var json = Uri.encodeFull(jsonEncode(jsonMap)); - //var uri = "http://test-app-server.dns.army:8080/bootstrap?json=$json"; // Testserver bei Torben var uri = "https://test-app-server.assecutor.org/bootstrap?json=$json"; Util.logInfo("Bootstrap request: $uri"); @@ -496,6 +513,31 @@ class MqttCore { message.received.ver = job.ver; message.received.utc = DateTime.now().millisecondsSinceEpoch; - MqttCore.instance.publishMessage('job_received', jsonEncode(message.toJson())); + 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)); + }); } }