Merge pull request #17 from clue-labs/unwrapped-write

Writing to closed unwrapped stream should return false (backpressure)

Author: jsor

src/UnwrapWritableStream.php

@@ -105,7 +105,7 @@ function ($e) use ($out, &$closed) {
     public function write($data)
     {
         if ($this->ending) {
-            return;
+            return false;
         }
 
         // forward to inner stream if possible

tests/UnwrapWritableTest.php

@@ -253,6 +253,55 @@ public function testForwardsNoDataWhenWritingAfterEndOncePromiseResolves()
         $this->loop->run();
     }
 
+    public function testWriteReturnsFalseWhenPromiseIsPending()
+    {
+        $promise = new \React\Promise\Promise(function () { });
+        $stream = Stream\unwrapWritable($promise);
+
+        $ret = $stream->write('nope');
+
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteReturnsTrueWhenUnwrappedStreamReturnsTrueForWrite()
+    {
+        $input = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $input->expects($this->once())->method('isWritable')->willReturn(true);
+        $input->expects($this->once())->method('write')->willReturn(true);
+
+        $promise = \React\Promise\resolve($input);
+        $stream = Stream\unwrapWritable($promise);
+
+        $ret = $stream->write('hello');
+
+        $this->assertTrue($ret);
+    }
+
+    public function testWriteReturnsFalseWhenUnwrappedStreamReturnsFalseForWrite()
+    {
+        $input = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $input->expects($this->once())->method('isWritable')->willReturn(true);
+        $input->expects($this->once())->method('write')->willReturn(false);
+
+        $promise = \React\Promise\resolve($input);
+        $stream = Stream\unwrapWritable($promise);
+
+        $ret = $stream->write('nope');
+
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteAfterCloseReturnsFalse()
+    {
+        $promise = new \React\Promise\Promise(function () { });
+        $stream = Stream\unwrapWritable($promise);
+
+        $stream->close();
+        $ret = $stream->write('nope');
+
+        $this->assertFalse($ret);
+    }
+
     public function testEmitsErrorAndClosesWhenInputEmitsError()
     {
         $input = new ThroughStream();
@@ -267,6 +316,20 @@ public function testEmitsErrorAndClosesWhenInputEmitsError()
         $this->assertFalse($stream->isWritable());
     }
 
+    public function testEmitsDrainWhenPromiseResolvesWithStreamWhenForwardingData()
+    {
+        $input = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();
+        $input->expects($this->once())->method('isWritable')->willReturn(true);
+        $input->expects($this->once())->method('write')->with('hello')->willReturn(true);
+
+        $deferred = new Deferred();
+        $stream = Stream\unwrapWritable($deferred->promise());
+        $stream->write('hello');
+
+        $stream->on('drain', $this->expectCallableOnce());
+        $deferred->resolve($input);
+    }
+
     public function testDoesNotEmitDrainWhenStreamBufferExceededAfterForwardingData()
     {
         $input = $this->getMockBuilder('React\Stream\WritableStreamInterface')->getMock();