Added Throttle::join loop to avoid overloading throttle.

Author: Bilge

composer.json

@@ -11,7 +11,7 @@
   "require": {
     "php": "^7.1",
     "amphp/amp": "^2",
-    "async/throttle": "^2",
+    "async/throttle": "dev-join",
     "psr/cache": "^1",
     "psr/container": "^1",
     "scriptfusion/retry": "^2.1",

src/Connector/ImportConnector.php

@@ -103,6 +103,10 @@ public function fetchAsync(AsyncDataSource $source): Promise
             $this->maxFetchAttempts,
             function () use ($source): Promise {
                 return call(function () use ($source): \Generator {
+                    while (!yield $this->throttle->join()) {
+                        // Throttle is choked. Wait for free slot.
+                    }
+
                     yield $this->throttle->await($response = $this->connector->fetchAsync($source));
 
                     return yield $response;

test/FixtureFactory.php

@@ -3,6 +3,7 @@
 
 namespace ScriptFUSIONTest;
 
+use Amp\Success;
 use ScriptFUSION\Async\Throttle\Throttle;
 use ScriptFUSION\Porter\Connector\Connector;
 use ScriptFUSION\Porter\Connector\ImportConnector;
@@ -26,6 +27,9 @@ public static function buildImportConnector(
             $maxFetchAttempts,
             $mustCache,
             \Mockery::mock(Throttle::class)
+                ->shouldReceive('join')
+                    ->andReturn(new Success(true))
+                ->getMock()
         );
     }
 }