Merge remote-tracking branch 'origin/master' into v10.0.0

# Conflicts:
#	packages/stream_chat/CHANGELOG.md
#	packages/stream_chat/test/src/client/channel_test.dart
#	packages/stream_chat_flutter/test/src/poll/creator/goldens/ci/stream_poll_creator_dialog_dark.png
#	packages/stream_chat_flutter/test/src/poll/creator/goldens/ci/stream_poll_creator_dialog_light.png
#	packages/stream_chat_flutter/test/src/poll/creator/goldens/ci/stream_poll_creator_full_screen_dialog_dark.png
#	packages/stream_chat_flutter/test/src/poll/creator/goldens/ci/stream_poll_creator_full_screen_dialog_light.png
#	packages/stream_chat_flutter/test/src/poll/goldens/ci/stream_poll_creator_dialog_dark.png
#	packages/stream_chat_flutter/test/src/poll/goldens/ci/stream_poll_creator_dialog_light.png
#	packages/stream_chat_flutter/test/src/poll/goldens/ci/stream_poll_creator_full_screen_dialog_dark.png
#	packages/stream_chat_flutter/test/src/poll/goldens/ci/stream_poll_creator_full_screen_dialog_light.png

Author: xsahil03x

packages/stream_chat/CHANGELOG.md

@@ -3,6 +3,8 @@
 🐞 Fixed
 
 - Fixed `skipPush` and `skipEnrichUrl` not preserving during message send or update retry
+- Fixed `Channel` methods to throw proper `StateError` exceptions instead of relying on assertions
+  for state validation.
 
 ## 10.0.0-beta.4
 

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

@@ -2132,15 +2132,18 @@ class Channel {
   void dispose() {
     client.state.removeChannel('$cid');
     state?.dispose();
+    state = null;
     _muteExpirationTimer?.cancel();
     _keyStrokeHandler.cancel();
   }
 
   void _checkInitialized() {
-    assert(
-      _initializedCompleter.isCompleted,
-      "Channel $cid hasn't been initialized yet. Make sure to call .watch()"
-      ' or to instantiate the client using [Channel.fromState]',
+    if (_initializedCompleter.isCompleted && state != null) return;
+
+    throw StateError(
+      "Channel $cid hasn't been initialized yet or has been disposed. "
+      'Make sure to call .watch() or instantiate the client using '
+      '[Channel.fromState]',
     );
   }
 }

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

@@ -129,6 +129,13 @@ class StreamChatClient {
           },
         );
 
+    _connectionStatusSubscription = wsConnectionStatusStream.pairwise().listen(
+      (statusPair) {
+        final [prevStatus, currStatus] = statusPair;
+        return _onConnectionStatusChanged(prevStatus, currStatus);
+      },
+    );
+
     state = ClientState(this);
   }
 
@@ -226,7 +233,7 @@ class StreamChatClient {
   ///```
   final LogHandlerFunction logHandlerFunction;
 
-  StreamSubscription<ConnectionStatus>? _connectionStatusSubscription;
+  StreamSubscription<List<ConnectionStatus>>? _connectionStatusSubscription;
 
   /// Stream of [Event] coming from [_ws] connection
   /// Listen to this or use the [on] method to filter specific event types
@@ -242,20 +249,14 @@ class StreamChatClient {
     ],
   );
 
-  final _wsConnectionStatusController =
-      BehaviorSubject.seeded(ConnectionStatus.disconnected);
-
-  set _wsConnectionStatus(ConnectionStatus status) =>
-      _wsConnectionStatusController.add(status);
-
   /// The current status value of the [_ws] connection
-  ConnectionStatus get wsConnectionStatus =>
-      _wsConnectionStatusController.value;
+  ConnectionStatus get wsConnectionStatus => _ws.connectionStatus;
 
   /// This notifies the connection status of the [_ws] connection.
   /// Listen to this to get notified when the [_ws] tries to reconnect.
