fix(core): Replaced all StreamController.add() with .safeAdd() to fix bad state errors.

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

packages/stream_chat_flutter_core/CHANGELOG.md

@@ -3,6 +3,7 @@
 🐞 Fixed
 
 - Do not move a channel to top if the new message is from a thread.
+- [[#848]](https://github.com/GetStream/stream-chat-flutter/issues/848) Fixed "Bad state: Cannot add new events after calling close" by replacing all `.add` methods with a new `.safeAdd`.
 
 ## 3.3.1
 

packages/stream_chat_flutter_core/lib/src/channels_bloc.dart

@@ -1,11 +1,11 @@
 import 'dart:async';
 
-import 'package:flutter/foundation.dart';
 import 'package:flutter/material.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/stream_chat.dart';
 import 'package:stream_chat_flutter_core/src/channel_list_core.dart';
 import 'package:stream_chat_flutter_core/src/stream_chat_core.dart';
+import 'package:stream_chat_flutter_core/src/stream_controller_extension.dart';
 
 /// Widget dedicated to the management of a channel list with pagination
 /// [ChannelsBloc] is used together with [ChannelListCore] to manage a list of
@@ -116,7 +116,7 @@ class ChannelsBlocState extends State<ChannelsBloc>
     }
 
     if (_channelsController.hasValue) {
-      _queryChannelsLoadingController.add(true);
+      _queryChannelsLoadingController.safeAdd(true);
     }
 
     try {
@@ -135,14 +135,14 @@ class ChannelsBlocState extends State<ChannelsBloc>
       )) {
         newChannels = channels;
         if (clear) {
-          _channelsController.add(channels);
+          _channelsController.safeAdd(channels);
         } else {
           final temp = oldChannels + channels;
-          _channelsController.add(temp);
+          _channelsController.safeAdd(temp);
         }
         if (_channelsController.hasValue &&
             _queryChannelsLoadingController.value) {
-          _queryChannelsLoadingController.sink.add(false);
+          _queryChannelsLoadingController.safeAdd(false);
         }
       }
       if (newChannels.isEmpty || newChannels.length < paginationParams.limit) {
@@ -150,11 +150,11 @@ class ChannelsBlocState extends State<ChannelsBloc>
       }
     } catch (e, stk) {
       // reset loading controller
-      _queryChannelsLoadingController.sink.add(false);
+      _queryChannelsLoadingController.safeAdd(false);
       if (_channelsController.hasValue) {
-        _queryChannelsLoadingController.addError(e, stk);
+        _queryChannelsLoadingController.safeAddError(e, stk);
       } else {
-        _channelsController.addError(e, stk);
+        _channelsController.safeAddError(e, stk);
       }
     }
   }
@@ -200,7 +200,7 @@ class ChannelsBlocState extends State<ChannelsBloc>
           if (widget.channelsComparator != null) {
             newChannels.sort(widget.channelsComparator);
           }
-          _channelsController.add(newChannels);
+          _channelsController.safeAdd(newChannels);
         }));
       }
 
