fix(llc): fix skipped message retries due to premature queue removal

packages/stream_chat/CHANGELOG.md

@@ -4,6 +4,8 @@
 
 - Fixed cached messages are cleared from channels with unread messages when accessed
   offline. [[#2083]](https://github.com/GetStream/stream-chat-flutter/issues/2083)
+- Fixed RetryQueue skipping messages due to premature removal from the
+  queue. [[#2308]](https://github.com/GetStream/stream-chat-flutter/pull/2308)
 
 ✅ Added
 

packages/stream_chat/lib/src/client/retry_queue.dart

@@ -2,7 +2,6 @@ import 'dart:async';
 
 import 'package:collection/collection.dart';
 import 'package:rxdart/rxdart.dart';
-import 'package:stream_chat/src/client/retry_policy.dart';
 import 'package:stream_chat/stream_chat.dart';
 
 /// The retry queue associated to a channel.
@@ -14,7 +13,6 @@ class RetryQueue {
   }) : client = channel.client {
     _retryPolicy = client.retryPolicy;
     _listenConnectionRecovered();
-    _listenMessageEvents();
   }
 
   /// The channel of this queue.
@@ -41,19 +39,6 @@ class RetryQueue {
     }).addTo(_compositeSubscription);
   }
 
-  void _listenMessageEvents() {
-    channel.on().where((event) => event.message != null).listen((event) {
-      final message = event.message!;
-      final containsMessage = _messageQueue.containsMessage(message);
-      if (!containsMessage) return;
-
-      if (message.state.isCompleted) {
-        logger?.info('Removing sent message from queue : ${message.id}');
-        return _messageQueue.removeMessage(message);
-      }
-    }).addTo(_compositeSubscription);
-  }
-
   /// Add a list of messages.
   void add(Iterable<Message> messages) {
     assert(
@@ -62,9 +47,7 @@ class RetryQueue {
     );
 
     // Filter out messages that are already in the queue.
-    final messagesToAdd = messages.where((it) {
-      return !_messageQueue.containsMessage(it);
-    });
+    final messagesToAdd = messages.whereNot(_messageQueue.containsMessage);
 
     // If there are no messages to add, return.
     if (messagesToAdd.isEmpty) return;
@@ -106,7 +89,7 @@ class RetryQueue {
         channel.state?.updateMessage(message);
       } finally {
         // remove the message from the queue after it's handled.
-        _messageQueue.removeFirst();
+        _messageQueue.removeMessage(message);
       }
     }
 
@@ -130,8 +113,8 @@ class RetryQueue {
     final date2 = _getMessageDate(m2);
 
     if (date1 == null && date2 == null) return 0;
-    if (date1 == null) return -1;
-    if (date2 == null) return 1;
+    if (date1 == null) return 1;
+    if (date2 == null) return -1;
     return date1.compareTo(date2);
   }
 
@@ -148,18 +131,21 @@ class RetryQueue {
 }
 
 extension on HeapPriorityQueue<Message> {
-  void removeMessage(Message message) {
-    final list = toUnorderedList();
-    final index = list.indexWhere((it) => it.id == message.id);
-    if (index == -1) return;
-    final element = list[index];
-    remove(element);
+  bool removeMessage(Message message) {
+    for (final element in unorderedElements) {
+      if (element.id == message.id) {
+        return remove(element);
+      }
+    }
+
+    return false;
   }
 
   bool containsMessage(Message message) {
-    final list = toUnorderedList();
-    final index = list.indexWhere((it) => it.id == message.id);
-    if (index == -1) return false;
-    return true;
+    for (final element in unorderedElements) {
+      if (element.id == message.id) return true;
+    }
+
+    return false;
   }
 }

packages/stream_chat/lib/stream_chat.dart

@@ -19,6 +19,7 @@ export 'package:uuid/uuid.dart';
 export 'src/client/channel.dart';
 export 'src/client/client.dart';
 export 'src/client/key_stroke_handler.dart';
+export 'src/client/retry_policy.dart';
 export 'src/core/api/attachment_file_uploader.dart';
 export 'src/core/api/requests.dart';
 export 'src/core/api/responses.dart';

packages/stream_chat/test/src/client/channel_test.dart

@@ -1,7 +1,6 @@
 // ignore_for_file: lines_longer_than_80_chars
 
 import 'package:mocktail/mocktail.dart';
-import 'package:stream_chat/src/client/retry_policy.dart';
 import 'package:stream_chat/stream_chat.dart';
 import 'package:test/test.dart';
 

packages/stream_chat/test/src/client/retry_queue_test.dart

@@ -1,5 +1,4 @@
 import 'package:mocktail/mocktail.dart';
-import 'package:stream_chat/src/client/retry_policy.dart';
 import 'package:stream_chat/src/client/retry_queue.dart';
 import 'package:stream_chat/stream_chat.dart';
 import 'package:test/test.dart';

sample_app/lib/app.dart

@@ -265,6 +265,12 @@ StreamChatClient buildStreamChatClient(String apiKey) {
     apiKey,
     logLevel: logLevel,
     logHandlerFunction: _sampleAppLogHandler,
+    retryPolicy: RetryPolicy(
+      maxRetryAttempts: 3,
+      shouldRetry: (client, attempt, error) {
+        return error is StreamChatNetworkError && error.isRetriable;
+      },
+    ),
     //baseURL: 'http://<local-ip>:3030',
     //baseWsUrl: 'ws://<local-ip>:8800',
   )..chatPersistenceClient = chatPersistentClient;