fix(events): parse newline-delimited JSON from Docker event stream

src/Client/DockerApiClientWrapper.php

@@ -5,25 +5,30 @@
 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
 {
     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);
                 }
             }
         }

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) {

tests/Unit/Client/DockerApiClientWrapperTest.php

@@ -4,9 +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 function strlen;
 
 final class DockerApiClientWrapperTest extends Unit
 {
@@ -24,4 +30,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<ContainerEvent> $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<ContainerEvent> $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<ContainerEvent> $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<string> $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);
+    }
 }