feat!: Report every connection failure and add an option to not reconnect

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 = <String, String>{
     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,20 +154,36 @@ 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<String> eventTypes,
       {Map<String, String> headers = defaultHeaders,
       Duration connectTimeout = defaultConnectTimeout,
       Duration readTimeout = defaultReadTimeout,
       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 = <String, String>{};
     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<Event>? sourceStream,
     Set<SSECapability>? capabilities,
   }) {
@@ -185,6 +214,7 @@ abstract class SSEClient {
         readTimeout: readTimeout,
         body: body,
         httpMethod: httpMethod,
+        autoReconnect: autoReconnect,
         sourceStream: sourceStream,
         capabilities: capabilities);
   }

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()';
+}

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<String> _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<String> 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<ld_message_event.Event>.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);

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<void> _resetRequest =
       StreamController<void>.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<Event>.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,13 +115,18 @@ class HttpSseClient implements SSEClient {
   @override
   Future close() async {
     _logger.debug('Closing SSE client permanently.');
+    _stateValues.closed = true;
     _messageEventsController.close();
     _connectionDesiredStateController.close();
     _resetRequest.close();
   }
 
   @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);

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.');