Merge branch 'develop' into 3rd-party-video-integration

packages/stream_chat/CHANGELOG.md

@@ -4,6 +4,10 @@
 
 - Added `StreamChatClient.getCallToken` and `StreamChatClient.createCall` methods.
 
+🐞 Fixed
+
+- Only listen to client events when the user is connected to the websocket.
+
 ## 4.5.0
 
 🐞 Fixed

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

@@ -397,6 +397,10 @@ class StreamChatClient {
         user,
         includeUserDetails: includeUserDetailsInConnectCall,
       );
+
+      // Start listening to events
+      state.subscribeToEvents();
+
       return user.merge(event.me);
     } catch (e, stk) {
       logger.severe('error connecting ws', e, stk);
@@ -418,6 +422,9 @@ class StreamChatClient {
     _connectionStatusSubscription?.cancel();
     _connectionStatusSubscription = null;
 
+    // Stop listening to events
+    state.cancelEventSubscription();
+
     _ws.disconnect();
   }
 
@@ -1490,29 +1497,40 @@ class StreamChatClient {
 /// The class that handles the state of the channel listening to the events
 class ClientState {
   /// Creates a new instance listening to events and updating the state
-  ClientState(this._client) {
+  ClientState(this._client);
-    _subscriptions.addAll([
+
-      _client
+  CompositeSubscription? _eventsSubscription;
+
+  /// Starts listening to the client events.
+  void subscribeToEvents() {
+    if (_eventsSubscription != null) {
+      cancelEventSubscription();
+    }
+
+    _eventsSubscription = CompositeSubscription();
+    _eventsSubscription!
+      ..add(_client
           .on()
           .where((event) =>
               event.me != null && event.type != EventType.healthCheck)
           .map((e) => e.me!)
-          .listen((user) => currentUser = currentUser?.merge(user) ?? user),
+          .listen((user) {
-      _client
+        currentUser = currentUser?.merge(user) ?? user;
+      }))
+      ..add(_client
           .on()
           .map((event) => event.unreadChannels)
           .whereType<int>()
           .listen((count) {
         currentUser = currentUser?.copyWith(unreadChannels: count);
-      }),
+      }))
-      _client
+      ..add(_client
           .on()
           .map((event) => event.totalUnreadCount)
           .whereType<int>()
           .listen((count) {
         currentUser = currentUser?.copyWith(totalUnreadCount: count);
-      }),
+      }));
-    ]);
 
     _listenChannelDeleted();
 
@@ -1523,56 +1541,73 @@ class ClientState {
     _listenAllChannelsRead();
   }
 
-  final _subscriptions = <StreamSubscription>[];
+  /// Stops listening to the client events.
+  void cancelEventSubscription() {
+    if (_eventsSubscription != null) {
+      _eventsSubscription!.cancel();
+      _eventsSubscription = null;
+    }
+  }
 
-  /// Used internally for optimistic update of unread count
+  /// Pauses listening to the client events.
-  set totalUnreadCount(int unreadCount) {
+  void pauseEventSubscription([Future<void>? resumeSignal]) {
-    _totalUnreadCountController.add(unreadCount);
+    _eventsSubscription?.pause(resumeSignal);
+  }
+
+  /// Resumes listening to the client events.
+  void resumeEventSubscription() {
+    _eventsSubscription?.resume();
   }
 
   void _listenChannelHidden() {
-    _subscriptions
+    _eventsSubscription?.add(
-        .add(_client.on(EventType.channelHidden).listen((event) async {
+      _client.on(EventType.channelHidden).listen((event) async {
-      final eventChannel = event.channel!;
+        final eventChannel = event.channel!;
-      await _client.chatPersistenceClient?.deleteChannels([eventChannel.cid]);
+        await _client.chatPersistenceClient?.deleteChannels([eventChannel.cid]);
-      channels[eventChannel.cid]?.dispose();
+        channels[eventChannel.cid]?.dispose();
-      channels = channels..remove(eventChannel.cid);
+        channels = channels..remove(eventChannel.cid);
-    }));
+      }),
+    );
   }
 
   void _listenUserUpdated() {
-    _subscriptions.add(_client.on(EventType.userUpdated).listen((event) {
+    _eventsSubscription?.add(
-      if (event.user!.id == currentUser!.id) {
+      _client.on(EventType.userUpdated).listen((event) {
-        currentUser = OwnUser.fromJson(event.user!.toJson());
+        if (event.user!.id == currentUser!.id) {
-      }
+          currentUser = OwnUser.fromJson(event.user!.toJson());
-      updateUser(event.user);
+        }
-    }));
+        updateUser(event.user);
+      }),
+    );
   }
 
   void _listenAllChannelsRead() {
-    _subscriptions
+    _eventsSubscription?.add(
-        .add(_client.on(EventType.notificationMarkRead).listen((event) {
+      _client.on(EventType.notificationMarkRead).listen((event) {
-      if (event.cid == null) {
+        if (event.cid == null) {
-        channels.forEach((key, value) {
+          channels.forEach((key, value) {
-          value.state?.unreadCount = 0;
+            value.state?.unreadCount = 0;
-        });
+          });
-      }
+        }
-    }));
+      }),
+    );
   }
 
   void _listenChannelDeleted() {
-    _subscriptions.add(_client
+    _eventsSubscription?.add(
-        .on(
+      _client
-      EventType.channelDeleted,
+          .on(
-      EventType.notificationRemovedFromChannel,
+        EventType.channelDeleted,
-      EventType.notificationChannelDeleted,
+        EventType.notificationRemovedFromChannel,
-    )
+        EventType.notificationChannelDeleted,
-        .listen((Event event) async {
+      )
-      final eventChannel = event.channel!;
+          .listen((Event event) async {
-      await _client.chatPersistenceClient?.deleteChannels([eventChannel.cid]);
+        final eventChannel = event.channel!;
-      channels[eventChannel.cid]?.dispose();
+        await _client.chatPersistenceClient?.deleteChannels([eventChannel.cid]);
-      channels = channels..remove(eventChannel.cid);
+        channels[eventChannel.cid]?.dispose();
-    }));
+        channels = channels..remove(eventChannel.cid);
+      }),
+    );
   }
 
   final StreamChatClient _client;
@@ -1634,6 +1669,11 @@ class ClientState {
     _channelsController.add(newChannels);
   }
 
+  /// Used internally for optimistic update of unread count
+  set totalUnreadCount(int unreadCount) {
+    _totalUnreadCountController.add(unreadCount);
+  }
+
   void _computeUnreadCounts(OwnUser? user) {
     final totalUnreadCount = user?.totalUnreadCount;
     if (totalUnreadCount != null) {
@@ -1654,7 +1694,7 @@ class ClientState {
 
   /// Call this method to dispose this object
   void dispose() {
-    _subscriptions.forEach((s) => s.cancel());
+    cancelEventSubscription();
     _currentUserController.close();
     _unreadChannelsController.close();
     _totalUnreadCountController.close();