Add backpressure support and support throttling

Author: clue

composer.json

@@ -12,7 +12,7 @@
     ],
     "require": {
         "php": ">=5.3",
-        "react/stream": "^1.0 || ^0.7 || ^0.6"
+        "react/stream": "^1.0 || ^0.7"
     },
     "require-dev": {
         "clue/hexdump": "~0.2.0",

src/Decoder.php

@@ -23,6 +23,7 @@ class Decoder extends EventEmitter implements WritableStreamInterface
     private $buffer = '';
     private $writable = true;
     private $closing = false;
+    private $paused = false;
     private $streaming = null;
     private $remaining = 0;
     private $padding = 0;
@@ -52,7 +53,7 @@ public function write($data)
 
             // entry still incomplete => wait for next chunk
             if ($this->streaming !== null) {
-                return true;
+                return !$this->paused;
             }
         }
 
@@ -93,6 +94,22 @@ public function write($data)
             $this->remaining = $header['size'];
             $this->padding   = $header['padding'];
 
+            // entry stream is not paused by default - unless explicitly paused
+            // emit "drain" even when entry stream is ready again to support backpressure
+            $that = $this;
+            $paused =& $this->paused;
+            $paused = false;
+            $this->streaming->on('drain', function () use (&$paused, $that) {
+                $paused = false;
+                $that->emit('drain');
+            });
+            $this->streaming->on('close', function () use (&$paused, $that) {
+                if ($paused) {
+                    $paused = false;
+                    $that->emit('drain');
+                }
+            });
+
             $this->emit('entry', array($header, $this->streaming));
 
             if ($this->remaining === 0) {
@@ -104,7 +121,7 @@ public function write($data)
 
             // incomplete entry => do not read next header
             if ($this->streaming !== null) {
-                return true;
+                return !$this->paused;
             }
 
             if ($this->padding !== 0) {
@@ -184,14 +201,19 @@ private function consumeEntry($buffer)
         $this->remaining -= $len;
 
         // emit chunk of data
-        $this->streaming->write($data);
+        $ret = $this->streaming->write($data);
 
         // nothing remaining => entry stream finished
         if ($this->remaining === 0) {
             $this->streaming->end();
             $this->streaming = null;
         }
 
+        // throttle input when streaming entry is still writable but returns false (backpressure)
+        if ($ret === false && $this->streaming !== null && $this->streaming->isWritable()) {
+            $this->paused = true;
+        }
+
         return $buffer;
     }
 

tests/DecoderTest.php

@@ -131,6 +131,108 @@ public function testWriteDataWhenStreamingEntryWillEmitDataOnEntryWithoutEndWhen
         $this->assertTrue($ret);
     }
 
+    public function testWriteDataWhenStreamingEntryIsClosedAlreadyWillNotEmitDataAndWillNotThrottleWhenMoreDataIsRemaining()
+    {
+        $entry = new ThroughStream();
+        $entry->close();
+        $entry->on('data', $this->expectCallableNever());
+
+        $ref = new \ReflectionProperty($this->decoder, 'streaming');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, $entry);
+
+        $ref = new \ReflectionProperty($this->decoder, 'remaining');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, 100);
+
+        $ret = $this->decoder->write('hi');
+
+        $this->assertTrue($ret);
+    }
+
+    public function testWriteDataWhenStreamingEntryIsPausedAlreadyWillEmitDataOnEntryAndThrottleWhenMoreDataIsRemaining()
+    {
+        $entry = new ThroughStream();
+        $entry->pause();
+        $entry->on('data', $this->expectCallableOnceWith('hi'));
+        $entry->on('end', $this->expectCallableNever());
+        $entry->on('close', $this->expectCallableNever());
+
+        $ref = new \ReflectionProperty($this->decoder, 'streaming');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, $entry);
+
+        $ref = new \ReflectionProperty($this->decoder, 'remaining');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, 100);
+
+        $ret = $this->decoder->write('hi');
+
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteDataWhenStreamingEntryIsPausedDuringDataWillEmitDataOnEntryAndThrottleWhenMoreDataIsRemaining()
+    {
+        $entry = new ThroughStream();
+        $entry->on('data', function () use ($entry) {
+            $entry->pause();
+        });
+        $entry->on('end', $this->expectCallableNever());
+        $entry->on('close', $this->expectCallableNever());
+
+        $ref = new \ReflectionProperty($this->decoder, 'streaming');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, $entry);
+
+        $ref = new \ReflectionProperty($this->decoder, 'remaining');
+        $ref->setAccessible(true);
+        $ref->setValue($this->decoder, 100);
+
+        $ret = $this->decoder->write('hi');
+
+        $this->assertFalse($ret);
+    }
+
+    public function testWriteDataWhenStreamingEntryIsPausedDuringDataAndResumeEntryAfterwardsWillEmitDrainEventOnDecoder()
+    {
+        $ref = null;
+        $this->decoder->on('entry', function (array $header, ReadableStreamInterface $stream) use (&$ref) {
+            $ref = $stream;
+            $stream->pause();
+        });
+        $this->decoder->on('entry', $this->expectCallableOnce());
+
+        $data = file_get_contents(__DIR__ . '/fixtures/alice-bob.tar', false, null, 0, 512 + 1);
+        $ret = $this->decoder->write($data);
+
+        $this->assertFalse($ret);
+
+        $this->decoder->on('drain', $this->expectCallableOnce());
+
+        $this->assertNotNull($ref);
+        $ref->resume();
+    }
+
+    public function testWriteDataWhenStreamingEntryIsPausedDuringDataAndCloseEntryAfterwardsWillEmitDrainEventOnDecoder()
+    {
+        $ref = null;
+        $this->decoder->on('entry', function (array $header, ReadableStreamInterface $stream) use (&$ref) {
+            $ref = $stream;
+            $stream->pause();
+        });
+        $this->decoder->on('entry', $this->expectCallableOnce());
+
+        $data = file_get_contents(__DIR__ . '/fixtures/alice-bob.tar', false, null, 0, 512 + 1);
+        $ret = $this->decoder->write($data);
+
+        $this->assertFalse($ret);
+
+        $this->decoder->on('drain', $this->expectCallableOnce());
+
+        $this->assertNotNull($ref);
+        $ref->close();
+    }
+
     public function testWriteDataWhenStreamingEntryWillEmitDataOnEntryAndEndAndCloseWhenRemainingDataIsMatched()
     {
         $entry = new ThroughStream();