Emit JsonProgressException for progress messages with an "error" key

Author: clue

README.md

@@ -174,6 +174,12 @@ $client->imageCreate('clue/streamripper')->then(
     },
     function ($error) {
         // an error occurred (possibly after receiving *some* elements)
+        
+        if ($error instanceof Io\JsonProgressException) {
+            // a progress message (usually the last) contains an error message
+        } else {
+            // any other error, like invalid request etc.
+        }
     }
 );
 ```
@@ -199,6 +205,8 @@ 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
+  * 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
@@ -218,6 +226,13 @@ $stream->on('close', function () {
 
 See also the [pull example](examples/pull.php) and the [push example](examples/push.php).
 
+### JsonProgressException
+
+The `Io\JsonProgressException` will be thrown by [JSON streaming](#json-streaming)
+endpoints if an individual progress message contains an error message.
+
+The `getData()` method can be used to obtain the progress message.
+
 ## Install
 
 The recommended way to install this library is [through composer](http://getcomposer.org). [New to composer?](http://getcomposer.org/doc/00-intro.md)

src/Io/JsonProgressException.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace Clue\React\Docker\Io;
+
+use RuntimeException;
+
+/**
+ * Will be used for JSON streaming endpoints where an individal progress event contains an "error" key
+ */
+class JsonProgressException extends RuntimeException
+{
+    private $data;
+
+    public function __construct($data, $message = null, $code = null, $previous = null)
+    {
+        if ($message === null) {
+            $message = $data['error'];
+        }
+        parent::__construct($message, $code, $previous);
+        $this->data = $data;
+    }
+
+    public function getData()
+    {
+        return $data;
+    }
+}

src/Io/StreamingParser.php

@@ -32,6 +32,11 @@ public function parseJsonStream(PromiseInterface $promise)
             $objects = $parser->push($data);
 
             foreach ($objects as $object) {
+                if (isset($object['error'])) {
+                    $out->emit('error', array(new JsonProgressException($object), $out));
+                    $out->close();
+                    return;
+                }
                 $out->emit('progress', array($object, $out));
             }
         });

tests/FunctionalClientTest.php

@@ -84,6 +84,21 @@ public function testImageTag()
         $ret = $this->waitFor($promise, $this->loop);
     }
 
+    public function testImageCreateStreamMissingWillEmitJsonError()
+    {
+        $stream = $this->client->imageCreateStream('clue/does-not-exist');
+
+        // one "progress" event, but no "data" events
+        $stream->on('progress', $this->expectCallableOnce());
+        $stream->on('data', $this->expectCallableNever());
+
+        // will emit "error" with JsonProgressException and close
+        $stream->on('error', $this->expectCallableOnceParameter('Clue\React\Docker\Io\JsonProgressException'));
+        $stream->on('close', $this->expectCallableOnce());
+
+        $this->loop->run();
+    }
+
     public function testInfo()
     {
         $this->expectPromiseResolve($this->client->info());