diff --git a/packages/stream_core/CHANGELOG.md b/packages/stream_core/CHANGELOG.md index 8b86e432..e0d30d7d 100644 --- a/packages/stream_core/CHANGELOG.md +++ b/packages/stream_core/CHANGELOG.md @@ -4,8 +4,14 @@ - Removed the `userId` parameter from `UserToken.anonymous`, anonymous tokens always use `User.anonymousUserId` - Removed the `TokenManager.tokenProvider` setter, use `setTokenProvider` instead +- `StreamWebSocketClient` now takes an `optionsBuilder` instead of `options`, and calls it for every connection attempt +- Renamed `StreamWebSocketClient.onConnectionEstablished` to `onAuthenticate`, which is what it is called for and when +- `StreamWebSocketClient.onAuthenticate` is now a `WebSocketAuthenticator`: it is handed a `WsSender` and returns a `Result`, so a failure to authenticate can be observed - `TokenManager.userId` is now nullable, and is `null` until an identity is configured - `User` now requires a user of type `UserType.anonymous` to carry `User.anonymousUserId` as its id. The constructor is `const`, so a mismatch in a const context fails to compile rather than throwing in debug mode +- `StreamApiError.isTokenExpiredError` now means the token expired (code 40) rather than any invalid-token code. The rest — not yet valid, used before issued, wrong signature — are `isInvalidTokenError`, along with a wrong API key, since another token does not fix any of them +- `StreamApiError.isClientError` compares the HTTP `statusCode` against 400..499 rather than the Stream error `code`, which never falls in that range and so never matched +- `Result.getOrElse`, `getOrDefault`, `recover` and `recoverCatching` no longer take a type parameter of their own and return the result's own type. They previously cast the value to the callback's type, which failed on a successful result — most visibly for a callback that only throws. Kotlin's equivalents widen through a `` bound that Dart cannot express; to widen here, name the wider type on the result (`Result widened = intResult`), which works because `Result` is covariant, or use `fold` ### ✨ Features @@ -15,6 +21,13 @@ - Added `User.anonymousUserId`, the id every anonymous user has - Added `TokenManager.unconfigured`, for a client that exists before its user does - Added `TokenManager.reset`, which drops the configured identity and its cached token +- Added `DisconnectionSource.connectTimeout`, reported when a connection attempt is abandoned before the connection is established; it is eligible for automatic reconnection, since a handshake that did not complete in time is the same failure as a connection that stops answering health checks +- Added `DisconnectionSource.authenticationFailed`, reported with its cause when a connection opens but cannot be authenticated +- `StreamWebSocketClient` now honours `WebSocketOptions.connectTimeout`, which is no longer nullable and defaults to `WebSocketOptions.defaultConnectTimeout`. This is a behaviour change as well as an API one: a connection previously waited indefinitely for its first health check, and is now abandoned — and reconnected — after 30 seconds, matching the wait the Swift SDK allows for the same handshake +- Added `WsSender`, the send capability handed to a `WebSocketAuthenticator` +- Added `WebSocketConnectionState.isExpiredTokenDisconnection`, so a caller that can replace the token knows when to. Automatic reconnection deliberately refuses this case: whoever retries it would present the token the server just refused, and only the caller can obtain another +- Added `ConnectUserDetailsRequest.fromUser`, which builds the details a client may send from a `User` +- Added `StreamWebSocketClient.dispose`, which closes the connection along with `events` and `connectionState`; the client is `Disposable`, so `isDisposed` reports whether it has been called - Added `teams` field to `User` class ### 🐛 Bug Fixes @@ -22,6 +35,14 @@ - Fixed `TokenManager.getToken()` contacting the `TokenProvider` on every call instead of returning the cached token - Fixed `DynamicTokenProvider` accepting a token issued for a different user than the one requested - Fixed `TokenManager` caching a token that finished loading after `expireToken` or `setTokenProvider` had invalidated it +- Fixed `StreamWebSocketClient.disconnect` completing before the socket was closed, so a `connect` straight afterwards raced the closure and saw the connection go down again +- Fixed a failure to close the socket leaving `StreamWebSocketClient` reporting itself as disconnecting for good, since the engine reports such a failure rather than notifying its listener +- Fixed a `WebSocketAuthenticator` that throws, rather than returning a failure, escaping as an unhandled error and leaving the connection authenticating until it timed out — losing the cause, which the timeout does not carry. The natural authenticator throws, since loading a token does +- Fixed `StreamWebSocketClient.disconnect` replacing the source of a closure already under way, which could turn a reconnectable `ServerInitiated` error into a permanent `ConnectTimeout` +- Fixed `isAutomaticReconnectionEnabled` neither refusing a deliberate server close nor refusing client errors: it compared the API error's code against the close code 1000, and against a 400..499 range that Stream codes never occupy, so both rules were dead. It now mirrors the iOS SDK — close code 1000, invalid tokens and 4xx all refuse, an expired token does not +- A connection the server closed because the request was rate limited is now eligible for automatic reconnection. The backend closes it with the rate-limit window's reset in the response headers and a one-minute window, so the condition clears on its own — unlike every other 4xx it can close a socket with, which carry no reset +- Fixed `ConnectionRecoveryHandler` retrying a first connection attempt, which failed the caller of `connect` and reconnected behind them at the same time — and made the caller's own retry fail with "connection already in progress". It now recovers only connections that have existed since the caller last asked for one, so a deliberate `disconnect` hands connecting back and the next `connect` is the caller's attempt again +- Fixed a health check arriving while disconnecting reporting the connection as established again, which replaced the disconnection source and could turn a deliberate disconnect into an automatic reconnect ### 🔄 Changed diff --git a/packages/stream_core/lib/src/errors/stream_api_error.dart b/packages/stream_core/lib/src/errors/stream_api_error.dart index 19c2fe05..f08f5052 100644 --- a/packages/stream_core/lib/src/errors/stream_api_error.dart +++ b/packages/stream_core/lib/src/errors/stream_api_error.dart @@ -69,16 +69,32 @@ class StreamApiError extends Equatable { ]; } -final _tokenInvalidErrorCodes = _range(40, 42); -final _clientErrorCodes = _range(400, 499); +// The token this was issued for has expired; another one is accepted. +const _expiredTokenCode = 40; + +// The token cannot be accepted for a reason another token does not fix: not +// valid yet, used before it was issued, or signed with the wrong secret. +final _invalidTokenCodes = _range(41, 43); + +// The API key itself is wrong, which no token repairs either. +const _accessKeyErrorCode = 2; + +final _clientErrorStatusCodes = _range(400, 499); /// Extension methods for [StreamApiError] to provide convenient error type checks. extension StreamApiErrorExtension on StreamApiError { - /// Whether this error indicates an expired or invalid token. - bool get isTokenExpiredError => _tokenInvalidErrorCodes.contains(code); + /// Whether the token has expired. + /// + /// Distinct from [isInvalidTokenError]: an expired token is replaced by asking + /// the provider for another, where an invalid one is a configuration problem + /// that a fresh token presents again. + bool get isTokenExpiredError => code == _expiredTokenCode; + + /// Whether the token, or the key it was signed with, cannot be accepted. + bool get isInvalidTokenError => _invalidTokenCodes.contains(code) || code == _accessKeyErrorCode; /// Whether this error is a client-side error (4xx status codes). - bool get isClientError => _clientErrorCodes.contains(code); + bool get isClientError => _clientErrorStatusCodes.contains(statusCode); /// Whether this error indicates rate limiting (429 status code). bool get isRateLimitError => statusCode == 429; diff --git a/packages/stream_core/lib/src/user/connect_user_details_request.dart b/packages/stream_core/lib/src/user/connect_user_details_request.dart index 553ba9d2..7e27d170 100644 --- a/packages/stream_core/lib/src/user/connect_user_details_request.dart +++ b/packages/stream_core/lib/src/user/connect_user_details_request.dart @@ -1,5 +1,7 @@ import 'package:json_annotation/json_annotation.dart'; +import 'user.dart'; + part 'connect_user_details_request.g.dart'; @JsonSerializable(createFactory: false) @@ -13,6 +15,24 @@ class ConnectUserDetailsRequest { this.custom, }); + /// Creates the details a client may send when connecting as [user]. + /// + /// A user's role and teams are left out: the server assigns both and does not + /// accept them from a client. + /// + /// Pass [includeDetails] as `false` to send the id alone. + factory ConnectUserDetailsRequest.fromUser( + User user, { + bool includeDetails = true, + }) { + return ConnectUserDetailsRequest( + id: user.id, + name: includeDetails ? user.originalName : null, + image: includeDetails ? user.image : null, + custom: includeDetails ? user.custom : null, + ); + } + final String id; final String? image; final bool? invisible; diff --git a/packages/stream_core/lib/src/utils/result.dart b/packages/stream_core/lib/src/utils/result.dart index 07c66ca7..0032e597 100644 --- a/packages/stream_core/lib/src/utils/result.dart +++ b/packages/stream_core/lib/src/utils/result.dart @@ -58,8 +58,8 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or `null` /// if it is [Failure]. /// - /// This function is a shorthand for `getOrElse(() => null)` or - /// `fold(onSuccess: (it) => it, onFailure: (_) => null)`. + /// This function is a shorthand for + /// `fold(onSuccess: (it) => it, onFailure: (_, _) => null)`. T? getOrNull() { return switch (this) { Success(:final data) => data, @@ -90,7 +90,7 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or throws the encapsulated error /// if it is [Failure]. /// - /// This function is a shorthand for `getOrElse((error) => throw error)`. + /// This function is a shorthand for `getOrElse((error, _) => throw error)`. T getOrThrow() { return switch (this) { Success(:final data) => data, @@ -107,9 +107,13 @@ extension PatternMatching on Result { /// Note, that this function rethrows any error thrown by [onFailure] function. /// /// This function is a shorthand for `fold(onSuccess: (it) => it, onFailure: onFailure)`. - R getOrElse(R Function(Object error, StackTrace? stackTrace) onFailure) { + /// + /// [onFailure] returns this result's own type. To fall back to a supertype, + /// widen the result first — `Result widened = intResult` — or use [fold], + /// which takes its return type from both branches. + T getOrElse(T Function(Object error, StackTrace? stackTrace) onFailure) { return switch (this) { - Success(:final data) => data as R, + Success(:final data) => data, Failure(:final error, :final stackTrace) => onFailure(error, stackTrace), }; } @@ -117,10 +121,13 @@ extension PatternMatching on Result { /// Returns the encapsulated value if this instance represents [Success] or the /// [defaultValue] if it is [Failure]. /// - /// This function is a shorthand for `getOrElse((_) => defaultValue)`. - R getOrDefault(R defaultValue) { + /// This function is a shorthand for `getOrElse((_, _) => defaultValue)`. + /// + /// [defaultValue] is of this result's own type; widen the result to fall back + /// to a supertype. + T getOrDefault(T defaultValue) { return switch (this) { - Success(:final data) => data as R, + Success(:final data) => data, Failure() => defaultValue, }; } @@ -180,11 +187,14 @@ extension PatternMatching on Result { /// /// Note, that this function rethrows any error thrown by [transform] function. /// See [recoverCatching] for an alternative that encapsulates errors. - Result recover( - R Function(Object error, StackTrace? stackTrace) transform, + /// + /// [transform] returns this result's own type, so widening is done on the way + /// in rather than on the way out: widen the result first, then recover. + Result recover( + T Function(Object error, StackTrace? stackTrace) transform, ) { return switch (this) { - Success(:final data) => Result.success(data as R), + Success() => this, Failure(:final error, :final stackTrace) => Result.success( transform(error, stackTrace), ), @@ -196,11 +206,14 @@ extension PatternMatching on Result { /// /// This function catches any error thrown by [transform] function and encapsulates it as a failure. /// See [recover] for an alternative that rethrows errors. - Result recoverCatching( - R Function(Object error, StackTrace? stackTrace) transform, + /// + /// [transform] returns this result's own type, so widening is done on the way + /// in rather than on the way out: widen the result first, then recover. + Result recoverCatching( + T Function(Object error, StackTrace? stackTrace) transform, ) { return switch (this) { - Success(:final data) => Result.success(data as R), + Success() => this, Failure(:final error, :final stackTrace) => runSafelySync( () => transform(error, stackTrace), ), diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart index 7abf7513..d44a0c00 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_engine.dart @@ -195,8 +195,6 @@ class WebSocketEngineException extends Equatable implements Exception { return null; } - static const stopErrorCode = 1000; - @override List get props => [reason, code, error]; } diff --git a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart index 0d2f86c0..e2bd5be9 100644 --- a/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart +++ b/packages/stream_core/lib/src/ws/client/engine/web_socket_options.dart @@ -22,7 +22,7 @@ class WebSocketOptions { /// Creates a new instance of [WebSocketOptions]. const WebSocketOptions({ required this.url, - this.connectTimeout, + this.connectTimeout = defaultConnectTimeout, this.protocols, this.queryParameters, }); @@ -35,9 +35,17 @@ class WebSocketOptions { /// Maximum time allowed for establishing the WebSocket connection. /// - /// When specified, the connection attempt will timeout if not completed - /// within this duration. If `null`, uses the platform default timeout. - final Duration? connectTimeout; + /// Covers the whole attempt, not just opening the socket: a connection that + /// opens but is never established is abandoned once this elapses. + /// + /// Defaults to [defaultConnectTimeout]. + final Duration connectTimeout; + + /// The [connectTimeout] used when none is given. + /// + /// Matches the wait the Swift SDK allows for the same handshake; the Android + /// one is stricter at ten seconds. + static const defaultConnectTimeout = Duration(seconds: 30); /// WebSocket sub-protocols to negotiate during the handshake. /// diff --git a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart index 55c4a0af..48df9c97 100644 --- a/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart +++ b/packages/stream_core/lib/src/ws/client/reconnect/connection_recovery_handler.dart @@ -18,6 +18,10 @@ import 'retry_strategy.dart'; /// when reconnection should occur, implementing exponential backoff with jitter for optimal /// retry behavior. /// +/// It recovers connections that existed: until [StreamWebSocketClient.connect] has established +/// one, connecting belongs to whoever called it and was handed the failure. A first attempt that +/// fails is therefore not retried here, including when the network returns. +/// /// ## Built-in Policies /// /// The handler automatically includes several reconnection policies: @@ -79,6 +83,11 @@ class ConnectionRecoveryHandler extends Disposable { late final _subscriptions = CompositeSubscription(); + // Whether a connection has been established since the caller last asked for + // one: it is what separates a drop this handler recovers from an attempt whose + // outcome the caller is waiting on. + var _hasConnected = false; + /// Attempts reconnection if policies allow it. /// /// Evaluates all configured policies and initiates reconnection when conditions are met. @@ -117,7 +126,15 @@ class ConnectionRecoveryHandler extends Disposable { _reconnectionTimer = null; } - bool _canBeReconnected() => _policies.every((it) => it.canBeReconnected()); + bool _canBeReconnected() { + // Until a connection has existed, connecting belongs to whoever called + // `connect` and was handed the failure. Retrying here as well would work + // behind a caller already told the attempt failed — and would race the + // retry that caller makes in response. + if (!_hasConnected) return false; + + return _policies.every((it) => it.canBeReconnected()); + } bool _canBeDisconnected() { return switch (_client.connectionState.value) { @@ -147,13 +164,29 @@ class ConnectionRecoveryHandler extends Disposable { void _onConnectionStateChanged(WebSocketConnectionState state) { return switch (state) { Connecting() => _cancelReconnection(), - Connected() => _reconnectStrategy.resetConsecutiveFailures(), - Disconnected() => _scheduleReconnectionIfNeeded(), + Connected() => _onConnectionEstablished(), + Disconnected(:final source) => _onConnectionLost(source), // These states do not require any action. Initialized() || Authenticating() || Disconnecting() => () {}, }; } + // A connection exists, so keeping it is this handler's job from here, and the + // failures the backoff had accumulated are behind us. + void _onConnectionEstablished() { + _hasConnected = true; + return _reconnectStrategy.resetConsecutiveFailures(); + } + + // A disconnect the caller asked for hands connecting back to them, so the next + // `connect` is a fresh attempt they await rather than a drop to recover. A + // system-initiated one is the opposite: backgrounding and network loss are what + // this handler exists to come back from. + void _onConnectionLost(DisconnectionSource source) { + if (source is UserInitiated) _hasConnected = false; + return _scheduleReconnectionIfNeeded(); + } + @override Future dispose() async { _cancelReconnection(); diff --git a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart index 3e6c4ec4..4c1d5ff4 100644 --- a/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart +++ b/packages/stream_core/lib/src/ws/client/stream_web_socket_client.dart @@ -18,6 +18,32 @@ WsRequest _defaultPingRequestBuilder([HealthCheckInfo? info]) { return HealthCheckPingEvent(connectionId: info?.connectionId); } +/// A function that builds the options for a connection attempt. +/// +/// Called once per attempt, so the options may carry values that change over the +/// client's lifetime. +/// +/// Returns the [WebSocketOptions] to open the connection with. +typedef WebSocketOptionsBuilder = WebSocketOptions Function(); + +/// A function that sends a request over a connection that is not usable yet. +/// +/// Handed to a [WebSocketAuthenticator], which runs while the connection is +/// still being established and so cannot be given the client itself. +/// +/// Returns a [Result] indicating whether the request was sent. +typedef WsSender = Result Function(WsRequest request); + +/// A function that authenticates a newly opened connection. +/// +/// Called once the socket is open, while the state is [Authenticating]. Sending +/// the credentials the server expects is this function's job. +/// +/// Returns a [Result] that fails when the credentials could not be sent, in +/// which case the connection is closed with [AuthenticationFailed] rather than +/// left waiting for a reply that never comes. +typedef WebSocketAuthenticator = Future> Function(WsSender send); + /// A WebSocket client with connection management and event handling. /// /// The primary interface for WebSocket connections in the Stream Core SDK that provides @@ -31,20 +57,18 @@ WsRequest _defaultPingRequestBuilder([HealthCheckInfo? info]) { /// ## Example /// ```dart /// final client = StreamWebSocketClient( -/// options: WebSocketOptions(url: 'wss://api.example.com'), +/// optionsBuilder: () => WebSocketOptions(url: 'wss://api.example.com'), /// messageCodec: JsonMessageCodec(), -/// onConnectionEstablished: () { -/// client.send(AuthRequest(token: authToken)); -/// }, +/// onAuthenticate: (send) async => send(AuthRequest(token: authToken)), /// ); /// /// await client.connect(); /// ``` -class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineListener { +class StreamWebSocketClient with Disposable implements WebSocketHealthListener, WebSocketEngineListener { /// Creates a new instance of [StreamWebSocketClient]. StreamWebSocketClient({ - required this.options, - this.onConnectionEstablished, + required this.optionsBuilder, + this.onAuthenticate, WebSocketProvider? wsProvider, this.pingRequestBuilder = _defaultPingRequestBuilder, required WebSocketMessageCodec messageCodec, @@ -58,18 +82,34 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL ); } - /// The WebSocket connection options including URL and configuration. - final WebSocketOptions options; + /// The function used to build the connection options for each attempt. + final WebSocketOptionsBuilder optionsBuilder; /// The function used to build ping requests for health checks. final PingRequestBuilder pingRequestBuilder; - /// Called when the WebSocket connection is established and ready for authentication. - final void Function()? onConnectionEstablished; + /// The function used to authenticate a newly opened connection. + final WebSocketAuthenticator? onAuthenticate; late final StreamWebSocketEngine _engine; late final _healthMonitor = WebSocketHealthMonitor(listener: this); + // Bounds a connection attempt that never reaches 'connected'. + Timer? _connectTimeoutTimer; + + void _startConnectTimeout(Duration timeout) { + _connectTimeoutTimer?.cancel(); + _connectTimeoutTimer = Timer(timeout, () { + const source = DisconnectionSource.connectTimeout(); + unawaited(disconnect(source: source)); + }); + } + + void _cancelConnectTimeout() { + _connectTimeoutTimer?.cancel(); + _connectTimeoutTimer = null; + } + /// The event emitter for WebSocket events. /// /// Use this to listen to incoming WebSocket events with type-safe event handling. @@ -80,11 +120,12 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL /// /// Emits state changes as the WebSocket transitions through different connection states. ConnectionStateEmitter get connectionState => _connectionStateEmitter; - late final _connectionStateEmitter = MutableConnectionStateEmitter( - const WebSocketConnectionState.initialized(), - ); + late final _connectionStateEmitter = MutableConnectionStateEmitter(const .initialized()); set _connectionState(WebSocketConnectionState connectionState) { + // Return early if the emitter is closed. + if (_connectionStateEmitter.isClosed) return; + // Return early if the state hasn't changed. if (_connectionStateEmitter.value == connectionState) return; @@ -103,20 +144,35 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL /// Establishes a WebSocket connection. /// /// The connection state can be monitored through [connectionState] for real-time updates. - /// If the connection is already established or in progress, this method returns immediately. + /// If the connection is already established or in progress, this method returns immediately, + /// as it does while a previous connection is still closing, and once [dispose] has been called. /// - /// Returns a [Future] that completes when the connection attempt finishes. + /// Returns a [Future] that completes once the socket is open — before the + /// connection is authenticated, and well before it is [Connected]. Watch + /// [connectionState] to know when it is usable. Future connect() async { + assert(!isDisposed, 'Cannot connect a disposed StreamWebSocketClient'); + if (isDisposed) return; + // If the connection is already established or in the process of connecting, // do not initiate a new connection. if (connectionState.value is Connecting) return; if (connectionState.value is Authenticating) return; if (connectionState.value is Connected) return; + // Nor while a previous connection is still closing: the socket it opened + // would be brought down by the old one's close event, which would also + // disarm the new attempt's timeout and leave it authenticating unwatched. + if (connectionState.value is Disconnecting) return; + // Update the connection state to 'connecting'. _connectionState = const WebSocketConnectionState.connecting(); - // Open the connection using the engine. + // Open the connection using the engine, with options built for this attempt. + final options = optionsBuilder.call(); + + // Time the whole handshake: nothing else watches 'authenticating'. + _startConnectTimeout(options.connectTimeout); final result = await _engine.open(options); // If some failure occurs, disconnect and rethrow the error. @@ -133,14 +189,45 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL CloseCode closeCode = CloseCode.normalClosure, DisconnectionSource source = const UserInitiated(), }) async { - // If the connection is already disconnected, do nothing. - if (connectionState.value is Disconnected) return; + // A connection already going down keeps the source it started going down + // with: whoever asked first described why. Without this the connect + // timeout could replace a `ServerInitiated` closure, which is + // reconnectable, with one that is not. + if (connectionState.value case Disconnected() || Disconnecting()) return; + + // Stop the timeout from firing later and replacing this source. + _cancelConnectTimeout(); // Update the connection state to 'disconnecting'. _connectionState = WebSocketConnectionState.disconnecting(source: source); - // Close the connection using the engine. - unawaited(_engine.close(closeCode, source.closeReason)); + // Awaited so the connection is closed, rather than merely closing, once this + // returns: a reconnect straight afterwards would otherwise race the close + // and see the connection go down again. + final result = await _engine.close(closeCode, source.closeReason); + + // The engine reports a failed close rather than throwing, and does not + // notify its listener on that path. The connection is unusable either way, + // so report it closed rather than leave it disconnecting for good. + if (result.isFailure) onClose(closeCode, source.closeReason); + } + + /// Releases every resource held by this client. + /// + /// Closes the connection along with [events] and [connectionState], after which + /// this client cannot be connected again. Use [disconnect] for a connection that + /// may be opened again. + /// + /// Returns a [Future] that completes once everything has been released. + @override + Future dispose() async { + await disconnect(); + _healthMonitor.stop(); + + await _events.close(); + await _connectionStateEmitter.close(); + + return super.dispose(); } @override @@ -148,13 +235,32 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL // Update the connection state to 'authenticating'. _connectionState = const WebSocketConnectionState.authenticating(); - // Notify that the connection has been established and we are ready - // to authenticate. - onConnectionEstablished?.call(); + // The socket is open, so authenticate before the connection is usable. + unawaited(_authenticate()); + } + + Future _authenticate() async { + final authenticate = onAuthenticate; + if (authenticate == null) return; + + // Guarded rather than awaited directly: an authenticator that awaits a + // token throws rather than returning a failure, and nothing observes this + // future, so the error would escape and leave the connection + // authenticating until the timeout reported a cause it does not know. + final outcome = await runSafely(() => authenticate(send)); + final result = outcome.flatten(); + + // Close the connection rather than wait for a reply that cannot come. + if (result.exceptionOrNull() case final error?) { + final source = DisconnectionSource.authenticationFailed(error: error); + return disconnect(source: source); + } } @override void onClose([int? closeCode, String? closeReason]) { + _cancelConnectTimeout(); + final source = switch (connectionState.value) { // If we were already disconnecting, keep the caller-provided source. Disconnecting(:final source) => source, @@ -217,6 +323,11 @@ class StreamWebSocketClient implements WebSocketHealthListener, WebSocketEngineL void _handleHealthCheckEvent(WsEvent event, HealthCheckInfo info) { print('WebSocketClient: Health check pong received: $info'); + // Ignore a pong that arrives once the connection is on its way down. + if (connectionState.value case Disconnecting() || Disconnected()) return; + + _cancelConnectTimeout(); + // Update the connection state with health check info. _connectionState = WebSocketConnectionState.connected(healthCheck: info); diff --git a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart index 87a6cac2..d18e8cf9 100644 --- a/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart +++ b/packages/stream_core/lib/src/ws/client/web_socket_connection_state.dart @@ -98,27 +98,38 @@ sealed class WebSocketConnectionState extends Equatable { /// /// ## Reconnection is disabled for: /// - User-initiated disconnections (explicit disconnect calls) - /// - Server errors with code 1000 (normal closure) - /// - Token expired/invalid errors - /// - Client errors (4xx status codes) + /// - A socket the server closed deliberately (close code 1000) + /// - Tokens another token would not fix, and a wrong API key + /// - Client errors (4xx status codes), other than a rate limit + /// - A failure to load or send credentials /// + /// Whether the server closed this connection because the token had expired. + /// + /// Reported so a caller that can replace the token knows to, since a + /// reconnection cannot: [isAutomaticReconnectionEnabled] refuses this, because + /// whoever retries it here would present the token that was just refused. + bool get isExpiredTokenDisconnection => switch (this) { + Disconnected(source: ServerInitiated(:final error)) => error?.apiError?.isTokenExpiredError ?? false, + _ => false, + }; + /// Returns `true` if automatic reconnection should be attempted. - bool get isAutomaticReconnectionEnabled { - return switch (this) { - Disconnected(:final source) => switch (source) { - ServerInitiated() => switch (source.error?.apiError) { - final error? when error.code == 1000 => false, - final error? when error.isTokenExpiredError => false, - final error? when error.isClientError => false, - _ => true, // Reconnect on other server initiated disconnections - }, - UnHealthyConnection() => true, - SystemInitiated() => true, - UserInitiated() => false, + bool get isAutomaticReconnectionEnabled => switch (this) { + Disconnected(:final source) => switch (source) { + ServerInitiated(:final error) when error?.code == CloseCode.normalClosure => false, + ServerInitiated(:final error) => switch (error?.apiError) { + final it? when it.isInvalidTokenError => false, + final it? when it.isClientError && !it.isRateLimitError => false, + _ => true, // Reconnect on other server initiated disconnections }, - _ => false, // No automatic reconnection for other states - }; - } + UnHealthyConnection() => true, + SystemInitiated() => true, + ConnectTimeout() => true, + UserInitiated() => false, + AuthenticationFailed() => false, + }, + _ => false, // No automatic reconnection for other states + }; @override List get props => []; @@ -252,6 +263,18 @@ sealed class DisconnectionSource extends Equatable { /// typically when ping requests do not receive pong responses. const factory DisconnectionSource.unHealthyConnection() = UnHealthyConnection; + /// Creates a [ConnectTimeout] disconnection source. + /// + /// Indicates that the connection never became usable within the allotted + /// time, so it was abandoned before it was ever established. + const factory DisconnectionSource.connectTimeout() = ConnectTimeout; + + /// Creates an [AuthenticationFailed] disconnection source. + /// + /// Indicates that the connection opened but could not be authenticated, so it + /// was closed without ever being usable. + const factory DisconnectionSource.authenticationFailed({Object? error}) = AuthenticationFailed; + /// A human-readable description of the disconnection source. /// /// Provides a descriptive string that explains why the connection was closed. @@ -264,6 +287,8 @@ sealed class DisconnectionSource extends Equatable { ServerInitiated() => 'Server initiated disconnection', SystemInitiated() => 'System initiated disconnection', UnHealthyConnection() => 'Unhealthy connection (no pong received)', + ConnectTimeout() => 'Timed out before the connection was established', + AuthenticationFailed() => 'Authentication failed', }; } @@ -320,3 +345,27 @@ final class UnHealthyConnection extends DisconnectionSource { /// Creates an [UnHealthyConnection] disconnection source. const UnHealthyConnection(); } + +/// A disconnection caused by the connection not becoming usable in time. +/// +/// This source indicates that the connection was abandoned while it was still +/// being established, so it was never usable. +final class ConnectTimeout extends DisconnectionSource { + /// Creates a [ConnectTimeout] disconnection source. + const ConnectTimeout(); +} + +/// A disconnection caused by the connection failing to authenticate. +/// +/// This source indicates that the socket opened but authentication did not +/// complete, so the connection was never usable. +final class AuthenticationFailed extends DisconnectionSource { + /// Creates an [AuthenticationFailed] disconnection source. + const AuthenticationFailed({this.error}); + + /// The error that prevented the connection from authenticating. + final Object? error; + + @override + List get props => [error]; +} diff --git a/packages/stream_core/test/user/connect_user_details_request_test.dart b/packages/stream_core/test/user/connect_user_details_request_test.dart new file mode 100644 index 00000000..2ff4858a --- /dev/null +++ b/packages/stream_core/test/user/connect_user_details_request_test.dart @@ -0,0 +1,59 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +void main() { + group('ConnectUserDetailsRequest.fromUser', () { + test('carries the fields the server accepts from a client', () { + const user = User( + id: 'user-1', + name: 'Bob', + image: 'https://example.com/bob.png', + custom: {'plan': 'pro'}, + ); + + final details = ConnectUserDetailsRequest.fromUser(user); + + expect(details.id, 'user-1'); + expect(details.name, 'Bob'); + expect(details.image, 'https://example.com/bob.png'); + expect(details.custom, {'plan': 'pro'}); + }); + + test('leaves out the fields the server decides itself', () { + const user = User(id: 'user-1', role: 'admin', teams: ['red']); + + final json = ConnectUserDetailsRequest.fromUser(user).toJson(); + + // Sending either is pointless: the server ignores both from a client. + expect(json, isNot(contains('role'))); + expect(json, isNot(contains('teams'))); + }); + + test('sends the id alone when details are excluded', () { + const user = User( + id: 'user-1', + name: 'Bob', + image: 'https://example.com/bob.png', + custom: {'plan': 'pro'}, + ); + + final details = ConnectUserDetailsRequest.fromUser(user, includeDetails: false); + + expect(details.id, 'user-1'); + expect(details.name, isNull); + expect(details.image, isNull); + expect(details.custom, isNull); + }); + + test('reports the name the user was created with, not the id fallback', () { + // `User.name` falls back to the id; the wire form must not, or a user + // with no name would be given the id as one. + const user = User(id: 'user-1'); + + final details = ConnectUserDetailsRequest.fromUser(user); + + expect(user.name, 'user-1'); + expect(details.name, isNull); + }); + }); +} diff --git a/packages/stream_core/test/utils/result_test.dart b/packages/stream_core/test/utils/result_test.dart new file mode 100644 index 00000000..92234e36 --- /dev/null +++ b/packages/stream_core/test/utils/result_test.dart @@ -0,0 +1,123 @@ +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; + +void main() { + group('Result.getOrElse', () { + test('returns the value when the fallback only throws', () { + const result = Result.success(42); + + // The value used to be cast to the fallback's return type, here `Never`. + final value = result.getOrElse((_, _) => throw StateError('unreachable')); + + expect(value, 42); + }); + + test('throws what the fallback throws, so an error can be reworded', () { + final result = Result.failure(Exception('original')); + + expect( + () => result.getOrElse((error, _) => throw StateError('$error')), + throwsA(isA()), + ); + }); + + test('returns what the fallback returns', () { + final result = Result.failure(Exception('original')); + + expect(result.getOrElse((_, _) => 0), 0); + }); + + test('hands the fallback the error and its stack trace', () { + final stackTrace = StackTrace.current; + final error = Exception('original'); + final result = Result.failure(error, stackTrace); + + Object? seenError; + StackTrace? seenStackTrace; + result.getOrElse((error, stackTrace) { + seenError = error; + seenStackTrace = stackTrace; + return 0; + }); + + expect(seenError, error); + expect(seenStackTrace, stackTrace); + }); + }); + + group('Result widening', () { + test('falls back to a supertype when the result is widened', () { + // Kotlin widens through a `` bound Dart has no equivalent for. + // Naming the wider type on the result gets there instead, since `Result` + // is covariant. + final Result widened = Result.failure(Exception('failed')); + + expect(widened.getOrElse((_, _) => 0.5), 0.5); + expect(widened.getOrDefault(0.5), 0.5); + expect(widened.recover((_, _) => 0.5).getOrNull(), 0.5); + }); + }); + + group('Result.getOrDefault', () { + test('returns the value when there is one', () { + const result = Result.success(42); + + expect(result.getOrDefault(0), 42); + }); + + test('returns the default when there is not', () { + final result = Result.failure(Exception('failed')); + + expect(result.getOrDefault(0), 0); + }); + }); + + group('Result.recover', () { + test('keeps the value when the transform only throws', () { + const result = Result.success(42); + + // Same cast as `getOrElse`. + final recovered = result.recover((_, _) => throw StateError('unreachable')); + + expect(recovered.getOrNull(), 42); + }); + + test('turns a failure into the value the transform returns', () { + final result = Result.failure(Exception('failed')); + + expect(result.recover((_, _) => 0).getOrNull(), 0); + }); + + test('rethrows an error from the transform', () { + final result = Result.failure(Exception('failed')); + + expect( + () => result.recover((_, _) => throw StateError('while recovering')), + throwsA(isA()), + ); + }); + }); + + group('Result.recoverCatching', () { + test('keeps the value when the transform only throws', () { + const result = Result.success(42); + + final recovered = result.recoverCatching( + (_, _) => throw StateError('unreachable'), + ); + + expect(recovered.getOrNull(), 42); + }); + + test('reports an error from the transform as a failure', () { + final result = Result.failure(Exception('failed')); + + final recovered = result.recoverCatching( + (_, _) => throw StateError('while recovering'), + ); + + // Unlike `recover`, the error replaces the original rather than escaping. + expect(recovered.exceptionOrNull(), isA()); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart new file mode 100644 index 00000000..234b4846 --- /dev/null +++ b/packages/stream_core/test/ws/client/reconnect/connection_recovery_handler_test.dart @@ -0,0 +1,207 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; + +class _MockWebSocketChannel extends Mock implements WebSocketChannel {} + +class _MockWebSocketSink extends Mock implements WebSocketSink {} + +/// A stream whose subscription cancels without going through the event loop, so +/// a close under `fakeAsync` runs to completion as it does in production. +class _CancellableStream extends Stream { + _CancellableStream(this._source); + + final Stream _source; + + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _CancellableSubscription( + _source.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError), + ); + } +} + +class _CancellableSubscription implements StreamSubscription { + _CancellableSubscription(this._delegate); + + final StreamSubscription _delegate; + + @override + Future cancel() { + _delegate.cancel().ignore(); + return Future.value(); + } + + @override + void onData(void Function(T data)? handleData) => _delegate.onData(handleData); + + @override + void onError(Function? handleError) => _delegate.onError(handleError); + + @override + void onDone(void Function()? handleDone) => _delegate.onDone(handleDone); + + @override + void pause([Future? resumeSignal]) => _delegate.pause(resumeSignal); + + @override + void resume() => _delegate.resume(); + + @override + bool get isPaused => _delegate.isPaused; + + @override + Future asFuture([E? futureValue]) => _delegate.asFuture(futureValue); +} + +class _NoopCodec implements WebSocketMessageCodec { + const _NoopCodec(); + + @override + Object encode(WsRequest message) => ''; + + @override + WsEvent decode(Object message) => const _HealthCheckEvent(); +} + +final class _HealthCheckEvent extends WsEvent { + const _HealthCheckEvent(); + + @override + HealthCheckInfo? get healthCheckInfo => const HealthCheckInfo(connectionId: 'connection-id'); +} + +final class _PingRequest extends WsRequest { + const _PingRequest(); + + @override + Map toJson() => const {}; + + @override + List get props => const []; +} + +/// A client whose socket opens but answers nothing, with a handler attached and +/// a count of the attempts it has made. +({StreamWebSocketClient client, int Function() attempts}) _client() { + final incoming = StreamController.broadcast(); + addTearDown(incoming.close); + + final channel = _MockWebSocketChannel(); + when(() => channel.ready).thenAnswer((_) async {}); + when(() => channel.stream).thenAnswer((_) => _CancellableStream(incoming.stream)); + final sink = _MockWebSocketSink(); + when(() => channel.sink).thenReturn(sink); + when(() => sink.close(any(), any())).thenAnswer((_) async {}); + + var attempts = 0; + final client = StreamWebSocketClient( + optionsBuilder: () { + attempts++; + return const WebSocketOptions(url: 'wss://example.com'); + }, + wsProvider: (_) => channel, + pingRequestBuilder: ([_]) => const _PingRequest(), + messageCodec: const _NoopCodec(), + ); + + final handler = ConnectionRecoveryHandler(client: client); + addTearDown(handler.dispose); + + return (client: client, attempts: () => attempts); +} + +void main() { + group('ConnectionRecoveryHandler', () { + test('does not retry a first attempt that never connected', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // The socket opened but the server never answers, so the attempt is + // abandoned — the failure the caller of `connect` is handed. + async.elapse(WebSocketOptions.defaultConnectTimeout); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + // Retrying here would work behind a caller already told it failed, and + // would race the retry that caller makes in response. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 1); + }); + }); + + test('retries a connection that dropped after being established', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + // The connection stops answering health checks — a drop rather than a + // failure to connect, so recovering it is this handler's job. The first + // retry carries no delay, so it is under way by the time this returns. + async.elapse(const Duration(seconds: 29)); + async.flushMicrotasks(); + + expect(attempts(), 2); + expect(client.connectionState.value, isA()); + }); + }); + + test('hands connecting back after the caller disconnected', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + client.disconnect().ignore(); + async.flushMicrotasks(); + + // A fresh attempt, awaited by whoever made it, that never connects. + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + async.flushMicrotasks(); + + // Having connected in the previous session does not make this failure + // the handler's to retry. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 2); + }); + }); + + test('does not retry a disconnect the caller asked for', () { + fakeAsync((async) { + final (:client, :attempts) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + + client.disconnect().ignore(); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + // Having been connected is not enough on its own: the source still says + // this was deliberate. + async.elapse(const Duration(minutes: 1)); + expect(attempts(), 1); + }); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart new file mode 100644 index 00000000..3b073e3c --- /dev/null +++ b/packages/stream_core/test/ws/client/stream_web_socket_client_test.dart @@ -0,0 +1,603 @@ +import 'dart:async'; + +import 'package:fake_async/fake_async.dart'; +import 'package:mocktail/mocktail.dart'; +import 'package:stream_core/stream_core.dart'; +import 'package:test/test.dart'; +import 'package:web_socket_channel/web_socket_channel.dart'; + +class _MockWebSocketChannel extends Mock implements WebSocketChannel {} + +class _MockWebSocketSink extends Mock implements WebSocketSink {} + +/// A codec that is never exercised: these tests drive the client through its +/// engine listener callbacks rather than through encoded frames. +class _NoopCodec implements WebSocketMessageCodec { + const _NoopCodec(); + + @override + Object encode(WsRequest message) => ''; + + @override + WsEvent decode(Object message) => const _HealthCheckEvent(); +} + +final class _HealthCheckEvent extends WsEvent { + const _HealthCheckEvent({this.connectionId = 'connection-id'}); + + final String? connectionId; + + @override + HealthCheckInfo? get healthCheckInfo { + return HealthCheckInfo(connectionId: connectionId); + } +} + +final class _PingRequest extends WsRequest { + const _PingRequest(); + + @override + Map toJson() => const {}; + + @override + List get props => const []; +} + +/// A stream whose subscription cancels without going through the event loop. +/// +/// `StreamController`'s cancel completes on the event loop, which `fakeAsync` +/// never drives, so a client awaiting one hangs in a test where it would not in +/// production — leaving the close half finished. +class _CancellableStream extends Stream { + _CancellableStream(this._source); + + final Stream _source; + + @override + StreamSubscription listen( + void Function(T event)? onData, { + Function? onError, + void Function()? onDone, + bool? cancelOnError, + }) { + return _CancellableSubscription( + _source.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError), + ); + } +} + +class _CancellableSubscription implements StreamSubscription { + _CancellableSubscription(this._delegate); + + final StreamSubscription _delegate; + + @override + Future cancel() { + _delegate.cancel().ignore(); + return Future.value(); + } + + @override + void onData(void Function(T data)? handleData) => _delegate.onData(handleData); + + @override + void onError(Function? handleError) => _delegate.onError(handleError); + + @override + void onDone(void Function()? handleDone) => _delegate.onDone(handleDone); + + @override + void pause([Future? resumeSignal]) => _delegate.pause(resumeSignal); + + @override + void resume() => _delegate.resume(); + + @override + bool get isPaused => _delegate.isPaused; + + @override + Future asFuture([E? futureValue]) => _delegate.asFuture(futureValue); +} + +/// Builds a client whose socket opens successfully but sends nothing, so the +/// handshake only progresses when a test drives it. +({ + StreamWebSocketClient client, + StreamController incoming, + int Function() optionsBuilt, + WebSocketSink sink, +}) +_client({ + Duration connectTimeout = WebSocketOptions.defaultConnectTimeout, + WebSocketAuthenticator? onAuthenticate, +}) { + final incoming = StreamController.broadcast(); + addTearDown(incoming.close); + + final channel = _MockWebSocketChannel(); + when(() => channel.ready).thenAnswer((_) async {}); + when(() => channel.stream).thenAnswer((_) => _CancellableStream(incoming.stream)); + final sink = _MockWebSocketSink(); + when(() => channel.sink).thenReturn(sink); + // A socket that closes cleanly; tests that need otherwise re-stub this. + when(() => sink.close(any(), any())).thenAnswer((_) async {}); + + var built = 0; + final client = StreamWebSocketClient( + optionsBuilder: () { + built++; + return WebSocketOptions( + url: 'wss://example.com', + connectTimeout: connectTimeout, + ); + }, + onAuthenticate: onAuthenticate, + wsProvider: (_) => channel, + pingRequestBuilder: ([_]) => const _PingRequest(), + messageCodec: const _NoopCodec(), + ); + + return (client: client, incoming: incoming, optionsBuilt: () => built, sink: sink); +} + +void main() { + group('StreamWebSocketClient.disconnect', () { + test('leaves the connection closed, not closing, once it returns', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + + await client.disconnect(); + + // A caller that reconnects straight away would otherwise race the close + // and see the connection go down again. + expect(client.connectionState.value, isA()); + }); + + test('reports the connection closed even when the socket close fails', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + when(() => sink.close(any(), any())).thenAnswer((_) => Future.error(Exception('close failed'))); + await client.connect(); + + await client.disconnect(); + + // The engine swallows the failure and never notifies its listener, which + // used to leave the connection disconnecting for good. + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + + test('does not open a socket while the previous one is still closing', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + // Held open so the connection is still closing when connect is called. + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); + await client.connect(); + + client.disconnect().ignore(); + expect(client.connectionState.value, isA()); + await client.connect(); + + // The old socket's close event would otherwise bring the new connection + // down and disarm the timeout meant to be watching it. + expect(optionsBuilt(), 1); + expect(client.connectionState.value, isA()); + closing.complete(); + }); + + test('can be followed by another connect', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + await client.connect(); + await client.disconnect(); + + await client.connect(); + + expect(client.connectionState.value, isA()); + expect(optionsBuilt(), 2); + }); + }); + + group('StreamWebSocketClient.dispose', () { + test('closes the connection and both emitters', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + + await client.dispose(); + + expect(client.isDisposed, isTrue); + expect(client.events.isClosed, isTrue); + expect(client.connectionState.isClosed, isTrue); + }); + + test('does nothing when called again', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + await client.dispose(); + + await expectLater(client.dispose(), completes); + }); + + test('refuses to connect again', () async { + final (:client, :sink, incoming: _, :optionsBuilt) = _client(); + await client.connect(); + await client.dispose(); + + // Asserts rather than throws: a reconnect can come from the recovery + // handler, which does not await it and cannot report an error. In a + // release build the assert is gone and the call is a no-op, which is what + // the untouched builder count pins. + await expectLater(client.connect(), throwsA(isA())); + expect(optionsBuilt(), 1); + }); + + test('ignores a socket event arriving after it', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + await client.connect(); + await client.dispose(); + + // The state emitter is closed, so a late event must not be reported into + // it rather than throwing. + expect(() => client.onClose(1000, 'late'), returnsNormally); + }); + }); + + group('StreamWebSocketClient.optionsBuilder', () { + test('is called for every connection attempt, not once per client', () async { + final (:client, :incoming, :optionsBuilt, sink: _) = _client(); + + await client.connect(); + expect(optionsBuilt(), 1); + + await client.disconnect(); + + await client.connect(); + expect(optionsBuilt(), 2); + }); + }); + + group('StreamWebSocketClient.onAuthenticate', () { + test('is called once the socket is open, while authenticating', () async { + WebSocketConnectionState? stateWhenCalled; + late StreamWebSocketClient client; + final built = _client( + onAuthenticate: (_) async { + stateWhenCalled = client.connectionState.value; + return const Result.success(null); + }, + ); + client = built.client; + + await client.connect(); + await pumpEventQueue(); + + expect(stateWhenCalled, isA()); + }); + + test('is called once per connection attempt', () async { + var calls = 0; + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async { + calls++; + return const Result.success(null); + }, + ); + + await client.connect(); + await pumpEventQueue(); + expect(calls, 1); + + await client.disconnect(); + + await client.connect(); + await pumpEventQueue(); + expect(calls, 2); + }); + + test('is handed a sender that puts the request on the socket', () async { + Result? sent; + final (:client, incoming: _, optionsBuilt: _, :sink) = _client( + onAuthenticate: (send) async => sent = send(const _PingRequest()), + ); + + await client.connect(); + await pumpEventQueue(); + + // The sender is only useful if it reaches the socket: an authenticator + // that cannot send has nothing to report but failure. + expect(sent, isA>()); + verify(() => sink.add(any())).called(1); + }); + + test('leaves the connection authenticating when it succeeds', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => const Result.success(null), + ); + + await client.connect(); + await pumpEventQueue(); + + // Sending the credentials does not establish the connection; the server + // answering does. + expect(client.connectionState.value, isA()); + + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + }); + + test('leaves the connection authenticating when there is no authenticator', () async { + // A socket that has nothing to send before it is usable, such as one whose + // protocol authenticates elsewhere. + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + await client.connect(); + await pumpEventQueue(); + + expect(client.connectionState.value, isA()); + + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + }); + }); + + group('StreamWebSocketClient authentication failure', () { + test('closes the connection instead of waiting for a reply', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => Result.failure(StateError('no token')), + ); + + await client.connect(); + await pumpEventQueue(); + + final state = client.connectionState.value; + expect( + state, + isA().having( + (it) => it.source, + 'source', + isA().having((it) => it.error, 'error', isStateError), + ), + ); + }); + + test('closes the connection when the authenticator throws', () async { + // The natural authenticator awaits a token, and loading one throws rather + // than returning a failure. Left unguarded the error escapes unhandled and + // the connection waits for the timeout, which knows no cause. + final (:client, :sink, incoming: _, optionsBuilt: _) = _client( + onAuthenticate: (_) async => throw StateError('token load failed'), + ); + + await client.connect(); + await pumpEventQueue(); + + expect( + client.connectionState.value, + isA().having( + (it) => it.source, + 'source', + isA().having((it) => it.error, 'error', isStateError), + ), + ); + }); + + test('is not retried, since the same credentials would fail again', () async { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) async => Result.failure(StateError('no token')), + ); + + await client.connect(); + await pumpEventQueue(); + + final state = client.connectionState.value; + expect(state, isA()); + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + }); + + group('StreamWebSocketClient connect timeout', () { + test('abandons an attempt that never becomes connected', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // The socket opened, so the client is authenticating with nothing else + // watching it. + expect(client.connectionState.value, isA()); + + // Still waiting a tick before the timeout is due. + async.elapse(WebSocketOptions.defaultConnectTimeout - const Duration(seconds: 1)); + expect(client.connectionState.value, isA()); + + async.elapse(const Duration(seconds: 1)); + final state = client.connectionState.value; + expect( + state, + isA().having((it) => it.source, 'source', isA()), + ); + + // An attempt abandoned here is retried: a first health check that never + // arrives is the same failure as one that stops arriving. + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + }); + + test('abandons an attempt whose authenticator never returns', () { + fakeAsync((async) { + // The realistic hang: an authenticator awaiting something that never + // resolves. Nothing else watches 'authenticating', so only this fires. + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client( + onAuthenticate: (_) => Completer>().future, + ); + + client.connect().ignore(); + async.flushMicrotasks(); + expect(client.connectionState.value, isA()); + + async.elapse(WebSocketOptions.defaultConnectTimeout); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('is armed again for a later attempt', () { + fakeAsync((async) { + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + expect(client.connectionState.value, isA()); + + client.connect().ignore(); + async.flushMicrotasks(); + async.elapse(WebSocketOptions.defaultConnectTimeout); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('honours a timeout given in the options', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client( + connectTimeout: const Duration(seconds: 2), + ); + + client.connect().ignore(); + async.flushMicrotasks(); + + async.elapse(const Duration(seconds: 2)); + + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + + test('does not fire once the connection is established', () { + fakeAsync((async) { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + + // A live connection answers its pings, which is what keeps the health + // monitor quiet across a connection that outlives this timeout. The + // answer arrives over the wire, so it lands after the monitor has armed + // its pong timeout rather than inside the send that triggered it. + when(() => sink.add(any())).thenAnswer((_) { + Timer(const Duration(milliseconds: 50), () { + client.onMessage(const _HealthCheckEvent()); + }); + }); + + client.connect().ignore(); + async.flushMicrotasks(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + // Past the timeout, and past a ping cycle with it. + async.elapse(WebSocketOptions.defaultConnectTimeout + const Duration(seconds: 10)); + + expect(client.connectionState.value, isA()); + }); + }); + + test('does not replace the source of a socket error that came first', () { + fakeAsync((async) { + final (:client, incoming: _, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + + // A socket error closes the connection without cancelling the timer. + client.onError(StateError('socket died')); + expect(client.connectionState.value, isA()); + + async.elapse(WebSocketOptions.defaultConnectTimeout * 2); + + // The peer closes after sending the error, as the socket protocol has + // it, which is what turns the state into a disconnection. + client.onClose(); + + // Replacing this source with `ConnectTimeout` used to make a + // reconnectable failure permanent. + final state = client.connectionState.value; + expect( + state, + isA().having((it) => it.source, 'source', isA()), + ); + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + }); + + test('does not replace the source of a disconnect that came first', () { + fakeAsync((async) { + final (:client, :incoming, optionsBuilt: _, sink: _) = _client(); + + client.connect().ignore(); + async.flushMicrotasks(); + client.disconnect().ignore(); + async.flushMicrotasks(); + + async.elapse(WebSocketOptions.defaultConnectTimeout * 2); + + // The timeout would otherwise report this deliberate disconnect as a + // timed-out attempt, which is retried where this is not. + expect( + client.connectionState.value, + isA().having((it) => it.source, 'source', isA()), + ); + }); + }); + }); + + group('StreamWebSocketClient health check while disconnecting', () { + test('does not report the connection as established again', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + // Held open so the connection is still closing when the pong arrives. + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); + + await client.connect(); + client.onMessage(const _HealthCheckEvent()); + expect(client.connectionState.value, isA()); + + client.disconnect().ignore(); + expect(client.connectionState.value, isA()); + + // Arrives before the socket finished closing. + client.onMessage(const _HealthCheckEvent(connectionId: 'late')); + + expect(client.connectionState.value, isA()); + closing.complete(); + }); + + test('leaves the disconnection source intact once the socket closes', () async { + final (:client, :sink, incoming: _, optionsBuilt: _) = _client(); + final closing = Completer(); + when(() => sink.close(any(), any())).thenAnswer((_) => closing.future); + + await client.connect(); + client.onMessage(const _HealthCheckEvent()); + client.disconnect().ignore(); + client.onMessage(const _HealthCheckEvent(connectionId: 'late')); + + closing.complete(); + await pumpEventQueue(); + + // Without the guard the late health check moves the state back to + // connected, and `onClose` then reports a server-initiated disconnect, + // which is eligible for an automatic reconnect. + final state = client.connectionState.value; + expect(state, isA().having((it) => it.source, 'source', isA())); + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + }); +} diff --git a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart index c8d9a6b9..5219159d 100644 --- a/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart +++ b/packages/stream_core/test/ws/client/web_socket_connection_state_test.dart @@ -1,13 +1,13 @@ import 'package:stream_core/stream_core.dart'; import 'package:test/test.dart'; -StreamApiError _apiError(int code) => StreamApiError( +StreamApiError _apiError(int code, {int statusCode = 401}) => StreamApiError( code: code, details: const [], duration: '0ms', message: 'error $code', moreInfo: '', - statusCode: 401, + statusCode: statusCode, ); Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( @@ -23,22 +23,113 @@ Disconnected _serverDisconnect(StreamApiError apiError) => Disconnected( void main() { group('WebSocketConnectionState.isAutomaticReconnectionEnabled', () { test( - 'is disabled when the server closes with a token-expired error, so an ' - 'expired (e.g. guest) token does not trigger a silent reconnect loop', + 'is disabled when the token has expired, since a retry here would present ' + 'the same one', () { - // Token-invalid error codes are 40..42; 40 = token expired. + // 40 = expired. Replacing it is the caller's to do, and it is the caller + // that retries — `isExpiredTokenDisconnection` is how they are told. final state = _serverDisconnect(_apiError(40)); expect(state.isAutomaticReconnectionEnabled, isFalse); + expect(state.isExpiredTokenDisconnection, isTrue); }, ); - test('is enabled for a generic, retryable server-initiated disconnection', () { - // A server error that is neither a normal closure (1000), a token error - // (40..42), nor a client error (400..499) should still reconnect. - final state = _serverDisconnect(_apiError(43)); + test('is disabled when another token would be refused too', () { + // 41 not valid yet, 42 used before issued, 43 signed with the wrong + // secret, 2 wrong API key — none of which a fresh token repairs. + for (final code in [41, 42, 43, 2]) { + expect( + _serverDisconnect(_apiError(code)).isAutomaticReconnectionEnabled, + isFalse, + reason: 'code $code', + ); + } + }); + + test('is enabled when the request was rate limited', () { + // 9 = rate limited, sent as 429. The server closes with the window's reset + // in the response headers, so the condition clears without the caller + // doing anything. + final state = _serverDisconnect(_apiError(9, statusCode: 429)); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + + test('is disabled for any other client error', () { + // 17 = not allowed. Nothing about retrying changes the answer. + final state = _serverDisconnect(_apiError(17, statusCode: 403)); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + + test('is enabled for a server-side failure', () { + // Stream error codes never fall in 400..499, so this is classified by the + // status code alone — which is what the ported rule got wrong. + final state = _serverDisconnect(_apiError(9, statusCode: 500)); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + + test('is disabled when the socket was closed deliberately', () { + const state = Disconnected( + source: ServerInitiated( + error: WebSocketEngineException( + code: CloseCode.normalClosure, + ), + ), + ); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }); + + test('is enabled when the server closed without saying why', () { + const state = Disconnected(source: ServerInitiated()); + + expect(state.isAutomaticReconnectionEnabled, isTrue); + }); + + test( + 'is disabled when a connection could not be authenticated, since the ' + 'same credentials would fail again', + () { + const state = Disconnected(source: AuthenticationFailed(error: 'no token')); + + expect(state.isAutomaticReconnectionEnabled, isFalse); + }, + ); + + test('is enabled when a connected socket stops answering health checks', () { + const state = Disconnected(source: UnHealthyConnection()); expect(state.isAutomaticReconnectionEnabled, isTrue); }); }); + + group('DisconnectionSource.closeReason', () { + test('reads differently for every source', () { + const sources = [ + UserInitiated(), + ServerInitiated(), + SystemInitiated(), + UnHealthyConnection(), + ConnectTimeout(), + AuthenticationFailed(error: 'no token'), + ]; + + final reasons = sources.map((it) => it.closeReason).toSet(); + + // A shared reason would report two different outcomes identically. + expect(reasons, hasLength(sources.length)); + }); + }); + + group('WebSocketOptions.defaultConnectTimeout', () { + test('is the timeout used when the options do not say', () { + const options = WebSocketOptions(url: 'wss://example.com'); + + expect(options.connectTimeout, WebSocketOptions.defaultConnectTimeout); + expect(WebSocketOptions.defaultConnectTimeout, const Duration(seconds: 30)); + }); + }); }