@@ -212,7 +212,7 @@ class ChannelsBlocState extends State<ChannelsBloc>
           if (channelIndex > -1) {
             final channel = newChannels.removeAt(channelIndex);
             _hiddenChannels.add(channel);
-            _channelsController.add(newChannels);
+            _channelsController.safeAdd(newChannels);
           }
         }))
         ..add(client
@@ -222,7 +222,7 @@ class ChannelsBlocState extends State<ChannelsBloc>
         )
             .listen((e) {
           final channel = e.channel;
-          _channelsController.add(List.from(
+          _channelsController.safeAdd(List.from(
             (channels ?? [])..removeWhere((c) => c.cid == channel?.cid),
           ));
         }));

packages/stream_chat_flutter_core/lib/src/message_search_bloc.dart

@@ -2,6 +2,7 @@ import 'package:flutter/material.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/stream_chat.dart';
 import 'package:stream_chat_flutter_core/src/stream_chat_core.dart';
+import 'package:stream_chat_flutter_core/src/stream_controller_extension.dart';
 
 /// [MessageSearchBloc] is used to manage a list of messages with pagination.
 /// This class can be used to load messages, perform queries, etc.
@@ -95,7 +96,7 @@ class MessageSearchBlocState extends State<MessageSearchBloc>
     }
 
     if (_messageResponses.hasValue) {
-      _queryMessagesLoadingController.add(true);
+      _queryMessagesLoadingController.safeAdd(true);
     }
     try {
       final oldMessages = List<GetMessageResponse>.from(messageResponses ?? []);
@@ -120,24 +121,24 @@ class MessageSearchBlocState extends State<MessageSearchBloc>
 
       final newMessages = response.results;
       if (clear) {
-        _messageResponses.add(newMessages);
+        _messageResponses.safeAdd(newMessages);
       } else {
         final temp = oldMessages + newMessages;
-        _messageResponses.add(temp);
+        _messageResponses.safeAdd(temp);
       }
       if (_messageResponses.hasValue && _queryMessagesLoadingController.value) {
-        _queryMessagesLoadingController.add(false);
+        _queryMessagesLoadingController.safeAdd(false);
       }
       if (newMessages.isEmpty || newMessages.length < pagination.limit) {
         _paginationEnded = true;
       }
     } catch (e, stk) {
       // reset loading controller
-      _queryMessagesLoadingController.add(false);
+      _queryMessagesLoadingController.safeAdd(false);
       if (_messageResponses.hasValue) {
-        _queryMessagesLoadingController.addError(e, stk);
+        _queryMessagesLoadingController.safeAddError(e, stk);
       } else {
-        _messageResponses.addError(e, stk);
+        _messageResponses.safeAddError(e, stk);
       }
     }
   }

packages/stream_chat_flutter_core/lib/src/stream_channel.dart

@@ -1,10 +1,10 @@
 import 'dart:async';
 
 import 'package:collection/collection.dart';
-import 'package:flutter/foundation.dart';
 import 'package:flutter/material.dart';
 import 'package:rxdart/rxdart.dart';
 import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_flutter_core/src/stream_controller_extension.dart';
 
 /// Specifies query direction for pagination
 enum QueryDirection {
@@ -92,10 +92,10 @@ class StreamChannelState extends State<StreamChannel> {
         channel.state == null) {
       return;
     }
-    _queryTopMessagesController.add(true);
+    _queryTopMessagesController.safeAdd(true);
 
     if (channel.state!.messages.isEmpty) {
-      return _queryTopMessagesController.add(false);
+      return _queryTopMessagesController.safeAdd(false);
     }
 
     final oldestMessage = channel.state!.messages.first;
@@ -109,9 +109,9 @@ class StreamChannelState extends State<StreamChannel> {
       if (state.messages.isEmpty || state.messages.length < limit) {
         _topPaginationEnded = true;
       }
-      _queryTopMessagesController.add(false);
+      _queryTopMessagesController.safeAdd(false);
     } catch (e, stk) {
-      _queryTopMessagesController.addError(e, stk);
+      _queryTopMessagesController.safeAddError(e, stk);
     }
   }
 
@@ -123,10 +123,10 @@ class StreamChannelState extends State<StreamChannel> {
         _queryBottomMessagesController.value ||
         channel.state == null ||
         channel.state!.isUpToDate) return;
-    _queryBottomMessagesController.add(true);
+    _queryBottomMessagesController.safeAdd(true);
 
     if (channel.state!.messages.isEmpty) {
-      return _queryBottomMessagesController.add(false);
+      return _queryBottomMessagesController.safeAdd(false);
     }
 
     final recentMessage = channel.state!.messages.last;
