Pool Manager: limit, pause and cancel requests

example/lib/main.dart

@@ -3,6 +3,7 @@ import 'dart:developer';
 import 'dart:io';
 
 import 'package:flutter/material.dart';
+import 'package:http/http.dart';
 import 'package:http_interceptor/http_interceptor.dart';
 import 'package:shared_preferences/shared_preferences.dart';
 

example/pubspec.yaml

@@ -12,6 +12,8 @@ dependencies:
   http_interceptor:
     path: ../
   shared_preferences: ^2.0.6
+  http: ^0.13.4
+
 
 dev_dependencies:
   flutter_test:

lib/exceptions/exceptions.dart

@@ -0,0 +1 @@
+export 'package:http_interceptor/exceptions/request_exceptions.dart';

lib/exceptions/request_exceptions.dart

@@ -0,0 +1,11 @@
+import 'package:http/http.dart';
+
+class RequestCancelledException implements Exception {
+  final BaseRequest request;
+
+  RequestCancelledException(this.request);
+
+  String toString() {
+    return 'The request has been cancelled (${request.url})';
+  }
+}

lib/http/intercepted_client.dart

@@ -2,10 +2,14 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:typed_data';
 
+import 'package:async/async.dart';
 import 'package:http/http.dart';
 import 'package:http_interceptor/extensions/extensions.dart';
+import 'package:http_interceptor/managers/pool_manager.dart';
 import 'package:http_interceptor/models/models.dart';
+import 'package:pool/pool.dart';
 
+import '../exceptions/exceptions.dart';
 import 'http_methods.dart';
 import 'interceptor_contract.dart';
 
