SK-402: Implement 'Typing...' feature (#77)

* add typingMessageListener

* add TypingManager

* add send typing

* add TypingIndicator

* update TypingIndicator

* fix stop typing when message received

---------

Co-authored-by: Magellan <magellan@connectycube.com>

Author: 2 people

lib/src/api/api.dart

@@ -2,6 +2,7 @@ export 'chats/attachments_api.dart';
 export 'chats/messages_api.dart';
 export 'chats/realtime/messages_manager.dart';
 export 'chats/realtime/models/models.dart';
+export 'chats/realtime/typing_manager.dart';
 export 'connection/connection.dart';
 export 'connection/exceptions.dart';
 export 'conversations/models/models.dart';

lib/src/api/chats/messages_api.dart

@@ -8,6 +8,7 @@ const String messageEditRequestName = 'message_edit';
 const String messagesListRequestName = 'message_list';
 const String messagesReadRequestName = 'message_read';
 const String messagesDeleteRequestName = 'message_delete';
+const String messageTypingName = 'typing';
 
 String linkPreviewUrl = dotenv.env['LINK_PREVIEW_URL'] ?? '';
 
@@ -91,3 +92,12 @@ Future<LinkPreview> linkPreviewData(String url) {
     return LinkPreview.fromJson(response);
   });
 }
+
+Future<bool> sendTypingStatus(TypingMessageStatus typing) {
+  return SamaConnectionService.instance
+      .sendRequest(messageTypingName, typing.toJson(), shouldAwaiting: false)
+      .then((response) {
+    print('sendTypingStatus response $response');
+    return true;
+  });
+}

lib/src/api/chats/realtime/messages_manager.dart

@@ -53,6 +53,12 @@ class MessagesManager {
   Stream<DeleteMessagesStatus> get deletedMessageStatusStream =>
       _deletedMessagesStatusController.stream;
 
+  final StreamController<TypingMessageStatus> _typingStatusController =
+      StreamController.broadcast();
+
+  Stream<TypingMessageStatus> get typingStatusStream =>
+      _typingStatusController.stream;
+
   _init() {
     if (dataListener != null) return;
 
@@ -69,6 +75,8 @@ class MessagesManager {
         _processDeleteMessagePackage(data['message_delete']);
       } else if (data['system_message'] != null) {
         _processSystemMessagePackage(data['system_message']);
+      } else if (data['typing'] != null) {
+        _processTypingPackage(data['typing']);
       }
     });
   }
@@ -114,4 +122,9 @@ class MessagesManager {
       _systemChatMessagesController.add(systemMessage);
     }
   }
+
+  void _processTypingPackage(Map<String, dynamic> data) {
+    var typingStatus = TypingMessageStatus.fromJson(data);
+    _typingStatusController.add(typingStatus);
+  }
 }

lib/src/api/chats/realtime/models/message_statuses.dart

@@ -88,3 +88,18 @@ class DeleteMessagesStatus extends MessageStatus {
 }
 
 enum DeleteMessageType { myself, all }
+
+class TypingMessageStatus {
+  final String? cid; // cid
+  final String? type; // c_type
+  final String? from; // from
+  final int? t; // t
+
+  TypingMessageStatus.fromJson(Map<String, dynamic> json)
+      : cid = json['cid'],
+        type = json['c_type'],
+        from = json['from'],
+        t = json['t'];
+
+  Map<String, dynamic> toJson() => {'cid': cid};
+}

lib/src/api/chats/realtime/typing_manager.dart

@@ -0,0 +1,76 @@
+import 'dart:async';
+
+import 'package:equatable/equatable.dart';
+
+import '../../api.dart';
+
+const stopTypingTime = 6;
+
+enum TypingState { start, stop }
+
+class TypingStatus extends Equatable {
+  final TypingState state;
+  final String? cid;
+  final String? from;
+
+  const TypingStatus(this.state, this.cid, this.from);
+
+  TypingStatus copyWith({
+    TypingState? state,
+    String? cid,
+    String? from,
+  }) {
+    return TypingStatus(
+      state ?? this.state,
+      cid ?? this.cid,
+      from ?? this.from,
+    );
+  }
+
+  @override
+  List<Object?> get props => [state, cid, from];
+}
+
+class TypingManager {
+  TypingManager._() {
+    _init();
+  }
+
+  static final _instance = TypingManager._();
+
+  static TypingManager get instance {
+    return _instance;
+  }
+
+  Map<String, Timer> clearTypingTimers = {};
+  StreamSubscription<TypingMessageStatus>? typingMessageSubscription;
+
+  final StreamController<TypingStatus> _typingStatusController =
+      StreamController.broadcast();
+
+  Stream<TypingStatus> get typingStatusStream => _typingStatusController.stream;
+
+  _init() {
+    typingMessageSubscription = MessagesManager.instance.typingStatusStream
+        .listen((typingStatus) async {
+      var typing =
+          TypingStatus(TypingState.start, typingStatus.cid, typingStatus.from);
+      _typingStatusController.add(typing);
+
+      restartTypingTimer(typingStatus.cid ?? '', () {
+        _typingStatusController.add(typing.copyWith(state: TypingState.stop));
+      });
+    });
+  }
+
+  void restartTypingTimer(String cid, void Function() callback) {
+    clearTypingTimers[cid]?.cancel();
+    var timer = Timer(const Duration(seconds: stopTypingTime), callback);
+    clearTypingTimers[cid] = timer;
+  }
+
+  destroy() {
+    typingMessageSubscription?.cancel();
+    clearTypingTimers.clear();
+  }
+}