@@ -140,9 +140,9 @@ class StreamChannelState extends State<StreamChannel> {
       if (state.messages.isEmpty || state.messages.length < limit) {
         _bottomPaginationEnded = true;
       }
-      _queryBottomMessagesController.add(false);
+      _queryBottomMessagesController.safeAdd(false);
     } catch (e, stk) {
-      _queryBottomMessagesController.addError(e, stk);
+      _queryBottomMessagesController.safeAddError(e, stk);
     }
   }
 
@@ -166,7 +166,7 @@ class StreamChannelState extends State<StreamChannel> {
     if (_topPaginationEnded ||
         _queryTopMessagesController.value ||
         channel.state == null) return;
-    _queryTopMessagesController.add(true);
+    _queryTopMessagesController.safeAdd(true);
 
     Message? message;
     if (channel.state!.threads.containsKey(parentId)) {
@@ -188,9 +188,9 @@ class StreamChannelState extends State<StreamChannel> {
       if (response.messages.isEmpty || response.messages.length < limit) {
         _topPaginationEnded = true;
       }
-      _queryTopMessagesController.add(false);
+      _queryTopMessagesController.safeAdd(false);
     } catch (e, stk) {
-      _queryTopMessagesController.addError(e, stk);
+      _queryTopMessagesController.safeAddError(e, stk);
     }
   }
 

packages/stream_chat_flutter_core/lib/src/stream_controller_extension.dart

@@ -0,0 +1,18 @@
+import 'dart:async';
+
+/// Extension on [StreamController] to safely add events and errors.
+extension StreamControllerX<T> on StreamController<T> {
+  /// Safely adds the event to the controller,
+  /// Returns early if the controller is closed.
+  void safeAdd(T event) {
+    if (isClosed) return;
+    add(event);
+  }
+
+  /// Safely adds the error to the controller,
+  /// Returns early if the controller is closed.
+  void safeAddError(Object error, [StackTrace? stackTrace]) {
+    if (isClosed) return;
+    addError(error, stackTrace);
+  }
+}

packages/stream_chat_flutter_core/lib/src/users_bloc.dart

@@ -1,7 +1,6 @@
-import 'package:flutter/foundation.dart';
 import 'package:flutter/material.dart';
 import 'package:rxdart/rxdart.dart';
-import 'package:stream_chat/stream_chat.dart';
+import 'package:stream_chat_flutter_core/src/stream_controller_extension.dart';
 import 'package:stream_chat_flutter_core/stream_chat_flutter_core.dart';
 
 /// Widget dedicated to the management of a users list with pagination.
@@ -82,7 +81,7 @@ class UsersBlocState extends State<UsersBloc>
     }
 
     if (_usersController.hasValue) {
-      _queryUsersLoadingController.add(true);
+      _queryUsersLoadingController.safeAdd(true);
     }
 
     try {
@@ -97,24 +96,24 @@ class UsersBlocState extends State<UsersBloc>
 
       final newUsers = usersResponse.users;
       if (clear) {
-        _usersController.add(usersResponse.users);
+        _usersController.safeAdd(usersResponse.users);
       } else {
         final temp = oldUsers + usersResponse.users;
-        _usersController.add(temp);
+        _usersController.safeAdd(temp);
       }
       if (_usersController.hasValue && _queryUsersLoadingController.value) {
-        _queryUsersLoadingController.add(false);
+        _queryUsersLoadingController.safeAdd(false);
       }
       if (newUsers.isEmpty || newUsers.length < pagination.limit) {
         _paginationEnded = true;
       }
     } catch (e, stk) {
       // reset loading controller
-      _queryUsersLoadingController.add(false);
+      _queryUsersLoadingController.safeAdd(false);
       if (_usersController.hasValue) {
-        _queryUsersLoadingController.addError(e, stk);
+        _queryUsersLoadingController.safeAddError(e, stk);
       } else {
-        _usersController.addError(e, stk);
+        _usersController.safeAddError(e, stk);
       }
     }
   }