Merge pull request #18 from clue-labs/backpressure

Add backpressure support and support pause()/resume()

Author: clue

src/TransformStream.php

@@ -16,17 +16,27 @@ class TransformStream extends EventEmitter implements DuplexStreamInterface
     private $readable = true;
     private $writable = true;
     private $closed = false;
+    private $paused = false;
+    private $drain = false;
 
     public function write($data)
     {
         if (!$this->writable || $data === '') {
-            return false;
+            return $this->writable;
         }
 
         try {
             $this->transformData($data);
+
+            if ($this->paused) {
+                $this->drain = true;
+                return false;
+            }
+
+            return true;
         } catch (Exception $e) {
             $this->forwardError($e);
+            return false;
         }
     }
 
@@ -73,12 +83,21 @@ public function isWritable()
 
     public function pause()
     {
+        if (!$this->readable) {
+            return;
+        }
 
+        $this->paused = true;
     }
 
     public function resume()
     {
+        $this->paused = false;
 
+        if ($this->drain && $this->writable) {
+            $this->drain = false;
+            $this->emit('drain');
+        }
     }
 
     public function pipe(WritableStreamInterface $dest, array $options = array())

tests/TransformStreamTest.php

@@ -1,6 +1,7 @@
 <?php
 
 use Clue\React\Zlib\TransformStream;
+use React\Stream\ThroughStream;
 
 class TransformStreamTest extends TestCase
 {
@@ -95,4 +96,101 @@ public function testDoesNotEmitIfAlreadyClosed(TransformStream $stream)
         $stream->end();
         $stream->close();
     }
+
+    public function testWriteReturnsTrueNormally()
+    {
+        $stream = new TransformStream();
+
+        $ret = $stream->write('hello');
+        $this->assertTrue($ret);
+    }
+
+    public function testWriteEmptyStringReturnsTrueNormally()
+    {
+        $stream = new TransformStream();
+
+        $ret = $stream->write('');
+        $this->assertTrue($ret);
+    }
+
+    public function testWriteReturnsFalseWhenClosed()
+    {
+        $stream = new TransformStream();
+        $stream->close();
+
+        $ret = $stream->write('hello');
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteEmptyStringReturnsFalseWhenClosed()
+    {
+        $stream = new TransformStream();
+        $stream->close();
+
+        $ret = $stream->write('');
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteReturnsFalseWhenPaused()
+    {
+        $stream = new TransformStream();
+        $stream->pause();
+
+        $ret = $stream->write('hello');
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteReturnsTrueWhenResumedAgain()
+    {
+        $stream = new TransformStream();
+        $stream->pause();
+        $stream->resume();
+
+        $ret = $stream->write('hello');
+        $this->assertTrue($ret);
+    }
+
+    public function testResumeEmitsDrainEventWhenPreviousWriteReturnedFalse()
+    {
+        $stream = new TransformStream();
+        $stream->pause();
+        $stream->write('hello');
+
+        $stream->on('drain', $this->expectCallableOnce());
+        $stream->resume();
+    }
+
+    public function testResumeDoesNotEmitDrainEventWhenNoPreviousWriteReturnedFalse()
+    {
+        $stream = new TransformStream();
+        $stream->pause();
+
+        $stream->on('drain', $this->expectCallableNever());
+        $stream->resume();
+    }
+
+    public function testPauseAndResumeIsNoOpWhenClosed()
+    {
+        $stream = new TransformStream();
+        $stream->close();
+
+        $stream->on('drain', $this->expectCallableNever());
+        $stream->pause();
+        $stream->resume();
+    }
+
+    public function testSupportsBackPressureInPipeChain()
+    {
+        $source = new ThroughStream();
+
+        $dest = new ThroughStream();
+        $dest->pause();
+
+        $stream = new TransformStream();
+
+        $source->pipe($stream)->pipe($dest);
+
+        $ret = $source->write('hello');
+        $this->assertFalse($ret);
+    }
 }