Support keyed response for amp

phanxipang · Sep 27, 2023 · 49ac713 · 49ac713
1 parent ab7edb4
commit 49ac713
Show file tree

Hide file tree

Showing 9 changed files with 171 additions and 51 deletions.
diff --git a/src/Concurrency/AmpTask.php b/src/Concurrency/AmpTask.php
@@ -0,0 +1,42 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fansipan\Peak\Concurrency;
+
+/**
+ * @template T
+ */
+final class AmpTask
+{
+    /**
+     * @var T
+     */
+    private mixed $value = null;
+
+    /**
+     * @param  \Closure(): T $task
+     */
+    public function __construct(
+        private readonly string|int $key,
+        private readonly \Closure $task
+    ) {
+    }
+
+    public function key(): string|int
+    {
+        return $this->key;
+    }
+
+    public function value(): mixed
+    {
+        return $this->value;
+    }
+
+    public function __invoke(): self
+    {
+        $this->value = ($this->task)();
+
+        return $this;
+    }
+}
diff --git a/src/Concurrency/AmpWorker.php b/src/Concurrency/AmpWorker.php
@@ -22,11 +22,26 @@ public function __construct(private readonly int $limit = 10)
 
     public function run(iterable $tasks): array
     {
-        $promises = Pipeline::fromIterable($tasks)
+        $promises = Pipeline::fromIterable(static function () use ($tasks): \Generator {
+            foreach ($tasks as $key => $task) {
+                yield new AmpTask($key, $task);
+            }
+        })
             ->concurrent($this->limit)
             ->unordered()
-            ->map(static fn (\Closure $task) => Amp\async($task));
+            ->map(static fn (AmpTask $task) => Amp\async(\Closure::fromCallable($task)));
 
-        return Future\awaitAll($promises)[1] ?? [];
+        $results = [];
+
+        foreach (Future::iterate($promises) as $promise) {
+            try {
+                /** @var AmpTask $t */
+                $t = $promise->await();
+                $results[$t->key()] = $t->value();
+            } catch (\Throwable) {
+            }
+        }
+
+        return $results;
     }
 }
diff --git a/tests/AmpPoolTest.php b/tests/AmpPoolTest.php
@@ -29,4 +29,11 @@ public function test_amp_pool_using_guzzle(): void
     {
         $this->performConnectorTests($this->createConnector($this->createGuzzleClient()));
     }
+
+    public function test_amp_pool_keyed_response(): void
+    {
+        $this->performKeyedResponseTests(
+            $this->mockSymfonyClient(new AmpDeferred())
+        );
+    }
 }
diff --git a/tests/DelayTest.php b/tests/DelayTest.php
@@ -6,24 +6,20 @@
 
 use Fansipan\Mock\MockResponse;
 use Fansipan\Peak\Client\AsyncClientInterface;
-use Fansipan\Peak\Client\GuzzleClient;
+use Fansipan\Peak\Client\Delayable;
 use Fansipan\Peak\Client\ReactClient;
-use Fansipan\Peak\Client\SymfonyClient;
-use Fansipan\Peak\Concurrency\Deferrable;
 use Fansipan\Peak\Concurrency\PslDeferred;
 use Fansipan\Peak\Concurrency\ReactDeferred;
 use Fansipan\Peak\PoolFactory;
-use GuzzleHttp\Client;
-use GuzzleHttp\Handler\MockHandler;
-use GuzzleHttp\HandlerStack;
 use Http\Discovery\Psr17FactoryDiscovery;
 use PHPUnit\Framework\TestCase;
 use Psr\Http\Message\RequestFactoryInterface;
 use Psr\Http\Message\ResponseInterface;
-use Symfony\Component\HttpClient\MockHttpClient;
 
 final class DelayTest extends TestCase
 {
+    use TestTrait;
+
     private RequestFactoryInterface $requestFactory;
 
     protected function setUp(): void
@@ -33,6 +29,26 @@ protected function setUp(): void
         $this->requestFactory = Psr17FactoryDiscovery::findRequestFactory();
     }
 
