Added Throttle support for async imports only.

Author: Bilge

composer.json

@@ -10,12 +10,13 @@
   "license": "LGPL-3.0",
   "require": {
     "php": "^7.1",
-    "scriptfusion/static-class": "^1",
-    "scriptfusion/retry": "^2.1",
-    "scriptfusion/retry-exception-handlers": "^1.2",
     "amphp/amp": "^2",
+    "async/throttle": "^2",
+    "psr/cache": "^1",
     "psr/container": "^1",
-    "psr/cache": "^1"
+    "scriptfusion/retry": "^2.1",
+    "scriptfusion/retry-exception-handlers": "^1.2",
+    "scriptfusion/static-class": "^1"
   },
   "require-dev": {
     "amphp/phpunit-util": "^1.1",

src/Connector/ImportConnector.php

@@ -4,10 +4,12 @@
 namespace ScriptFUSION\Porter\Connector;
 
 use Amp\Promise;
+use ScriptFUSION\Async\Throttle\Throttle;
 use ScriptFUSION\Porter\Cache\CacheUnavailableException;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableException;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
 use ScriptFUSION\Porter\Connector\Recoverable\StatelessRecoverableExceptionHandler;
+use function Amp\call;
 use function Amp\Promise\all;
 use function ScriptFUSION\Retry\retry;
 use function ScriptFUSION\Retry\retryAsync;
@@ -40,6 +42,8 @@ final class ImportConnector implements ConnectorWrapper
 
     private $maxFetchAttempts;
 
+    private $throttle;
+
     /**
      * @param Connector|AsyncConnector $connector Wrapped connector.
      * @param RecoverableExceptionHandler $recoverableExceptionHandler User's recoverable exception handler.
@@ -50,7 +54,8 @@ public function __construct(
         $connector,
         RecoverableExceptionHandler $recoverableExceptionHandler,
         int $maxFetchAttempts,
-        bool $mustCache
+        bool $mustCache,
+        ?Throttle $throttle
     ) {
         if ($mustCache && !$connector instanceof CachingConnector) {
             throw CacheUnavailableException::createUnsupported();
@@ -64,6 +69,7 @@ public function __construct(
         );
         $this->userExceptionHandler = $recoverableExceptionHandler;
         $this->maxFetchAttempts = $maxFetchAttempts;
+        $this->throttle = $throttle;
     }
 
     /**
@@ -96,7 +102,11 @@ public function fetchAsync(AsyncDataSource $source): Promise
         return retryAsync(
             $this->maxFetchAttempts,
             function () use ($source): Promise {
-                return $this->connector->fetchAsync($source);
+                return call(function () use ($source): \Generator {
+                    yield $this->throttle->await($response = $this->connector->fetchAsync($source));
+
+                    return yield $response;
+                });
             },
             $this->createExceptionHandler()
         );

src/Connector/ImportConnectorFactory.php

@@ -3,6 +3,7 @@
 
 namespace ScriptFUSION\Porter\Connector;
 
+use ScriptFUSION\Porter\Specification\AsyncImportSpecification;
 use ScriptFUSION\Porter\Specification\Specification;
 use ScriptFUSION\StaticClass;
 
@@ -22,7 +23,8 @@ public static function create($connector, Specification $specification): ImportC
             $connector,
             $specification->getRecoverableExceptionHandler(),
             $specification->getMaxFetchAttempts(),
-            $specification->mustCache()
+            $specification->mustCache(),
+            $specification instanceof AsyncImportSpecification ? $specification->getThrottle() : null
         );
     }
 }

src/Specification/AsyncImportSpecification.php

@@ -3,6 +3,7 @@
 
 namespace ScriptFUSION\Porter\Specification;
 
+use ScriptFUSION\Async\Throttle\Throttle;
 use ScriptFUSION\Porter\Connector\Recoverable\ExponentialAsyncDelayRecoverableExceptionHandler;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
 use ScriptFUSION\Porter\Provider\Resource\AsyncResource;
@@ -15,6 +16,9 @@ class AsyncImportSpecification extends Specification
 {
     private $asyncResource;
 
+    /** @var Throttle */
+    private $throttle;
+
     /**
      * Initializes this instance with the specified asynchronous resource.
      *
@@ -53,4 +57,16 @@ protected static function createDefaultRecoverableExceptionHandler(): Recoverabl
     {
         return new ExponentialAsyncDelayRecoverableExceptionHandler;
     }
+
+    final public function getThrottle(): Throttle
+    {
+        return $this->throttle ?? $this->throttle = new Throttle();
+    }
+
+    final public function setThrottle(Throttle $throttle): self
+    {
+        $this->throttle = $throttle;
+
+        return $this;
+    }
 }

test/FixtureFactory.php

@@ -3,6 +3,7 @@
 
 namespace ScriptFUSIONTest;
 
+use ScriptFUSION\Async\Throttle\Throttle;
 use ScriptFUSION\Porter\Connector\Connector;
 use ScriptFUSION\Porter\Connector\ImportConnector;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
@@ -23,7 +24,8 @@ public static function buildImportConnector(
             $connector,
             $recoverableExceptionHandler ?: \Mockery::spy(RecoverableExceptionHandler::class),
             $maxFetchAttempts,
-            $mustCache
+            $mustCache,
+            \Mockery::mock(Throttle::class)
         );
     }
 }