add HttpClient force close feature

CHANGELOG.md

@@ -1,6 +1,10 @@
+## 0.0.4+2
+
+* Added `HttpClient` force close feature.
+
 ## 0.0.4+1
 
-* Fix benchmarking instructions.
+* Fixed benchmarking instructions.
 
 ## 0.0.4
 

lib/src/http_callback_handler.dart

@@ -48,11 +48,14 @@ class CallbackHandler {
   /// Registers the [NativePort] to the cronet side.
   CallbackHandler(this.executor, this.receivePort);
 
-  /// [Stream] controller for [HttpClientResponse]
+  /// [Stream] for [HttpClientResponse].
   Stream<List<int>> get stream {
     return _controller.stream;
   }
 
+  /// [Stream] controller for [HttpClientResponse].
+  StreamController<List<int>> get controller => _controller;
+
   // Clean up tasks for a request.
   //
   // We need to call this then whenever we are done with the request.

lib/src/http_client.dart

@@ -42,7 +42,7 @@ class HttpClient {
   final Pointer<Cronet_Engine> _cronetEngine;
   // Keep all the request reference in a list so if the client is being
   // explicitly closed, we can clean up the requests.
-  final _requests = List<HttpClientRequest>.empty(growable: true);
+  final _requests = List<HttpClientRequestImpl>.empty(growable: true);
   var _stop = false;
 
   static const int defaultHttpPort = 80;
@@ -118,11 +118,24 @@ class HttpClient {
 
   /// Shuts down the [HttpClient].
   ///
-  /// The HttpClient will be kept alive until all active connections are done.
-  /// Trying to establish a new connection after calling close, will throw an
-  /// exception.
-  void close() {
+  /// The [HttpClient] will be kept alive until all active connections are done.
+  /// If [force] is true any active connections will be closed to immediately
+  /// release all resources. These closed connections will receive an error
+  /// event to indicate that the client was shut down. Trying to establish a
+  /// new connection after calling close, will throw an [Exception].
+  void close({bool force = false}) {
+    if (_stop) return;
     _stop = true;
+    if (force) {
+      // Deep copying the list because the original list may get modified
+      // during the traversal as cronet sends onCancel callbacks.
+      final requests = _requests.toList();
+      for (final request in requests) {
+        cronet.Cronet_UrlRequest_Cancel(request.requestPtr);
+        request.callbackHandler.controller
+            .addError(HttpException('HttpClient: Force Closed'));
+      }
+    }
   }
 
   /// Constructs [Uri] from [host], [port] & [path].

lib/src/http_client_request.dart

@@ -76,6 +76,16 @@ class HttpClientRequestImpl implements HttpClientRequest {
   /// Implemented by: http_client.dart.
   final void Function(HttpClientRequest) _clientCleanup;
 
+  /// Pointer associated with [this] request.
+  ///
+  /// This is not a part of public api.
+  Pointer<Cronet_UrlRequest> get requestPtr => _request;
+
+  /// [CallbackHandler] handling this request.
+  ///
+  /// This is not a part of public api.
+  CallbackHandler get callbackHandler => _callbackHandler;
+
   @override
   Encoding encoding;
 

pubspec.yaml

@@ -3,7 +3,7 @@
 # BSD-style license that can be found in the LICENSE file.
 
 name: cronet
-version: 0.0.4+1
+version: 0.0.4+2
 homepage: https://github.com/google/cronet.dart
 description: Experimental Cronet dart bindings.
 

test/http_client_close_test.dart

@@ -40,7 +40,24 @@ void main() {
       final resp = await request.close();
       client.close();
       final dataStream = resp.transform(utf8.decoder);
-      expect(dataStream, emitsInOrder(<dynamic>[equals(sentData), emitsDone]));
+      expect(dataStream, emitsInOrder(<Matcher>[equals(sentData), emitsDone]));
+    });
+
+    test(
+        'Force closing after starting a request cancels previous connections'
+        ' with error event', () async {
+      final client = HttpClient();
+      final request =
+          await client.openUrl('GET', Uri.parse('http://$host:$port'));
+      final resp = await request.close();
+      client.close(force: true);
+      final dataStream = resp.transform(utf8.decoder);
+      expect(
+          dataStream,
+          emitsInAnyOrder(<Matcher>[
+            mayEmit(isA<Stream<String>>()),
+            emitsInOrder(<Matcher>[emitsError(isA<HttpException>()), emitsDone])
+          ]));
     });
 
     tearDown(() {