Change JSON stream to always report data events instead of progress

Author: clue

README.md

@@ -330,18 +330,15 @@ $stream = $client->eventsStream();
 
 The resulting stream will emit the following events:
 
-* `progress`: for *each* element in the update stream
-* `error`:    once if an error occurs, will close() stream then
+* `data`:  for *each* element in the update stream
+* `error`: once if an error occurs, will close() stream then
   * Will emit an [`Io\JsonProgressException`](#jsonprogressexception) if an individual progress message contains an error message
   * Any other `Exception` in case of an transport error, like invalid request etc.
-* `close`:    once the stream ends (either finished or after "error")
-
-Please note that the resulting stream does not emit any "data" events, so
-you will not be able to pipe() its events into another `WritableStream`.
+* `close`: once the stream ends (either finished or after "error")
 
 ```php
 $stream = $client->imageCreateStream('clue/redis-benchmark');
-$stream->on('progress', function ($data) {
+$stream->on('data', function ($data) {
     // data will be emitted for *each* complete element in the JSON stream
     echo $data['status'] . PHP_EOL;
 });

examples/events.php

@@ -20,7 +20,7 @@
 // stream all events for 10 seconds
 $stream = $client->eventsStream(null, microtime(true) + 10.0);
 
-$stream->on('progress', function ($event) {
+$stream->on('data', function ($event) {
     echo json_encode($event) . PHP_EOL;
 });
 

examples/pull.php

@@ -15,7 +15,7 @@
 
 $stream = $client->imageCreateStream($image);
 
-$stream->on('progress', function ($progress) {
+$stream->on('data', function ($progress) {
     echo 'progress: '. json_encode($progress) . PHP_EOL;
 });
 

src/Client.php

@@ -127,8 +127,7 @@ public function version()
     public function events($since = null, $until = null, $filters = array())
     {
         return $this->streamingParser->deferredStream(
-            $this->eventsStream($since, $until, $filters),
-            'progress'
+            $this->eventsStream($since, $until, $filters)
         );
     }
 
@@ -138,12 +137,9 @@ public function events($since = null, $until = null, $filters = array())
      * This is a JSON streaming API endpoint that returns a stream instance.
      *
      * The resulting stream will emit the following events:
-     * - progress: for *each* element in the update stream
-     * - error:    once if an error occurs, will close() stream then
-     * - close:    once the stream ends (either finished or after "error")
-     *
-     * Please note that the resulting stream does not emit any "data" events, so
-     * you will not be able to pipe() its events into another `WritableStream`.
+     * - data:  for *each* element in the update stream
+     * - error: once if an error occurs, will close() stream then
+     * - close: once the stream ends (either finished or after "error")
      *
      * The optional `$filters` parameter can be used to only get events for
      * certain event types, images and/or containers etc. like this:
@@ -758,9 +754,9 @@ public function imageList($all = false)
      */
     public function imageCreate($fromImage = null, $fromSrc = null, $repo = null, $tag = null, $registry = null, $registryAuth = null)
     {
-        $stream = $this->imageCreateStream($fromImage, $fromSrc, $repo, $tag, $registry, $registryAuth);
-
-        return $this->streamingParser->deferredStream($stream, 'progress');
+        return $this->streamingParser->deferredStream(
+            $this->imageCreateStream($fromImage, $fromSrc, $repo, $tag, $registry, $registryAuth)
+        );
     }
 
     /**
@@ -769,12 +765,9 @@ public function imageCreate($fromImage = null, $fromSrc = null, $repo = null, $t
      * This is a JSON streaming API endpoint that returns a stream instance.
      *
      * The resulting stream will emit the following events:
-     * - progress: for *each* element in the update stream
-     * - error:    once if an error occurs, will close() stream then
-     * - close:    once the stream ends (either finished or after "error")
-     *
-     * Please note that the resulting stream does not emit any "data" events, so
-     * you will not be able to pipe() its events into another `WritableStream`.
+     * - data:  for *each* element in the update stream
+     * - error: once if an error occurs, will close() stream then
+     * - close: once the stream ends (either finished or after "error").
      *
      * Pulling a private image from a remote registry will likely require authorization, so make
      * sure to pass the $registryAuth parameter, see `self::authHeaders()` for
@@ -871,9 +864,9 @@ public function imageHistory($image)
      */
     public function imagePush($image, $tag = null, $registry = null, $registryAuth = null)
     {
-        $stream = $this->imagePushStream($image, $tag, $registry, $registryAuth);
-
-        return $this->streamingParser->deferredStream($stream, 'progress');
+        return $this->streamingParser->deferredStream(
+            $this->imagePushStream($image, $tag, $registry, $registryAuth)
+        );
     }
 
     /**
@@ -882,12 +875,9 @@ public function imagePush($image, $tag = null, $registry = null, $registryAuth =
      * This is a JSON streaming API endpoint that returns a stream instance.
      *
      * The resulting stream will emit the following events:
-     * - progress: for *each* element in the update stream
-     * - error:    once if an error occurs, will close() stream then
-     * - close:    once the stream ends (either finished or after "error")
-     *
-     * Please note that the resulting stream does not emit any "data" events, so
-     * you will not be able to pipe() its events into another `WritableStream`.
+     * - data:  for *each* element in the update stream
+     * - error: once if an error occurs, will close() stream then
+     * - close: once the stream ends (either finished or after "error")
      *
      * Pushing to a remote registry will likely require authorization, so make
      * sure to pass the $registryAuth parameter, see `self::authHeaders()` for

src/Io/StreamingParser.php

@@ -52,7 +52,7 @@ public function parseJsonStream(PromiseInterface $promise)
                     $out->close();
                     return;
                 }
-                $out->emit('progress', array($object, $out));
+                $out->emit('data', array($object, $out));
             }
         });
 
@@ -108,13 +108,12 @@ public function bufferedStream(ReadableStreamInterface $stream)
     }
 
     /**
-     * Returns a promise which resolves with an array of all "progress" events
+     * Returns a promise which resolves with an array of all "data" events
      *
      * @param ReadableStreamInterface $stream
-     * @param string                  $progressEventName the name of the event to collect
      * @return PromiseInterface Promise<array, Exception>
      */
-    public function deferredStream(ReadableStreamInterface $stream, $progressEventName)
+    public function deferredStream(ReadableStreamInterface $stream)
     {
         // cancelling the deferred will (try to) close the stream
         $deferred = new Deferred(function () use ($stream) {
@@ -126,7 +125,7 @@ public function deferredStream(ReadableStreamInterface $stream, $progressEventNa
         if ($stream->isReadable()) {
             // buffer all data events for deferred resolving
             $buffered = array();
-            $stream->on($progressEventName, function ($data) use (&$buffered) {
+            $stream->on('data', function ($data) use (&$buffered) {
                 $buffered []= $data;
             });
 

tests/ClientTest.php

@@ -106,7 +106,7 @@ public function testEvents()
 
         $this->expectRequest('GET', '/events', $this->createResponseJsonStream($json));
         $this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
-        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
+        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream))->will($this->returnPromise($json));
 
         $this->expectPromiseResolveWith($json, $this->client->events());
     }
@@ -118,7 +118,7 @@ public function testEventsArgs()
 
         $this->expectRequest('GET', '/events?since=10&until=20&filters=%7B%22image%22%3A%5B%22busybox%22%2C%22ubuntu%22%5D%7D', $this->createResponseJsonStream($json));
         $this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
-        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
+        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream))->will($this->returnPromise($json));
 
         $this->expectPromiseResolveWith($json, $this->client->events(10, 20, array('image' => array('busybox', 'ubuntu'))));
     }
@@ -357,7 +357,7 @@ public function testImageCreate()
 
         $this->expectRequest('post', '/images/create?fromImage=busybox', $this->createResponseJsonStream($json));
         $this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
-        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
+        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream))->will($this->returnPromise($json));
 
         $this->expectPromiseResolveWith($json, $this->client->imageCreate('busybox'));
     }
@@ -395,7 +395,7 @@ public function testImagePush()
 
         $this->expectRequest('post', '/images/123/push', $this->createResponseJsonStream($json));
         $this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
-        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
+        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream))->will($this->returnPromise($json));
 
         $this->expectPromiseResolveWith($json, $this->client->imagePush('123'));
     }
@@ -419,7 +419,7 @@ public function testImagePushCustomRegistry()
 
         $this->expectRequest('post', '/images/demo.acme.com%3A5000/123/push?tag=test', $this->createResponseJsonStream($json));
         $this->streamingParser->expects($this->once())->method('parseJsonStream')->will($this->returnValue($stream));
-        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream), $this->equalTo('progress'))->will($this->returnPromise($json));
+        $this->streamingParser->expects($this->once())->method('deferredStream')->with($this->equalTo($stream))->will($this->returnPromise($json));
 
         $this->expectPromiseResolveWith($json, $this->client->imagePush('123', 'test', 'demo.acme.com:5000', $auth));
     }

tests/Io/StreamingParserTest.php

@@ -73,19 +73,19 @@ public function testDeferredClosedStreamWillReject()
         $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
         $stream->expects($this->once())->method('isReadable')->will($this->returnValue(false));
 
-        $promise = $this->parser->deferredStream($stream, 'anything');
+        $promise = $this->parser->deferredStream($stream);
         $this->expectPromiseReject($promise);
     }
 
     public function testDeferredStreamEventsWillBeEmittedAndBuffered()
     {
         $stream = new ReadableStream();
 
-        $promise = $this->parser->deferredStream($stream, 'demo');
+        $promise = $this->parser->deferredStream($stream);
 
         $stream->emit('ignored', array('ignored'));
-        $stream->emit('demo', array('a'));
-        $stream->emit('demo', array('b'));
+        $stream->emit('data', array('a'));
+        $stream->emit('data', array('b'));
 
         $stream->close();
 
@@ -96,11 +96,11 @@ public function testDeferredStreamErrorEventWillRejectPromise()
     {
         $stream = new ReadableStream();
 
-        $promise = $this->parser->deferredStream($stream, 'demo');
+        $promise = $this->parser->deferredStream($stream);
 
         $stream->emit('ignored', array('ignored'));
 
-        $stream->emit('demo', array('a'));
+        $stream->emit('data', array('a'));
 
         $stream->emit('error', array('value', 'ignord'));
 
@@ -115,7 +115,7 @@ public function testDeferredCancelingPromiseWillCloseStream()
         $stream = $this->getMockBuilder('React\Stream\ReadableStreamInterface')->getMock();
         $stream->expects($this->once())->method('isReadable')->willReturn(true);
 
-        $promise = $this->parser->deferredStream($stream, 'anything');
+        $promise = $this->parser->deferredStream($stream);
         if (!($promise instanceof CancellablePromiseInterface)) {
             $this->markTestSkipped('Requires Promise v2 API and has no effect on v1 API');
         }