docker-php · joelwurtz · Feb 17, 2019 · May 12, 2018 · May 12, 2018 · May 12, 2018
diff --git a/src/Client/AmpArtaxStreamEndpoint.php b/src/Client/AmpArtaxStreamEndpoint.php
@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Docker\Client;
+
+use Amp\Artax\Response;
+use Amp\CancellationTokenSource;
+use Amp\Promise;
+use Jane\OpenApiRuntime\Client\Client;
+use Symfony\Component\Serializer\SerializerInterface;
+
+interface AmpArtaxStreamEndpoint
+{
+    /**
+     * Parse and transform an Artax InputStream chunk into a different object.
+     *
+     * Implementations may vary depending the status code of the response and the fetch mode used.
+     */
+    public function parseArtaxStreamResponse(
+        Response $response,
+        SerializerInterface $serializer,
+        CancellationTokenSource $cancellationTokenSource,
+        string $fetchMode = Client::FETCH_OBJECT
+    ): Promise;
+}
diff --git a/src/Client/AmpArtaxStreamEndpointTrait.php b/src/Client/AmpArtaxStreamEndpointTrait.php
@@ -0,0 +1,41 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Docker\Client;
+
+use Amp\Artax\Response;
+use Amp\CancellationTokenSource;
+use Amp\Promise;
+use Docker\Stream\ArtaxCallbackStream;
+use Jane\OpenApiRuntime\Client\Client;
+use Jane\OpenApiRuntime\Client\Exception\InvalidFetchModeException;
+use Symfony\Component\Serializer\SerializerInterface;
+use function Amp\call;
+
+trait AmpArtaxStreamEndpointTrait
+{
+    abstract protected function transformResponseBody(string $body, int $status, SerializerInterface $serializer);
+
+    public function parseArtaxStreamResponse(
+        Response $response,
+        SerializerInterface $serializer,
+        CancellationTokenSource $cancellationTokenSource,
+        string $fetchMode = Client::FETCH_OBJECT
+    ): Promise {
+        if (!\in_array($fetchMode, [Client::FETCH_OBJECT, Client::FETCH_RESPONSE], true)) {
+            throw new InvalidFetchModeException(\sprintf('Fetch mode %s is not supported', $fetchMode));
+        }
+
+        return call(function () use ($response, $serializer, $fetchMode, $cancellationTokenSource) {
+            $responseTransformer = null;
+            if (Client::FETCH_OBJECT === $fetchMode) {
+                $responseTransformer = function ($chunk) use ($response, $serializer) {
+                    return $this->transformResponseBody($chunk, $response->getStatus(), $serializer);
+                };
+            }
+
+            return new ArtaxCallbackStream($response->getBody(), $cancellationTokenSource, $responseTransformer);
+        });
+    }
+}
diff --git a/src/Client/ProvideAmpArtaxClientOptions.php b/src/Client/ProvideAmpArtaxClientOptions.php
@@ -0,0 +1,15 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ElevenLabs\Docker\Client;
+
+namespace Docker\Client;
+
+interface ProvideAmpArtaxClientOptions
+{
+    /**
+     * Return a list of options for the Artax client.
+     */
+    public function getAmpArtaxClientOptions(): array;
+}
diff --git a/src/DockerAsync.php b/src/DockerAsync.php
@@ -4,7 +4,15 @@
 
 namespace Docker;
 
+use Amp\Artax\Request;
+use Amp\CancellationTokenSource;
+use Amp\Promise;
 use Docker\API\ClientAsync;