+    public function test_amp_delay(): void
+    {
+        $request = $this->requestFactory->createRequest('GET', 'http://localhost');
+
+        $client = $this->mockGuzzleClient(new PslDeferred(), [
+            MockResponse::create(''),
+        ]);
+
+        $reflection = new \ReflectionProperty($client, 'delay');
+        $reflection->setAccessible(true);
+
+        $client->delay(1000);
+
+        $this->assertSame(1000, $reflection->getValue($client));
+
+        $client->sendRequest($request);
+
+        $this->assertSame(0, $reflection->getValue($client));
+    }
+
     public function test_psl_delay(): void
     {
         $request = $this->requestFactory->createRequest('GET', 'http://localhost');
@@ -97,7 +113,7 @@ private function runPoolDelayTests(AsyncClientInterface $client, int $totalReque
     {
         $requests = function (int $total) use ($delay) {
             for ($i = 0; $i < $total; $i++) {
-                yield function (AsyncClientInterface $client) use ($delay): ResponseInterface {
+                yield function (AsyncClientInterface&Delayable $client) use ($delay): ResponseInterface {
                     $client->delay($delay);
 
                     return $client->sendRequest($this->requestFactory->createRequest('GET', 'http://localhost'));
@@ -108,17 +124,4 @@ private function runPoolDelayTests(AsyncClientInterface $client, int $totalReque
         $pool = PoolFactory::createFromClient($client);
         $pool->send($requests($totalRequests));
     }
-
-    private function mockGuzzleClient(Deferrable $defer, ?array $response = null): GuzzleClient
-    {
-        $handler = new MockHandler($response);
-        $handlerStack = HandlerStack::create($handler);
-
-        return new GuzzleClient($defer, new Client(['handler' => $handlerStack]));
-    }
-
-    private function mockSymfonyClient(Deferrable $defer, mixed $response = null): SymfonyClient
-    {
-        return new SymfonyClient($defer, new MockHttpClient($response));
-    }
 }
diff --git a/tests/PslPoolTest.php b/tests/PslPoolTest.php
@@ -20,13 +20,20 @@ private function createGuzzleClient(): GuzzleClient
         return new GuzzleClient(new PslDeferred());
     }
 
-    public function test_react_pool_using_symfony_http_client(): void
+    public function test_psl_pool_using_symfony_http_client(): void
     {
         $this->performConnectorTests($this->createConnector($this->createSymfonyClient()));
     }
 
-    public function test_react_pool_using_guzzle(): void
+    public function test_psl_pool_using_guzzle(): void
     {
         $this->performConnectorTests($this->createConnector($this->createGuzzleClient()));
     }
+
+    public function test_psl_pool_keyed_response(): void
+    {
+        $this->performKeyedResponseTests(
+            $this->mockSymfonyClient(new PslDeferred())
+        );
+    }
 }
diff --git a/tests/ReactPoolTest.php b/tests/ReactPoolTest.php
@@ -33,4 +33,11 @@ public function test_react_pool_using_guzzle(): void
     {
         $this->performConnectorTests($this->createConnector($this->createGuzzleClient()));
     }
+
+    public function test_react_pool_keyed_response(): void
+    {
+        $this->performKeyedResponseTests(
+            $this->mockSymfonyClient(new ReactDeferred())
+        );
+    }
 }
diff --git a/tests/TestCase.php b/tests/TestCase.php
@@ -7,6 +7,8 @@
 use Fansipan\Peak\Client\AsyncClientInterface;
 use Fansipan\Peak\ClientPool;
 use Fansipan\Peak\ConnectorPool;
+use Fansipan\Peak\PoolFactory;
+use Http\Discovery\Psr17FactoryDiscovery;
 use Jenky\Atlas\Contracts\ConnectorInterface;
 use Jenky\Atlas\GenericConnector;
 use Jenky\Atlas\Middleware\Interceptor;
@@ -17,7 +19,7 @@
 
 abstract class TestCase extends BaseTestCase
 {
-    use TestRequestTrait;
+    use TestTrait;
 
     protected function createConnector(?ClientInterface $client = null): ConnectorInterface
     {
@@ -54,4 +56,21 @@ protected function performConnectorTests(ConnectorInterface $connector): void
         $this->assertInstanceOf(Response::class, $responses[0]);
         $this->assertSame('bar', $responses[0]->header('X-Foo'));
     }
+
+    protected function performKeyedResponseTests(AsyncClientInterface $client): void
+    {
+        $requestFactory = Psr17FactoryDiscovery::findRequestFactory();
+
+        $responses = PoolFactory::createFromClient($client)
+            ->send([
+                'foo' => $requestFactory->createRequest('GET', 'http://localhost/foo'),
+                'bar' => $requestFactory->createRequest('GET', 'http://localhost/bar'),
+            ]);
+
+        $this->assertArrayHasKey('foo', $responses);
+        $this->assertArrayHasKey('bar', $responses);
+
+        $this->assertSame(200, $responses['foo']->getStatusCode());
+        $this->assertSame(200, $responses['bar']->getStatusCode());
+    }
 }
diff --git a/tests/TestRequestTrait.php b/tests/TestRequestTrait.php
diff --git a/tests/TestTrait.php b/tests/TestTrait.php
@@ -0,0 +1,44 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Fansipan\Peak\Tests;
+
+use Fansipan\Peak\Client\GuzzleClient;
+use Fansipan\Peak\Client\SymfonyClient;
+use Fansipan\Peak\Concurrency\Deferrable;
+use GuzzleHttp\Client;
+use GuzzleHttp\Handler\MockHandler;
+use GuzzleHttp\HandlerStack;
+use Jenky\Atlas\Util;
+use Symfony\Component\HttpClient\MockHttpClient;
+
+trait TestTrait
+{
+    protected function createRequests(int $total): iterable
+    {
+        for ($i = 1; $i <= $total; $i++) {
+            yield new AkamaiTileRequest($i);
+        }
+    }
+
+    protected function createPsrRequests(int $total): iterable
+    {
+        for ($i = 1; $i <= $total; $i++) {
+            yield Util::request(new AkamaiTileRequest($i));
+        }
+    }
+
+    protected function mockGuzzleClient(Deferrable $defer, ?array $response = null): GuzzleClient
+    {
+        $handler = new MockHandler($response);
+        $handlerStack = HandlerStack::create($handler);
+
+        return new GuzzleClient($defer, new Client(['handler' => $handlerStack]));
+    }
+
+    protected function mockSymfonyClient(Deferrable $defer, mixed $response = null): SymfonyClient
+    {
+        return new SymfonyClient($defer, new MockHttpClient($response));
+    }
+}