Merge pull request #1 from clue-labs/close

Closing stream should also close input stream

Author: clue

src/UnwrapReadableStream.php

@@ -56,9 +56,11 @@ function (ReadableStreamInterface $stream) use ($out, &$closed) {
                 });
 
                 // forward end events and close
-                $stream->on('end', function () use ($out) {
-                    $out->emit('end', array($out));
-                    $out->close();
+                $stream->on('end', function () use ($out, &$closed) {
+                    if (!$closed) {
+                        $out->emit('end', array($out));
+                        $out->close();
+                    }
                 });
 
                 // error events cancel output stream
@@ -67,10 +69,9 @@ function (ReadableStreamInterface $stream) use ($out, &$closed) {
                     $out->close();
                 });
 
-                // close output stream once input closes
-                $stream->on('close', function () use ($out) {
-                    $out->close();
-                });
+                // close both streams once either side closes
+                $stream->on('close', array($out, 'close'));
+                $out->on('close', array($stream, 'close'));
 
                 return $stream;
             },

tests/UnwrapReadableTest.php

@@ -211,6 +211,18 @@ public function testPipingStreamWillForwardDataEvents()
         $output->promise()->then($this->expectCallableOnceWith('helloworld'));
     }
 
+    public function testClosingStreamWillCloseInputStream()
+    {
+        $input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $input->expects($this->once())->method('isReadable')->willReturn(true);
+        $input->expects($this->once())->method('close');
+
+        $promise = Promise\resolve($input);
+        $stream = Stream\unwrapReadable($promise);
+
+        $stream->close();
+    }
+
     public function testClosingStreamWillCloseStreamIfItIgnoredCancellationAndResolvesLater()
     {
         $input = new ReadableStream();