feat: support aborting HTTP requests

pkgs/http/CHANGELOG.md

@@ -1,5 +1,6 @@
-## 1.4.1
+## 1.5.0
 
+* Added support for abortion of requests
 * Clarify that some header names may not be sent/received.
 
 ## 1.4.0

pkgs/http/README.md

@@ -113,6 +113,87 @@ the [`RetryClient()`][new RetryClient] constructor.
 
 [new RetryClient]: https://pub.dev/documentation/http/latest/retry/RetryClient/RetryClient.html
 
+## Aborting requests
+
+Some clients, such as [`BrowserClient`][browserclient], [`IOClient`][ioclient], and
+[`RetryClient`][retryclient], support abortion of requests in-flight.
+
+Abortion in this way can only be performed when using [`Client.send`][clientsend] or
+[`BaseRequest.send`][baserequestsend] with an [`Abortable`][abortable] request (such
+as [`AbortableRequest`][abortablerequest]).
+
+To abort a request, complete the [`Abortable.abortTrigger`][aborttrigger].
+
+Depending on when the abortion is triggered, an [`AbortedRequest`][abortedrequest] may be thrown in different places.
+
+```dart
+import 'dart:async';
+import 'package:http/http.dart' as http;
+
+Future<void> main() async {
+  final abortTrigger = Completer<void>();
+  final client = Client();
+  final request = AbortableRequest(
+    'GET',
+    Uri.parse('http://example.org'),
+    abortTrigger: abortTrigger.future,
+  );
+
+  // Whenever abortion is required:
+  // > abortTrigger.complete();
+
+  // Send request
+  final StreamedResponse response;
+  try {
+    response = await client.send(request);
+  } on AbortedRequest {
+    // request aborted before it was fully sent
+    rethrow;
+  }
+
+  // Using full response bytes listener
+  response.stream.listen(
+    (data) {
+      // consume response bytes
+    },
+    onError: (Object err) {
+      if (err is AbortedRequest) {
+        // request aborted whilst response bytes are being streamed;
+        // the stream will always be finished early
+      }
+    },
+    onDone: () {
+      // response bytes consumed, or partially consumed if finished
+      // early due to abortion
+    },
+  );
+
+  // Alternatively, using `asFuture`
+  try {
+    await response.stream.listen(
+      (data) {
+        // consume response bytes
+      },
+    ).asFuture<void>();
+  } on AbortedRequest {
+    // request aborted whilst response bytes are being streamed
+    rethrow;
+  }
+    // response bytes fully consumed
+}
+```
+
+[browserclient]: https://pub.dev/documentation/http/latest/browser_client/BrowserClient-class.html
+[ioclient]: https://pub.dev/documentation/http/latest/io_client/IOClient-class.html
+[retryclient]: https://pub.dev/documentation/http/latest/retry/RetryClient-class.html
+[clientsend]: https://pub.dev/documentation/http/latest/http/Client/send.html
+[baserequestsend]: https://pub.dev/documentation/http/latest/http/BaseRequest/send.html
+[abortable]: https://pub.dev/documentation/http/latest/http/Abortable-class.html
+[abortablerequest]: https://pub.dev/documentation/http/latest/http/AbortableRequest-class.html
+[aborttrigger]: https://pub.dev/documentation/http/latest/http/Abortable/abortTrigger.html
+[abortedrequest]: https://pub.dev/documentation/http/latest/http/AbortedRequest-class.html
+
+
 ## Choosing an implementation
 
 There are multiple implementations of the `package:http` [`Client`][client] interface. By default, `package:http` uses [`BrowserClient`][browserclient] on the web and [`IOClient`][ioclient] on all other platforms. You can choose a different [`Client`][client] implementation based on the needs of your application.

pkgs/http/lib/http.dart

@@ -19,6 +19,7 @@ export 'src/base_request.dart';
 export 'src/base_response.dart'
     show BaseResponse, BaseResponseWithUrl, HeadersWithSplitValues;
 export 'src/byte_stream.dart';
+export 'src/abortable.dart';
 export 'src/client.dart' hide zoneClient;
 export 'src/exception.dart';
 export 'src/multipart_file.dart';

pkgs/http/lib/retry.dart

