Support back-pressure (forward pause() and resume() calls)

Author: clue

src/UnwrapReadableStream.php

@@ -28,8 +28,6 @@ public function __construct(PromiseInterface $promise)
     {
         $out = $this;
 
-        // TODO: support backpressure
-
         $this->promise = $promise->then(
             function ($stream) {
                 if (!($stream instanceof ReadableStreamInterface)) {
@@ -78,10 +76,16 @@ public function isReadable()
 
     public function pause()
     {
+        $this->promise->then(function (ReadableStreamInterface $stream) {
+            $stream->pause();
+        });
     }
 
     public function resume()
     {
+        $this->promise->then(function (ReadableStreamInterface $stream) {
+            $stream->resume();
+        });
     }
 
     public function pipe(WritableStreamInterface $dest, array $options = array())

tests/UnwrapReadableTest.php

@@ -164,4 +164,26 @@ public function testEmitsCloseOnlyOnceWhenClosingStreamMultipleTimes()
         $stream->close();
         $stream->close();
     }
+
+    public function testForwardsPauseToInputStream()
+    {
+        $input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $input->expects($this->once())->method('pause');
+
+        $promise = Promise\resolve($input);
+        $stream = Stream\unwrapReadable($promise);
+
+        $stream->pause();
+    }
+
+    public function testForwardsResumeToInputStream()
+    {
+        $input = $this->getMock('React\Stream\ReadableStreamInterface');
+        $input->expects($this->once())->method('resume');
+
+        $promise = Promise\resolve($input);
+        $stream = Stream\unwrapReadable($promise);
+
+        $stream->resume();
+    }
 }