Added concurrent-fiber throttle test.

Author: Bilge

composer.json

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

src/Specification/AsyncImportSpecification.php

@@ -3,6 +3,7 @@
 
 namespace ScriptFUSION\Porter\Specification;
 
+use ScriptFUSION\Async\Throttle\NullThrottle;
 use ScriptFUSION\Async\Throttle\Throttle;
 use ScriptFUSION\Porter\Connector\Recoverable\ExponentialAsyncDelayRecoverableExceptionHandler;
 use ScriptFUSION\Porter\Connector\Recoverable\RecoverableExceptionHandler;
@@ -61,7 +62,7 @@ protected static function createDefaultRecoverableExceptionHandler(): Recoverabl
 
     final public function getThrottle(): Throttle
     {
-        return $this->throttle ?? $this->throttle = new Throttle();
+        return $this->throttle ?? $this->throttle = new NullThrottle;
     }
 
     final public function setThrottle(Throttle $throttle): self

test/Integration/PorterAsyncTest.php

@@ -6,6 +6,8 @@
 use Amp\Iterator;
 use Amp\Loop;
 use Amp\Producer;
+use Amp\Promise;
+use ScriptFUSION\Async\Throttle\DualThrottle;
 use ScriptFUSION\Porter\Collection\AsyncRecordCollection;
 use ScriptFUSION\Porter\Collection\CountableAsyncPorterRecords;
 use ScriptFUSION\Porter\Collection\CountableAsyncProviderRecords;
@@ -22,6 +24,7 @@
 use ScriptFUSION\Porter\Transform\AsyncTransformer;
 use ScriptFUSION\Porter\Transform\FilterTransformer;
 use ScriptFUSIONTest\MockFactory;
+use function Amp\call;
 
 /**
  * @see Porter
@@ -200,20 +203,44 @@ public function testPorterAwareAsyncTransformer(): void
     }
 
     /**
-     * Tests that the throttle is invoked during fetch operations.
+     * Tests that a working throttle implementation is invoked during fetch operations.
      */
     public function testThrottle(): \Generator
     {
-        $throttle = $this->specification->getThrottle();
-        $throttle->setMaxPerSecond(1);
-
-        $start = microtime(true);
+        $this->specification->setThrottle($throttle = new DualThrottle);
+        $throttle->setMaxConcurrency(1);
 
         $records = $this->porter->importAsync($this->specification);
         self::assertTrue($throttle->isThrottling());
 
         yield $records->advance();
         self::assertFalse($throttle->isThrottling());
-        self::assertGreaterThan(1, microtime(true) - $start);
+    }
+
+    /**
+     * Tests that a working throttle implementation can be called from multiple fibers queueing excess objects.
+     */
+    public function testThrottleConcurrentFibers(): \Generator
+    {
+        $this->specification->setThrottle($throttle = new DualThrottle);
+        $throttle->setMaxPerSecond(1);
+
+        $import = function (): Promise {
+            return call(
+                function (): \Generator {
+                    $records = $this->porter->importAsync($this->specification);
+
+                    while (yield $records->advance()) {
+                        // Do nothing.
+                    }
+                }
+            );
+        };
+
+        $start = microtime(true);
+
+        yield [$import(), $import(), $import()];
+
+        self::assertGreaterThan(3, microtime(true) - $start);
     }
 }