diff --git a/lib/main.dart b/lib/main.dart index 29fa9d8..7f035e4 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -7,6 +7,7 @@ import 'package:flutter/services.dart'; import 'package:geolocator/geolocator.dart'; import 'package:path_provider/path_provider.dart'; import 'package:shared_map/shared_map.dart'; +import 'package:stadtbote/content_views/assigned_jobs_page.dart'; import 'package:stadtbote/messages/pop_content_view_message.dart'; import 'package:stadtbote/model/log_entry.dart'; import 'package:stadtbote/popups/accept_jobs_popup.dart'; @@ -529,7 +530,8 @@ class _MainState extends State
with SingleTickerProviderStateMixin, Produc void _startCheckForNewState0JobsJobsTimer() { _checkForNewState0JobsJobsTimer ??= Timer.periodic(const Duration(seconds: 5), (Timer t) async { - if (await _jobMap!.length() > 0 && !_dialogOpened) { + // Auf der Auftragsliste können neue Aufträge direkt beantwortet werden. + if (await _jobMap!.length() > 0 && !_dialogOpened && _currentContent is! AssignedJobsPage) { _dialogOpened = true; await showAcceptJobsPopup(_jobMap!.sharedReference(), () async { _dialogOpened = false; diff --git a/lib/tools/mqtt_core.dart b/lib/tools/mqtt_core.dart index 1560caf..d52ce73 100644 --- a/lib/tools/mqtt_core.dart +++ b/lib/tools/mqtt_core.dart @@ -6,6 +6,7 @@ import 'dart:math'; import 'package:archive/archive.dart'; import 'package:dart_mq/dart_mq.dart'; import 'package:device_info_plus/device_info_plus.dart'; +import 'package:flutter/foundation.dart'; import 'package:geolocator/geolocator.dart'; import 'package:http/http.dart' as http; import 'package:mqtt5_client/mqtt5_client.dart'; @@ -98,6 +99,22 @@ class MqttCore with ProducerMixin, ConsumerMixin { _processBootstrap(); } + /// Ermöglicht Tests des Nachrichtenflusses ohne Bootstrap und Netzwerkzugriff. + @visibleForTesting + MqttCore.withClient({ + required MqttServerClient client, + required String clientId, + required String fromServerTopic, + required String toServerTopic, + }) : _mqttClient = client, + _clientId = clientId, + _fromServerTopic = fromServerTopic, + _toServerTopic = toServerTopic { + instance = this; + initMessages(); + _setupMqtt(); + } + ConnectionStateEnum getConnectionStatus() { return connectionState; } @@ -185,15 +202,24 @@ class MqttCore with ProducerMixin, ConsumerMixin { } } - bool _publishMessage(String topic, Map message, bool acknowledgeRequired, {bool enqueueOnError = true}) { + bool _publishMessage(String topic, Map message, bool acknowledgeRequired) { try { - String messageId = const Uuid().v4(); - String messageText = jsonEncode(message); + final messageId = const Uuid().v4(); if (acknowledgeRequired) { - _enqueueMessage(messageId, topic, messageText); + // Vor dem ersten Versand speichern; erst die Serverbestätigung entfernt den Eintrag. + _enqueueMessage(messageId, topic, jsonEncode(message)); } + return _sendMessage(topic, message, acknowledgeRequired, messageId); + } catch (ex) { + Util.logInfo("send failed: $ex"); + return false; + } + } + + bool _sendMessage(String topic, Map message, bool acknowledgeRequired, String messageId) { + try { var encapsulatedMessageText = _encapsulateMessage(messageId, acknowledgeRequired, message); var buffer = Uint8Buffer(); @@ -761,7 +787,7 @@ class MqttCore with ProducerMixin, ConsumerMixin { } } - void _enqueueMessage(String uuid, String receiver, String payload) async { + void _enqueueMessage(String uuid, String receiver, String payload) { var enqueuedMessage = EnqueuedMessage(); enqueuedMessage.messageId = uuid; enqueuedMessage.topic = receiver; @@ -780,9 +806,8 @@ class MqttCore with ProducerMixin, ConsumerMixin { try { var json = jsonDecode(enqueuedMessage.payload); - if (_publishMessage(enqueuedMessage.topic, json, false, enqueueOnError: false)) { - box.remove(enqueuedMessage.id); - } + // Wiederholungen behalten ihre ID und benötigen weiterhin eine Bestätigung. + _sendMessage(enqueuedMessage.topic, json, true, enqueuedMessage.messageId); } catch (ex) { Util.logInfo("resend enqueued message failed: $ex"); } diff --git a/test/tools/mqtt_core_test.dart b/test/tools/mqtt_core_test.dart new file mode 100644 index 0000000..0f8d549 --- /dev/null +++ b/test/tools/mqtt_core_test.dart @@ -0,0 +1,317 @@ +import 'dart:async'; +import 'dart:convert'; + +import 'package:archive/archive.dart'; +import 'package:dart_mq/dart_mq.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:mqtt5_client/mqtt5_client.dart'; +import 'package:mqtt5_client/mqtt5_server_client.dart'; +import 'package:shared_preferences/shared_preferences.dart'; +import 'package:stadtbote/main.dart'; +import 'package:stadtbote/model/enqueued_message.dart'; +import 'package:stadtbote/model/job.dart'; +import 'package:stadtbote/model/log_entry.dart'; +import 'package:stadtbote/objectbox.g.dart'; +import 'package:stadtbote/tools/constants.dart'; +import 'package:stadtbote/tools/message_header.dart'; +import 'package:stadtbote/tools/messages/publish_payload_message.dart'; +import 'package:stadtbote/tools/mqtt_core.dart'; +import 'package:stadtbote/tools/util.dart'; +// ignore: depend_on_referenced_packages +import 'package:typed_data/typed_buffers.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + late _MessageStore store; + late _MqttClient client; + + setUp(() async { + SharedPreferences.setMockInitialValues({}); + await Util.init(); + MQClient.initialize(); + for (final topic in [ + Constants.messageTopicPublishPayload, + Constants.messageTopicDisconnectClient, + Constants.messageTopicConnectClient, + Constants.messageTopicRequestConnectionStatus, + Constants.messageTopicConnectionStatusChanged, + ]) { + MQClient.instance.declareQueue(topic); + } + store = _MessageStore(); + MyApp.objectBoxStore = store; + client = _MqttClient(); + MqttCore.withClient( + client: client, + clientId: 'test-device', + fromServerTopic: 'server/a/b/c/d/e/', + toServerTopic: 'client/a/b/c/d/e/', + ); + }); + + tearDown(() async { + await client.incoming.close(); + MQClient.instance.close(); + }); + + Future publish(WidgetTester tester, {bool requiresAck = true}) async { + MQClient.instance.sendMessage( + routingKey: Constants.messageTopicPublishPayload, + message: Message( + payload: PublishPayloadMessage( + topic: 'mssg', + message: {'text': 'Auftrag erledigt'}, + acknowledgeRequired: requiresAck, + ), + ), + ); + await tester.pump(); + } + + Future retry(WidgetTester tester) async { + client.receive('sync_request', {}); + await tester.pump(); + } + + testWidgets('behält ID, Inhalt und Bestätigungspflicht bei Wiederholungen', ( + tester, + ) async { + // Auch beim ersten Sendeversuch muss die Nachricht bereits gespeichert sein. + List? queuedAtFirstSend; + client.onPublish = () => queuedAtFirstSend = store.messages.getAll(); + await publish(tester); + client.onPublish = null; + final original = client.chatMessages.single; + final queued = store.messages.rows.single; + expect(queuedAtFirstSend, hasLength(1)); + expect(queuedAtFirstSend!.single.messageId, original.messageId); + expect(queued.messageId, original.messageId); + expect(jsonDecode(queued.payload), original.payload); + + await retry(tester); + await retry(tester); + + expect(client.chatMessages, hasLength(3)); + for (final message in client.chatMessages) { + expect(message.toJson(), original.toJson()); + expect(message.requiresAck, isTrue); + } + expect(store.messages.rows.single.id, queued.id); + expect(store.messages.putCount, 1); + }); + + testWidgets('sendet bestätigte Nachrichten nicht erneut', (tester) async { + await publish(tester); + final messageId = client.chatMessages.single.messageId; + await retry(tester); + expect(store.messages.rows, hasLength(1)); + + client.receive('acknowledge', {'messageId': messageId}); + await tester.pump(); + expect(store.messages.rows, isEmpty); + + await retry(tester); + expect(client.chatMessages, hasLength(2)); + }); + + testWidgets('behält unbestätigte Nachrichten nach fehlgeschlagenem Versand', ( + tester, + ) async { + client.failPublish = true; + await publish(tester); + final messageId = store.messages.rows.single.messageId; + await retry(tester); + expect(store.messages.rows.single.messageId, messageId); + expect(client.chatMessages, isEmpty); + + client.failPublish = false; + await retry(tester); + expect(client.chatMessages.single.messageId, messageId); + expect(client.chatMessages.single.requiresAck, isTrue); + expect(store.messages.rows, hasLength(1)); + }); + + testWidgets('verwendet die ID bereits gespeicherter Nachrichten', ( + tester, + ) async { + store.messages.put( + EnqueuedMessage() + ..messageId = 'persisted-message-id' + ..topic = 'mssg' + ..payload = '{"text":"Vor dem Neustart gespeichert"}', + ); + + await retry(tester); + + final message = client.chatMessages.single; + expect(message.messageId, 'persisted-message-id'); + expect(message.requiresAck, isTrue); + expect(message.payload['text'], 'Vor dem Neustart gespeichert'); + expect(store.messages.rows, hasLength(1)); + expect(store.messages.putCount, 1); + }); + + testWidgets('wiederholt Nachrichten ohne Bestätigungspflicht nicht', ( + tester, + ) async { + await publish(tester, requiresAck: false); + await retry(tester); + + expect(client.chatMessages, hasLength(1)); + expect(client.chatMessages.single.requiresAck, isFalse); + expect(store.messages.rows, isEmpty); + }); + + testWidgets( + 'vergibt für neue Nachrichten auch bei gleichem Inhalt eigene IDs', + (tester) async { + await publish(tester); + await publish(tester); + + final ids = client.chatMessages + .map((message) => message.messageId) + .toList(); + expect(ids.toSet(), hasLength(2)); + await retry(tester); + expect(client.chatMessages.map((message) => message.messageId), [ + ...ids, + ...ids, + ]); + expect(store.messages.rows, hasLength(2)); + }, + ); + + testWidgets('sendet bei fehlgeschlagener Speicherung keine neue Nachricht', ( + tester, + ) async { + store.messages.failPut = true; + await publish(tester); + + expect(client.chatMessages, isEmpty); + expect(store.messages.rows, isEmpty); + }); +} + +class _MqttClient extends Fake implements MqttServerClient { + final incoming = + StreamController>>.broadcast(); + final chatMessages = []; + bool failPublish = false; + void Function()? onPublish; + + @override + Stream>> get updates => incoming.stream; + + @override + MqttSubscription? subscribe(String topic, MqttQos qosLevel) => null; + + @override + int publishMessage( + String topic, + MqttQos qualityOfService, + Uint8Buffer data, { + bool retain = false, + List? userProperties, + }) { + onPublish?.call(); + if (failPublish) throw StateError('Keine Verbindung'); + if (topic.endsWith('/mssg')) { + chatMessages.add( + MessageHeader.fromJson( + jsonDecode(utf8.decode(GZipDecoder().decodeBytes(data))), + ), + ); + } + return 1; + } + + void receive(String topic, Map payload) { + final data = Uint8Buffer() + ..addAll( + GZipEncoder().encode( + utf8.encode( + jsonEncode({ + 'messageId': 'server-message-id', + 'requiresAck': false, + 'payload': payload, + }), + ), + ), + ); + final message = MqttPublishMessage() + .toTopic('server/a/b/c/d/e/test-device/$topic') + .publishData(data); + incoming.add([MqttReceivedMessage(topic, message)]); + } +} + +class _MessageStore extends Fake implements Store { + final messages = _MessageBox(); + final logs = _MemoryBox(); + final jobs = _MemoryBox(); + + @override + Box box() { + if (T == EnqueuedMessage) return messages as Box; + if (T == LogEntry) return logs as Box; + if (T == Job) return jobs as Box; + throw StateError('Unerwarteter Entity-Typ: $T'); + } +} + +class _MemoryBox extends Fake implements Box { + final rows = []; + + @override + List getAll() => List.of(rows); + + @override + int put(T object, {PutMode mode = PutMode.put}) => 0; + + @override + QueryBuilder query([Condition? qc]) => _MemoryQueryBuilder(rows); +} + +class _MessageBox extends _MemoryBox { + int putCount = 0; + bool failPut = false; + + @override + int put(EnqueuedMessage object, {PutMode mode = PutMode.put}) { + if (failPut) throw StateError('Speicherung fehlgeschlagen'); + object.id = ++putCount; + rows.add(object); + return object.id; + } + + @override + int removeMany(List ids) { + final count = rows.length; + rows.removeWhere((row) => ids.contains(row.id)); + return count - rows.length; + } +} + +// Die nativen Datenbankzugriffe sind ersetzt. Im Bestätigungstest enthält +// die Warteschlange genau die eine zum ACK passende Nachricht. +class _MemoryQueryBuilder extends Fake implements QueryBuilder { + final List rows; + + _MemoryQueryBuilder(this.rows); + + @override + Query build() => _MemoryQuery(rows); +} + +class _MemoryQuery extends Fake implements Query { + final List rows; + + _MemoryQuery(this.rows); + + @override + List find() => List.of(rows); + + @override + void close() {} +}