lib/src/api/connection/connection.dart

@@ -121,16 +121,19 @@ class SamaConnectionService {
     String? retryRequestId,
     Completer<Map<String, dynamic>>? retryCompleter,
     bool shouldRetry = true,
+    bool shouldAwaiting = true,
   }) {
     var requestId = retryRequestId ??= const Uuid().v4().toString();
 
     var requestCompleter = retryCompleter ??= Completer<Map<String, dynamic>>();
 
-    awaitingRequests[requestId] = RequestInfo(
-        name: requestName,
-        data: requestData,
-        completer: requestCompleter,
-        shouldRetry: shouldRetry);
+    if (shouldAwaiting) {
+      awaitingRequests[requestId] = RequestInfo(
+          name: requestName,
+          data: requestData,
+          completer: requestCompleter,
+          shouldRetry: shouldRetry);
+    }
 
     var request = {
       'request': {

lib/src/features/conversation/bloc/conversation_bloc.dart

@@ -26,7 +26,14 @@ EventTransformer<E> throttleDroppable<E>(Duration duration) {
   };
 }
 
-EventTransformer<Event> debounce<Event>({
+EventTransformer<E> typingThrottleDroppable<E>() {
+  Duration duration = const Duration(milliseconds: 5000);
+  return (events, mapper) {
+    return droppable<E>().call(events.throttle(duration), mapper);
+  };
+}
+
+EventTransformer<Event> readDebounce<Event>({
   Duration duration = const Duration(milliseconds: 500),
 }) {
   return (events, mapper) => events.debounce(duration).switchMap(mapper);
@@ -40,6 +47,7 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
 
   StreamSubscription<ChatMessage>? incomingMessagesSubscription;
   StreamSubscription<MessageSendStatus>? statusMessagesSubscription;
+  StreamSubscription<TypingStatus>? typingMessageSubscription;
   StreamSubscription<Map<String, dynamic>>? lastActivitySubscription;
   StreamSubscription<ConversationModel?>? conversationWatcher;
 
@@ -76,7 +84,7 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
     );
     on<_ReadStatusReceived>(
       _onReadStatusReceived,
-      transformer: debounce(),
+      transformer: readDebounce(),
     );
     on<_FailedStatusReceived>(
       _onFailedStatusReceived,
@@ -87,6 +95,13 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
     on<ConversationDeleted>(
       _onConversationDeleted,
     );
+    on<TypingStatusStartReceived>(
+      _onTypingStatusStartReceived,
+      transformer: typingThrottleDroppable(),
+    );
+    on<TypingStatusStopReceived>(
+      _onTypingStatusStopReceived,
+    );
 
     add(const ParticipantsReceived());
 
@@ -126,6 +141,17 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
       }
     });
 
+    typingMessageSubscription =
+        messagesRepository.typingMessageStream.listen((typing) async {
+      if (typing.cid == currentConversation.id) {
+        if (typing.state == TypingState.start) {
+          add(TypingStatusStartReceived(typing.from!));
+        } else if (typing.state == TypingState.stop) {
+          add(TypingStatusStopReceived(typing.from!));
+        }
+      }
+    });
+
     lastActivitySubscription =
         userRepository.lastActivityStream.listen((data) async {
       var recentActivity = data[currentConversation.opponent?.id];
@@ -270,6 +296,20 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
         : emit(state.copyWith(status: ConversationStatus.failure));
   }
 
+  Future<void> _onTypingStatusStartReceived(
+      TypingStatusStartReceived event, Emitter<ConversationState> emit) async {
+    var user = await userRepository.getUserById(event.from);
+    emit(state.copyWith(
+        typingStatus: TypingMessageStatus(TypingState.start, user)));
+  }
+
+  Future<void> _onTypingStatusStopReceived(
+      TypingStatusStopReceived event, Emitter<ConversationState> emit) async {
+    var user = await userRepository.getUserById(event.from);
+    emit(state.copyWith(
+        typingStatus: TypingMessageStatus(TypingState.stop, user)));
+  }
+
   FutureOr<void> _onMessageReceived(
       _MessageReceived event, Emitter<ConversationState> emit) {
     var messages = [...state.messages];
@@ -365,6 +405,7 @@ class ConversationBloc extends Bloc<ConversationEvent, ConversationState> {
     unsubscribeOpponentLastActivity();
     incomingMessagesSubscription?.cancel();
     statusMessagesSubscription?.cancel();
+    typingMessageSubscription?.cancel();
     lastActivitySubscription?.cancel();
     conversationWatcher?.cancel();
     return super.close();

lib/src/features/conversation/bloc/conversation_event.dart

@@ -71,3 +71,15 @@ final class _ConversationUpdated extends ConversationEvent {
 final class ConversationDeleted extends ConversationEvent {
   const ConversationDeleted();
 }
+
+final class TypingStatusStartReceived extends ConversationEvent {
+  final String from;
+
+  const TypingStatusStartReceived(this.from);
+}
+
+final class TypingStatusStopReceived extends ConversationEvent {
+  final String from;
+
+  const TypingStatusStopReceived(this.from);
+}

lib/src/features/conversation/bloc/conversation_state.dart

@@ -2,6 +2,13 @@ part of 'conversation_bloc.dart';
 
 enum ConversationStatus { initial, success, failure, delete }
 
+class TypingMessageStatus {
+  final TypingState typingState;
+  final UserModel? user;
+
+  TypingMessageStatus(this.typingState, this.user);
+}
+
 final class ConversationState extends Equatable {
   const ConversationState({
     required this.conversation,
@@ -11,6 +18,7 @@ final class ConversationState extends Equatable {
     this.hasReachedMax = false,
     this.initial = false,
     this.draftMessage,
+    this.typingStatus,
   });
 
   final ConversationModel conversation;
@@ -20,6 +28,7 @@ final class ConversationState extends Equatable {
   final bool hasReachedMax;
   final bool initial;
   final MessageModel? draftMessage;
+  final TypingMessageStatus? typingStatus;
 
   ConversationState copyWith({
     ConversationModel? conversation,
@@ -29,6 +38,7 @@ final class ConversationState extends Equatable {
     bool? hasReachedMax,
     bool? initial,
     MessageModel? Function()? draftMessage,
+    TypingMessageStatus? typingStatus,
   }) {
     return ConversationState(
       conversation: conversation ?? this.conversation,
@@ -38,6 +48,7 @@ final class ConversationState extends Equatable {
       hasReachedMax: hasReachedMax ?? this.hasReachedMax,
       initial: initial ?? this.initial,
       draftMessage: draftMessage != null ? draftMessage() : this.draftMessage,
+      typingStatus: typingStatus,
     );
   }
 
@@ -54,6 +65,7 @@ final class ConversationState extends Equatable {
         hasReachedMax,
         initial,
         participants,
-        draftMessage
+        draftMessage,
+        typingStatus
       ];
 }

lib/src/features/conversation/bloc/send_message/send_message_bloc.dart

@@ -1,7 +1,9 @@
 import 'dart:async';
 
+import 'package:bloc_concurrency/bloc_concurrency.dart';
 import 'package:equatable/equatable.dart';
 import 'package:flutter_bloc/flutter_bloc.dart';
+import 'package:stream_transform/stream_transform.dart';
 
 import '../../../../api/api.dart';
 import '../../../../db/models/conversation_model.dart';
@@ -12,6 +14,15 @@ part 'send_message_event.dart';
 
 part 'send_message_state.dart';
 
+const typingThrottleDuration = 5;
+
+EventTransformer<E> typingThrottleDroppable<E>() {
+  Duration duration = const Duration(seconds: typingThrottleDuration);
+  return (events, mapper) {
+    return droppable<E>().call(events.throttle(duration), mapper);
+  };
+}
+
 class SendMessageBloc extends Bloc<SendMessageEvent, SendMessageState> {
   final ConversationModel currentConversation;
   final ConversationRepository conversationRepository;
@@ -34,6 +45,8 @@ class SendMessageBloc extends Bloc<SendMessageEvent, SendMessageState> {
     on<SendStatusReadMessages>(
       _onSendStatusReadMessages,
     );
+    on<SendTypingChanged>(_onSendTypingChanged,
+        transformer: typingThrottleDroppable());
   }
 
   Future<void> _onSendTextMessage(
@@ -75,6 +88,13 @@ class SendMessageBloc extends Bloc<SendMessageEvent, SendMessageState> {
     } catch (_) {}
   }
 
+  Future<FutureOr<void>> _onSendTypingChanged(
+      SendTypingChanged event, Emitter<SendMessageState> emit) async {
+    try {
+      await messagesRepository.sendTypingStatus(currentConversation.id);
+    } catch (_) {}
+  }
+
   saveDraftIfExist() {
     if (state.text.isNotEmpty) {
       messagesRepository.saveDraftMessage(state.text, currentConversation.id);