Update to stream responses without dependency on react/promise-stream

composer.json

@@ -16,7 +16,6 @@
         "react/event-loop": "^1.6",
         "react/http": "^1.11",
         "react/promise": "^3.3 || ^2.11 || ^1.3",
-        "react/promise-stream": "^1.6",
         "react/socket": "^1.17",
         "react/stream": "^1.4"
     },

src/Client.php

@@ -2,6 +2,7 @@
 
 namespace Clue\React\Docker;
 
+use Clue\React\Docker\Io\ReadableUnwrapStream;
 use Clue\React\Docker\Io\ResponseParser;
 use Clue\React\Docker\Io\StreamingParser;
 use React\EventLoop\LoopInterface;
@@ -365,7 +366,7 @@ public function containerLogsStream($container, $follow = false, $stdout = true,
         ));
 
         // first inspect container to check TTY setting, then request logs with appropriate log parser
-        return \React\Promise\Stream\unwrapReadable($this->containerInspect($container)->then(function ($info) use ($url, $browser, $parser, $stderrEvent) {
+        return new ReadableUnwrapStream($this->containerInspect($container)->then(function ($info) use ($url, $browser, $parser, $stderrEvent) {
             $stream = $parser->parsePlainStream($browser->requestStreaming('GET', $url));
 
             if (!$info['Config']['Tty']) {
@@ -729,7 +730,7 @@ public function containerAttachStream($container, $logs = false, $stream = false
         ));
 
         // first inspect container to check TTY setting, then attach with appropriate log parser
-        return \React\Promise\Stream\unwrapReadable($this->containerInspect($container)->then(function ($info) use ($url, $browser, $parser, $stderrEvent) {
+        return new ReadableUnwrapStream($this->containerInspect($container)->then(function ($info) use ($url, $browser, $parser, $stderrEvent) {
             $stream = $parser->parsePlainStream($browser->requestStreaming('POST', $url));
 
             if (!$info['Config']['Tty']) {

src/Io/ReadableUnwrapStream.php

@@ -0,0 +1,127 @@
+<?php
+
+namespace Clue\React\Docker\Io;
+
+use Evenement\EventEmitter;
+use React\Promise\PromiseInterface;
+use React\Stream\ReadableStreamInterface;
+use React\Stream\Util;
+use React\Stream\WritableStreamInterface;
+
+/**
+ * Readable stream for the given promise which resolves with a readable stream
+ *
+ * @internal
+ * @link https://github.com/reactphp/promise-stream/blob/v1.7.0/src/UnwrapReadableStream.php Based on react/promise-stream
+ */
+class ReadableUnwrapStream extends EventEmitter implements ReadableStreamInterface
+{
+    /** @var ?PromiseInterface<ReadableStreamInterface> */
+    private $promise;
+
+    /** @var ?ReadableStreamInterface */
+    private $input;
+
+    private $paused = false;
+    private $closed = false;
+
+    /**
+     * @param PromiseInterface<ReadableStreamInterface> $promise
+     */
+    public function __construct(PromiseInterface $promise)
+    {
+        $this->promise = $promise;
+
+        $out = $this;
+        $input =& $this->input;
+        $paused =& $this->paused;
+
+        $promise->then(
+            function (ReadableStreamInterface $stream) use ($out, &$input, &$paused) {
+                // output already closed while waiting or input already closed => close other side
+                if (!$out->isReadable() || !$stream->isReadable()) {
+                    $stream->close();
+                    $out->close();
+                    return;
+                }
+
+                $input = $stream;
+                if ($paused) {
+                    $stream->pause();
+                }
+
+                // forward data event to output
+                $stream->on('data', function ($data) use ($out) {
+                    $out->emit('data', array($data));
+                });
+
+                // forward end event to output and close
+                $stream->on('end', function () use ($out) {
+                    $out->emit('end');
+                    $out->close();
+                });
+
+                // forward error event to output and close
+                $stream->on('error', function ($e) use ($out) {
+                    $out->emit('error', array($e));
+                    $out->close();
+                });
+
+                // close both streams once either side closes
+                $stream->on('close', array($out, 'close'));
+                $out->on('close', array($stream, 'close'));
+            },
+            function ($e) use ($out) {
+                // ignore rejection after output is closed (cancellation)
+                if ($out->isReadable()) {
+                    $out->emit('error', array($e));
+                    $out->close();
+                }
+            }
+        );
+    }
+
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+    public function pause()
+    {
+        $this->paused = true;
+        if ($this->input !== null) {
+            $this->input->pause();
+        }
+    }
+
+    public function resume()
+    {
+        $this->paused = false;
+        if ($this->input !== null) {
+            $this->input->resume();
+        }
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        return Util::pipe($this, $dest, $options);
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+
+        // try to cancel promise once the stream closes
+        if ($this->promise !== null && method_exists($this->promise, 'cancel')) {
+            $this->promise->cancel();
+        }
+        $this->promise = null;
+
+        $this->emit('close');
+        $this->removeAllListeners();
+    }
+}

src/Io/StreamingParser.php

@@ -5,7 +5,6 @@
 use Psr\Http\Message\ResponseInterface;
 use React\Promise\Deferred;
 use React\Promise\PromiseInterface;
-use React\Promise\Stream;
 use React\Stream\ReadableStreamInterface;
 use RuntimeException;
 
@@ -41,7 +40,7 @@ public function parsePlainStream(PromiseInterface $promise)
     {
         // text/plain
 
-        return Stream\unwrapReadable($promise->then(function (ResponseInterface $response) {
+        return new ReadableUnwrapStream($promise->then(function (ResponseInterface $response) {
             return $response->getBody();
         }));
     }
@@ -62,11 +61,44 @@ public function demultiplexStream(ReadableStreamInterface $input, $stderrEvent =
      * Returns a promise which resolves with the buffered stream contents of the given stream
      *
      * @param ReadableStreamInterface $stream
-     * @return PromiseInterface Promise<string, Exception>
+     * @return PromiseInterface<string>
      */
     public function bufferedStream(ReadableStreamInterface $stream)
     {
-        return Stream\buffer($stream);
+        // stream already closed => resolve with empty buffer
+        if (!$stream->isReadable()) {
+            return \React\Promise\resolve('');
+        }
+
+        $buffer = '';
+        $bufferer = function ($data) use (&$buffer) {
+            $buffer .= $data;
+        };
+        $stream->on('data', $bufferer);
+
+        // cancelling the deferred will stop buffering and clear buffered data
+        $deferred = new Deferred(function () use ($stream, $bufferer, &$buffer) {
+            $stream->removeListener('data', $bufferer);
+            $buffer = '';
+
+            throw new RuntimeException('Cancelled buffering');
+        });
+
+        // error event rejects (stream closes afterwards)
+        $stream->on('error', function ($e) use ($deferred) {
+            $deferred->reject(new RuntimeException(
+                'An error occured on the underlying stream while buffering: ' . $e->getMessage(),
+                $e->getCode(),
+                $e
+            ));
+        });
+
+        // close event resolves with buffered data (unless already rejected)
+        $stream->on('close', function () use ($deferred, &$buffer) {
+            $deferred->resolve($buffer);
+        });
+
+        return $deferred->promise();
     }
 
     /**

tests/FunctionalClientTest.php

@@ -3,19 +3,21 @@
 namespace Clue\Tests\React\Docker;
 
 use Clue\React\Docker\Client;
+use Clue\React\Docker\Io\StreamingParser;
 use React\EventLoop\Loop;
-use React\Promise\Stream;
 
 class FunctionalClientTest extends TestCase
 {
     private $client;
+    private $parser;
 
     /**
      * @before
      */
     public function setUpClient()
     {
         $this->client = new Client();
+        $this->parser = new StreamingParser();
 
         $promise = $this->client->ping();
 
@@ -221,7 +223,7 @@ public function testExecStreamOutputInMultipleChunksWhileRunning($container)
         $stream->once('data', $this->expectCallableOnceWith('hello'));
         $stream->on('end', $this->expectCallableOnce());
 
-        $output = \React\Async\await(Stream\buffer($stream));
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
 
         $this->assertEquals('helloworld', $output);
     }
@@ -278,7 +280,7 @@ public function testExecStreamCommandWithTtyAndStderrOutputWhileRunning($contain
         $stream->once('data', $this->expectCallableOnce('hello world'));
         $stream->on('end', $this->expectCallableOnce());
 
-        $output = \React\Async\await(Stream\buffer($stream));
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
 
         $this->assertEquals('hello world', $output);
     }
@@ -301,7 +303,7 @@ public function testExecStreamStderrCustomEventWhileRunning($container)
         $stream->on('error', $this->expectCallableNever());
         $stream->on('end', $this->expectCallableOnce());
 
-        $output = \React\Async\await(Stream\buffer($stream));
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
 
         $this->assertEquals('', $output);
     }
@@ -321,7 +323,7 @@ public function testExecStreamEmptyOutputWhileRunning($container)
         $stream = $this->client->execStartStream($exec['Id'], true);
         $stream->on('end', $this->expectCallableOnce());
 
-        $output = \React\Async\await(Stream\buffer($stream));
+        $output = \React\Async\await($this->parser->bufferedStream($stream));
 
         $this->assertEquals('', $output);
     }