Add ability to close a StreamedRequest

pkgs/http/lib/src/browser_client.dart

@@ -8,9 +8,9 @@ import 'dart:typed_data';
 
 import 'base_client.dart';
 import 'base_request.dart';
+import 'browser_streamed_response.dart';
 import 'byte_stream.dart';
 import 'exception.dart';
-import 'streamed_response.dart';
 
 /// Create a [BrowserClient].
 ///
@@ -39,7 +39,7 @@ class BrowserClient extends BaseClient {
 
   /// Sends an HTTP request and asynchronously returns the response.
   @override
-  Future<StreamedResponse> send(BaseRequest request) async {
+  Future<BrowserStreamedResponse> send(BaseRequest request) async {
     var bytes = await request.finalize().toBytes();
     var xhr = HttpRequest();
     _xhrs.add(xhr);
@@ -49,16 +49,17 @@ class BrowserClient extends BaseClient {
       ..withCredentials = withCredentials;
     request.headers.forEach(xhr.setRequestHeader);
 
-    var completer = Completer<StreamedResponse>();
+    var completer = Completer<BrowserStreamedResponse>();
 
     unawaited(xhr.onLoad.first.then((_) {
       var body = (xhr.response as ByteBuffer).asUint8List();
-      completer.complete(StreamedResponse(
+      completer.complete(BrowserStreamedResponse(
           ByteStream.fromBytes(body), xhr.status!,
           contentLength: body.length,
           request: request,
           headers: xhr.responseHeaders,
-          reasonPhrase: xhr.statusText));
+          reasonPhrase: xhr.statusText,
+          inner: xhr));
     }));
 
     unawaited(xhr.onError.first.then((_) {

pkgs/http/lib/src/browser_streamed_response.dart

@@ -0,0 +1,42 @@
+// Copyright (c) 2020, 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:html';
+
+import 'base_request.dart';
+import 'streamed_response.dart';
+
+/// An HTTP response where the response body is received asynchronously after
+/// the headers have been received.
+class BrowserStreamedResponse extends StreamedResponse {
+  final HttpRequest? _inner;
+
+  /// Creates a new streaming response.
+  ///
+  /// [stream] should be a single-subscription stream.
+  BrowserStreamedResponse(Stream<List<int>> stream, int statusCode,
+      {int? contentLength,
+      BaseRequest? request,
+      Map<String, String> headers = const {},
+      bool isRedirect = false,
+      bool persistentConnection = true,
+      String? reasonPhrase,
+      HttpRequest? inner})
+      : _inner = inner,
+        super(stream, statusCode,
+            contentLength: contentLength,
+            request: request,
+            headers: headers,
+            isRedirect: isRedirect,
+            persistentConnection: persistentConnection,
+            reasonPhrase: reasonPhrase);
+
+  /// Closes the underlying HTTP Request
+  ///
+  /// Will throw if `inner` was not set or `null` when `this` was created.
+  @override
+  Future<void> close() async {
+    _inner!.abort();
+  }
+}

pkgs/http/lib/src/io_streamed_response.dart

@@ -38,4 +38,12 @@ class IOStreamedResponse extends StreamedResponse {
   ///
   /// Will throw if `inner` was not set or `null` when `this` was created.
   Future<Socket> detachSocket() async => _inner!.detachSocket();
+
+  /// Closes the underlying HTTP Request
+  ///
+  /// Will throw if `inner` was not set or `null` when `this` was created.
+  @override
+  Future<void> close() async {
+    (await detachSocket()).destroy();
+  }
 }

pkgs/http/lib/src/streamed_response.dart

@@ -33,4 +33,12 @@ class StreamedResponse extends BaseResponse {
             isRedirect: isRedirect,
             persistentConnection: persistentConnection,
             reasonPhrase: reasonPhrase);
+
+  /// Closes the underlying HTTP Request
+  ///
+  /// This will throw an [UnimplementedError] for the base [StreamedResponse]
+  /// class
+  Future<void> close() {
+    throw UnimplementedError();
+  }
 }