+use Docker\Client\AmpArtaxStreamEndpoint;
+use Docker\Client\ProvideAmpArtaxClientOptions;
+use Docker\Endpoint\SystemEvents;
+use Jane\OpenApiRuntime\Client\AmpArtaxEndpoint;
+use function Amp\call;
 
 /**
  * Docker\Docker.
@@ -19,4 +27,48 @@ public static function create($httpClient = null)
 
         return parent::create($httpClient);
     }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function systemEvents(array $queryParameters = [], string $fetch = self::FETCH_OBJECT): Promise
+    {
+        return $this->executeArtaxEndpoint(new SystemEvents($queryParameters), $fetch);
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function executeArtaxEndpoint(AmpArtaxEndpoint $endpoint, string $fetch = self::FETCH_OBJECT): Promise
+    {
+        return call(function () use ($endpoint, $fetch) {
+            [$bodyHeaders, $body] = $endpoint->getBody($this->serializer);
+            $queryString = $endpoint->getQueryString();
+            $uri = '' !== $queryString ? $endpoint->getUri().'?'.$queryString : $endpoint->getUri();
+            $request = new Request($uri, $endpoint->getMethod());
+            $request = $request->withBody($body);
+            $request = $request->withHeaders($endpoint->getHeaders($bodyHeaders));
+            $options = [];
+            if ($endpoint instanceof ProvideAmpArtaxClientOptions) {
+                $options = $endpoint->getAmpArtaxClientOptions();
+            }
+
+            if ($endpoint instanceof AmpArtaxStreamEndpoint) {
+                $cancellationTokenSource = new CancellationTokenSource();
+
+                return $endpoint->parseArtaxStreamResponse(
+                    yield $this->httpClient->request($request, $options, $cancellationTokenSource->getToken()),
+                    $this->serializer,
+                    $cancellationTokenSource,
+                    $fetch
+                );
+            }
+
+            return $endpoint->parseArtaxResponse(
+                yield $this->httpClient->request($request, $options),
+                $this->serializer,
+                $fetch
+            );
+        });
+    }
 }
diff --git a/src/Endpoint/SystemEvents.php b/src/Endpoint/SystemEvents.php
@@ -4,15 +4,26 @@
 
 namespace Docker\Endpoint;
 
+use Amp\Artax\Client as ArtaxClient;
 use Docker\API\Endpoint\SystemEvents as BaseEndpoint;
+use Docker\Client\AmpArtaxStreamEndpoint;
+use Docker\Client\AmpArtaxStreamEndpointTrait;
+use Docker\Client\ProvideAmpArtaxClientOptions;
 use Docker\Stream\EventStream;
 use Jane\OpenApiRuntime\Client\Client;
 use Jane\OpenApiRuntime\Client\Exception\InvalidFetchModeException;
 use Psr\Http\Message\ResponseInterface;
 use Symfony\Component\Serializer\SerializerInterface;
 
-class SystemEvents extends BaseEndpoint
+class SystemEvents extends BaseEndpoint implements ProvideAmpArtaxClientOptions, AmpArtaxStreamEndpoint
 {
+    use AmpArtaxStreamEndpointTrait;
+
+    public function getAmpArtaxClientOptions(): array
+    {
+        return [ArtaxClient::OP_TRANSFER_TIMEOUT => 0];
+    }
+
     public function parsePSR7Response(ResponseInterface $response, SerializerInterface $serializer, string $fetchMode = Client::FETCH_OBJECT)
     {
         if (Client::FETCH_OBJECT === $fetchMode) {

diff --git a/src/Stream/ArtaxCallbackStream.php b/src/Stream/ArtaxCallbackStream.php
@@ -0,0 +1,78 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Docker\Stream;
+
+use Amp\ByteStream\InputStream;
+use Amp\CancellationTokenSource;
+use Amp\Promise;
+use function Amp\call;
+
+class ArtaxCallbackStream
+{
+    private $stream;
+    private $onNewFrameCallables = [];
+    private $chunkTransformer;
+    private $cancellationTokenSource;
+
+    public function __construct(
+        InputStream $stream,
+        CancellationTokenSource $cancellationTokenSource,
+        ?callable $chunkTransformer
+    ) {
+        $this->stream = $stream;
+        $this->cancellationTokenSource = $cancellationTokenSource;
+        $this->chunkTransformer = $chunkTransformer;
+    }
+
+    /**
+     * Called when there is a new frame from the stream.
+     *
+     * @param callable $onNewFrame
+     */
+    public function onFrame(callable $onNewFrame): void
+    {
+        $this->onNewFrameCallables[] = $onNewFrame;
+    }
+
+    /**
+     * Consume stream chunks.
+     *
+     * @return Promise
+     */
+    public function listen(): Promise
+    {
+        return call(function () {
+            while (null !== ($chunk = yield $this->stream->read())) {
+                foreach ($this->onNewFrameCallables as $newFrameCallable) {
+                    $newFrameCallable($this->transformChunk($chunk));
+                }
+            }
+        });
+    }
+
+    /**
+     * Stop consuming stream chunks.
+     */
+    public function cancel(): void
+    {
+        $this->cancellationTokenSource->cancel();
+    }
+
+    /**
+     * Transform stream chunks if required.
+     *
+     * @param string $chunk
+     *
+     * @return mixed The raw chunk or the transformed chunk
+     */
+    private function transformChunk(string $chunk)
+    {
+        if (null === $this->chunkTransformer) {
+            return $chunk;
+        }
+
+        return \call_user_func($this->chunkTransformer, $chunk);
+    }
+}
diff --git a/tests/DockerAsyncTest.php b/tests/DockerAsyncTest.php
@@ -6,7 +6,9 @@
 
 use Amp\Loop;
 use Docker\API\Model\ContainersCreatePostBody;
+use Docker\API\Model\EventsGetResponse200;
 use Docker\DockerAsync;
+use Docker\Stream\ArtaxCallbackStream;
 
 class DockerAsyncTest extends \PHPUnit\Framework\TestCase
 {
@@ -39,4 +41,46 @@ public function testAsync(): void
             $this->assertSame($containerCreate->getId(), $containerInfo->getId());
         });
     }
+
+    public function testSystemEventsAllowTheConsumptionOfDockerEvents(): void
+    {
+        $matchedEvents = [];
+
+        Loop::run(function () use (&$matchedEvents) {
+            $docker = DockerAsync::create();
+
+            /** @var ArtaxCallbackStream $events */
+            $events = yield $docker->systemEvents([
+                'filters' => \json_encode(
+                    [
+                        'type' => ['container'],
+                        'action' => ['create'],
+                    ]
+                ),
+            ]);
+            $events->onFrame(function ($event) use (&$matchedEvents): void {
+                if (\is_object($event)
+                    && $event instanceof EventsGetResponse200
+                    && 'create' === $event->getAction()
+                    && 'container' === $event->getType()
+                ) {
+                    $matchedEvents[] = $event;
+                }
+            });
+
+            $events->listen();
+
+            $containerConfig = new ContainersCreatePostBody();
+            $containerConfig->setImage('busybox:latest');
+            $containerConfig->setCmd(['echo', '-n', 'output']);
+
+            yield $docker->containerCreate($containerConfig);
+
+            Loop::delay(1000, function (): void {
+                Loop::stop();
+            });
+        });
+
+        $this->assertCount(1, $matchedEvents);
+    }
 }