From 14cb823e5542d6f040424f2771bf00b56242fabe Mon Sep 17 00:00:00 2001 From: Benjamin Fahl Date: Mon, 13 Jul 2026 21:59:50 +0200 Subject: [PATCH 1/2] fix(events): parse newline-delimited JSON from Docker event stream Docker's /events endpoint returns newline-delimited JSON with Content-Type: application/json, not Server-Sent Events. The old loop only handled ServerSentEvent chunks, which a plain HttpClient never emits, so listenForEvents() never invoked the callback and container events were silently dropped. - Buffer raw data chunks and split on newlines, decoding each complete JSON line and denormalizing the decoded array (no second JSON parse). - Drop the duplicate /events request issued on every loop iteration. - Inject an optional event-stream HttpClient so the method is unit testable via MockHttpClient; production still builds the socket-bound client by default. - Add unit tests for dispatch filtering, chunk buffering and invalid JSON handling. Drive-by: cs:fix adds a missing closure return type in EventsListenCommand. --- src/Client/DockerApiClientWrapper.php | 57 ++++++----- src/Command/EventsListenCommand.php | 2 +- .../Client/DockerApiClientWrapperTest.php | 97 +++++++++++++++++++ 3 files changed, 130 insertions(+), 26 deletions(-) diff --git a/src/Client/DockerApiClientWrapper.php b/src/Client/DockerApiClientWrapper.php index 8ae4fd4..f70674c 100644 --- a/src/Client/DockerApiClientWrapper.php +++ b/src/Client/DockerApiClientWrapper.php @@ -5,18 +5,22 @@ namespace WebProject\DockerApiClient\Client; use JsonException; -use Symfony\Component\HttpClient\Chunk\ServerSentEvent; use Symfony\Component\HttpClient\HttpClient; use Symfony\Component\HttpClient\Psr18Client; use Symfony\Component\Serializer\Encoder\JsonEncoder; use Symfony\Component\Serializer\Normalizer\ObjectNormalizer; use Symfony\Component\Serializer\Serializer; use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface; +use Symfony\Contracts\HttpClient\HttpClientInterface; use Webmozart\Assert\Assert; use WebProject\DockerApi\Library\Generated\Client; use WebProject\DockerApiClient\Event\ContainerEvent; use function is_array; use function json_decode; +use function json_validate; +use function strpos; +use function substr; +use function trim; final class DockerApiClientWrapper { @@ -24,6 +28,7 @@ public function __construct( private readonly string $baseUri, private readonly string $socketPath, private readonly Client $client, + private readonly ?HttpClientInterface $eventStreamClient = null, ) { } @@ -35,7 +40,7 @@ public function __construct( */ public function listenForEvents(callable $eventCallback): void { - $client = HttpClient::create([ + $client = $this->eventStreamClient ?? HttpClient::create([ 'base_uri' => $this->baseUri, 'bindto' => $this->socketPath, 'timeout' => null, @@ -46,35 +51,37 @@ public function listenForEvents(callable $eventCallback): void encoders: [new JsonEncoder()] ); - // Connect to the Docker API event stream + // Connect to the Docker API event stream. Docker streams + // newline-delimited JSON objects (Content-Type: application/json), + // NOT Server-Sent Events, so the payload arrives as plain data chunks + // that must be buffered and split on newlines before decoding. $source = $client->request(method: 'GET', url: '/events'); - while ($client->request(method: 'GET', url: '/events')) { - foreach ($client->stream(responses: $source, timeout: 2) as $r => $chunk) { - if ($chunk->isTimeout()) { - // Handle timeout - continue; - } + $buffer = ''; + foreach ($client->stream(responses: $source) as $chunk) { + if ($chunk->isTimeout()) { + continue; + } - if ($chunk->isLast()) { - // Handle end of stream - return; - } + if ($chunk->isLast()) { + return; + } - // Process the ServerSentEvent chunk - if ($chunk instanceof ServerSentEvent) { - // Do something with the event data - $content = $chunk->getContent(); - if (!json_validate($content)) { - continue; - } + $buffer .= $chunk->getContent(); + + while (false !== ($newlinePos = strpos($buffer, "\n"))) { + $line = trim(substr($buffer, 0, $newlinePos)); + $buffer = substr($buffer, $newlinePos + 1); + + if ('' === $line || !json_validate($line)) { + continue; + } - $eventObject = json_decode(json: $content, associative: true, depth: 512, flags: JSON_THROW_ON_ERROR); - if (is_array($eventObject) && ($eventObject['Type'] ?? false) === 'container') { - $event = $serializer->deserialize(data: $content, type: ContainerEvent::class, format: 'json'); + $eventObject = json_decode(json: $line, associative: true, depth: 512, flags: JSON_THROW_ON_ERROR); + if (is_array($eventObject) && ($eventObject['Type'] ?? false) === 'container') { + $event = $serializer->denormalize(data: $eventObject, type: ContainerEvent::class, format: 'json'); - $eventCallback($event); - } + $eventCallback($event); } } } diff --git a/src/Command/EventsListenCommand.php b/src/Command/EventsListenCommand.php index eb8f6d9..61c17e5 100644 --- a/src/Command/EventsListenCommand.php +++ b/src/Command/EventsListenCommand.php @@ -39,7 +39,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int 'stop', ]; - $service->listenForEvents(function (ContainerEvent $event) use ($service, $actions, $io) { + $service->listenForEvents(function (ContainerEvent $event) use ($service, $actions, $io): void { $container = $this->containers[$event->Actor->ID] ?? null; $prefix = '[event]'; if ($container) { diff --git a/tests/Unit/Client/DockerApiClientWrapperTest.php b/tests/Unit/Client/DockerApiClientWrapperTest.php index 910aff6..97cef5f 100644 --- a/tests/Unit/Client/DockerApiClientWrapperTest.php +++ b/tests/Unit/Client/DockerApiClientWrapperTest.php @@ -5,8 +5,15 @@ namespace WebProject\DockerApiClient\Tests\Unit\Client; use Codeception\Test\Unit; +use Symfony\Component\HttpClient\MockHttpClient; +use Symfony\Component\HttpClient\Response\MockResponse; use WebProject\DockerApi\Library\Generated\Client; use WebProject\DockerApiClient\Client\DockerApiClientWrapper; +use WebProject\DockerApiClient\Event\ContainerEvent; + +use function json_encode; + +use const JSON_THROW_ON_ERROR; final class DockerApiClientWrapperTest extends Unit { @@ -24,4 +31,94 @@ public function testCreate(): void $this->assertInstanceOf(DockerApiClientWrapper::class, $wrapper); $this->assertInstanceOf(Client::class, $wrapper->getDockerClient()); } + + public function testListenForEventsDispatchesOnlyContainerEvents(): void + { + // Docker streams newline-delimited JSON objects; only container events + // must reach the callback, non-container events are ignored. + $wrapper = $this->createWrapperStreaming([ + $this->containerEventJson('start', 'abc123') . "\n", + json_encode(['Type' => 'network', 'Action' => 'connect'], JSON_THROW_ON_ERROR) . "\n", + $this->containerEventJson('stop', 'def456') . "\n", + ]); + + /** @var list $received */ + $received = []; + $wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void { + $received[] = $event; + }); + + $this->assertCount(2, $received); + $this->assertSame('start', $received[0]->Action); + $this->assertSame('abc123', $received[0]->Actor->ID); + $this->assertSame('stop', $received[1]->Action); + $this->assertSame('def456', $received[1]->Actor->ID); + } + + public function testListenForEventsBuffersJsonSplitAcrossChunks(): void + { + // A single JSON object may be split across two transport chunks and + // must be reassembled from the buffer before decoding. + $json = $this->containerEventJson('start', 'split-id') . "\n"; + $half = (int) (strlen($json) / 2); + + $wrapper = $this->createWrapperStreaming([ + substr($json, 0, $half), + substr($json, $half), + ]); + + /** @var list $received */ + $received = []; + $wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void { + $received[] = $event; + }); + + $this->assertCount(1, $received); + $this->assertSame('split-id', $received[0]->Actor->ID); + } + + public function testListenForEventsSkipsInvalidJsonLines(): void + { + $wrapper = $this->createWrapperStreaming([ + "not-json-at-all\n", + $this->containerEventJson('die', 'valid-id') . "\n", + ]); + + /** @var list $received */ + $received = []; + $wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void { + $received[] = $event; + }); + + $this->assertCount(1, $received); + $this->assertSame('die', $received[0]->Action); + $this->assertSame('valid-id', $received[0]->Actor->ID); + } + + /** + * @param list $chunks + */ + private function createWrapperStreaming(array $chunks): DockerApiClientWrapper + { + $mockHttpClient = new MockHttpClient(new MockResponse($chunks)); + + return new DockerApiClientWrapper( + baseUri: 'http://localhost', + socketPath: '/var/run/docker.sock', + client: $this->createMock(Client::class), + eventStreamClient: $mockHttpClient, + ); + } + + private function containerEventJson(string $action, string $id): string + { + return json_encode([ + 'Type' => 'container', + 'Action' => $action, + 'Actor' => ['ID' => $id, 'Attributes' => ['name' => 'test']], + 'scope' => 'local', + 'time' => 1_700_000_000, + 'timeNano' => 1_700_000_000_000_000_000, + ], JSON_THROW_ON_ERROR); + } } From cf60c675804b8ef5472869ead5ba8431dcd91c68 Mon Sep 17 00:00:00 2001 From: Benjamin Fahl Date: Mon, 13 Jul 2026 22:23:12 +0200 Subject: [PATCH 2/2] style(tests): apply php-cs-fixer import ordering to event stream tests --- tests/Unit/Client/DockerApiClientWrapperTest.php | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/tests/Unit/Client/DockerApiClientWrapperTest.php b/tests/Unit/Client/DockerApiClientWrapperTest.php index 97cef5f..73a5345 100644 --- a/tests/Unit/Client/DockerApiClientWrapperTest.php +++ b/tests/Unit/Client/DockerApiClientWrapperTest.php @@ -4,16 +4,15 @@ namespace WebProject\DockerApiClient\Tests\Unit\Client; +use const JSON_THROW_ON_ERROR; use Codeception\Test\Unit; use Symfony\Component\HttpClient\MockHttpClient; use Symfony\Component\HttpClient\Response\MockResponse; use WebProject\DockerApi\Library\Generated\Client; use WebProject\DockerApiClient\Client\DockerApiClientWrapper; use WebProject\DockerApiClient\Event\ContainerEvent; - use function json_encode; - -use const JSON_THROW_ON_ERROR; +use function strlen; final class DockerApiClientWrapperTest extends Unit {