Annahmedialog auf Auftragsliste unterdrückt und MQTT-Wiederholungen abgesichert

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<Main> 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;

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<String, dynamic> message, bool acknowledgeRequired, {bool enqueueOnError = true}) {
+  bool _publishMessage(String topic, Map<String, dynamic> message, bool acknowledgeRequired) {
     try {
-      String messageId = const Uuid().v4();
+      final messageId = const Uuid().v4();
-      String messageText = jsonEncode(message);
 
       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<String, dynamic> 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)) {
+          // Wiederholungen behalten ihre ID und benötigen weiterhin eine Bestätigung.
-            box.remove(enqueuedMessage.id);
+          _sendMessage(enqueuedMessage.topic, json, true, enqueuedMessage.messageId);
-          }
         } catch (ex) {
           Util.logInfo("resend enqueued message failed: $ex");
         }

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<void> 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<void> 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<EnqueuedMessage>? 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<List<MqttReceivedMessage<MqttMessage>>>.broadcast();
+  final chatMessages = <MessageHeader>[];
+  bool failPublish = false;
+  void Function()? onPublish;
+
+  @override
+  Stream<List<MqttReceivedMessage<MqttMessage>>> 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<MqttUserProperty>? 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<String, dynamic> 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<LogEntry>();
+  final jobs = _MemoryBox<Job>();
+
+  @override
+  Box<T> box<T>() {
+    if (T == EnqueuedMessage) return messages as Box<T>;
+    if (T == LogEntry) return logs as Box<T>;
+    if (T == Job) return jobs as Box<T>;
+    throw StateError('Unerwarteter Entity-Typ: $T');
+  }
+}
+
+class _MemoryBox<T> extends Fake implements Box<T> {
+  final rows = <T>[];
+
+  @override
+  List<T> getAll() => List.of(rows);
+
+  @override
+  int put(T object, {PutMode mode = PutMode.put}) => 0;
+
+  @override
+  QueryBuilder<T> query([Condition<T>? qc]) => _MemoryQueryBuilder(rows);
+}
+
+class _MessageBox extends _MemoryBox<EnqueuedMessage> {
+  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<int> 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<T> extends Fake implements QueryBuilder<T> {
+  final List<T> rows;
+
+  _MemoryQueryBuilder(this.rows);
+
+  @override
+  Query<T> build() => _MemoryQuery(rows);
+}
+
+class _MemoryQuery<T> extends Fake implements Query<T> {
+  final List<T> rows;
+
+  _MemoryQuery(this.rows);
+
+  @override
+  List<T> find() => List.of(rows);
+
+  @override
+  void close() {}
+}