diff --git a/packages/event_source_client/lib/launchdarkly_event_source_client.dart b/packages/event_source_client/lib/launchdarkly_event_source_client.dart index 802d49aa..04532bc5 100644 --- a/packages/event_source_client/lib/launchdarkly_event_source_client.dart +++ b/packages/event_source_client/lib/launchdarkly_event_source_client.dart @@ -12,7 +12,14 @@ import 'src/sse_client_stub.dart' if (dart.library.js_interop) 'src/sse_client_html.dart'; import 'src/test_sse_client.dart'; -export 'src/errors.dart' show SseHttpError; +export 'src/errors.dart' + show + SseContentTypeError, + SseHttpError, + SseStreamClosedError, + SseTimeoutError, + SseTimeoutPhase, + SseTransportError; export 'src/events.dart' show Event, MessageEvent, OpenEvent; export 'src/test_sse_client.dart' show TestSseClient; export 'src/logging.dart' @@ -65,16 +72,18 @@ enum SseHttpMethod { /// You can receive [Events]s by listening to the [stream] object. The SSEClient /// will connect when there is a nonzero number of subscribers on the [stream] /// and will disconnect when there are zero subscribers on the [stream]. -/// In certain cases, unrecoverable errors will be reported on the [stream] at -/// which point the stream will be done. /// -/// On `html` platforms the client is incapable of reporting unrecoverable -/// errors: the browser's native `EventSource` exposes no HTTP status codes -/// or response headers, so every failure is treated as recoverable and -/// retried with backoff indefinitely, and no error is ever reported on the -/// [stream]. +/// The client reports every connection failure as an error on the [stream]. +/// The error types are [SseHttpError], [SseTimeoutError], +/// [SseTransportError], [SseContentTypeError] and [SseStreamClosedError]. +/// A close that the owner starts is not reported. /// -/// The [SSEClient] will make best effort to maintain the streaming connection. +/// On `html` platforms the browser's native `EventSource` exposes no HTTP +/// status codes or response headers, so every failure is reported as an +/// [SseTransportError]. +/// +/// By default, the [SSEClient] makes best effort to maintain the streaming +/// connection. See the `autoReconnect` parameter of the [SSEClient] factory. abstract class SSEClient { static const defaultHeaders = { HttpHeaders.userAgentHeader: HttpHeaders.defaultAgentHeaderValue, @@ -95,6 +104,9 @@ abstract class SSEClient { /// Request that the SSEClient drops the current connection and then /// establishes a new connection respecting delay/backoff as if this was /// an error condition with the connection. + /// + /// Has no effect on a client made with `autoReconnect: false`. To + /// reconnect, close that client and make a new one. void restart(); /// Whether this implementation supports the given [capability]. Lets @@ -142,6 +154,12 @@ abstract class SSEClient { /// only when no provider is given. This allows query parameters to /// vary between attempts (e.g. a state selector that advances as data /// is received). Supported on all platforms. + /// + /// An optional [autoReconnect], true by default. When true, the client + /// reconnects with backoff after a failure. When false, the client + /// reports the first failure and then stops. It does not reconnect by + /// itself. The owner closes it and makes a new client when it wants to + /// connect again. Supported on all platforms. factory SSEClient(Uri uri, Set eventTypes, {Map headers = defaultHeaders, Duration connectTimeout = defaultConnectTimeout, @@ -149,13 +167,23 @@ abstract class SSEClient { String? body, SseHttpMethod httpMethod = SseHttpMethod.get, EventSourceLogger? logger, - Uri Function()? uriProvider}) { + Uri Function()? uriProvider, + bool autoReconnect = true}) { // merge headers so consumer gets reasonable defaults var mergedHeaders = {}; mergedHeaders.addAll(defaultHeaders); mergedHeaders.addAll(headers); - return getSSEClient(uri, eventTypes, mergedHeaders, connectTimeout, - readTimeout, body, httpMethod.toString(), logger, uriProvider); + return getSSEClient( + uri, + eventTypes, + mergedHeaders, + connectTimeout, + readTimeout, + body, + httpMethod.toString(), + logger, + uriProvider, + autoReconnect); } /// Get an SSE client for use in unit tests. @@ -176,6 +204,7 @@ abstract class SSEClient { Duration readTimeout = defaultReadTimeout, String? body, SseHttpMethod httpMethod = SseHttpMethod.get, + bool autoReconnect = true, Stream? sourceStream, Set? capabilities, }) { @@ -185,6 +214,7 @@ abstract class SSEClient { readTimeout: readTimeout, body: body, httpMethod: httpMethod, + autoReconnect: autoReconnect, sourceStream: sourceStream, capabilities: capabilities); } diff --git a/packages/event_source_client/lib/src/errors.dart b/packages/event_source_client/lib/src/errors.dart index 80395b5b..b9afd76d 100644 --- a/packages/event_source_client/lib/src/errors.dart +++ b/packages/event_source_client/lib/src/errors.dart @@ -8,7 +8,8 @@ /// reconnect with backoff. The subscription stays open; the error is an /// advisory the consumer may act on (or ignore and let the retry run). /// - `false`: the client will not retry. It stops reconnecting until -/// connection desire changes. +/// connection desire changes. A client made with automatic reconnect +/// turned off always reports `false`. /// /// Only produced by implementations whose transport can observe HTTP /// responses. The browser's native `EventSource` cannot, so on `html` @@ -33,3 +34,54 @@ final class SseHttpError implements Exception { String toString() => 'SseHttpError(statusCode: $statusCode, recoverable: $recoverable)'; } + +/// The part of a connection in which an [SseTimeoutError] occurs. +enum SseTimeoutPhase { + /// The server did not respond within the connect timeout. + connect, + + /// The connection was silent for longer than the read timeout. + read, +} + +/// Reported on the event stream when a connection times out. +final class SseTimeoutError implements Exception { + /// The part of the connection that timed out. + final SseTimeoutPhase phase; + + const SseTimeoutError(this.phase); + + @override + String toString() => 'SseTimeoutError(phase: ${phase.name})'; +} + +/// Reported on the event stream when a connection fails below the HTTP +/// level. Examples are a DNS failure, a TLS failure, a connection reset, +/// and a redirect loop. +final class SseTransportError implements Exception { + /// The error from the transport. Its string form can contain the request + /// URL, so do not log it directly. + final Object cause; + + const SseTransportError(this.cause); + + @override + String toString() => 'SseTransportError(${cause.runtimeType})'; +} + +/// Reported on the event stream when the server responds with a 200 +/// status, but the content type is not `text/event-stream`. +final class SseContentTypeError implements Exception { + const SseContentTypeError(); + + @override + String toString() => 'SseContentTypeError()'; +} + +/// Reported on the event stream when the server ends the stream. +final class SseStreamClosedError implements Exception { + const SseStreamClosedError(); + + @override + String toString() => 'SseStreamClosedError()'; +} diff --git a/packages/event_source_client/lib/src/sse_client_html.dart b/packages/event_source_client/lib/src/sse_client_html.dart index dbe7be2c..ab4aa370 100644 --- a/packages/event_source_client/lib/src/sse_client_html.dart +++ b/packages/event_source_client/lib/src/sse_client_html.dart @@ -11,12 +11,11 @@ import 'events.dart' as ld_message_event; /// An [SSEClient] that uses the [web.EventSource] available on most browsers for web platform support. /// /// The native `EventSource` API does not expose HTTP status codes or -/// response headers, so this implementation is incapable of reporting -/// terminal errors: every failure is treated as recoverable and retried -/// with backoff indefinitely, and no error is ever reported on the -/// [stream]. Consumers that need to react to unrecoverable statuses -/// (e.g. invalid credentials) must detect them through a transport that -/// can observe HTTP responses, such as a polling request. +/// response headers, so this implementation reports every failure as an +/// [SseTransportError]. With automatic reconnect on, every failure is +/// retried with backoff indefinitely. With automatic reconnect off, the +/// client closes the browser `EventSource` after the first failure and +/// does not reconnect. class HtmlSseClient implements SSEClient { /// The underlying eventsource web.EventSource? _eventSource; @@ -31,6 +30,7 @@ class HtmlSseClient implements SSEClient { final Uri _uri; final Uri Function()? _uriProvider; final Set _eventTypes; + final bool _autoReconnect; int? _activeSince; Timer? _retryTimer; @@ -42,10 +42,11 @@ class HtmlSseClient implements SSEClient { /// a provider is given. The fixed [uri] is used only when no provider /// is given. HtmlSseClient(Uri uri, Set eventTypes, EventSourceLogger? logger, - {Uri Function()? uriProvider}) + {Uri Function()? uriProvider, bool autoReconnect = true}) : _uri = uri, _uriProvider = uriProvider, - _eventTypes = eventTypes { + _eventTypes = eventTypes, + _autoReconnect = autoReconnect { _logger = logger ?? NoOpLogger(); _messageEventsController = StreamController.broadcast( @@ -82,10 +83,20 @@ class HtmlSseClient implements SSEClient { } void _handleError(web.Event event) { - // The browser event source errors are reasonably opaque, if we could - // determine the type of condition, then this is where we would - // determine if this was a temporary or permanent failure. - restart(); + if (_messageEventsController.isClosed) { + return; + } + // The browser event source errors are opaque, so every failure is a + // transport failure. + _messageEventsController.sink + .addError(const SseTransportError('EventSource error')); + if (!_autoReconnect) { + // Close the browser EventSource so that it does not reconnect by + // itself. + _closeConnection(); + return; + } + _reconnect(); } void _handleOpen(web.Event event) { @@ -116,6 +127,14 @@ class HtmlSseClient implements SSEClient { @override void restart() { + if (!_autoReconnect) { + _logger.debug('Ignoring restart, automatic reconnect is off.'); + return; + } + _reconnect(); + } + + void _reconnect() { _closeConnection(); final delay = _backoff.getRetryDelay(_activeSince); _activeSince = null; @@ -147,6 +166,8 @@ SSEClient getSSEClient( String? body, String method, EventSourceLogger? logger, - Uri Function()? uriProvider) => + Uri Function()? uriProvider, + bool autoReconnect) => // dropping unsupported configuration options - HtmlSseClient(uri, eventTypes, logger, uriProvider: uriProvider); + HtmlSseClient(uri, eventTypes, logger, + uriProvider: uriProvider, autoReconnect: autoReconnect); diff --git a/packages/event_source_client/lib/src/sse_client_http.dart b/packages/event_source_client/lib/src/sse_client_http.dart index 8255773d..28f11161 100644 --- a/packages/event_source_client/lib/src/sse_client_http.dart +++ b/packages/event_source_client/lib/src/sse_client_http.dart @@ -9,9 +9,9 @@ import 'state_value_object.dart'; /// An [SSEClient] that uses the [http.Client] for platform support. /// -/// The following cases will be considered unrecoverable and the [SSEClient] will report -/// an error on the [stream]. -/// - HTTP Status Codes other than 200, 400, 408, 429, 500 - 599 will cause an error +/// The client reports every connection failure on the [stream]. With +/// automatic reconnect on, these cases are not retried: +/// - HTTP Status Codes other than 200, 400, 408, 429, 500 - 599. /// - Cases in which a redirect loop is detected or a redirect is malformed. class HttpSseClient implements SSEClient { static const defaultConnectTimeout = Duration(seconds: 30); @@ -29,6 +29,8 @@ class HttpSseClient implements SSEClient { final StreamController _resetRequest = StreamController.broadcast(); + late final StateValues _stateValues; + /// Creates an instance of an SSEClient that will connect in the future /// to the [uri]. HttpSseClient( @@ -40,7 +42,8 @@ class HttpSseClient implements SSEClient { String? body, String httpMethod, EventSourceLogger? logger, - {Uri Function()? uriProvider}) + {Uri Function()? uriProvider, + bool autoReconnect = true}) : this.internal( uri, eventTypes, @@ -53,7 +56,8 @@ class HttpSseClient implements SSEClient { body, httpMethod, logger, - uriProvider: uriProvider); + uriProvider: uriProvider, + autoReconnect: autoReconnect); /// An internal constructor for injecting necessary dependencies for testing. HttpSseClient.internal( @@ -68,16 +72,22 @@ class HttpSseClient implements SSEClient { String? body, String httpMethod, EventSourceLogger? logger, - {Uri Function()? uriProvider}) { + {Uri Function()? uriProvider, + bool autoReconnect = true}) { _logger = logger ?? NoOpLogger(); _messageEventsController = StreamController.broadcast( // this is triggered when first listener subscribes onListen: () => _connectionDesiredStateController.add(true), - // this is triggered when last listener unsubscribes - onCancel: () => _connectionDesiredStateController.add(false), + // this is triggered when last listener unsubscribes, which also + // happens after close + onCancel: () { + if (!_connectionDesiredStateController.isClosed) { + _connectionDesiredStateController.add(false); + } + }, ); - StateIdle.run(StateValues( + _stateValues = StateValues( uri, eventTypes, headers, @@ -92,7 +102,9 @@ class HttpSseClient implements SSEClient { httpMethod, _resetRequest.stream, logger ?? NoOpLogger(), - uriProvider: uriProvider)); + uriProvider: uriProvider, + autoReconnect: autoReconnect); + StateIdle.run(_stateValues); } /// Subscribe to this [stream] to receive events and sometimes errors. The first @@ -103,6 +115,7 @@ class HttpSseClient implements SSEClient { @override Future close() async { _logger.debug('Closing SSE client permanently.'); + _stateValues.closed = true; _messageEventsController.close(); _connectionDesiredStateController.close(); _resetRequest.close(); @@ -110,6 +123,10 @@ class HttpSseClient implements SSEClient { @override void restart() { + if (!_stateValues.autoReconnect) { + _logger.debug('Ignoring restart, automatic reconnect is off.'); + return; + } _logger.debug('Restarting SSE client.'); if (_resetRequest.hasListener) { _resetRequest.sink.add(null); @@ -139,7 +156,8 @@ SSEClient getSSEClient( String? body, String method, EventSourceLogger? logger, - Uri Function()? uriProvider) => + Uri Function()? uriProvider, + bool autoReconnect) => HttpSseClient(uri, eventTypes, headers, connectTimeout, readTimeout, body, method, logger, - uriProvider: uriProvider); + uriProvider: uriProvider, autoReconnect: autoReconnect); diff --git a/packages/event_source_client/lib/src/sse_client_stub.dart b/packages/event_source_client/lib/src/sse_client_stub.dart index 97e64d84..b76067d4 100644 --- a/packages/event_source_client/lib/src/sse_client_stub.dart +++ b/packages/event_source_client/lib/src/sse_client_stub.dart @@ -10,6 +10,7 @@ SSEClient getSSEClient( String? body, String method, EventSourceLogger? logger, - Uri Function()? uriProvider) => + Uri Function()? uriProvider, + bool autoReconnect) => throw UnsupportedError( 'LaunchDarkly SSE Client is not supported on this platform.'); diff --git a/packages/event_source_client/lib/src/state_connected.dart b/packages/event_source_client/lib/src/state_connected.dart index 2789f870..7180ee8b 100644 --- a/packages/event_source_client/lib/src/state_connected.dart +++ b/packages/event_source_client/lib/src/state_connected.dart @@ -3,9 +3,11 @@ import 'dart:collection'; import 'dart:convert'; import 'package:http/http.dart' as http; +import 'errors.dart'; import 'events.dart'; import 'state_backoff.dart'; import 'state_idle.dart'; +import 'state_stopped.dart'; import 'state_value_object.dart'; import 'stateful_sse_parser.dart'; @@ -18,14 +20,14 @@ class StateConnected { // record transition to this state for testing/logging svo.transitionSink.add(StateConnected); svo.logger.debug('Transitioned to StateConnected'); - svo.eventSink.add(OpenEvent( + svo.emit(OpenEvent( headers: svo.connectHeaders != null ? UnmodifiableMapView(svo.connectHeaders!) : null)); // wait for either the stream to terminate or desired connection change to transition final transition = await Future.any([ - _processStream(svo, stream), + _processStream(svo, client, stream), _monitorConnectionNoLongerDesired(svo, client), _monitorReset(svo, client), ]); @@ -38,7 +40,7 @@ class StateConnected { // other errors at this point are considered recoverable. That may not be the case // but the Connecting state will figure that out during the next attempt. static Future _processStream( - StateValues svo, Stream> stream) async { + StateValues svo, http.Client client, Stream> stream) async { var recordedActiveSince = false; final sseParser = StatefulSSEParser(); try { @@ -63,7 +65,7 @@ class StateConnected { // only emit events that have event types the sse client was configured to use if (svo.eventTypes.contains(event.type)) { - svo.eventSink.add(event); + svo.emit(event); } default: break; @@ -71,20 +73,18 @@ class StateConnected { } return () { - svo.logger.info('Connection closed, will retry with backoff.'); - StateBackoff.run(svo); + svo.logger.info('Connection closed by the server.'); + handleFailure(svo, client, const SseStreamClosedError()); }; } on TimeoutException { return () { - svo.logger.info( - 'Connection TimeoutException occurred, will retry with backoff.'); - StateBackoff.run(svo); + svo.logger.info('Connection TimeoutException occurred.'); + handleFailure(svo, client, const SseTimeoutError(SseTimeoutPhase.read)); }; } catch (error) { return () { - svo.logger.info( - 'Connection error occurred, will retry with backoff. Error was: $error'); - StateBackoff.run(svo); + svo.logger.info('Connection error occurred. Error was: $error'); + handleFailure(svo, client, SseTransportError(error)); }; } } diff --git a/packages/event_source_client/lib/src/state_connecting.dart b/packages/event_source_client/lib/src/state_connecting.dart index d4a995de..3df41c70 100644 --- a/packages/event_source_client/lib/src/state_connecting.dart +++ b/packages/event_source_client/lib/src/state_connecting.dart @@ -5,9 +5,9 @@ import 'package:http/http.dart' as http; import 'error_utils.dart'; import 'errors.dart'; import 'http_consts.dart'; -import 'state_backoff.dart'; import 'state_connected.dart'; import 'state_idle.dart'; +import 'state_stopped.dart'; import 'state_value_object.dart'; /// This is the active state while we are sending the connection request to the @@ -61,23 +61,11 @@ class StateConnecting { // Report every error response with its status and headers (which // may carry a service directive). The recoverable flag tells the // consumer whether we will retry on our own. - final recoverable = + final recoverable = svo.autoReconnect && ErrorUtils.isHttpStatusCodeRecoverable(response.statusCode); final error = SseHttpError(response.statusCode, response.headers, recoverable: recoverable); - if (!recoverable) { - // Not retried: report the error and go idle until something - // changes. - return () => StateIdle.run(svo, errorCause: error); - } - // Retried: report the error, then back off and try again. The - // consumer may react (e.g. close the client) before the retry runs. - return () { - svo.logger.info( - 'Recoverable HTTP status code ${response.statusCode}, will retry'); - svo.eventSink.addError(error); - StateBackoff.run(svo); - }; + return () => handleFailure(svo, client, error, retry: recoverable); } final isEventStream = response.headers.entries.any((e) => @@ -85,10 +73,10 @@ class StateConnecting { e.value.toLowerCase().contains(MimeTypes.textEventStream)); if (!isEventStream) { // non event-stream content types are considered recoverable since it may be a service issue. + // The content type comes from the server, so it is not logged. return () { - svo.logger.info( - 'Connection received non-event-stream content type, will retry with backoff.'); - StateBackoff.run(svo); + svo.logger.info('Connection received non-event-stream content type.'); + handleFailure(svo, client, const SseContentTypeError()); }; } @@ -96,25 +84,17 @@ class StateConnecting { return () => StateConnected.run(svo, client, response.stream); } on TimeoutException { - // didn't connect in a timely manner, so backoff then we'll try again - - client.close; return () { - svo.logger.info( - 'Connection TimeoutException occurred, will retry with backoff.'); - StateBackoff.run(svo); + svo.logger.info('Connection TimeoutException occurred.'); + handleFailure( + svo, client, const SseTimeoutError(SseTimeoutPhase.connect)); }; } catch (err) { - client.close(); - if (!ErrorUtils.isConnectionErrorRecoverable(err)) { - // go to idle relaying the error that caused us to go idle. - return () => StateIdle.run(svo, errorCause: err); - } - // we err on the side of caution and say that errors are recoverable. The worst // thing that happens is we retry wastefully. The alternative of assuming // errors are NOT recoverable is far worse! - return () => StateBackoff.run(svo); + return () => handleFailure(svo, client, SseTransportError(err), + retry: ErrorUtils.isConnectionErrorRecoverable(err)); } } diff --git a/packages/event_source_client/lib/src/state_idle.dart b/packages/event_source_client/lib/src/state_idle.dart index a2716bea..fcad7c1c 100644 --- a/packages/event_source_client/lib/src/state_idle.dart +++ b/packages/event_source_client/lib/src/state_idle.dart @@ -2,16 +2,16 @@ import 'state_connecting.dart'; import 'state_value_object.dart'; /// This is the active state when there is no desire to be connected -/// and also the active state when we have encountered an unrecoverable error. +/// and also the active state after a failure that the client does not retry. class StateIdle { static Future run(StateValues svo, {Object? errorCause}) async { // record transition to this state for testing/logging svo.transitionSink.add(StateIdle); - // unrecoverable errors are reported to subscribers of the client. + // failures that the client does not retry are reported to subscribers. if (errorCause != null) { svo.logger.info('Transitioned to StateIdle due to error: $errorCause'); - svo.eventSink.addError(errorCause); + svo.reportError(errorCause); } else { svo.logger.debug('Transitioned to StateIdle'); } diff --git a/packages/event_source_client/lib/src/state_stopped.dart b/packages/event_source_client/lib/src/state_stopped.dart new file mode 100644 index 00000000..9e01d04d --- /dev/null +++ b/packages/event_source_client/lib/src/state_stopped.dart @@ -0,0 +1,37 @@ +import 'package:http/http.dart' as http; + +import 'state_backoff.dart'; +import 'state_idle.dart'; +import 'state_value_object.dart'; + +/// This is the final state after a failure when automatic reconnect is off. +/// The client reports the failure, and then does nothing until it is closed. +class StateStopped { + static void run(StateValues svo, Object error) { + // record transition to this state for testing/logging + svo.transitionSink.add(StateStopped); + svo.logger.info('Transitioned to StateStopped due to error: $error'); + svo.reportError(error); + } +} + +/// Closes [client] and moves the state machine on after a connection +/// failure. Every path reports [error]. +/// +/// With automatic reconnect off, the client stops. Otherwise it backs off +/// and tries again when [retry] is true, and goes idle when [retry] is +/// false. +void handleFailure(StateValues svo, http.Client client, Object error, + {bool retry = true}) { + client.close(); + if (!svo.autoReconnect) { + StateStopped.run(svo, error); + return; + } + if (!retry) { + StateIdle.run(svo, errorCause: error); + return; + } + svo.reportError(error); + StateBackoff.run(svo); +} diff --git a/packages/event_source_client/lib/src/state_value_object.dart b/packages/event_source_client/lib/src/state_value_object.dart index ce81f034..39680a8a 100644 --- a/packages/event_source_client/lib/src/state_value_object.dart +++ b/packages/event_source_client/lib/src/state_value_object.dart @@ -46,6 +46,10 @@ class StateValues { final Backoff backoff; final EventSourceLogger logger; + /// When false, the state machine stops after the first failure instead of + /// reconnecting. + final bool autoReconnect; + // Transient data /// The timestamp when the last active connection was established. @@ -56,6 +60,10 @@ class StateValues { /// Headers received from the connection. Map? connectHeaders; + /// Set when the owner closes the client. After that, nothing is sent to + /// subscribers. + bool closed = false; + /// The URI to use for the next connection attempt. Uri get connectUri => uriProvider?.call() ?? uri; @@ -75,6 +83,21 @@ class StateValues { this.httpMethod, this.resetRequest, this.logger, - {this.uriProvider}) + {this.uriProvider, + this.autoReconnect = true}) : backoff = Backoff(random); + + /// Sends [event] to subscribers, unless the client is closed. + void emit(Event event) { + if (!closed) { + eventSink.add(event); + } + } + + /// Reports [error] to subscribers, unless the client is closed. + void reportError(Object error) { + if (!closed) { + eventSink.addError(error); + } + } } diff --git a/packages/event_source_client/lib/src/test_sse_client.dart b/packages/event_source_client/lib/src/test_sse_client.dart index c8f497e7..045e468a 100644 --- a/packages/event_source_client/lib/src/test_sse_client.dart +++ b/packages/event_source_client/lib/src/test_sse_client.dart @@ -15,6 +15,7 @@ final class TestSseClient implements SSEClient { final Duration readTimeout; final String? body; final SseHttpMethod httpMethod; + final bool autoReconnect; final Set _capabilities; late final Stream? _sourceStream; StreamSubscription? _sourceStreamSubscription; @@ -76,6 +77,7 @@ final class TestSseClient implements SSEClient { required this.readTimeout, required this.body, required this.httpMethod, + this.autoReconnect = true, Stream? sourceStream, Set? capabilities, }) : _capabilities = capabilities ?? {SSECapability.requestHeaders} { diff --git a/packages/event_source_client/test/failure_reporting_test.dart b/packages/event_source_client/test/failure_reporting_test.dart new file mode 100644 index 00000000..a5ec0954 --- /dev/null +++ b/packages/event_source_client/test/failure_reporting_test.dart @@ -0,0 +1,302 @@ +// ignore_for_file: close_sinks + +import 'dart:async'; +import 'dart:convert'; +import 'dart:math' as math; + +import 'package:http/http.dart' as http; +import 'package:http/testing.dart'; +import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart'; +import 'package:launchdarkly_event_source_client/src/http_consts.dart'; +import 'package:launchdarkly_event_source_client/src/sse_client_http.dart'; +import 'package:launchdarkly_event_source_client/src/state_backoff.dart'; +import 'package:launchdarkly_event_source_client/src/state_stopped.dart'; +import 'package:test/test.dart'; + +/// A content type from the server. The client must not log it. +const _serverContentType = 'text/html; injected-text'; + +const _eventStreamHeaders = { + HttpHeaders.contentTypeHeader: MimeTypes.textEventStream +}; + +/// Each case makes one kind of failure on every request. +final class _FailureCase { + final String name; + final http.Client Function() clientFactory; + final Matcher error; + final Duration connectTimeout; + final Duration readTimeout; + + _FailureCase(this.name, this.clientFactory, this.error, + {this.connectTimeout = const Duration(days: 1), + this.readTimeout = const Duration(days: 1)}); +} + +http.Client _statusClient(int status) => MockClient.streaming( + (request, body) async => http.StreamedResponse(const Stream.empty(), status, + headers: const {'x-ld-fd-fallback': 'true'})); + +http.Client _contentTypeClient() => + MockClient.streaming((request, body) async => http.StreamedResponse( + Stream.value(utf8.encode('{}')), 200, + headers: const {HttpHeaders.contentTypeHeader: _serverContentType})); + +http.Client _blockingClient() => MockClient.streaming((request, body) async { + await Completer().future; + throw StateError('unreachable'); + }); + +http.Client _silentStreamClient() => + MockClient.streaming((request, body) async => http.StreamedResponse( + StreamController>().stream, 200, + headers: _eventStreamHeaders)); + +http.Client _transportErrorClient() => MockClient.streaming( + (request, body) async => throw http.ClientException('connection reset')); + +http.Client _closingStreamClient() => + MockClient.streaming((request, body) async => http.StreamedResponse( + Stream.value(utf8.encode('event:put\ndata:hello\n\n')), 200, + headers: _eventStreamHeaders)); + +final _failureCases = [ + _FailureCase( + 'a 401 response', + () => _statusClient(401), + isA() + .having((e) => e.statusCode, 'statusCode', 401) + .having((e) => e.headers['x-ld-fd-fallback'], 'header', 'true')), + _FailureCase('a 503 response', () => _statusClient(503), + isA().having((e) => e.statusCode, 'statusCode', 503)), + _FailureCase( + 'a connect timeout', + _blockingClient, + isA() + .having((e) => e.phase, 'phase', SseTimeoutPhase.connect), + connectTimeout: const Duration(milliseconds: 10)), + _FailureCase( + 'a read timeout', + _silentStreamClient, + isA() + .having((e) => e.phase, 'phase', SseTimeoutPhase.read), + readTimeout: const Duration(milliseconds: 10)), + _FailureCase( + 'a transport error', _transportErrorClient, isA()), + _FailureCase( + 'a wrong content type', + _contentTypeClient, + isA().having( + (e) => e.toString(), 'toString', isNot(contains('injected-text')))), + _FailureCase('a stream closed by the server', _closingStreamClient, + isA()), +]; + +HttpSseClient _makeClient(_FailureCase failure, + {required bool autoReconnect, + required StreamController transitions, + void Function()? onRequest, + EventSourceLogger logger = const NoOpLogger()}) { + return HttpSseClient.internal(Uri.parse('/path'), {'put'}, {}, + failure.connectTimeout, failure.readTimeout, transitions.sink, () { + onRequest?.call(); + return failure.clientFactory(); + }, math.Random(), null, 'GET', logger, autoReconnect: autoReconnect); +} + +/// Keeps every message that the client logs. +final class _RecordingLogger implements EventSourceLogger { + final List messages = []; + + @override + void debug(String message) => messages.add(message); + + @override + void info(String message) => messages.add(message); + + @override + void warn(String message) => messages.add(message); + + @override + void error(String message) => messages.add(message); +} + +void main() { + for (final autoReconnect in [true, false]) { + test( + 'it does not log the content type from the server ' + '(autoReconnect: $autoReconnect)', () async { + final logger = _RecordingLogger(); + final client = _makeClient( + _FailureCase('a wrong content type', _contentTypeClient, + isA()), + autoReconnect: autoReconnect, + transitions: StreamController.broadcast(), + logger: logger); + + final firstError = Completer(); + client.stream.listen((_) {}, onError: (Object err) { + if (!firstError.isCompleted) { + firstError.complete(err); + } + }); + await firstError.future; + await client.close(); + + expect(logger.messages, isNotEmpty); + expect(logger.messages, everyElement(isNot(contains('injected-text')))); + }); + } + + group('with automatic reconnect off', () { + for (final failure in _failureCases) { + test('it reports ${failure.name} once and stops', () async { + final transitions = StreamController.broadcast(); + final seenTransitions = []; + transitions.stream.listen(seenTransitions.add); + var requests = 0; + final client = _makeClient(failure, + autoReconnect: false, + transitions: transitions, + onRequest: () => requests++); + + final errors = []; + client.stream.listen((_) {}, onError: errors.add); + + await Future.delayed(const Duration(milliseconds: 1500)); + + expect(errors, [failure.error]); + expect(requests, 1); + expect(seenTransitions, contains(StateStopped)); + expect(seenTransitions, isNot(contains(StateBackoff))); + await client.close(); + }); + } + + test('it ignores a restart request', () async { + final transitions = StreamController.broadcast(); + final seenTransitions = []; + transitions.stream.listen(seenTransitions.add); + final client = _makeClient( + _FailureCase('an open stream', _silentStreamClient, isA()), + autoReconnect: false, + transitions: transitions); + + final events = []; + client.stream.listen(events.add); + await Future.delayed(const Duration(milliseconds: 20)); + client.restart(); + await Future.delayed(const Duration(milliseconds: 20)); + + expect(events, [isA()]); + expect(seenTransitions, isNot(contains(StateBackoff))); + await client.close(); + }); + }); + + group('with automatic reconnect on', () { + for (final failure in _failureCases) { + test('it reports ${failure.name} once per attempt', () async { + final transitions = StreamController.broadcast(); + final client = + _makeClient(failure, autoReconnect: true, transitions: transitions); + + final firstError = Completer(); + client.stream.listen((_) {}, onError: (Object err) { + if (!firstError.isCompleted) { + firstError.complete(err); + } + }); + + expect(await firstError.future, failure.error); + await client.close(); + }); + } + + test('it reconnects after the server closes the stream', () async { + final transitions = StreamController.broadcast(); + var requests = 0; + final secondRequest = Completer(); + final client = _makeClient( + _FailureCase('a stream closed by the server', _closingStreamClient, + isA()), + autoReconnect: true, + transitions: transitions, onRequest: () { + requests++; + if (requests == 2) { + secondRequest.complete(); + } + }); + + client.stream.listen((_) {}, onError: (_) {}); + + await secondRequest.future.timeout(const Duration(seconds: 5)); + await client.close(); + }); + + test('it marks a status it retries as recoverable', () async { + final transitions = StreamController.broadcast(); + final client = _makeClient( + _FailureCase( + 'a 503 response', () => _statusClient(503), isA()), + autoReconnect: true, + transitions: transitions); + + final firstError = Completer(); + client.stream.listen((_) {}, onError: (Object err) { + if (!firstError.isCompleted) { + firstError.complete(err); + } + }); + + expect( + await firstError.future, + isA() + .having((e) => e.recoverable, 'recoverable', true)); + await client.close(); + }); + + test('it marks a status as not recoverable when it does not reconnect', + () async { + final transitions = StreamController.broadcast(); + final client = _makeClient( + _FailureCase( + 'a 503 response', () => _statusClient(503), isA()), + autoReconnect: false, + transitions: transitions); + + final firstError = Completer(); + client.stream.listen((_) {}, onError: firstError.complete); + + expect( + await firstError.future, + isA() + .having((e) => e.recoverable, 'recoverable', false)); + await client.close(); + }); + }); + + group('when the owner closes the client', () { + for (final autoReconnect in [true, false]) { + test('it reports nothing (autoReconnect: $autoReconnect)', () async { + final transitions = StreamController.broadcast(); + final client = _makeClient( + _FailureCase('an open stream', _silentStreamClient, isA()), + autoReconnect: autoReconnect, + transitions: transitions); + + final errors = []; + final done = Completer(); + client.stream + .listen((_) {}, onError: errors.add, onDone: done.complete); + await Future.delayed(const Duration(milliseconds: 20)); + + await client.close(); + await done.future; + await Future.delayed(const Duration(milliseconds: 20)); + + expect(errors, isEmpty); + }); + } + }); +} diff --git a/packages/event_source_client/test/sse_client_http_test.dart b/packages/event_source_client/test/sse_client_http_test.dart index b9a639c7..194832c2 100644 --- a/packages/event_source_client/test/sse_client_http_test.dart +++ b/packages/event_source_client/test/sse_client_http_test.dart @@ -148,8 +148,9 @@ void main() { headers: {'content-type': 'text/event-stream'}); }); - // Start listening so the SSE client conntects. - sseClientUnderTest.stream.listen((event) {}); + // Start listening so the SSE client conntects. The stream ends after + // one event, so the client reports an error. + sseClientUnderTest.stream.listen((event) {}, onError: (_) {}); final res = await untilCalled(() => mockClient.send(any())); expect((res.positionalArguments[0] as http.Request).method, method); @@ -183,8 +184,9 @@ void main() { headers: {'content-type': 'text/event-stream'}); }); - // Start listening so the SSE client conntects. - sseClientUnderTest.stream.listen((event) {}); + // Start listening so the SSE client conntects. The stream ends after one + // event, so the client reports an error. + sseClientUnderTest.stream.listen((event) {}, onError: (_) {}); final res = await untilCalled(() => mockClient.send(any())); expect((res.positionalArguments[0] as http.Request).headers,