@@ -133,7 +133,18 @@ final class RetryClient extends BaseClient {
 
   /// Returns a copy of [original] with the given [body].
   StreamedRequest _copyRequest(BaseRequest original, Stream<List<int>> body) {
-    final request = StreamedRequest(original.method, original.url)
+    final StreamedRequest request;
+    if (original case Abortable(:final abortTrigger?)) {
+      request = AbortableStreamedRequest(
+        original.method,
+        original.url,
+        abortTrigger: abortTrigger,
+      );
+    } else {
+      request = StreamedRequest(original.method, original.url);
+    }
+
+    request
       ..contentLength = original.contentLength
       ..followRedirects = original.followRedirects
       ..headers.addAll(original.headers)

pkgs/http/lib/src/abortable.dart

@@ -0,0 +1,43 @@
+// Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'base_request.dart';
+import 'client.dart';
+import 'streamed_response.dart';
+
+/// Enables a request to be recognised by a [Client] as abortable
+abstract mixin class Abortable implements BaseRequest {
+  /// Completion of this future aborts this request (if the client supports
+  /// abortion)
+  ///
+  /// Requests/responses may be aborted at any time during their lifecycle.
+  ///
+  ///  * If completed before the request has been finalized and sent,
+  ///    [Client.send] completes with an [AbortedRequest] exception
+  ///  * If completed after the response headers are available, or whilst
+  ///    streaming response bytes, clients inject [AbortedRequest] into the
+  ///    [StreamedResponse.stream] then finish it early
+  ///  * If completed after the response is fully complete, there is no effect
+  ///
+  /// A common pattern is aborting a request when another event occurs (such as
+  /// a user action): use a [Completer] to implement this. To implement a
+  /// timeout (to abort the request after a set time has elapsed), use
+  /// [Future.delayed].
+  ///
+  /// This future must not complete to an error.
+  ///
+  /// Some clients may not support abortion, or may not support this trigger.
+  abstract final Future<void>? abortTrigger;
+}
+
+/// Thrown when an HTTP request is aborted
+///
+/// Usually, this is due to [Abortable.abortTrigger]. See documentation on that
+/// property for more info.
+class AbortedRequest implements Exception {
+  /// Indicator that the request has been aborted
+  const AbortedRequest();
+}

pkgs/http/lib/src/base_request.dart

@@ -7,6 +7,7 @@ import 'dart:collection';
 import 'package:meta/meta.dart';
 
 import '../http.dart' show ClientException, get;
+import 'abortable.dart';
 import 'base_client.dart';
 import 'base_response.dart';
 import 'byte_stream.dart';
@@ -20,6 +21,10 @@ import 'utils.dart';
 /// [BaseClient.send], which allows the user to provide fine-grained control
 /// over the request properties. However, usually it's easier to use convenience
 /// methods like [get] or [BaseClient.get].
+///
+/// Subclasses/implementers should mixin/implement [Abortable] to support
+/// abortion of requests. A future breaking version of 'package:http' will
+/// merge [Abortable] into [BaseRequest], making it a requirement.
 abstract class BaseRequest {
   /// The HTTP method of the request.
   ///

pkgs/http/lib/src/browser_client.dart

@@ -8,12 +8,14 @@ import 'dart:js_interop';
 import 'package:web/web.dart'
     show
         AbortController,
+        DOMException,
         HeadersInit,
         ReadableStreamDefaultReader,
         RequestInfo,
         RequestInit,
         Response;
 
+import 'abortable.dart';
 import 'base_client.dart';
 import 'base_request.dart';
 import 'exception.dart';
@@ -49,15 +51,14 @@ external JSPromise<Response> _fetch(
 /// Responses are streamed but requests are not. A request will only be sent
 /// once all the data is available.
 class BrowserClient extends BaseClient {
-  final _abortController = AbortController();
-
   /// Whether to send credentials such as cookies or authorization headers for
   /// cross-site requests.
   ///
   /// Defaults to `false`.
   bool withCredentials = false;
 
   bool _isClosed = false;
+  final _openRequestAbortControllers = <AbortController>[];
 
   /// Sends an HTTP request and asynchronously returns the response.
   @override
@@ -67,8 +68,16 @@ class BrowserClient extends BaseClient {
           'HTTP request failed. Client is already closed.', request.url);
     }
 
+    final _abortController = AbortController();
+    _openRequestAbortControllers.add(_abortController);
+
     final bodyBytes = await request.finalize().toBytes();
     try {
+      Future<void>? canceller;
+      if (request case Abortable(:final abortTrigger?)) {
+        canceller = abortTrigger.whenComplete(() => _abortController.abort());
+      }
+
       final response = await _fetch(
         '${request.url}'.toJS,
         RequestInit(
@@ -86,6 +95,8 @@ class BrowserClient extends BaseClient {
         ),
       ).toDart;
 
+      canceller?.ignore();
+
       final contentLengthHeader = response.headers.get('content-length');
 
       final contentLength = contentLengthHeader != null
@@ -114,18 +125,29 @@ class BrowserClient extends BaseClient {
         url: Uri.parse(response.url),
         reasonPhrase: response.statusText,
       );
+    } on DOMException catch (e, st) {
+      if (e.name == 'AbortError') {
+        Error.throwWithStackTrace(const AbortedRequest(), st);
+      }
+
+      _rethrowAsClientException(e, st, request);
     } catch (e, st) {
       _rethrowAsClientException(e, st, request);
+    } finally {
+      _openRequestAbortControllers.remove(_abortController);
     }
   }
 
   /// Closes the client.
   ///
-  /// This terminates all active requests.
+  /// This terminates all active requests, which may cause them to throw
+  /// [AbortedRequest] or [ClientException].
   @override
   void close() {
+    for (final abortController in _openRequestAbortControllers) {
+      abortController.abort();
+    }
     _isClosed = true;
-    _abortController.abort();
   }
 }
 
@@ -158,6 +180,14 @@ Stream<List<int>> _readBody(BaseRequest request, Response response) async* {
       }
       yield (chunk.value! as JSUint8Array).toDart;
     }
+  } on DOMException catch (e, st) {
+    isError = true;
+
+    if (e.name == 'AbortError') {
+      Error.throwWithStackTrace(const AbortedRequest(), st);
+    }
+
+    _rethrowAsClientException(e, st, request);
   } catch (e, st) {
     isError = true;
     _rethrowAsClientException(e, st, request);

pkgs/http/lib/src/io_client.dart

@@ -2,8 +2,10 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:io';
 
+import 'abortable.dart';
 import 'base_client.dart';
 import 'base_request.dart';
 import 'base_response.dart';
@@ -123,7 +125,45 @@ class IOClient extends BaseClient {
         ioRequest.headers.set(name, value);
       });
 
-      var response = await stream.pipe(ioRequest) as HttpClientResponse;
+      // We can only abort the actual connection up until the point we obtain
+      // the response.
+      // After that point, the full response bytes are always available.
+      // However, we instead inject an error into the response stream to match
+      // the behaviour of `BrowserClient`.
+
+      StreamSubscription<List<int>>? subscription;
+      final controller = StreamController<List<int>>(sync: true);
+
+      if (request case Abortable(:final abortTrigger?)) {
+        abortTrigger.whenComplete(() async {
+          if (subscription == null) {
+            ioRequest.abort(const AbortedRequest());
+          } else {
+            if (!controller.isClosed) {
+              controller.addError(const AbortedRequest());
+            }
+            await subscription.cancel();
+          }
+          await controller.close();
+        });
+      }
+
+      final response = await stream.pipe(ioRequest) as HttpClientResponse;
+
+      subscription = response.listen(
+        controller.add,
+        onDone: controller.close,
+        onError: (Object err, StackTrace stackTrace) {
+          if (err is HttpException) {
+            controller.addError(
+              ClientException(err.message, err.uri),
+              stackTrace,
+            );
+          } else {
+            controller.addError(err, stackTrace);
+          }
+        },
+      );
 
       var headers = <String, String>{};
       response.headers.forEach((key, values) {
@@ -134,22 +174,20 @@ class IOClient extends BaseClient {
       });
 
       return _IOStreamedResponseV2(
-          response.handleError((Object error) {
-            final httpException = error as HttpException;
-            throw ClientException(httpException.message, httpException.uri);
-          }, test: (error) => error is HttpException),
-          response.statusCode,
-          contentLength:
-              response.contentLength == -1 ? null : response.contentLength,
-          request: request,
-          headers: headers,
-          isRedirect: response.isRedirect,
-          url: response.redirects.isNotEmpty
-              ? response.redirects.last.location
-              : request.url,
-          persistentConnection: response.persistentConnection,
-          reasonPhrase: response.reasonPhrase,
-          inner: response);
+        controller.stream,
+        response.statusCode,
+        contentLength:
+            response.contentLength == -1 ? null : response.contentLength,
+        request: request,
+        headers: headers,
+        isRedirect: response.isRedirect,
+        url: response.redirects.isNotEmpty
+            ? response.redirects.last.location
+            : request.url,
+        persistentConnection: response.persistentConnection,
+        reasonPhrase: response.reasonPhrase,
+        inner: response,
+      );
     } on SocketException catch (error) {
       throw _ClientSocketException(error, request.url);
     } on HttpException catch (error) {
@@ -159,8 +197,9 @@ class IOClient extends BaseClient {
 
   /// Closes the client.
   ///
-  /// Terminates all active connections. If a client remains unclosed, the Dart
-  /// process may not terminate.
+  /// Terminates all active connections, which may cause them to throw
+  /// [AbortedRequest] or [ClientException]/[SocketException]. If a client
+  /// remains unclosed, the Dart process may not terminate.
   @override
   void close() {
     if (_inner != null) {