Skip to content

Commit 2100c93

Browse files
Fahl-DesignFahl-Design
committed
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.
1 parent a17b285 commit 2100c93

3 files changed

Lines changed: 130 additions & 26 deletions

File tree

‎src/Client/DockerApiClientWrapper.php‎

Lines changed: 32 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -5,25 +5,30 @@
55
namespace WebProject\DockerApiClient\Client;
66

77
use JsonException;
8-
use Symfony\Component\HttpClient\Chunk\ServerSentEvent;
98
use Symfony\Component\HttpClient\HttpClient;
109
use Symfony\Component\HttpClient\Psr18Client;
1110
use Symfony\Component\Serializer\Encoder\JsonEncoder;
1211
use Symfony\Component\Serializer\Normalizer\ObjectNormalizer;
1312
use Symfony\Component\Serializer\Serializer;
1413
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
14+
use Symfony\Contracts\HttpClient\HttpClientInterface;
1515
use Webmozart\Assert\Assert;
1616
use WebProject\DockerApi\Library\Generated\Client;
1717
use WebProject\DockerApiClient\Event\ContainerEvent;
1818
use function is_array;
1919
use function json_decode;
20+
use function json_validate;
21+
use function strpos;
22+
use function substr;
23+
use function trim;
2024

2125
final class DockerApiClientWrapper
2226
{
2327
public function __construct(
2428
private readonly string $baseUri,
2529
private readonly string $socketPath,
2630
private readonly Client $client,
31+
private readonly ?HttpClientInterface $eventStreamClient = null,
2732
) {
2833
}
2934

@@ -35,7 +40,7 @@ public function __construct(
3540
*/
3641
public function listenForEvents(callable $eventCallback): void
3742
{
38-
$client = HttpClient::create([
43+
$client = $this->eventStreamClient ?? HttpClient::create([
3944
'base_uri' => $this->baseUri,
4045
'bindto' => $this->socketPath,
4146
'timeout' => null,
@@ -46,35 +51,37 @@ public function listenForEvents(callable $eventCallback): void
4651
encoders: [new JsonEncoder()]
4752
);
4853

49-
// Connect to the Docker API event stream
54+
// Connect to the Docker API event stream. Docker streams
55+
// newline-delimited JSON objects (Content-Type: application/json),
56+
// NOT Server-Sent Events, so the payload arrives as plain data chunks
57+
// that must be buffered and split on newlines before decoding.
5058
$source = $client->request(method: 'GET', url: '/events');
5159

52-
while ($client->request(method: 'GET', url: '/events')) {
53-
foreach ($client->stream(responses: $source, timeout: 2) as $r => $chunk) {
54-
if ($chunk->isTimeout()) {
55-
// Handle timeout
56-
continue;
57-
}
60+
$buffer = '';
61+
foreach ($client->stream(responses: $source) as $chunk) {
62+
if ($chunk->isTimeout()) {
63+
continue;
64+
}
5865

59-
if ($chunk->isLast()) {
60-
// Handle end of stream
61-
return;
62-
}
66+
if ($chunk->isLast()) {
67+
return;
68+
}
6369

64-
// Process the ServerSentEvent chunk
65-
if ($chunk instanceof ServerSentEvent) {
66-
// Do something with the event data
67-
$content = $chunk->getContent();
68-
if (!json_validate($content)) {
69-
continue;
70-
}
70+
$buffer .= $chunk->getContent();
71+
72+
while (false !== ($newlinePos = strpos($buffer, "\n"))) {
73+
$line = trim(substr($buffer, 0, $newlinePos));
74+
$buffer = substr($buffer, $newlinePos + 1);
75+
76+
if ('' === $line || !json_validate($line)) {
77+
continue;
78+
}
7179

72-
$eventObject = json_decode(json: $content, associative: true, depth: 512, flags: JSON_THROW_ON_ERROR);
73-
if (is_array($eventObject) && ($eventObject['Type'] ?? false) === 'container') {
74-
$event = $serializer->deserialize(data: $content, type: ContainerEvent::class, format: 'json');
80+
$eventObject = json_decode(json: $line, associative: true, depth: 512, flags: JSON_THROW_ON_ERROR);
81+
if (is_array($eventObject) && ($eventObject['Type'] ?? false) === 'container') {
82+
$event = $serializer->denormalize(data: $eventObject, type: ContainerEvent::class, format: 'json');
7583

76-
$eventCallback($event);
77-
}
84+
$eventCallback($event);
7885
}
7986
}
8087
}

‎src/Command/EventsListenCommand.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
3939
'stop',
4040
];
4141

42-
$service->listenForEvents(function (ContainerEvent $event) use ($service, $actions, $io) {
42+
$service->listenForEvents(function (ContainerEvent $event) use ($service, $actions, $io): void {
4343
$container = $this->containers[$event->Actor->ID] ?? null;
4444
$prefix = '[event]';
4545
if ($container) {

‎tests/Unit/Client/DockerApiClientWrapperTest.php‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,15 @@
55
namespace WebProject\DockerApiClient\Tests\Unit\Client;
66

77
use Codeception\Test\Unit;
8+
use Symfony\Component\HttpClient\MockHttpClient;
9+
use Symfony\Component\HttpClient\Response\MockResponse;
810
use WebProject\DockerApi\Library\Generated\Client;
911
use WebProject\DockerApiClient\Client\DockerApiClientWrapper;
12+
use WebProject\DockerApiClient\Event\ContainerEvent;
13+
14+
use function json_encode;
15+
16+
use const JSON_THROW_ON_ERROR;
1017

1118
final class DockerApiClientWrapperTest extends Unit
1219
{
@@ -24,4 +31,94 @@ public function testCreate(): void
2431
$this->assertInstanceOf(DockerApiClientWrapper::class, $wrapper);
2532
$this->assertInstanceOf(Client::class, $wrapper->getDockerClient());
2633
}
34+
35+
public function testListenForEventsDispatchesOnlyContainerEvents(): void
36+
{
37+
// Docker streams newline-delimited JSON objects; only container events
38+
// must reach the callback, non-container events are ignored.
39+
$wrapper = $this->createWrapperStreaming([
40+
$this->containerEventJson('start', 'abc123') . "\n",
41+
json_encode(['Type' => 'network', 'Action' => 'connect'], JSON_THROW_ON_ERROR) . "\n",
42+
$this->containerEventJson('stop', 'def456') . "\n",
43+
]);
44+
45+
/** @var list<ContainerEvent> $received */
46+
$received = [];
47+
$wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void {
48+
$received[] = $event;
49+
});
50+
51+
$this->assertCount(2, $received);
52+
$this->assertSame('start', $received[0]->Action);
53+
$this->assertSame('abc123', $received[0]->Actor->ID);
54+
$this->assertSame('stop', $received[1]->Action);
55+
$this->assertSame('def456', $received[1]->Actor->ID);
56+
}
57+
58+
public function testListenForEventsBuffersJsonSplitAcrossChunks(): void
59+
{
60+
// A single JSON object may be split across two transport chunks and
61+
// must be reassembled from the buffer before decoding.
62+
$json = $this->containerEventJson('start', 'split-id') . "\n";
63+
$half = (int) (strlen($json) / 2);
64+
65+
$wrapper = $this->createWrapperStreaming([
66+
substr($json, 0, $half),
67+
substr($json, $half),
68+
]);
69+
70+
/** @var list<ContainerEvent> $received */
71+
$received = [];
72+
$wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void {
73+
$received[] = $event;
74+
});
75+
76+
$this->assertCount(1, $received);
77+
$this->assertSame('split-id', $received[0]->Actor->ID);
78+
}
79+
80+
public function testListenForEventsSkipsInvalidJsonLines(): void
81+
{
82+
$wrapper = $this->createWrapperStreaming([
83+
"not-json-at-all\n",
84+
$this->containerEventJson('die', 'valid-id') . "\n",
85+
]);
86+
87+
/** @var list<ContainerEvent> $received */
88+
$received = [];
89+
$wrapper->listenForEvents(static function (ContainerEvent $event) use (&$received): void {
90+
$received[] = $event;
91+
});
92+
93+
$this->assertCount(1, $received);
94+
$this->assertSame('die', $received[0]->Action);
95+
$this->assertSame('valid-id', $received[0]->Actor->ID);
96+
}
97+
98+
/**
99+
* @param list<string> $chunks
100+
*/
101+
private function createWrapperStreaming(array $chunks): DockerApiClientWrapper
102+
{
103+
$mockHttpClient = new MockHttpClient(new MockResponse($chunks));
104+
105+
return new DockerApiClientWrapper(
106+
baseUri: 'http://localhost',
107+
socketPath: '/var/run/docker.sock',
108+
client: $this->createMock(Client::class),
109+
eventStreamClient: $mockHttpClient,
110+
);
111+
}
112+
113+
private function containerEventJson(string $action, string $id): string
114+
{
115+
return json_encode([
116+
'Type' => 'container',
117+
'Action' => $action,
118+
'Actor' => ['ID' => $id, 'Attributes' => ['name' => 'test']],
119+
'scope' => 'local',
120+
'time' => 1_700_000_000,
121+
'timeNano' => 1_700_000_000_000_000_000,
122+
], JSON_THROW_ON_ERROR);
123+
}
27124
}

0 commit comments

Comments
 (0)