-  Stream<ConnectionStatus> get wsConnectionStatusStream =>
-      _wsConnectionStatusController.stream.distinct();
+  Stream<ConnectionStatus> get wsConnectionStatusStream {
+    return _ws.connectionStatusStream.distinct();
+  }
 
   /// Default log handler function for the [StreamChatClient] logger.
   static void defaultLogHandler(LogRecord record) {
@@ -451,16 +452,6 @@ class StreamChatClient {
       throw StreamChatError('Connection already available for ${user.id}');
     }
 
-    _wsConnectionStatus = ConnectionStatus.connecting;
-
-    // skipping `ws` seed connection status -> ConnectionStatus.disconnected
-    // 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 {
       final event = await _ws.connect(
         user,
@@ -483,13 +474,7 @@ class StreamChatClient {
   /// This will not trigger default auto-retry mechanism for reconnection.
   /// You need to call [openConnection] to reconnect to [_ws].
   void closeConnection() {
-    if (wsConnectionStatus == ConnectionStatus.disconnected) return;
-
     logger.info('Closing web-socket connection for ${state.currentUser?.id}');
-    _wsConnectionStatus = ConnectionStatus.disconnected;
-
-    _connectionStatusSubscription?.cancel();
-    _connectionStatusSubscription = null;
 
     // Stop listening to events
     state.cancelEventSubscription();
@@ -517,19 +502,25 @@ class StreamChatClient {
     return _eventController.add(event);
   }
 
-  void _connectionStatusHandler(ConnectionStatus status) async {
-    final previousState = wsConnectionStatus;
-    final currentState = _wsConnectionStatus = status;
+  void _onConnectionStatusChanged(
+    ConnectionStatus prevStatus,
+    ConnectionStatus currStatus,
+  ) async {
+    // If the status hasn't changed, we don't need to do anything.
+    if (prevStatus == currStatus) return;
 
-    if (previousState != currentState) {
-      handleEvent(Event(
-        type: EventType.connectionChanged,
-        online: status == ConnectionStatus.connected,
-      ));
-    }
+    final wasConnected = prevStatus == ConnectionStatus.connected;
+    final isConnected = currStatus == ConnectionStatus.connected;
+
+    // Notify the connection status change event
+    handleEvent(Event(
+      type: EventType.connectionChanged,
+      online: isConnected,
+    ));
 
-    if (currentState == ConnectionStatus.connected &&
-        previousState != ConnectionStatus.connected) {
+    final connectionRecovered = !wasConnected && isConnected;
+
+    if (connectionRecovered) {
       // connection recovered
       final cids = state.channels.keys.toList(growable: false);
       if (cids.isNotEmpty) {
@@ -2046,6 +2037,9 @@ class StreamChatClient {
   Future<void> disconnectUser({bool flushChatPersistence = false}) async {
     logger.info('Disconnecting user : ${state.currentUser?.id}');
 
+    // closing web-socket connection
+    closeConnection();
+
     // resetting state.
     state.dispose();
     state = ClientState(this);
@@ -2056,27 +2050,17 @@ class StreamChatClient {
     _connectionIdManager.reset();
 
     // closing persistence connection.
-    await closePersistenceConnection(flush: flushChatPersistence);
-
-    // closing web-socket connection
-    return closeConnection();
+    return closePersistenceConnection(flush: flushChatPersistence);
   }
 
   /// Call this function to dispose the client
   Future<void> dispose() async {
-    logger.info('Disposing new StreamChatClient');
-
-    // disposing state.
-    state.dispose();
-
-    // closing persistence connection.
-    await closePersistenceConnection();
-
-    // closing web-socket connection.
-    closeConnection();
+    logger.info('Disposing StreamChatClient');
 
+    await disconnectUser();
+    await _ws.dispose();
     await _eventController.close();
-    await _wsConnectionStatusController.close();
+    await _connectionStatusSubscription?.cancel();
   }
 }
 

packages/stream_chat/lib/src/ws/websocket.dart

@@ -43,6 +43,7 @@ class WebSocket with TimerHelper {
     this.reconnectionMonitorInterval = 10,
     this.healthCheckInterval = 20,
     this.reconnectionMonitorTimeout = 40,
+    this.maxReconnectAttempts = 6,
     this.queryParameters = const {},
   }) : _logger = logger;
 
@@ -93,6 +94,11 @@ class WebSocket with TimerHelper {
   /// connection unhealthy
   final int reconnectionMonitorTimeout;
 
+  /// Maximum number of reconnection attempts before giving up.
+  ///
+  /// Default is 6 attempts. ~30 seconds of reconnection attempts
+  final int maxReconnectAttempts;
+
   User? _user;
   String? _connectionId;
   DateTime? _lastEventAt;
@@ -105,18 +111,20 @@ class WebSocket with TimerHelper {
   ///
   String? get connectionId => _connectionId;
 
-  final _connectionStatusController =
-      BehaviorSubject.seeded(ConnectionStatus.disconnected);
-
-  set _connectionStatus(ConnectionStatus status) =>
-      _connectionStatusController.safeAdd(status);
+  final _connectionStatusController = BehaviorSubject.seeded(
+    ConnectionStatus.disconnected,
+  );
 
   /// The current connection status value
   ConnectionStatus get connectionStatus => _connectionStatusController.value;
+  set _connectionStatus(ConnectionStatus status) {
+    _connectionStatusController.safeAdd(status);
+  }
 
   /// This notifies of connection status changes
-  Stream<ConnectionStatus> get connectionStatusStream =>
-      _connectionStatusController.stream.distinct();
+  Stream<ConnectionStatus> get connectionStatusStream {
+    return _connectionStatusController.stream.distinct();
+  }
 
   void _initWebSocketChannel(Uri uri) {
     _logger?.info('Initiating connection with $baseUrl');
@@ -250,6 +258,12 @@ class WebSocket with TimerHelper {
   void _reconnect({bool refreshToken = false}) async {
     _logger?.info('Retrying connection : $_reconnectAttempt');
     if (_reconnectRequestInProgress) return;
+
+    if (_reconnectAttempt >= maxReconnectAttempts) {
+      _logger?.severe('Max reconnect attempts reached: $maxReconnectAttempts');
+      return disconnect();
+    }
+
     _reconnectRequestInProgress = true;
 
     _stopMonitoringEvents();
@@ -495,4 +509,14 @@ class WebSocket with TimerHelper {
 
     _closeWebSocketChannel();
   }
+
+  /// Disposes the web-socket connection and releases resources
+  Future<void> dispose() async {
+    _logger?.info('Disposing web-socket connection');
+
+    _stopMonitoringEvents();
+    _unsubscribeFromWebSocketChannel();
+    _closeWebSocketChannel();
+    _connectionStatusController.close();
+  }
 }