Skip to content

Commit 5c064e6

Browse files
clueclue
committed
Add imageCreateStream()
Also serves as a basis for imageCreate() Emit each progress event via the Promise progress API and resolve Promise with buffered array of all progress events
1 parent fca44a4 commit 5c064e6

4 files changed

Lines changed: 147 additions & 18 deletions

File tree

‎src/Client.php‎

Lines changed: 35 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
use Clue\React\Docker\Io\ResponseParser;
88
use React\Promise\PromiseInterface as Promise;
99
use Clue\React\Docker\Io\StreamingParser;
10+
use React\Stream\ReadableStreamInterface;
1011

1112
/**
1213
* Docker Remote API client
@@ -301,6 +302,9 @@ public function imageList($all = false)
301302
/**
302303
* Create an image, either by pulling it from the registry or by importing it
303304
*
305+
* Will resolve with an array of all progress events. These can also be
306+
* accessed via the Promise progress handler.
307+
*
304308
* @param string|null $fromImage name of the image to pull
305309
* @param string|null $fromSrc source to import, - means stdin
306310
* @param string|null $repo repository
@@ -309,13 +313,42 @@ public function imageList($all = false)
309313
* @param array|null $registryAuth AuthConfig object (to send as X-Registry-Auth header)
310314
* @return Promise Promise<array> stream of message objects
311315
* @link https://docs.docker.com/reference/api/docker_remote_api_v1.15/#create-an-image
316+
* @see self::imageCreateStream()
312317
*/
313318
public function imageCreate($fromImage = null, $fromSrc = null, $repo = null, $tag = null, $registry = null, $registryAuth = null)
314319
{
315-
return $this->streamingParser->parseResponse($this->browser->post(
320+
$stream = $this->imageCreateStream($fromImage, $fromSrc, $repo, $tag, $registry, $registryAuth);
321+
322+
return $this->streamingParser->deferredStream($stream, 'progress');
323+
}
324+
325+
/**
326+
* Create an image, either by pulling it from the registry or by importing it
327+
*
328+
* The resulting stream will emit the following events:
329+
* - progress: for *each* element in the update stream
330+
* - error: once if an error occurs, will close() stream then
331+
* - close: once the stream ends (either finished or after "error")
332+
*
333+
* Please note that the resulting stream does not emit any "data" events, so
334+
* you will not be able to pipe() its events into another `WritableStream`.
335+
*
336+
* @param string|null $fromImage name of the image to pull
337+
* @param string|null $fromSrc source to import, - means stdin
338+
* @param string|null $repo repository
339+
* @param string|null $tag (optional) (obsolete) tag, use $repo and $fromImage in the "name:tag" instead
340+
* @param string|null $registry the registry to pull from
341+
* @param array|null $registryAuth AuthConfig object (to send as X-Registry-Auth header)
342+
* @return ReadableStreamInterface
343+
* @link https://docs.docker.com/reference/api/docker_remote_api_v1.15/#create-an-image
344+
* @see self::imageCreate()
345+
*/
346+
public function imageCreateStream($fromImage = null, $fromSrc = null, $repo = null, $tag = null, $registry = null, $registryAuth = null)
347+
{
348+
return $this->streamingParser->parseJsonStream($this->browser->post(
316349
$this->url('/images/create?fromImage=%s&fromSrc=%s&repo=%s&tag=%s&registry=%s', $fromImage, $fromSrc, $repo, $tag, $registry),
317350
$this->authHeaders($registryAuth)
318-
))->then(array($this->parser, 'expectJson'));
351+
));
319352
}
320353

321354
/**

‎src/Io/StreamingParser.php‎

Lines changed: 70 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -5,32 +5,88 @@
55
use React\Promise\PromiseInterface;
66
use Clue\JsonStream\StreamingJsonParser;
77
use React\Promise\Deferred;
8+
use React\Stream\ReadableStream;
9+
use React\Stream\ReadableStreamInterface;
10+
use RuntimeException;
11+
use React\Promise\CancellablePromiseInterface;
812

913
class StreamingParser
1014
{
11-
public function parseResponse(PromiseInterface $promise)
15+
public function parseJsonStream(PromiseInterface $promise)
1216
{
17+
// TODO: assert expect tcp stream
18+
1319
$parser = new StreamingJsonParser();
1420

15-
$deferred = new Deferred();
21+
$out = new ReadableStream();
22+
23+
// try to cancel promise once the stream closes
24+
if ($promise instanceof CancellablePromiseInterface) {
25+
$out->on('close', function() use ($promise) {
26+
$promise->cancel();
27+
});
28+
}
29+
30+
$promise->then(
31+
function ($response) use ($out) {
32+
$out->close();
33+
},
34+
function ($error) use ($out) {
35+
$out->emit('error', array($error, $out));
36+
$out->close();
37+
},
38+
function ($progress) use ($parser, $out) {
39+
if (is_array($progress) && isset($progress['responseStream'])) {
40+
$stream = $progress['responseStream'];
41+
/* @var $stream React\Stream\Stream */
1642

