Add option to emit STDERR data as custom event

Author: clue

README.md

@@ -221,6 +221,23 @@ $stream->on('close', function () {
 });
 ```
 
+Note that by default the output of both STDOUT and STDERR will be emitted
+as normal `data` events. You can optionally pass a custom event name which
+will be used to emit STDERR data so that it can be handled separately.
+Note that the normal streaming primitives likely do not know about this
+event, so special care may have to be taken.
+Also note that this option has no effect if you execute with a TTY.
+
+```php
+$stream = $client->execStartStream($exec, $tty, 'stderr');
+$stream->on('data', function ($data) {
+    echo 'STDOUT data: ' . $data;
+});
+$stream->on('stderr', function ($data) {
+    echo 'STDERR data: ' . $data;
+});
+```
+
 See also the [streaming exec example](examples/exec-stream.php) and the [exec benchmark example](examples/benchmark-exec.php).
 
 The TTY mode should be set depending on whether your command needs a TTY

examples/exec-stream.php

@@ -27,13 +27,26 @@
 $out = new Stream(STDOUT, $loop);
 $out->pause();
 
+$stderr = new Stream(STDERR, $loop);
+$stderr->pause();
+
 // unkown exit code by default
 $exit = 1;
 
-$client->execCreate($container, $cmd)->then(function ($info) use ($client, $out, &$exit) {
-    $stream = $client->execStartStream($info['Id']);
+$client->execCreate($container, $cmd)->then(function ($info) use ($client, $out, $stderr, &$exit) {
+    $stream = $client->execStartStream($info['Id'], false, 'stderr');
     $stream->pipe($out);
 
+    // forward custom stderr event to STDERR stream
+    $stream->on('stderr', function ($data) use ($stderr, $stream) {
+        if ($stderr->write($data) === false) {
+            $stream->pause();
+            $stderr->once('drain', function () use ($stream) {
+                $stream->resume();
+            });
+        }
+    });
+
     $stream->on('error', 'printf');
 
     // remember exit code of executed command once it closes

src/Client.php

@@ -1008,14 +1008,22 @@ public function execStartDetached($exec, $tty = false)
      * This works for command output of any size as only small chunks have to
      * be kept in memory.
      *
-     * @param string  $exec exec ID
-     * @param boolean $tty  tty mode
+     * Note that by default the output of both STDOUT and STDERR will be emitted
+     * as normal "data" events. You can optionally pass a custom event name which
+     * will be used to emit STDERR data so that it can be handled separately.
+     * Note that the normal streaming primitives likely do not know about this
+     * event, so special care may have to be taken.
+     * Also note that this option has no effect if you execute with a TTY.
+     *
+     * @param string  $exec        exec ID
+     * @param boolean $tty         tty mode
+     * @param string  $stderrEvent custom event to emit for STDERR data (otherwise emits as "data")
      * @return ReadableStreamInterface stream of exec data
      * @link https://docs.docker.com/reference/api/docker_remote_api_v1.15/#exec-start
      * @see self::execStart()
      * @see self::execStartDetached()
      */
-    public function execStartStream($exec, $tty = false)
+    public function execStartStream($exec, $tty = false, $stderrEvent = null)
     {
         $stream = $this->streamingParser->parsePlainStream(
             $this->browser->withOptions(array('streaming' => true))->post(
@@ -1036,7 +1044,7 @@ public function execStartStream($exec, $tty = false)
 
         // this is a multiplexed stream unless this is started with a TTY
         if (!$tty) {
-            $stream = $this->streamingParser->demultiplexStream($stream);
+            $stream = $this->streamingParser->demultiplexStream($stream, $stderrEvent);
         }
 
         return $stream;

src/Io/StreamingParser.php

@@ -88,18 +88,27 @@ public function parsePlainStream(PromiseInterface $promise)
      * Returns a readable plain text stream for the given multiplexed stream using Docker's "attach multiplexing protocol"
      *
      * @param ReadableStreamInterface $input
+     * @param string $stderrEvent
      * @return ReadableStreamInterface
      */
-    public function demultiplexStream(ReadableStreamInterface $input)
+    public function demultiplexStream(ReadableStreamInterface $input, $stderrEvent = null)
     {
+        if ($stderrEvent === null) {
+            $stderrEvent = 'data';
+        }
+
         $out = new ReadableStream();
         $parser = new MultiplexStreamParser();
 
         // pass all input data chunks through the parser
-        $input->on('data', function ($chunk) use ($parser, $out) {
+        $input->on('data', function ($chunk) use ($parser, $out, $stderrEvent) {
             // once parser emits, forward to output stream
-            $parser->push($chunk, function ($stream, $data) use ($out) {
-                $out->emit('data', array($data));
+            $parser->push($chunk, function ($stream, $data) use ($out, $stderrEvent) {
+                if ($stream === 2) {
+                    $out->emit($stderrEvent, array($data));
+                } else {
+                    $out->emit('data', array($data));
+                }
             });
         });
 

tests/ClientTest.php

@@ -438,6 +438,18 @@ public function testExecStartStreamWithTtyWillNotDemultiplex()
         $this->assertSame($stream, $this->client->execStartStream(123, $config));
     }
 
+    public function testExecStartStreamWithCustomStderrEvent()
+    {
+        $config = array();
+        $stream = $this->getMock('React\Stream\ReadableStreamInterface');
+
+        $this->expectRequest('POST', '/exec/123/start', $this->createResponse());
+        $this->streamingParser->expects($this->once())->method('parsePlainStream')->will($this->returnValue($stream));
+        $this->streamingParser->expects($this->once())->method('demultiplexStream')->with($stream, 'stderr')->willReturn($stream);
+
+        $this->assertSame($stream, $this->client->execStartStream(123, $config, 'stderr'));
+    }
+
     public function testExecResize()
     {
         $this->expectRequestFlow('POST', '/exec/123/resize?w=800&h=600', $this->createResponse(), 'expectEmpty');

tests/FunctionalClientTest.php

@@ -171,6 +171,27 @@ public function testExecStringCommandWithOutputWhileRunning($container)
         $this->assertEquals('hello world', $output);
     }
 
+    /**
+     * @depends testStartRunning
+     * @param string $container
+     */
+    public function testExecStreamOutputInMultipleChunksWhileRunning($container)
+    {
+        $promise = $this->client->execCreate($container, 'echo -n hello && sleep 0 && echo -n world');
+        $exec = Block\await($promise, $this->loop);
+
+        $this->assertTrue(is_array($exec));
+        $this->assertTrue(is_string($exec['Id']));
+
+        $stream = $this->client->execStartStream($exec['Id']);
+        $stream->once('data', $this->expectCallableOnceWith('hello'));
+        //$stream->on('end', $this->expectCallableOnce());
+
+        $output = Block\await(Stream\buffer($stream), $this->loop);
+
+        $this->assertEquals('helloworld', $output);
+    }
+
     /**
      * @depends testStartRunning
      * @param string $container
@@ -228,6 +249,29 @@ public function testExecStreamCommandWithTtyAndStderrOutputWhileRunning($contain
         $this->assertEquals('hello world', $output);
     }
 
+    /**
+     * @depends testStartRunning
+     * @param string $container
+     */
+    public function testExecStreamStderrCustomEventWhileRunning($container)
+    {
+        $promise = $this->client->execCreate($container, 'echo -n hello world >&2');
+        $exec = Block\await($promise, $this->loop);
+
+        $this->assertTrue(is_array($exec));
+        $this->assertTrue(is_string($exec['Id']));
+
+        $stream = $this->client->execStartStream($exec['Id'], false, 'err');
+        $stream->on('err', $this->expectCallableOnceWith('hello world'));
+        $stream->on('data', $this->expectCallableNever());
+        $stream->on('error', $this->expectCallableNever());
+        //$stream->on('end', $this->expectCallableOnce());
+
+        $output = Block\await(Stream\buffer($stream), $this->loop);
+
+        $this->assertEquals('', $output);
+    }
+
     /**
      * @depends testStartRunning
      * @param string $container