add client_test.dart

Signed-off-by: Sahil Kumar <[email protected]>

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

@@ -22,9 +22,9 @@ class Channel {
     this._client,
     this._type,
     this._id, {
-    Map<String, Object?> extraData = const {},
+    Map<String, Object?>? extraData,
   })  : _cid = _id != null ? '$_type:$_id' : null,
-        _extraData = extraData {
+        _extraData = extraData ?? {} {
     _client.logger.info('New Channel instance not initialized created');
   }
 
@@ -202,8 +202,13 @@ class Channel {
   }
 
   /// Channel extra data
-  Map<String, dynamic> get extraData =>
+  Map<String, Object?> get extraData {
-      state?._channelState.channel?.extraData ?? _extraData;
+    var data = state?._channelState.channel?.extraData;
+    if (data == null || data.isEmpty) {
+      data = _extraData;
+    }
+    return data;
+  }
 
   /// Channel extra data as a stream
   Stream<Map<String, dynamic>> get extraDataStream {
@@ -942,7 +947,7 @@ class Channel {
     state = ChannelClientState(this, channelState);
 
     if (cid != null) {
-      client.state.channels[cid!] = this;
+      client.state.channels = {cid!: this};
     }
     if (!_initializedCompleter.isCompleted) {
       _initializedCompleter.complete(true);
@@ -959,8 +964,8 @@ class Channel {
   /// Set [preferOffline] to true to avoid the api call if the data is already
   /// in the offline storage
   Future<QueryRepliesResponse> getReplies(
-    String parentId,
+    String parentId, {
-    PaginationParams options, {
+    PaginationParams? options,
     bool preferOffline = false,
   }) async {
     final cachedReplies = await _client.chatPersistenceClient?.getReplies(
@@ -973,19 +978,22 @@ class Channel {
         return QueryRepliesResponse()..messages = cachedReplies;
       }
     }
-    final repliesResponse = await _client.getReplies(parentId, options);
+    final repliesResponse = await _client.getReplies(
+      parentId,
+      options: options,
+    );
     state?.updateThreadInfo(parentId, repliesResponse.messages);
     return repliesResponse;
   }
 
   /// List the reactions for a message in the channel
   Future<QueryReactionsResponse> getReactions(
-    String messageId,
+    String messageId, {
-    PaginationParams options,
+    PaginationParams? options,
-  ) =>
+  }) =>
       _client.getReactions(
         messageId,
-        options,
+        options: options,
       );
 
   /// Retrieves a list of messages by ID

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

@@ -63,7 +63,7 @@ class StreamChatClient {
   /// application.
   StreamChatClient(
     String apiKey, {
-    this.logLevel = Level.ALL,
+    this.logLevel = Level.WARNING,
     LogHandlerFunction? logHandlerFunction,
     RetryPolicy? retryPolicy,
     Location? location,
@@ -71,9 +71,11 @@ class StreamChatClient {
     Duration connectTimeout = const Duration(seconds: 6),
     Duration receiveTimeout = const Duration(seconds: 6),
     StreamChatApi? chatApi,
+    WebSocket? ws,
     AttachmentFileUploader? attachmentFileUploader,
   }) {
-    _setupLogger(logHandlerFunction);
+    this.logHandlerFunction = logHandlerFunction ?? _defaultLogHandler;
+    logger.info('Initiating new StreamChatClient');
 
     final options = StreamHttpClientOptions(
       baseUrl: baseURL,
@@ -92,13 +94,14 @@ class StreamChatClient {
           logger: detachedLogger('🕸️'),
         );
 
-    _ws = WebSocket(
+    _ws = ws ??
-      apiKey: apiKey,
+        WebSocket(
-      baseUrl: options.baseUrl,
+          apiKey: apiKey,
-      tokenManager: _tokenManager,
+          baseUrl: options.baseUrl,
-      handler: handleEvent,
+          tokenManager: _tokenManager,
-      logger: detachedLogger('🔌'),
+          handler: handleEvent,
-    );
+          logger: detachedLogger('🔌'),
+        );
 
     _retryPolicy = retryPolicy ??
         RetryPolicy(
@@ -107,8 +110,6 @@ class StreamChatClient {
         );
 
     state = ClientState(this);
-
-    logger.info('instantiating new client');
   }
 
   late final StreamChatApi _chatApi;
@@ -160,7 +161,7 @@ class StreamChatClient {
   /// Client specific logger instance.
   /// Refer to the class [Logger] to learn more about the specific
   /// implementation.
-  final Logger logger = Logger.detached('📡');
+  late final Logger logger = detachedLogger('📡');
 
   /// A function that has a parameter of type [LogRecord].
   /// This is called on every new log record.
@@ -185,7 +186,7 @@ class StreamChatClient {
 
   final _eventController = BehaviorSubject<Event>();
 
-  /// Stream of [Event] coming from websocket connection
+  /// Stream of [Event] coming from [_ws] connection
   /// Listen to this or use the [on] method to filter specific event types
   Stream<Event> get eventStream => _eventController.stream;
 
@@ -195,12 +196,12 @@ class StreamChatClient {
   set _wsConnectionStatus(ConnectionStatus status) =>
       _wsConnectionStatusController.add(status);
 
-  /// The current status value of the websocket connection
+  /// The current status value of the [_ws] connection
   ConnectionStatus get wsConnectionStatus =>
       _wsConnectionStatusController.value;
 
-  /// This notifies the connection status of the websocket connection.
+  /// This notifies the connection status of the [_ws] connection.
-  /// Listen to this to get notified when the websocket tries to reconnect.
+  /// Listen to this to get notified when the [_ws] tries to reconnect.
   Stream<ConnectionStatus> get wsConnectionStatusStream =>
       _wsConnectionStatusController.stream.distinct();
 
@@ -215,19 +216,9 @@ class StreamChatClient {
       };
 
   ///
-  Logger detachedLogger(
+  Logger detachedLogger(String name) => Logger.detached(name)
-    String name,
+    ..level = logLevel
-  ) =>
+    ..onRecord.listen(logHandlerFunction);
-      Logger.detached(name)
-        ..level = logLevel
-        ..onRecord.listen(logHandlerFunction);
-
-  void _setupLogger(LogHandlerFunction? logHandlerFunction) {
-    logger.level = logLevel;
-    this.logHandlerFunction = logHandlerFunction ?? _defaultLogHandler;
-    logger.onRecord.listen(this.logHandlerFunction);
-    logger.info('logger setup');
-  }
 
   /// Connects the current user, this triggers a connection to the API.
   /// It returns a [Future] that resolves when the connection is setup.
@@ -325,9 +316,13 @@ class StreamChatClient {
 
     _wsConnectionStatus = ConnectionStatus.connecting;
 
-    _connectionStatusSubscription = _ws.connectionStatusStream.listen(
+    // skipping `ws` seed connection status -> ConnectionStatus.disconnected
-      _connectionStatusHandler,
+    // otherwise `client.wsConnectionStatusStream` will emit in order
-    );
+    // 1. ConnectionStatus.disconnected -> client seed status
+    // 2. ConnectionStatus.connecting -> client connecting status
+    // 3. ConnectionStatus.disconnected -> ws seed status
+    _connectionStatusSubscription =
+        _ws.connectionStatusStream.skip(1).listen(_connectionStatusHandler);
 
     try {
       return await _ws.connect(user);
@@ -768,13 +763,13 @@ class StreamChatClient {
       _chatApi.device.removeDevice(id);
 
   /// Get a development token
-  String devToken(String userId) => Token.development(userId).rawValue;
+  Token devToken(String userId) => Token.development(userId);
 
   /// Returns a channel client with the given type, id and custom data.
   Channel channel(
     String type, {
     String? id,
-    Map<String, Object?> extraData = const {},
+    Map<String, Object?>? extraData,
   }) {
     if (id != null && state.channels.containsKey('$type:$id')) {
       return state.channels['$type:$id']!;
@@ -796,6 +791,7 @@ class StreamChatClient {
       );
 
   /// watches the provided channel
+  /// Creates first if not yet created
   Future<ChannelState> watchChannel(
     String channelType, {
     String? channelId,
@@ -809,6 +805,7 @@ class StreamChatClient {
       );
 
   /// Query the API, get messages, members or other channel fields
+  /// Creates the channel first if not yet created
   Future<ChannelState> queryChannel(
     String channelType, {
     bool state = true,
@@ -1139,22 +1136,22 @@ class StreamChatClient {
 
   /// Lists all the message replies for the [parentId]
   Future<QueryRepliesResponse> getReplies(
-    String parentId,
+    String parentId, {
-    PaginationParams options,
+    PaginationParams? options,
-  ) =>
+  }) =>
       _chatApi.message.getReplies(
         parentId,
-        options,
+        options: options,
       );
 
   /// Get all the reactions for a [messageId]
   Future<QueryReactionsResponse> getReactions(
-    String messageId,
+    String messageId, {
-    PaginationParams options,
+    PaginationParams? options,
-  ) =>
+  }) =>
       _chatApi.message.getReactions(
         messageId,
-        options,
+        options: options,
       );
 
   /// Update the given message

packages/stream_chat/lib/src/core/api/message_api.dart

@@ -142,13 +142,13 @@ class MessageApi {
 
   /// Get all the reactions for a [messageId]
   Future<QueryReactionsResponse> getReactions(
-    String messageId,
+    String messageId, {
-    PaginationParams options,
+    PaginationParams? options,
-  ) async {
+  }) async {
     final response = await _client.get(
       '/messages/$messageId/reactions',
       queryParameters: {
-        ...options.toJson(),
+        if (options != null) ...options.toJson(),
       },
     );
     return QueryReactionsResponse.fromJson(response.data);
@@ -168,13 +168,13 @@ class MessageApi {
 
   /// Lists all the message replies for the [parentId]
   Future<QueryRepliesResponse> getReplies(
-    String parentId,
+    String parentId, {
-    PaginationParams options,
+    PaginationParams? options,
-  ) async {
+  }) async {
     final response = await _client.get(
       '/messages/$parentId/replies',
       queryParameters: {
-        ...options.toJson(),
+        if (options != null) ...options.toJson(),
       },
     );
     return QueryRepliesResponse.fromJson(response.data);

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

@@ -1490,7 +1490,6 @@ void main() {
 
     test('`.getReplies`', () async {
       const parentId = 'test-parent-id';
-      const options = PaginationParams();
 
       final messages = List.generate(
         3,
@@ -1500,22 +1499,21 @@ void main() {
         ),
       );
 
-      when(() => client.getReplies(parentId, options)).thenAnswer(
+      when(() => client.getReplies(parentId)).thenAnswer(
         (_) async => QueryRepliesResponse()..messages = messages,
       );
 
-      final res = await channel.getReplies(parentId, options);
+      final res = await channel.getReplies(parentId);
 
       expect(res, isNotNull);
       expect(res.messages.length, messages.length);
       expect(res.messages.every((it) => it.parentId == parentId), isTrue);
 
-      verify(() => client.getReplies(parentId, options)).called(1);
+      verify(() => client.getReplies(parentId)).called(1);
     });
 
     test('`.getReactions`', () async {
       const messageId = 'test-message-id';
-      const options = PaginationParams();
 
       final reactions = List.generate(
         3,
@@ -1525,17 +1523,17 @@ void main() {
         ),
       );
 
-      when(() => client.getReactions(messageId, options)).thenAnswer(
+      when(() => client.getReactions(messageId)).thenAnswer(
         (_) async => QueryReactionsResponse()..reactions = reactions,
       );
 
-      final res = await channel.getReactions(messageId, options);
+      final res = await channel.getReactions(messageId);
 
       expect(res, isNotNull);
       expect(res.reactions.length, reactions.length);
       expect(res.reactions.every((it) => it.messageId == messageId), isTrue);
 
-      verify(() => client.getReactions(messageId, options)).called(1);
+      verify(() => client.getReactions(messageId)).called(1);
     });
 
     test('`.getMessagesById`', () async {

packages/stream_chat/test/src/core/api/message_api_test.dart

@@ -252,7 +252,7 @@ void main() {
               'reactions': [...reactions.map((it) => it.toJson())]
             }));
 
-    final res = await messageApi.getReactions(messageId, options);
+    final res = await messageApi.getReactions(messageId, options: options);
 
     expect(res, isNotNull);
     expect(res.reactions.length, reactions.length);
@@ -311,7 +311,7 @@ void main() {
               'messages': [...messages.map((it) => it.toJson())]
             }));
 
-    final res = await messageApi.getReplies(parentId, options);
+    final res = await messageApi.getReplies(parentId, options: options);
 
     expect(res, isNotNull);
     expect(res.messages.length, messages.length);

packages/stream_chat/test/src/fakes.dart

@@ -1,5 +1,8 @@
+import 'dart:async';
+
 import 'package:dio/dio.dart';
 import 'package:mocktail/mocktail.dart';
+import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/src/core/api/channel_api.dart';
 import 'package:stream_chat/src/core/api/device_api.dart';
 import 'package:stream_chat/src/core/api/general_api.dart';
@@ -10,6 +13,7 @@ import 'package:stream_chat/src/core/api/user_api.dart';
 import 'package:stream_chat/src/core/api/guest_api.dart';
 import 'package:stream_chat/src/core/http/token.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
+import 'package:stream_chat/src/ws/websocket.dart';
 import 'package:stream_chat/stream_chat.dart';
 
 import 'mocks.dart';
@@ -96,3 +100,88 @@ class FakeMessage extends Fake implements Message {}
 class FakeAttachmentFile extends Fake implements AttachmentFile {}
 
 class FakeEvent extends Fake implements Event {}
+
+class FakeUser extends Fake implements User {}
+
+class FakeWebSocket extends Fake implements WebSocket {
+  BehaviorSubject<ConnectionStatus>? _connectionStatusController;
+
+  BehaviorSubject<ConnectionStatus> get connectionStatusController =>
+      _connectionStatusController ??=
+          BehaviorSubject.seeded(ConnectionStatus.disconnected);
+
+  set connectionStatus(ConnectionStatus value) {
+    connectionStatusController.add(value);
+  }
+
+  @override
+  ConnectionStatus get connectionStatus => connectionStatusController.value;
+
+  @override
+  Stream<ConnectionStatus> get connectionStatusStream =>
+      connectionStatusController.stream;
+
+  @override
+  Completer<Event>? connectionCompleter;
+
+  @override
+  Future<Event> connect(User user) async {
+    connectionStatus = ConnectionStatus.connecting;
+    final event = Event(
+      type: EventType.healthCheck,
+      connectionId: 'fake-connection-id',
+      me: OwnUser.fromUser(user),
+    );
+    connectionCompleter = Completer()..complete(event);
+    connectionStatus = ConnectionStatus.connected;
+    return connectionCompleter!.future;
+  }
+
+  @override
+  void disconnect() {
+    connectionStatus = ConnectionStatus.disconnected;
+    connectionCompleter = null;
+    _connectionStatusController?.close();
+    _connectionStatusController = null;
+  }
+}
+
+class FakeWebSocketWithConnectionError extends Fake implements WebSocket {
+  BehaviorSubject<ConnectionStatus>? _connectionStatusController;
+
+  BehaviorSubject<ConnectionStatus> get connectionStatusController =>
+      _connectionStatusController ??=
+          BehaviorSubject.seeded(ConnectionStatus.disconnected);
+
+  set connectionStatus(ConnectionStatus value) {
+    connectionStatusController.add(value);
+  }
+
+  @override
+  ConnectionStatus get connectionStatus => connectionStatusController.value;
+
+  @override
+  Stream<ConnectionStatus> get connectionStatusStream =>
+      connectionStatusController.stream;
+
+  @override
+  Completer<Event>? connectionCompleter;
+
+  @override
+  Future<Event> connect(User user) async {
+    connectionStatus = ConnectionStatus.connecting;
+    const error = StreamWebSocketError('Error Connecting');
+    connectionCompleter = Completer()..completeError(error);
+    return connectionCompleter!.future;
+  }
+
+  @override
+  void disconnect() {
+    connectionStatus = ConnectionStatus.disconnected;
+    connectionCompleter = null;
+    _connectionStatusController?.close();
+    _connectionStatusController = null;
+  }
+}
+
+class FakeChannelState extends Fake implements ChannelState {}

packages/stream_chat/test/src/matchers.dart

@@ -1,7 +1,11 @@
 import 'package:collection/collection.dart';
 import 'package:dio/dio.dart' show MultipartFile;
+import 'package:stream_chat/src/client/channel.dart';
+import 'package:stream_chat/src/core/models/channel_model.dart';
+import 'package:stream_chat/src/core/models/channel_state.dart';
 import 'package:stream_chat/src/core/models/event.dart';
 import 'package:stream_chat/src/core/models/message.dart';
+import 'package:stream_chat/src/core/models/user.dart';
 import 'package:test/test.dart';
 
 Matcher isSameMultipartFileAs(MultipartFile targetFile) =>
@@ -96,3 +100,35 @@ class _IsSameMessageAs extends Matcher {
     return matches;
   }
 }
+
+Matcher isSameUserAs(User targetUser) => _IsSameUserAs(targetUser: targetUser);
+
+class _IsSameUserAs extends Matcher {
+  const _IsSameUserAs({required this.targetUser});
+
+  final User targetUser;
+
+  @override
+  Description describe(Description description) =>
+      description.add('is same user as $targetUser');
+
+  @override
+  bool matches(covariant User user, Map matchState) => user.id == targetUser.id;
+}
+
+Matcher isCorrectChannelFor(ChannelState channelState) =>
+    _IsCorrectChannelFor(channelState: channelState);
+
+class _IsCorrectChannelFor extends Matcher {
+  const _IsCorrectChannelFor({required this.channelState});
+
+  final ChannelState channelState;
+
+  @override
+  Description describe(Description description) =>
+      description.add('is correct channel for $channelState');
+
+  @override
+  bool matches(covariant Channel channel, Map matchState) =>
+      channel.cid == channelState.channel?.cid;
+}

packages/stream_chat/test/src/mocks.dart

@@ -18,6 +18,7 @@ import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:stream_chat/src/core/models/channel_config.dart';
 import 'package:stream_chat/src/core/models/channel_model.dart';
 import 'package:stream_chat/src/db/chat_persistence_client.dart';
+import 'package:stream_chat/src/ws/websocket.dart';
 import 'package:web_socket_channel/web_socket_channel.dart';
 
 import 'db/chat_persistence_client_test.dart';
@@ -66,7 +67,13 @@ class MockGeneralApi extends Mock implements GeneralApi {}
 class MockAttachmentFileUploader extends Mock
     implements AttachmentFileUploader {}
 
-class MockPersistenceClient extends Mock implements ChatPersistenceClient {}
+class MockPersistenceClient extends Mock implements ChatPersistenceClient {
+  @override
+  Future<void> connect(String userId) => Future.value();
+
+  @override
+  Future<void> disconnect({bool flush = false}) => Future.value();
+}
 
 class MockStreamChatClient extends Mock implements StreamChatClient {
   @override
@@ -105,3 +112,5 @@ class MockRetryQueueChannel extends Mock implements Channel {
   @override
   StreamChatClient get client => _client ??= MockStreamChatClient();
 }
+
+class MockWebSocket extends Mock implements WebSocket {}

packages/stream_chat/test/src/utils.dart

@@ -13,3 +13,12 @@ Directory get currentDirectory {
   }
   return directory;
 }
+
+// Extension function to convert int into durations
+extension IntX on num {
+  Duration toDuration() => Duration(milliseconds: toInt());
+}
+
+// Top level util function to delay the code execution
+Future delay(num milliseconds) =>
+    Future.delayed(Duration(milliseconds: milliseconds.toInt()));

packages/stream_chat_flutter/lib/src/message_list_view.dart

@@ -791,7 +791,7 @@ class _MessageListViewState extends State<MessageListView> {
           final channel = streamChannel.channel;
           if (_upToDate &&
               channel.config?.readEvents == true &&
-              channel.state!.unreadCount! > 0) {
+              channel.state!.unreadCount > 0) {
             streamChannel.channel.markRead();
           }
         }

packages/stream_chat_flutter_core/lib/src/stream_channel.dart

@@ -174,7 +174,7 @@ class StreamChannelState extends State<StreamChannel> {
     try {
       final response = await channel.getReplies(
         parentId,
-        PaginationParams(
+        options: PaginationParams(
           lessThan: message?.id,
           limit: limit,
         ),