diff --git a/packages/stream_chat/lib/src/client/channel.dart b/packages/stream_chat/lib/src/client/channel.dart index d610ddd5..7565b02b 100644 --- a/packages/stream_chat/lib/src/client/channel.dart +++ b/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 - /// based on the users keystrokes. Call this on every keystroke. + /// Sends the [Event.typingStart] event and schedules a timer to invoke the + /// [Event.typingStop] event. + /// + /// This is meant to be called every time the user presses a key. Future keyStroke([String? parentId]) async { - if (config?.typingEvents == false) { - return; - } + if (config?.typingEvents == false) return; - client.logger.info('start typing'); - final now = DateTime.now(); - - if (_lastTypingEvent == null || - now.difference(_lastTypingEvent!).inSeconds >= 2) { - _lastTypingEvent = now; - await sendEvent(Event( - type: EventType.typingStart, - parentId: parentId, - )); - } + client.logger.info('KeyStroke received'); + return _keyStrokeHandler(parentId); } - /// Sets last typing to null and sends the typing.stop event. + /// Sends the [EventType.typingStart] event. + Future 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 stopTyping([String? parentId]) async { - if (config?.typingEvents == false) { - return; - } + if (config?.typingEvents == false) 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 = []; + 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 retryFailedMessages() async { - final failedMessages = - [...messages, ...threads.values.expand((v) => v)] - .where( - (message) => - message.status != MessageSendingStatus.sent && - message.createdAt.isBefore( - DateTime.now().subtract( - const Duration( - seconds: 5, - ), - ), - ), - ) - .toList(); + final failedMessages = [...messages, ...threads.values.expand((v) => v)] + .where( + (message) => + message.status != MessageSendingStatus.sent && + message.createdAt.isBefore( + DateTime.now().subtract(const Duration(seconds: 5)), + ), + ) + .toList(); _retryQueue.add(failedMessages); } @@ -1986,12 +1989,7 @@ class ChannelClientState { } _subscriptions.add( - _channel - .on( - EventType.messageRead, - EventType.notificationMarkRead, - ) - .listen( + _channel.on(EventType.messageRead, EventType.notificationMarkRead).listen( (event) { final readList = List.from(_channelState.read ?? []); final userReadIndex = @@ -2252,34 +2250,29 @@ class ChannelClientState { ); } - /// Channel related typing users last value. - Map get typingEvents => _typingEventsController.value; - /// Channel related typing users stream. Stream> get typingEventsStream => _typingEventsController.stream; - final BehaviorSubject> _typingEventsController = - BehaviorSubject.seeded({}); - - final Channel _channel; - final Map _typings = {}; + /// Channel related typing users last value. + Map get typingEvents => _typingEventsController.value; + final _typingEventsController = BehaviorSubject.seeded({}); void _listenTypingEvents() { - if (_channelState.channel?.config.typingEvents == false) { - return; - } + if (_channelState.channel?.config.typingEvents == false) 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!; - if (user.id != _channel.client.state.currentUser?.id) { - _typings[user] = event; - _typingEventsController.add(_typings); - } + final user = event.user; + if (user != null && user.id != currentUser.id) { + final events = {...typingEvents}; + events[user] = event; + _typingEventsController.add(events); } }, ), @@ -2287,112 +2280,109 @@ class ChannelClientState { ..add( _channel.on(EventType.typingStop).listen( (event) { - if (event.user != null) { - final user = event.user!; - if (user.id != _channel.client.state.currentUser?.id) { - _typings.remove(event.user); - _typingEventsController.add(_typings); - } + final user = event.user; + if (user != null && user.id != currentUser.id) { + final events = {...typingEvents}..remove(user); + _typingEventsController.add(events); } }, ), ) ..add( - _channel - .on() - .where((event) => - event.user != null && - members.any((m) => m.userId == event.user!.id)) - .listen( + _channel.on().where((event) { + final user = event.user; + if (user == null) return false; + return members.any((m) => m.userId == user.id); + }).listen( (event) { final newMembers = List.from(members); final oldMemberIndex = newMembers.indexWhere((m) => m.userId == event.user!.id); if (oldMemberIndex > -1) { final oldMember = newMembers.removeAt(oldMemberIndex); - updateChannelState(ChannelState( - members: [ - ...newMembers, - oldMember.copyWith( - user: event.user, - ), - ], - )); + updateChannelState( + ChannelState( + members: [ + ...newMembers, + oldMember.copyWith( + user: event.user, + ), + ], + ), + ); } }, ), ); } - Timer? _cleaningTimer; + Timer? _staleTypingEventsCleanerTimer; - void _startCleaning() { - if (_channelState.channel?.config.typingEvents == false) { - return; - } + // Checks and removes stale typing events that were not explicitly stopped by + // the sender due to technical difficulties. e.g. process death, loss of + // Internet connection or custom implementation. + void _startCleaningStaleTypingEvents() { + if (_channelState.channel?.config.typingEvents == false) return; - _cleaningTimer = Timer.periodic(const Duration(seconds: 1), (_) { - final now = DateTime.now(); - - if (_channel._lastTypingEvent != null && - now.difference(_channel._lastTypingEvent!).inSeconds > 1) { - _channel.stopTyping(); - } - - _clean(); - }); + _staleTypingEventsCleanerTimer = Timer.periodic( + const Duration(seconds: 1), + (_) { + final now = DateTime.now(); + typingEvents.forEach((user, event) { + if (now.difference(event.createdAt).inSeconds > + incomingTypingStartEventTimeout) { + _channel.client.handleEvent( + Event( + type: EventType.typingStop, + user: user, + cid: _channel.cid, + parentId: event.parentId, + ), + ); + } + }); + }, + ); } - late Timer _pinnedMessagesTimer; + Timer? _stalePinnedMessagesCleanerTimer; - void _startCleaningPinnedMessages() { - _pinnedMessagesTimer = Timer.periodic(const Duration(seconds: 30), (_) { - final now = DateTime.now(); - var expiredMessages = channelState.pinnedMessages - ?.where((m) => m.pinExpires?.isBefore(now) == true) - .toList(); - if (expiredMessages != null && expiredMessages.isNotEmpty) { - expiredMessages = expiredMessages - .map((m) => m.copyWith( - pinExpires: null, - pinned: false, - )) + // Checks and removes stale pinned messages that are not valid anymore. + void _startCleaningStalePinnedMessages() { + _stalePinnedMessagesCleanerTimer = Timer.periodic( + const Duration(seconds: 30), + (_) { + final now = DateTime.now(); + var expiredMessages = channelState.pinnedMessages + ?.where((m) => m.pinExpires?.isBefore(now) == true) .toList(); + if (expiredMessages != null && expiredMessages.isNotEmpty) { + expiredMessages = expiredMessages + .map((m) => m.copyWith( + pinExpires: null, + pinned: false, + )) + .toList(); - updateChannelState(_channelState.copyWith( - pinnedMessages: pinnedMessages.where(_pinIsValid).toList(), - 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, - ), - ); - } - }); + updateChannelState(_channelState.copyWith( + pinnedMessages: pinnedMessages.where(_pinIsValid).toList(), + messages: expiredMessages, + )); + } + }, + ); } /// 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(); - _pinnedMessagesTimer.cancel(); + _staleTypingEventsCleanerTimer?.cancel(); + _stalePinnedMessagesCleanerTimer?.cancel(); _typingEventsController.close(); } } diff --git a/packages/stream_chat/lib/src/client/key_stroke_handler.dart b/packages/stream_chat/lib/src/client/key_stroke_handler.dart new file mode 100644 index 00000000..f67017b9 --- /dev/null +++ b/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 Function([String? parentId]) onStartTyping; + + /// Called when a `typingStop` event needs to be send. + final Future Function([String? parentId]) onStopTyping; + + Timer? _keyStrokeTimer; + String? _currentParentId; + DateTime? _lastTypingEvent; + Completer? _keyStrokeCompleter; + + Future _startTyping(String? parentId) { + _currentParentId = parentId; + _lastTypingEvent = DateTime.now(); + return onStartTyping(parentId); + } + + Future _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 _resetKeyStrokeCompleter() { + _completeKeyStrokeCompleterIfRequired(); + return _keyStrokeCompleter = Completer(); + } + + // 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 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; + } +} diff --git a/packages/stream_chat/lib/stream_chat.dart b/packages/stream_chat/lib/stream_chat.dart index 2755224f..7f4c22ef 100644 --- a/packages/stream_chat/lib/stream_chat.dart +++ b/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'; diff --git a/packages/stream_chat/test/src/client/channel_test.dart b/packages/stream_chat/test/src/client/channel_test.dart index f90430f2..cf2fc042 100644 --- a/packages/stream_chat/test/src/client/channel_test.dart +++ b/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, diff --git a/packages/stream_chat/test/src/client/key_stroke_handler_test.dart b/packages/stream_chat/test/src/client/key_stroke_handler_test.dart new file mode 100644 index 00000000..d0a13352 --- /dev/null +++ b/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 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); + }); + }); +} diff --git a/packages/stream_chat_flutter/CHANGELOG.md b/packages/stream_chat_flutter/CHANGELOG.md index f098c65c..be7ffa2b 100644 --- a/packages/stream_chat_flutter/CHANGELOG.md +++ b/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 diff --git a/packages/stream_chat_flutter/lib/src/message_input.dart b/packages/stream_chat_flutter/lib/src/message_input.dart index 76031e5f..f58d7bad 100644 --- a/packages/stream_chat_flutter/lib/src/message_input.dart +++ b/packages/stream_chat_flutter/lib/src/message_input.dart @@ -824,9 +824,14 @@ class MessageInputState extends State { value = value.trim(); final channel = StreamChannel.of(context).channel; - if (value.isNotEmpty) { - // ignore: no-empty-block - channel.keyStroke(widget.parentMessage?.id).catchError((e) {}); + if (value.isNotEmpty && + channel.ownCapabilities.contains(PermissionType.sendTypingEvents)) { + // 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; diff --git a/packages/stream_chat_flutter/lib/src/v4/message_input/stream_message_input.dart b/packages/stream_chat_flutter/lib/src/v4/message_input/stream_message_input.dart index 56ffcfe1..db1c1925 100644 --- a/packages/stream_chat_flutter/lib/src/v4/message_input/stream_message_input.dart +++ b/packages/stream_chat_flutter/lib/src/v4/message_input/stream_message_input.dart @@ -945,12 +945,14 @@ class StreamMessageInputState extends State value = value.trim(); final channel = StreamChannel.of(context).channel; - if (channel.ownCapabilities.contains(PermissionType.sendTypingEvents) && - value.isNotEmpty) { - channel - .keyStroke(_effectiveController.value.parentId) - // ignore: no-empty-block - .catchError((e) {}); + if (value.isNotEmpty && + channel.ownCapabilities.contains(PermissionType.sendTypingEvents)) { + // Notify the server that the user started typing. + channel.keyStroke(_effectiveController.message.parentId).onError( + (error, stackTrace) { + widget.onError?.call(error!, stackTrace); + }, + ); } var actionsLength = widget.actions.length;