refactor(llc): safeAdd data to streamControllers (#2135)

* refactor(llc): safeAdd data to streamControllers

* chore: update at more places

Author: xsahil03x

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

@@ -2195,7 +2195,7 @@ class ChannelClientState {
   /// [EventType.messageNew] will not be pushed on to message list.
   bool get isUpToDate => _isUpToDateController.value;
 
-  set isUpToDate(bool isUpToDate) => _isUpToDateController.add(isUpToDate);
+  set isUpToDate(bool isUpToDate) => _isUpToDateController.safeAdd(isUpToDate);
 
   /// [isUpToDate] flag count as a stream.
   Stream<bool> get isUpToDateStream => _isUpToDateController.stream;
@@ -2967,7 +2967,7 @@ class ChannelClientState {
   final Debounce _debouncedUpdatePersistenceChannelState;
 
   set _channelState(ChannelState v) {
-    _channelStateController.add(v);
+    _channelStateController.safeAdd(v);
     _debouncedUpdatePersistenceChannelState.call([v]);
   }
 
@@ -2982,7 +2982,7 @@ class ChannelClientState {
       BehaviorSubject.seeded({});
 
   set _threads(Map<String, List<Message>> threads) {
-    _threadsController.add(threads);
+    _threadsController.safeAdd(threads);
     _channel.client.chatPersistenceClient?.updateChannelThreads(
       _channel.cid!,
       threads,
@@ -3011,7 +3011,7 @@ class ChannelClientState {
             if (user != null && user.id != currentUser.id) {
               final events = {...typingEvents};
               events[user] = event;
-              _typingEventsController.add(events);
+              _typingEventsController.safeAdd(events);
             }
           },
         ),
@@ -3022,7 +3022,7 @@ class ChannelClientState {
             final user = event.user;
             if (user != null && user.id != currentUser.id) {
               final events = {...typingEvents}..remove(user);
-              _typingEventsController.add(events);
+              _typingEventsController.safeAdd(events);
             }
           },
         ),

packages/stream_chat/lib/src/core/util/extension.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:http_parser/http_parser.dart';
 import 'package:mime/mime.dart';
 
@@ -27,3 +29,20 @@ extension StringX on String {
     return MediaType.parse(mimeType);
   }
 }
+
+/// 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/lib/src/ws/websocket.dart

@@ -10,6 +10,7 @@ import 'package:stream_chat/src/core/http/system_environment_manager.dart';
 import 'package:stream_chat/src/core/http/token_manager.dart';
 import 'package:stream_chat/src/core/models/event.dart';
 import 'package:stream_chat/src/core/models/user.dart';
+import 'package:stream_chat/src/core/util/extension.dart';
 import 'package:stream_chat/src/event_type.dart';
 import 'package:stream_chat/src/ws/connection_status.dart';
 import 'package:stream_chat/src/ws/timer_helper.dart';
@@ -108,7 +109,7 @@ class WebSocket with TimerHelper {
       BehaviorSubject.seeded(ConnectionStatus.disconnected);
 
   set _connectionStatus(ConnectionStatus status) =>
-      _connectionStatusController.add(status);
+      _connectionStatusController.safeAdd(status);
 
   /// The current connection status value
   ConnectionStatus get connectionStatus => _connectionStatusController.value;

packages/stream_chat_flutter_core/lib/src/stream_channel.dart

@@ -4,7 +4,6 @@ import 'package:collection/collection.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 {