From b7547613510e78bc976bee221f934e93c64bf192 Mon Sep 17 00:00:00 2001 From: VelikovPetar Date: Thu, 30 Jul 2026 11:33:32 +0200 Subject: [PATCH] fix(llc): Ignore events dispatched after the client is disposed Co-Authored-By: Claude Opus 4.8 --- packages/stream_chat/CHANGELOG.md | 1 + .../stream_chat/lib/src/client/client.dart | 5 +- .../test/src/client/client_test.dart | 79 +++++++++++++++++++ 3 files changed, 84 insertions(+), 1 deletion(-) diff --git a/packages/stream_chat/CHANGELOG.md b/packages/stream_chat/CHANGELOG.md index 96208f121..a063de46f 100644 --- a/packages/stream_chat/CHANGELOG.md +++ b/packages/stream_chat/CHANGELOG.md @@ -21,6 +21,7 @@ - Fixed watchers not being removed from `ChannelClientState.watchers` on `user.watching.stop`. - Fixed `Channel.name`/`image`/`extraData` setters throwing after a *failed* initialization; they now only throw once the channel is successfully initialized. - Fixed `Channel.initialized` staying errored after a failed init; it now reflects a subsequent successful (re)initialization. +- Fixed a `StateError` (`Cannot add new events after calling close`) thrown when the client is disposed while a reconnect recovery is still in flight. ## 10.2.0 diff --git a/packages/stream_chat/lib/src/client/client.dart b/packages/stream_chat/lib/src/client/client.dart index 32967c242..d56a37c21 100644 --- a/packages/stream_chat/lib/src/client/client.dart +++ b/packages/stream_chat/lib/src/client/client.dart @@ -576,11 +576,14 @@ class StreamChatClient { /// Method called to add a new event to the [_eventController]. void handleEvent(Event event) { + // Ignore events that arrive after the client has been disposed. + if (_eventController.isClosed) return; + if (event.type == EventType.healthCheck) { return _handleHealthCheckEvent(event); } state.updateUser(event.user); - return _eventController.add(event); + return _eventController.safeAdd(event); } void _onConnectionStatusChanged( diff --git a/packages/stream_chat/test/src/client/client_test.dart b/packages/stream_chat/test/src/client/client_test.dart index 3ee4f0352..c341e23c8 100644 --- a/packages/stream_chat/test/src/client/client_test.dart +++ b/packages/stream_chat/test/src/client/client_test.dart @@ -1,5 +1,7 @@ // ignore_for_file: avoid_redundant_argument_values, lines_longer_than_80_chars +import 'dart:async'; + import 'package:mocktail/mocktail.dart'; import 'package:stream_chat/src/core/http/token.dart'; import 'package:stream_chat/stream_chat.dart'; @@ -5423,6 +5425,83 @@ void main() { }); }); + group('dispose during reconnect recovery', () { + const apiKey = 'test-api-key'; + final user = User(id: 'test-user-id'); + final token = Token.development(user.id).rawValue; + + late FakeChatApi api; + late FakeWebSocket ws; + late StreamChatClient client; + var disposed = false; + + setUpAll(() { + registerFallbackValue(const PaginationParams()); + registerFallbackValue(Filter.equal('cid', '')); + }); + + setUp(() { + api = FakeChatApi(); + ws = FakeWebSocket(); + disposed = false; + }); + + // The test disposes the client itself; avoid disposing it a second time. + tearDown(() async { + if (!disposed) await client.dispose(); + }); + + // Disposing the client while a reconnect is still recovering must complete + // cleanly: recovery work that finishes after disposal is discarded, never + // surfacing as an error. + test('disposing mid-recovery does not surface a late recovery event', () async { + // Keep the recovery's channel query pending so the client is still + // mid-recovery at the moment it is disposed. + final pendingQuery = Completer(); + when( + () => api.channel.queryChannels( + filter: any(named: 'filter'), + sort: any(named: 'sort'), + state: any(named: 'state'), + watch: any(named: 'watch'), + presence: any(named: 'presence'), + memberLimit: any(named: 'memberLimit'), + messageLimit: any(named: 'messageLimit'), + paginationParams: any(named: 'paginationParams'), + ), + ).thenAnswer((_) => pendingQuery.future); + + client = StreamChatClient(apiKey, chatApi: api, ws: ws); + await client.connectUser(user, token); + await delay(300); + + // Track a channel so reconnecting triggers channel recovery, which then + // blocks on the pending query above. + final channel = Channel.fromState( + client, + ChannelState(channel: ChannelModel(cid: 'messaging:c1')), + ); + client.state.addChannels({'messaging:c1': channel}); + + // Drop then restore the connection to start a reconnect recovery. + ws.connectionStatus = ConnectionStatus.disconnected; + await delay(100); + ws.connectionStatus = ConnectionStatus.connected; + await delay(100); + + // Dispose while the recovery is still in flight. + await client.dispose(); + disposed = true; + + // Let the now-orphaned recovery finish. Its trailing work must be + // discarded silently instead of thrown as an unhandled async error. + pendingQuery.complete(QueryChannelsResponse()..channels = []); + await delay(300); + + expect(client.wsConnectionStatus, ConnectionStatus.disconnected); + }); + }); + group('WS events', () { late StreamChatClient client;