diff --git a/src/Client/AmpArtaxStreamEndpoint.php b/src/Client/AmpArtaxStreamEndpoint.php new file mode 100644 index 00000000..ec7f3b9a --- /dev/null +++ b/src/Client/AmpArtaxStreamEndpoint.php @@ -0,0 +1,26 @@ +transformResponseBody($chunk, $response->getStatus(), $serializer); + }; + } + + return new ArtaxCallbackStream($response->getBody(), $cancellationTokenSource, $responseTransformer); + }); + } +} diff --git a/src/Client/ProvideAmpArtaxClientOptions.php b/src/Client/ProvideAmpArtaxClientOptions.php new file mode 100644 index 00000000..cd1aa87d --- /dev/null +++ b/src/Client/ProvideAmpArtaxClientOptions.php @@ -0,0 +1,15 @@ +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 index ade82394..712e67da 100644 --- 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 new file mode 100644 index 00000000..f85d5219 --- /dev/null +++ b/src/Stream/ArtaxCallbackStream.php @@ -0,0 +1,78 @@ +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 index 2fd15108..29b2f3bf 100644 --- 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); + } }