Fix custom $stderrEvent for container attach and logs streams

src/Client.php

@@ -374,7 +374,7 @@ public function containerLogsStream($container, $follow = false, $stdout = true,
             }
 
             return $stream;
-        }));
+        }), $stderrEvent);
     }
 
     /**
@@ -738,7 +738,7 @@ public function containerAttachStream($container, $logs = false, $stream = false
             }
 
             return $stream;
-        }));
+        }), $stderrEvent);
     }
 
     /**

src/Io/ReadableUnwrapStream.php

@@ -27,8 +27,9 @@ class ReadableUnwrapStream extends EventEmitter implements ReadableStreamInterfa
 
     /**
      * @param PromiseInterface<ReadableStreamInterface> $promise
+     * @param ?string $stderrEvent custom event to forward for STDERR data (otherwise only forwards "data")
      */
-    public function __construct(PromiseInterface $promise)
+    public function __construct(PromiseInterface $promise, $stderrEvent = null)
     {
         $this->promise = $promise;
 
@@ -37,7 +38,7 @@ public function __construct(PromiseInterface $promise)
         $paused =& $this->paused;
 
         $promise->then(
-            function (ReadableStreamInterface $stream) use ($out, &$input, &$paused) {
+            function (ReadableStreamInterface $stream) use ($out, &$input, &$paused, $stderrEvent) {
                 // output already closed while waiting or input already closed => close other side
                 if (!$out->isReadable() || !$stream->isReadable()) {
                     $stream->close();
@@ -55,6 +56,13 @@ function (ReadableStreamInterface $stream) use ($out, &$input, &$paused) {
                     $out->emit('data', array($data));
                 });
 
+                // forward custom STDERR event if given
+                if ($stderrEvent !== null && !in_array($stderrEvent, array('data', 'end', 'error', 'close'), true)) {
+                    $stream->on($stderrEvent, function ($data) use ($out, $stderrEvent) {
+                        $out->emit($stderrEvent, array($data));
+                    });
+                }
+
                 // forward end event to output and close
                 $stream->on('end', function () use ($out) {
                     $out->emit('end');

tests/ClientTest.php

@@ -242,6 +242,25 @@ public function testContainerLogsStreamReturnStreamWhenInspectingContainerResolv
         $response->write('output');
     }
 
+    public function testContainerLogsStreamWithCustomStderrEventReturnStreamThatEmitsCustomStderrEventWhenInspectingContainerResolvesWithoutTtyAndContainerLogsRequestResolves()
+    {
+        $this->browser->expects($this->once())->method('get')->with('containers/123/json')->willReturn(\React\Promise\resolve(new Response(200, array(), '{"Config":{"Tty":false}}')));
+        $this->browser->expects($this->once())->method('requestStreaming')->with('GET', 'containers/123/logs?stdout=1&stderr=1')->willReturn(\React\Promise\resolve(new Response(200, array(), '')));
+
+        $response = new ThroughStream();
+        $demultiplexed = new ThroughStream();
+        $this->parser->expects($this->once())->method('expectJson')->willReturn(array('Config' => array('Tty' => false)));
+        $this->streamingParser->expects($this->once())->method('parsePlainStream')->willReturn($response);
+        $this->streamingParser->expects($this->once())->method('demultiplexStream')->with($response, 'stderr')->willReturn($demultiplexed);
+
+        $stream = $this->client->containerLogsStream('123', false, true, true, 0, false, null, 'stderr');
+
+        $stream->on('data', $this->expectCallableOnceWith('output'));
+        $stream->on('stderr', $this->expectCallableOnceWith('error'));
+        $demultiplexed->write('output');
+        $demultiplexed->emit('stderr', array('error'));
+    }
+
     public function testContainerExport()
     {
         $data = 'tar stream';
@@ -437,6 +456,25 @@ public function testContainerAttachStreamReturnStreamWhenInspectingContainerReso
         $response->write('output');
     }
 
+    public function testContainerAttachStreamWithCustomStderrEventReturnStreamThatEmitsCustomStderrEventWhenInspectingContainerResolvesWithoutTtyAndContainerAttachRequestResolves()
+    {
+        $this->browser->expects($this->once())->method('get')->with('containers/123/json')->willReturn(\React\Promise\resolve(new Response(200, array(), '{"Config":{"Tty":false}}')));
+        $this->browser->expects($this->once())->method('requestStreaming')->with('POST', 'containers/123/attach?logs=1&stream=1&stdout=1&stderr=1')->willReturn(\React\Promise\resolve(new Response(200, array(), '')));
+
+        $response = new ThroughStream();
+        $demultiplexed = new ThroughStream();
+        $this->parser->expects($this->once())->method('expectJson')->willReturn(array('Config' => array('Tty' => false)));
+        $this->streamingParser->expects($this->once())->method('parsePlainStream')->willReturn($response);
+        $this->streamingParser->expects($this->once())->method('demultiplexStream')->with($response, 'stderr')->willReturn($demultiplexed);
+
+        $stream = $this->client->containerAttachStream('123', true, true, true, true, 'stderr');
+
+        $stream->on('data', $this->expectCallableOnceWith('output'));
+        $stream->on('stderr', $this->expectCallableOnceWith('error'));
+        $demultiplexed->write('output');
+        $demultiplexed->emit('stderr', array('error'));
+    }
+
     public function testContainerRemove()
     {
         $this->expectRequestFlow('delete', 'containers/123', $this->createResponse(), 'expectEmpty');

tests/FunctionalClientTest.php

@@ -108,6 +108,45 @@ public function testCreateStartAndRemoveContainer()
         $this->assertEquals(array('start', 'die', 'attach', 'destroy'), $actions);
     }
 
+    /**
+     * @depends testImageInspectCheckIfBusyboxExists
+     */
+    public function testCreateStartAndRemoveContainerWithStderrOutputEmitsCustomEventForLogsAndAttachStreams()
+    {
+        $config = array(
+            'Image' => 'busybox',
+            'Cmd' => array('sh', '-c', 'echo out; echo err >&2')
+        );
+
+        $promise = $this->client->containerCreate($config);
+        $container = \React\Async\await($promise);
+
+        $promise = $this->client->containerStart($container['Id']);
+        \React\Async\await($promise);
+
+        $promise = $this->client->containerWait($container['Id']);
+        \React\Async\await($promise);
+
+        $stream = $this->client->containerLogsStream($container['Id'], false, true, true, 0, false, null, 'err');
+        $stream->on('err', $this->expectCallableOnceWith("err\n"));
+        $stream->on('end', $this->expectCallableOnce());
+
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
+
+        $this->assertEquals("out\n", $output);
+
+        $stream = $this->client->containerAttachStream($container['Id'], true, false, true, true, 'err');
+        $stream->on('err', $this->expectCallableOnceWith("err\n"));
+        $stream->on('end', $this->expectCallableOnce());
+
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
+
+        $this->assertEquals("out\n", $output);
+
+        $promise = $this->client->containerRemove($container['Id'], false, true);
+        \React\Async\await($promise);
+    }
+
     /**
      * @depends testImageInspectCheckIfBusyboxExists
      */

tests/Io/ReadableUnwrapStreamTest.php

@@ -123,6 +123,52 @@ public function testEmitsDataWhenInputEmitsData()
         $input->emit('data', array('hello world'));
     }
 
+    public function testEmitsCustomEventWhenInputEmitsCustomEvent()
+    {
+        $input = new ThroughStream();
+        $stream = new ReadableUnwrapStream(Promise\resolve($input), 'stderr');
+
+        $stream->on('data', $this->expectCallableNever());
+        $stream->on('stderr', $this->expectCallableOnceWith('hello world'));
+        $input->emit('stderr', array('hello world'));
+    }
+
+    public function testEmitsDataOnlyOnceWhenCustomEventIsData()
+    {
+        $input = new ThroughStream();
+        $stream = new ReadableUnwrapStream(Promise\resolve($input), 'data');
+
+        $stream->on('data', $this->expectCallableOnceWith('hello world'));
+        $input->emit('data', array('hello world'));
+    }
+
+    public function testEmitsEndOnlyOnceWhenCustomEventIsEnd()
+    {
+        $input = new ThroughStream();
+        $stream = new ReadableUnwrapStream(Promise\resolve($input), 'end');
+
+        $stream->on('end', $this->expectCallableOnce());
+        $input->end();
+    }
+
+    public function testEmitsErrorOnlyOnceWhenCustomEventIsError()
+    {
+        $input = new ThroughStream();
+        $stream = new ReadableUnwrapStream(Promise\resolve($input), 'error');
+
+        $stream->on('error', $this->expectCallableOnceWith(new \RuntimeException()));
+        $input->emit('error', array(new \RuntimeException()));
+    }
+
+    public function testEmitsCloseOnlyOnceWhenCustomEventIsClose()
+    {
+        $input = new ThroughStream();
+        $stream = new ReadableUnwrapStream(Promise\resolve($input), 'close');
+
+        $stream->on('close', $this->expectCallableOnce());
+        $input->close();
+    }
+
     public function testEmitsErrorAndClosesWhenInputEmitsError()
     {
         $input = new ThroughStream();