17-
$promise->then(null, null, function ($progress) use ($parser, $deferred) {
18-
if (is_array($progress) && isset($progress['response'])) {
19-
$stream = $progress['response'];
20-
/* @var $stream React\Stream\Stream */
43+
// hack to do not buffer stream contents in body
44+
$stream->removeAllListeners('data');
2145

22-
// got a streaming HTTP reponse => forward each data chunk to the streaming JSON parser
23-
$stream->on('data', function ($data) use ($parser, $deferred) {
24-
$objects = $parser->push($data);
46+
// got a streaming HTTP reponse => forward each data chunk to the streaming JSON parser
47+
$stream->on('data', function ($data) use ($parser, $out) {
48+
$objects = $parser->push($data);
2549

26-
foreach ($objects as $object) {
27-
$deferred->progress($object);
28-
}
29-
});
50+
foreach ($objects as $object) {
51+
$out->emit('progress', array($object, $out));
52+
}
53+
});
54+
}
3055
}
56+
);
57+
58+
return $out;
59+
}
60+
61+
public function deferredStream(ReadableStreamInterface $stream, $progressEventName)
62+
{
63+
// cancelling the deferred will (try to) close the stream
64+
$deferred = new Deferred(function () use ($stream) {
65+
$stream->close();
66+
67+
throw new RuntimeException('Cancelled');
3168
});
3269

33-
$promise->then(array($deferred, 'resolve'), array($deferred, 'reject'));
70+
if ($stream->isReadable()) {
71+
// buffer all data events and emit as progress
72+
$buffered = array();
73+
$stream->on($progressEventName, function ($data) use ($deferred, &$buffered) {
74+
$buffered []= $data;
75+
$deferred->progress($data);
76+
});
77+
78+
// error event rejects
79+
$stream->on('error', function ($error) use ($deferred) {
80+
$deferred->reject($error);
81+
});
82+
83+
// close event resolves with buffered events (unless already error'ed)
84+
$stream->on('close', function () use ($deferred, &$buffered) {
85+
$deferred->resolve($buffered);
86+
});
87+
} else {
88+
$deferred->reject(new RuntimeException('Stream already ended, looks like it could not be opened'));
89+
}
3490

3591
return $deferred->promise();
3692
}

‎tests/ClientTest.php‎

Lines changed: 25 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -185,12 +185,25 @@ public function testImageList()
185185
public function testImageCreate()
186186
{
187187
$json = array();
188-
$this->streamingParser->expects($this->once())->method('parseResponse')->will($this->returnArgument(0));
189-
$this->expectRequestFlow('post', '/images/create?fromImage=busybox&fromSrc=&repo=&tag=&registry=', $this->createResponseJson($json), 'expectJson');
188+
$stream = $this->getMock('React\Stream\ReadableStreamInterface');
189+
190+
$this->expectRequest('post', '/images/create?fromImage=busybox&fromSrc=&repo=&tag=&registry=', $this->createResponseJsonStream($json));
191+
$this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
192+
$this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
190193

191194
$this->expectPromiseResolveWith($json, $this->client->imageCreate('busybox'));
192195
}
193196

197+
public function testImageCreateStream()
198+
{
199+
$stream = $this->getMock('React\Stream\ReadableStreamInterface');
200+
201+
$this->expectRequest('post', '/images/create?fromImage=busybox&fromSrc=&repo=&tag=&registry=', $this->createResponseJsonStream(array()));
202+
$this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
203+
204+
$this->assertSame($stream, $this->client->imageCreateStream('busybox'));
205+
}
206+
194207
public function testImageInspect()
195208
{
196209
$json = array();
@@ -283,6 +296,11 @@ private function expectRequestFlow($method, $url, Response $response, $parser)
283296
$this->parser->expects($this->once())->method($parser)->with($this->equalTo($response))->will($this->returnValue($return));
284297
}
285298

299+
private function expectRequest($method, $url, Response $response)
300+
{
301+
$this->browser->expects($this->once())->method($method)->with($this->equalTo($url))->will($this->returnPromise($response));
302+
}
303+
286304
private function createResponse($body = '')
287305
{
288306
return new Response('HTTP/1.0', 200, 'OK', null, new Body($body));
@@ -293,6 +311,11 @@ private function createResponseJson($json)
293311
return $this->createResponse(json_encode($json));
294312
}
295313

314+
private function createResponseJsonStream($json)
315+
{
316+
return $this->createResponse(implode('', array_map('json_encode', $json)));
317+
}
318+
296319
private function returnPromise($for)
297320
{
298321
$deferred = new Deferred();

‎tests/bootstrap.php‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
use React\EventLoop\LoopInterface;
44
use React\Promise\PromiseInterface;
5+
use React\Promise\Deferred;
56

67
require_once __DIR__ . '/../vendor/autoload.php';
78

@@ -105,6 +106,22 @@ protected function waitFor(PromiseInterface $promise, LoopInterface $loop)
105106

106107
return $resolved;
107108
}
109+
110+
protected function createPromiseResolved($value = null)
111+
{
112+
$deferred = new Deferred();
113+
$deferred->resolve($value);
114+
115+
return $deferred->promise();
116+
}
117+
118+
protected function createPromiseRejected($value = null)
119+
{
120+
$deferred = new Deferred();
121+
$deferred->reject($value);
122+
123+
return $deferred->promise();
124+
}
108125
}
109126

110127
class CallableStub

0 commit comments

Comments
 (0)