fix(ui): fix stopTyping unhandled exceptions when network is off or spotty. (#1296)

* feat(llc, ui): Introduce `keyStrokeHandler` to properly handle keyStrokes.

Signed-off-by: xsahil03x <[email protected]>

* chore(ui): update CHANGELOG.md

Signed-off-by: xsahil03x <[email protected]>

* test(llc): add key_stroke_handler_test.dart

Signed-off-by: xsahil03x <[email protected]>

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

@@ -8,6 +8,10 @@ import 'package:stream_chat/src/client/retry_queue.dart';
 import 'package:stream_chat/src/core/util/utils.dart';
 import 'package:stream_chat/stream_chat.dart';
 
+/// The maximum time the incoming [Event.typingStart] event is valid before a
+/// [Event.typingStop] event is emitted automatically.
+const incomingTypingStartEventTimeout = 7;
+
 /// Class that manages a specific channel.
 ///
 /// #### Channel name
@@ -1493,36 +1497,38 @@ class Channel {
           )
           .where((e) => e.cid == cid);
 
-  DateTime? _lastTypingEvent;
+  late final _keyStrokeHandler = KeyStrokeHandler(
+    onStartTyping: startTyping,
+    onStopTyping: stopTyping,
+  );
 
-  /// First of the [EventType.typingStart] and [EventType.typingStop] events
+  /// Sends the [Event.typingStart] event and schedules a timer to invoke the
-  /// based on the users keystrokes. Call this on every keystroke.
+  /// [Event.typingStop] event.
+  ///
+  /// This is meant to be called every time the user presses a key.
   Future<void> keyStroke([String? parentId]) async {
-    if (config?.typingEvents == false) {
+    if (config?.typingEvents == false) return;
-      return;
-    }
 
-    client.logger.info('start typing');
+    client.logger.info('KeyStroke received');
-    final now = DateTime.now();
+    return _keyStrokeHandler(parentId);
-
-    if (_lastTypingEvent == null ||
-        now.difference(_lastTypingEvent!).inSeconds >= 2) {
-      _lastTypingEvent = now;
-      await sendEvent(Event(
-        type: EventType.typingStart,
-        parentId: parentId,
-      ));
-    }
   }
 
-  /// Sets last typing to null and sends the typing.stop event.
+  /// Sends the [EventType.typingStart] event.
+  Future<void> startTyping([String? parentId]) async {
+    if (config?.typingEvents == false) return;
+
+    client.logger.info('start typing');
+    await sendEvent(Event(
+      type: EventType.typingStart,
+      parentId: parentId,
+    ));
+  }
+
+  /// Sends the [EventType.typingStop] event.
   Future<void> stopTyping([String? parentId]) async {
-    if (config?.typingEvents == false) {
+    if (config?.typingEvents == false) return;
-      return;
-    }
 
     client.logger.info('stop typing');
-    _lastTypingEvent = null;
     await sendEvent(Event(
       type: EventType.typingStop,
       parentId: parentId,
@@ -1532,6 +1538,7 @@ class Channel {
   /// Call this method to dispose the channel client.
   void dispose() {
     state?.dispose();
+    _keyStrokeHandler.cancel();
   }
 
   void _checkInitialized() {
@@ -1593,9 +1600,9 @@ class ChannelClientState {
 
     _listenMemberUnbanned();
 
-    _startCleaning();
+    _startCleaningStaleTypingEvents();
 
-    _startCleaningPinnedMessages();
+    _startCleaningStalePinnedMessages();
 
     _channel._client.chatPersistenceClient
         ?.getChannelThreads(_channel.cid!)
@@ -1613,7 +1620,8 @@ class ChannelClientState {
     });
   }
 
-  final _subscriptions = <StreamSubscription>[];
+  final Channel _channel;
+  final _subscriptions = CompositeSubscription();
 
   void _checkExpiredAttachmentMessages(ChannelState channelState) async {
     final expiredAttachmentMessagesId = channelState.messages
@@ -1786,20 +1794,15 @@ class ChannelClientState {
 
   /// Retry failed message.
   Future<void> retryFailedMessages() async {
-    final failedMessages =
+    final failedMessages = [...messages, ...threads.values.expand((v) => v)]
-        <Message>[...messages, ...threads.values.expand((v) => v)]
+        .where(
-            .where(
+          (message) =>
-              (message) =>
+              message.status != MessageSendingStatus.sent &&
-                  message.status != MessageSendingStatus.sent &&
+              message.createdAt.isBefore(
-                  message.createdAt.isBefore(
+                DateTime.now().subtract(const Duration(seconds: 5)),
-                    DateTime.now().subtract(
+              ),
-                      const Duration(
+        )
-                        seconds: 5,
+        .toList();
-                      ),
-                    ),
-                  ),
-            )
-            .toList();
 
     _retryQueue.add(failedMessages);
   }
@@ -1986,12 +1989,7 @@ class ChannelClientState {
     }
 
     _subscriptions.add(
-      _channel
+      _channel.on(EventType.messageRead, EventType.notificationMarkRead).listen(
-          .on(
-        EventType.messageRead,
-        EventType.notificationMarkRead,
-      )
-          .listen(
         (event) {
           final readList = List<Read>.from(_channelState.read ?? []);
           final userReadIndex =
@@ -2252,34 +2250,29 @@ class ChannelClientState {
     );
   }
 
-  /// Channel related typing users last value.
-  Map<User, Event> get typingEvents => _typingEventsController.value;
-
   /// Channel related typing users stream.
   Stream<Map<User, Event>> get typingEventsStream =>
       _typingEventsController.stream;
 
-  final BehaviorSubject<Map<User, Event>> _typingEventsController =
+  /// Channel related typing users last value.
-      BehaviorSubject.seeded({});
+  Map<User, Event> get typingEvents => _typingEventsController.value;
-
+  final _typingEventsController = BehaviorSubject.seeded(<User, Event>{});
-  final Channel _channel;
-  final Map<User, Event> _typings = {};
 
   void _listenTypingEvents() {
-    if (_channelState.channel?.config.typingEvents == false) {
+    if (_channelState.channel?.config.typingEvents == false) return;
-      return;
+
-    }
+    final currentUser = _channel.client.state.currentUser;
+    if (currentUser == null) return;
 
     _subscriptions
       ..add(
         _channel.on(EventType.typingStart).listen(
           (event) {
-            if (event.user != null) {
+            final user = event.user;
-              final user = event.user!;
+            if (user != null && user.id != currentUser.id) {
-              if (user.id != _channel.client.state.currentUser?.id) {
+              final events = {...typingEvents};
-                _typings[user] = event;
+              events[user] = event;
-                _typingEventsController.add(_typings);
+              _typingEventsController.add(events);
-              }
             }
           },
         ),
@@ -2287,112 +2280,109 @@ class ChannelClientState {
       ..add(
         _channel.on(EventType.typingStop).listen(
           (event) {
-            if (event.user != null) {
+            final user = event.user;
-              final user = event.user!;
+            if (user != null && user.id != currentUser.id) {
-              if (user.id != _channel.client.state.currentUser?.id) {
+              final events = {...typingEvents}..remove(user);
-                _typings.remove(event.user);
+              _typingEventsController.add(events);
-                _typingEventsController.add(_typings);
-              }
             }
           },
         ),
       )
       ..add(
-        _channel
+        _channel.on().where((event) {
-            .on()
+          final user = event.user;
-            .where((event) =>
+          if (user == null) return false;
-                event.user != null &&
+          return members.any((m) => m.userId == user.id);
-                members.any((m) => m.userId == event.user!.id))
+        }).listen(
-            .listen(
           (event) {
             final newMembers = List<Member>.from(members);
             final oldMemberIndex =
                 newMembers.indexWhere((m) => m.userId == event.user!.id);
             if (oldMemberIndex > -1) {
               final oldMember = newMembers.removeAt(oldMemberIndex);
-              updateChannelState(ChannelState(
+              updateChannelState(
-                members: [
+                ChannelState(
-                  ...newMembers,
+                  members: [
-                  oldMember.copyWith(
+                    ...newMembers,
-                    user: event.user,
+                    oldMember.copyWith(
-                  ),
+                      user: event.user,
-                ],
+                    ),
-              ));
+                  ],
+                ),
+              );
             }
           },
         ),
       );
   }
 
-  Timer? _cleaningTimer;
+  Timer? _staleTypingEventsCleanerTimer;
 
-  void _startCleaning() {
+  // Checks and removes stale typing events that were not explicitly stopped by
-    if (_channelState.channel?.config.typingEvents == false) {
+  // the sender due to technical difficulties. e.g. process death, loss of
-      return;
+  // Internet connection or custom implementation.
-    }
+  void _startCleaningStaleTypingEvents() {
+    if (_channelState.channel?.config.typingEvents == false) return;
 
-    _cleaningTimer = Timer.periodic(const Duration(seconds: 1), (_) {
+    _staleTypingEventsCleanerTimer = Timer.periodic(
-      final now = DateTime.now();
+      const Duration(seconds: 1),
-
+      (_) {
-      if (_channel._lastTypingEvent != null &&
+        final now = DateTime.now();
-          now.difference(_channel._lastTypingEvent!).inSeconds > 1) {
+        typingEvents.forEach((user, event) {
-        _channel.stopTyping();
+          if (now.difference(event.createdAt).inSeconds >
-      }
+              incomingTypingStartEventTimeout) {
-
+            _channel.client.handleEvent(
-      _clean();
+              Event(
-    });
+                type: EventType.typingStop,
+                user: user,
+                cid: _channel.cid,
+                parentId: event.parentId,
+              ),
+            );
+          }
+        });
+      },
+    );
   }
 
-  late Timer _pinnedMessagesTimer;
+  Timer? _stalePinnedMessagesCleanerTimer;
 
-  void _startCleaningPinnedMessages() {
+  // Checks and removes stale pinned messages that are not valid anymore.
-    _pinnedMessagesTimer = Timer.periodic(const Duration(seconds: 30), (_) {
+  void _startCleaningStalePinnedMessages() {
-      final now = DateTime.now();
+    _stalePinnedMessagesCleanerTimer = Timer.periodic(
-      var expiredMessages = channelState.pinnedMessages
+      const Duration(seconds: 30),
-          ?.where((m) => m.pinExpires?.isBefore(now) == true)
+      (_) {
-          .toList();
+        final now = DateTime.now();
-      if (expiredMessages != null && expiredMessages.isNotEmpty) {
+        var expiredMessages = channelState.pinnedMessages
-        expiredMessages = expiredMessages
+            ?.where((m) => m.pinExpires?.isBefore(now) == true)
-            .map((m) => m.copyWith(
-                  pinExpires: null,
-                  pinned: false,
-                ))
             .toList();
+        if (expiredMessages != null && expiredMessages.isNotEmpty) {
+          expiredMessages = expiredMessages
+              .map((m) => m.copyWith(
+                    pinExpires: null,
+                    pinned: false,
+                  ))
+              .toList();
 
-        updateChannelState(_channelState.copyWith(
+          updateChannelState(_channelState.copyWith(
-          pinnedMessages: pinnedMessages.where(_pinIsValid).toList(),
+            pinnedMessages: pinnedMessages.where(_pinIsValid).toList(),
-          messages: expiredMessages,
+            messages: expiredMessages,
-        ));
+          ));
-      }
+        }
-    });
+      },
-  }
+    );
-
-  void _clean() {
-    final now = DateTime.now();
-    _typings.forEach((user, event) {
-      if (now.difference(event.createdAt).inSeconds > 7) {
-        _channel.client.handleEvent(
-          Event(
-            type: EventType.typingStop,
-            user: user,
-            cid: _channel.cid,
-            parentId: event.parentId,
-          ),
-        );
-      }
-    });
   }
 
   /// Call this method to dispose this object.
   void dispose() {
     _debouncedUpdatePersistenceChannelState.cancel();
     _retryQueue.dispose();
-    _subscriptions.forEach((s) => s.cancel());
+    _subscriptions.cancel();
     _channelStateController.close();
     _isUpToDateController.close();
     _threadsController.close();
-    _cleaningTimer?.cancel();
+    _staleTypingEventsCleanerTimer?.cancel();
-    _pinnedMessagesTimer.cancel();
+    _stalePinnedMessagesCleanerTimer?.cancel();
     _typingEventsController.close();
   }
 }

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

@@ -0,0 +1,114 @@
+import 'dart:async';
+
+/// A class that manages buffering typing events and call [onTypingStarted] and
+/// [onTypingStopped] accordingly in a timed manner.
+///
+/// This class is used by [Channel] to manage typing events.
+class KeyStrokeHandler {
+  /// Creates a new instance of [KeyStrokeHandler].
+  KeyStrokeHandler({
+    this.startTypingEventTimeout = 1,
+    this.startTypingResendInterval = 3,
+    required this.onStartTyping,
+    required this.onStopTyping,
+  });
+
+  /// The number of seconds from the last [onStartTyping] callback until
+  /// the [onStopTyping] callback is automatically invoked.
+  final int startTypingEventTimeout;
+
+  /// The number of seconds after the last [onStartTyping] callback before
+  /// the [onStartTyping] callback is automatically invoked again.
+  final int startTypingResendInterval;
+
+  /// Called when a `typingStart` event needs to be send.
+  final Future<void> Function([String? parentId]) onStartTyping;
+
+  /// Called when a `typingStop` event needs to be send.
+  final Future<void> Function([String? parentId]) onStopTyping;
+
+  Timer? _keyStrokeTimer;
+  String? _currentParentId;
+  DateTime? _lastTypingEvent;
+  Completer<void>? _keyStrokeCompleter;
+
+  Future<void> _startTyping(String? parentId) {
+    _currentParentId = parentId;
+    _lastTypingEvent = DateTime.now();
+    return onStartTyping(parentId);
+  }
+
+  Future<void> _stopTyping(String? parentId) {
+    _currentParentId = null;
+    _lastTypingEvent = null;
+    return onStopTyping(parentId);
+  }
+
+  // Completes the key stroke completer if it is not yet completed.
+  void _completeKeyStrokeCompleterIfRequired() {
+    final completer = _keyStrokeCompleter;
+    if (completer != null && !completer.isCompleted) completer.complete();
+  }
+
+  // Completes the completer if available and not yet completed then creates a
+  // new completer and returns it.
+  Completer<void> _resetKeyStrokeCompleter() {
+    _completeKeyStrokeCompleterIfRequired();
+    return _keyStrokeCompleter = Completer<void>();
+  }
+
+  // Cancels the key stroke timer if it is running.
+  void _cancelKeyStrokeTimer() {
+    _keyStrokeTimer?.cancel();
+    _keyStrokeTimer = null;
+  }
+
+  /// Cancels the handler and stops the typing event.
+  void cancel() {
+    // If the user is typing, stop typing.
+    // This is needed to prevent the user from being stuck in typing mode.
+    if (_lastTypingEvent != null) {
+      // We don't need to handle the error here
+      _stopTyping(_currentParentId).catchError((_) {});
+    }
+    _cancelKeyStrokeTimer();
+    _completeKeyStrokeCompleterIfRequired();
+  }
+
+  /// Invokes the [onStartTyping] callback and schedules a timer to invoke the
+  /// [onStopTyping] callback.
+  ///
+  /// This is meant to be called every time the user presses a key. The method
+  /// will manage requests and timer as needed.
+  Future<void> call([String? parentId]) async {
+    final completer = _resetKeyStrokeCompleter();
+
+    _cancelKeyStrokeTimer();
+
+    _keyStrokeTimer = Timer(Duration(seconds: startTypingEventTimeout), () {
+      _stopTyping(parentId).then((_) {
+        if (completer.isCompleted) return;
+        completer.complete();
+      }).onError((error, stackTrace) {
+        if (completer.isCompleted) return;
+        completer.completeError(error!, stackTrace);
+      });
+    });
+
+    // If the user is typing too long, it should call [onStartTyping] again.
+    final now = DateTime.now();
+    final lastTypingEvent = _lastTypingEvent;
+    if (lastTypingEvent == null ||
+        now.difference(lastTypingEvent).inMilliseconds >
+            // startTypingResendInterval in milliseconds
+            startTypingResendInterval * 1000) {
+      _startTyping(parentId).onError((error, stackTrace) {
+        _cancelKeyStrokeTimer();
+        if (completer.isCompleted) return;
+        completer.completeError(error!, stackTrace);
+      });
+    }
+
+    return completer.future;
+  }
+}

packages/stream_chat/lib/stream_chat.dart

@@ -38,6 +38,7 @@ export './src/permission_type.dart';
 export './src/ws/connection_status.dart';
 export 'src/client/channel.dart';
 export 'src/client/client.dart';
+export 'src/client/key_stroke_handler.dart';
 export 'src/core/api/attachment_file_uploader.dart' show AttachmentFileUploader;
 export 'src/core/api/requests.dart';
 export 'src/core/api/responses.dart';

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

@@ -2658,16 +2658,22 @@ void main() {
         });
 
         test(
-          '''should send `typingStart` event if there is not already a typingEvent or the difference between the two is >= 2 seconds''',
+          '''should send `typingStart` event if there is not already a typingEvent or the difference between the two is > 3 seconds''',
           () async {
-            final typingEvent = Event(type: EventType.typingStart);
+            final startTypingEvent = Event(type: EventType.typingStart);
+            final stopTypingEvent = Event(type: EventType.typingStop);
 
             when(() => channel.config?.typingEvents).thenReturn(true);
 
             when(() => client.sendEvent(
                   channelId,
                   channelType,
-                  any(that: isSameEventAs(typingEvent)),
+                  any(that: isSameEventAs(startTypingEvent)),
+                )).thenAnswer((_) async => EmptyResponse());
+            when(() => client.sendEvent(
+                  channelId,
+                  channelType,
+                  any(that: isSameEventAs(stopTypingEvent)),
                 )).thenAnswer((_) async => EmptyResponse());
 
             await channel.keyStroke();
@@ -2675,7 +2681,12 @@ void main() {
             verify(() => client.sendEvent(
                   channelId,
                   channelType,
-                  any(that: isSameEventAs(typingEvent)),
+                  any(that: isSameEventAs(startTypingEvent)),
+                )).called(1);
+            verify(() => client.sendEvent(
+                  channelId,
+                  channelType,
+                  any(that: isSameEventAs(stopTypingEvent)),
                 )).called(1);
           },
         );
@@ -2688,7 +2699,7 @@ void main() {
 
         final typingStopEvent = Event(type: EventType.typingStop);
 
-        await channel.keyStroke();
+        await channel.stopTyping();
 
         verifyNever(() => client.sendEvent(
               channelId,

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

@@ -0,0 +1,73 @@
+// ignore_for_file: avoid_redundant_argument_values
+
+import 'package:mocktail/mocktail.dart';
+import 'package:stream_chat/stream_chat.dart';
+import 'package:test/test.dart';
+
+mixin OnKeyStrokeEvent {
+  Future<void> call([String? parentId]);
+}
+
+class OnStartTyping extends Mock implements OnKeyStrokeEvent {}
+
+class OnStopTyping extends Mock implements OnKeyStrokeEvent {}
+
+void main() {
+  final onStartTyping = OnStartTyping();
+  final onStopTyping = OnStopTyping();
+  late KeyStrokeHandler keyStrokeHandler;
+
+  const startTypingEventTimeout = 1;
+  const startTypingResendInterval = 2;
+
+  setUp(() {
+    when(() => onStartTyping(any())).thenAnswer((_) => Future.value());
+    when(() => onStopTyping(any())).thenAnswer((_) => Future.value());
+
+    keyStrokeHandler = KeyStrokeHandler(
+      onStartTyping: onStartTyping,
+      onStopTyping: onStopTyping,
+      startTypingEventTimeout: startTypingEventTimeout,
+      startTypingResendInterval: startTypingResendInterval,
+    );
+  });
+
+  tearDown(() {
+    keyStrokeHandler.cancel();
+    clearInteractions(onStartTyping);
+    clearInteractions(onStopTyping);
+  });
+
+  group('call', () {
+    test('should work fine', () {
+      expect(keyStrokeHandler.call(), completes);
+    });
+
+    test('should call onStartTyping', () async {
+      keyStrokeHandler.call();
+      verify(() => onStartTyping(any())).called(1);
+    });
+
+    test('should call onStopTyping', () async {
+      keyStrokeHandler
+        ..call()
+        ..cancel();
+      verify(() => onStopTyping(any())).called(1);
+    });
+
+    test('should call onStartTyping after startTypingResendInterval', () async {
+      final watch = Stopwatch()..start();
+      while (watch.elapsed.inSeconds <= startTypingResendInterval) {
+        keyStrokeHandler.call();
+      }
+      watch.stop();
+      verify(() => onStartTyping(any())).called(2);
+    });
+
+    test('should call onStopTyping after startTypingEventTimeout', () async {
+      keyStrokeHandler.call();
+      await Future.delayed(const Duration(seconds: startTypingEventTimeout));
+      verify(() => onStopTyping(any())).called(1);
+    });
+  });
+}

packages/stream_chat_flutter/CHANGELOG.md

@@ -1,3 +1,10 @@
+## Upcoming
+
+🐞 Fixed
+
+- [[#882]](https://github.com/GetStream/stream-chat-flutter/issues/882) Lots of unhandled exceptions
+  when network is off or spotty.
+
 ## 4.4.1
 
 🐞 Fixed

packages/stream_chat_flutter/lib/src/message_input.dart

@@ -824,9 +824,14 @@ class MessageInputState extends State<MessageInput> {
       value = value.trim();
 
       final channel = StreamChannel.of(context).channel;
-      if (value.isNotEmpty) {
+      if (value.isNotEmpty &&
-        // ignore: no-empty-block
+          channel.ownCapabilities.contains(PermissionType.sendTypingEvents)) {
-        channel.keyStroke(widget.parentMessage?.id).catchError((e) {});
+        // Notify the server that the user started typing.
+        channel.keyStroke(widget.parentMessage?.id).onError(
+          (error, stackTrace) {
+            widget.onError?.call(error!, stackTrace);
+          },
+        );
       }
 
       var actionsLength = widget.actions.length;

packages/stream_chat_flutter/lib/src/v4/message_input/stream_message_input.dart

@@ -945,12 +945,14 @@ class StreamMessageInputState extends State<StreamMessageInput>
       value = value.trim();
 
       final channel = StreamChannel.of(context).channel;
-      if (channel.ownCapabilities.contains(PermissionType.sendTypingEvents) &&
+      if (value.isNotEmpty &&
-          value.isNotEmpty) {
+          channel.ownCapabilities.contains(PermissionType.sendTypingEvents)) {
-        channel
+        // Notify the server that the user started typing.
-            .keyStroke(_effectiveController.value.parentId)
+        channel.keyStroke(_effectiveController.message.parentId).onError(
-            // ignore: no-empty-block
+          (error, stackTrace) {
-            .catchError((e) {});
+            widget.onError?.call(error!, stackTrace);
+          },
+        );
       }
 
       var actionsLength = widget.actions.length;