@@ -38,6 +42,8 @@ import 'interceptor_contract.dart';
 /// Don't forget to close the client once you are done, as a client keeps
 /// the connection alive with the server by default.
 class InterceptedClient extends BaseClient {
+  static const String kSkipPoolHeader = 'Intercepted-Client-Skip-Pool-Manager';
+
   /// List of interceptors that will be applied to the requests and responses.
   final List<InterceptorContract> interceptors;
 
@@ -48,13 +54,17 @@ class InterceptedClient extends BaseClient {
   /// retry. This is useful for implementing JWT token expiration
   final RetryPolicy? retryPolicy;
 
-  int _retryCount = 0;
+  /// Manage the requests in a Pool.
+  PoolManager? poolManager;
+
+  Map<BaseRequest, int> _retryCount = {};
   late Client _inner;
 
   InterceptedClient._internal({
     required this.interceptors,
     this.requestTimeout,
     this.retryPolicy,
+    this.poolManager,
     Client? client,
   }) : _inner = client ?? Client();
 
@@ -80,12 +90,14 @@ class InterceptedClient extends BaseClient {
     Duration? requestTimeout,
     RetryPolicy? retryPolicy,
     Client? client,
+    PoolManager? poolManager,
   }) =>
       InterceptedClient._internal(
         interceptors: interceptors,
         requestTimeout: requestTimeout,
         retryPolicy: retryPolicy,
         client: client,
+        poolManager: poolManager,
       );
 
   @override
@@ -205,11 +217,25 @@ class InterceptedClient extends BaseClient {
 
   @override
   Future<StreamedResponse> send(BaseRequest request) async {
-    final response = await _attemptRequest(request);
+    PoolResource? poolResource = await _addToRequestPool(request.headers);
+    request.headers.remove(kSkipPoolHeader);
 
-    final interceptedResponse = await _interceptResponse(response);
+    try {
+      final interceptedRequest = await _interceptRequest(request);
 
-    return interceptedResponse as StreamedResponse;
+      final response = await _cancellableSendRequest(
+        interceptedRequest,
+        poolResource,
+      );
+
+      final interceptedResponse = await _interceptResponse(response);
+
+      await _releasePoolRequest(poolResource);
+      return interceptedResponse as StreamedResponse;
+    } catch (_) {
+      await _releasePoolRequest(poolResource);
+      rethrow;
+    }
   }
 
   Future<BaseResponse> _sendUnstreamed({
@@ -237,7 +263,8 @@ class InterceptedClient extends BaseClient {
       }
     }
 
-    var response = await _attemptRequest(request);
+    PoolResource? poolResource = await _addToRequestPool(request.headers);
+    var response = await _cancellableAttemptRequest(request, poolResource);
 
     // Intercept response
     response = await _interceptResponse(response);
@@ -256,7 +283,14 @@ class InterceptedClient extends BaseClient {
 
   /// Attempts to perform the request and intercept the data
   /// of the response
-  Future<BaseResponse> _attemptRequest(BaseRequest request) async {
+  Future<BaseResponse> _attemptRequest(
+      BaseRequest request, PoolResource? poolResource) async {
+    request.headers.remove(kSkipPoolHeader);
+
+    if (!_retryCount.containsKey(request)) {
+      _retryCount[request] = 0;
+    }
+
     BaseResponse response;
     try {
       // Intercept request
@@ -270,26 +304,75 @@ class InterceptedClient extends BaseClient {
           request is Request ? await Response.fromStream(stream) : stream;
 
       if (retryPolicy != null &&
-          retryPolicy!.maxRetryAttempts > _retryCount &&
+          retryPolicy!.maxRetryAttempts > _retryCount[request]! &&
           await retryPolicy!.shouldAttemptRetryOnResponse(response)) {
-        _retryCount += 1;
-        return _attemptRequest(request);
+        _retryCount[request] = _retryCount[request]! + 1;
+        response = await _attemptRequest(request, poolResource);
+        return response;
       }
     } on Exception catch (error) {
+      if (error is RequestCancelledException) {
+        _retryCount.remove(request);
+        await _releasePoolRequest(poolResource);
+        rethrow;
+      }
       if (retryPolicy != null &&
-          retryPolicy!.maxRetryAttempts > _retryCount &&
+          retryPolicy!.maxRetryAttempts > _retryCount[request]! &&
           retryPolicy!.shouldAttemptRetryOnException(error, request)) {
-        _retryCount += 1;
-        return _attemptRequest(request);
+        _retryCount[request] = _retryCount[request]! + 1;
+        try {
+          response = await _attemptRequest(request, poolResource);
+          return response;
+        } on Exception catch (_) {
+          _retryCount.remove(request);
+          rethrow;
+        }
       } else {
+        _retryCount.remove(request);
+        await _releasePoolRequest(poolResource);
         rethrow;
       }
     }
 
-    _retryCount = 0;
+    _retryCount.remove(request);
+    await _releasePoolRequest(poolResource);
     return response;
   }
 
+  /// Attempt a request, but allow it to be cancelled.
+  Future<BaseResponse> _cancellableAttemptRequest(
+      BaseRequest request, PoolResource? poolResource) async {
+    CancelableCompleter completer = CancelableCompleter();
+
+    poolManager?.addCancelableRequest(completer.operation);
+
+    completer.complete(_attemptRequest(request, poolResource));
+    return await completer.operation.valueOrCancellation().then((value) {
+      poolManager?.removeCancelableRequest(completer.operation);
+      if (completer.isCanceled && poolManager?.throwCanceledException == true) {
+        throw RequestCancelledException(request);
+      }
+      return value;
+    });
+  }
+
+  /// Attempt a send request, but allow it to be cancelled.
+  Future<StreamedResponse> _cancellableSendRequest(
+      BaseRequest request, PoolResource? poolResource) async {
+    CancelableCompleter completer = CancelableCompleter();
+
+    poolManager?.addCancelableRequest(completer.operation);
+
+    completer.complete(_inner.send(request));
+    return await completer.operation.valueOrCancellation().then((value) {
+      poolManager?.removeCancelableRequest(completer.operation);
+      if (completer.isCanceled && poolManager?.throwCanceledException == true) {
+        throw RequestCancelledException(request);
+      }
+      return value;
+    });
+  }
+
   /// This internal function intercepts the request.
   Future<BaseRequest> _interceptRequest(BaseRequest request) async {
     BaseRequest interceptedRequest = request.copyWith();
@@ -317,4 +400,26 @@ class InterceptedClient extends BaseClient {
   void close() {
     _inner.close();
   }
+
+  /// Add a new request to the pool.
+  /// If [kSkipPoolHeader] is found in the headers, the pool is skipped so
+  /// the request is executed immediately.
+  Future<PoolResource?> _addToRequestPool(Map<String, String>? headers) async {
+    if (headers?.keys.contains(kSkipPoolHeader) == true) {
+      return null;
+    }
+
+    if (poolManager == null) {
+      return null;
+    }
+    return await poolManager?.request();
+  }
+
+  /// Release a [PoolRequest]
+  Future<void> _releasePoolRequest(PoolResource? poolResource) async {
+    if (poolResource == null) {
+      return;
+    }
+    await poolManager?.release(poolResource);
+  }
 }

lib/http/intercepted_http.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:typed_data';
 
+import 'package:http/http.dart';
 import 'package:http_interceptor/http_interceptor.dart';
 
 /// Class to be used by the user as a replacement for 'http' with interceptor

lib/http_interceptor.dart

@@ -1,6 +1,7 @@
 library http_interceptor;
 
-export 'package:http/http.dart';
+export 'package:http_interceptor/exceptions/exceptions.dart';
 export 'package:http_interceptor/extensions/extensions.dart';
 export 'package:http_interceptor/http/http.dart';
+export 'package:http_interceptor/managers/managers.dart';
 export 'package:http_interceptor/models/models.dart';

lib/managers/managers.dart

@@ -0,0 +1 @@
+export 'pool_manager.dart';