Silently close input stream if promise does not reject after cancellation

Author: clue

src/UnwrapReadableStream.php

@@ -27,6 +27,7 @@ class UnwrapReadableStream extends EventEmitter implements ReadableStreamInterfa
     public function __construct(PromiseInterface $promise)
     {
         $out = $this;
+        $closed =& $this->closed;
 
         $this->promise = $promise->then(
             function ($stream) {
@@ -36,12 +37,19 @@ function ($stream) {
                 return $stream;
             }
         )->then(
-            function (ReadableStreamInterface $stream) use ($out) {
+            function (ReadableStreamInterface $stream) use ($out, &$closed) {
+                // stream is already closed, make sure to close output stream
                 if (!$stream->isReadable()) {
                     $out->close();
                     return $stream;
                 }
 
+                // resolves but output is already closed, make sure to close stream silently
+                if ($closed) {
+                    $stream->close();
+                    return $stream;
+                }
+
                 // stream any writes into output stream
                 $stream->on('data', function ($data) use ($out) {
                     $out->emit('data', array($data, $out));

tests/UnwrapReadableTest.php

@@ -210,4 +210,43 @@ public function testPipingStreamWillForwardDataEvents()
 
         $output->promise()->then($this->expectCallableOnceWith('helloworld'));
     }
+
+    public function testClosingStreamWillCloseStreamIfItIgnoredCancellationAndResolvesLater()
+    {
+        $input = new ReadableStream();
+
+        $loop = $this->loop;
+        $promise = new Promise\Promise(function ($resolve) use ($loop, $input) {
+            $loop->addTimer(0.001, function () use ($resolve, $input) {
+                $resolve($input);
+            });
+        });
+
+        $stream = Stream\unwrapReadable($promise);
+
+        $stream->on('close', $this->expectCallableOnce());
+
+        $stream->close();
+
+        Block\await($promise, $this->loop);
+
+        $this->assertFalse($input->isReadable());
+    }
+
+    public function testClosingStreamWillCloseStreamFromCancellationHandler()
+    {
+        $input = new ReadableStream();
+
+        $promise = new \React\Promise\Promise(function () { }, function ($resolve) use ($input) {
+            $resolve($input);
+        });
+
+        $stream = Stream\unwrapReadable($promise);
+
+        $stream->on('close', $this->expectCallableOnce());
+
+        $stream->close();
+
+        $this->assertFalse($input->isReadable());
+    }
 }