Fix issue with high memory consumption in readable streams.

Author: fcabralpacheco

src/Response/SapiStreamEmitter.php

@@ -58,15 +58,18 @@ private function emitBody(ResponseInterface $response, $maxBufferLength)
     {
         $body = $response->getBody();
 
-        if (! $body->isSeekable()) {
-            echo $body;
-            return;
+        if ($body->isSeekable()) {
+            $body->rewind();
         }
 
-        $body->rewind();
-        while (! $body->eof()) {
-            echo $body->read($maxBufferLength);
+        if ($body->isReadable())
+        {
+            while (! $body->eof())
+                echo $body->read($maxBufferLength);
+            return;
         }
+
+        echo $body;
     }
 
     /**
@@ -82,25 +85,31 @@ private function emitBodyRange(array $range, ResponseInterface $response, $maxBu
 
         $body = $response->getBody();
 
-        if (! $body->isSeekable()) {
-            $contents = $body->getContents();
-            echo substr($contents, $first, $last - $first + 1);
-            return;
+        $length = $last - $first + 1;
+
+        if ($body->isSeekable()) {
+            $body->seek($first);
+            $first = 0;
         }
 
-        $body = new RelativeStream($body, $first);
-        $body->rewind();
-        $pos = 0;
-        $length = $last - $first + 1;
-        while (! $body->eof() && $pos < $length) {
-            if (($pos + $maxBufferLength) > $length) {
-                echo $body->read($length - $pos);
-                break;
+        if ($body->isReadable())
+        {
+            for ($remaining = $length; 
+                 ($remaining >= $maxBufferLength) && (!$body->eof());
+                 $remaining -= strlen($contents)) {
+                echo ($contents = $body->read($maxBufferLength));
+
+            }
+
+            if (($remaining > 0) && (!$body->eof())) {
+                echo $body->read($remaining);
             }
 
-            echo $body->read($maxBufferLength);
-            $pos = $body->tell();
+            return;
         }
+
+        $contents = $body->getContents();
+        echo substr($contents, $first, $length);
     }
 
     /**

test/Response/SapiStreamEmitterTest.php

@@ -10,6 +10,7 @@
 namespace ZendTest\Diactoros\Response;
 
 use PHPUnit_Framework_TestCase as TestCase;
+use Prophecy\Argument;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\StreamInterface;
 use Zend\Diactoros\CallbackStream;
@@ -43,6 +44,7 @@ public function testDoesNotInjectContentLengthHeaderIfStreamSizeIsUnknown()
         $stream = $this->prophesize('Psr\Http\Message\StreamInterface');
         $stream->__toString()->willReturn('Content!');
         $stream->isSeekable()->willReturn(false);
+        $stream->isReadable()->willReturn(false);
         $stream->eof()->willReturn(true);
         $stream->rewind()->willReturn(true);
         $stream->getSize()->willReturn(null);
@@ -58,6 +60,101 @@ public function testDoesNotInjectContentLengthHeaderIfStreamSizeIsUnknown()
         }
     }
 
+    public function emitBodyProvider()
+    {
+        return [
+            [true,   '01234567890123456789'    , 10, 2],
+            [true,   '012345678901234567890123', 10, 3],
+            [false,  '01234567890123456789'    , 10, 2],
+            [false,  '012345678901234567890123', 10, 3],
+        ];
+    }
+
+    /**
+     * @dataProvider emitBodyProvider
+    */
+    public function testEmitBody($seekable, $contents, $maxBufferLength, $expectedReads = 0)
+    {
+        $position = 0;
+
+        $stream = $this->prophesize('Psr\Http\Message\StreamInterface');
+        $stream->getSize()->willReturn(strlen($contents));
+        $stream->isSeekable()->willReturn($seekable);
+        $stream->isReadable()->willReturn(true);
+        $stream->rewind()->willReturn(true);
+
+        $stream->eof()->will(function () use (&$contents, &$position){
+            return !isset($contents[$position]);
+        });
+
+        $stream->read(Argument::type('integer'))->will(function ($args) use (&$contents, &$position){
+                $data = substr($contents, $position, $args[0]);
+                $position += strlen($data);
+                return $data;
+            });
+
+        $response = (new Response())
+            ->withStatus(200)
+            ->withBody($stream->reveal());
+
+        ob_start();
+        $this->emitter->emit($response, $maxBufferLength);
+        $stream->rewind()->shouldBeCalledTimes($seekable ? 1 : 0);
+        $stream->read(Argument::type('integer'))->shouldBeCalledTimes($expectedReads);
+        $this->assertEquals($contents, ob_get_clean());
+
+    }
+
+    public function emitBodyRangeProvider()
+    {
+        return [
+            [true,  '01234567890123456789'    , ['bytes', 10, 20, '*'], 10, 1],
+            [true,  '012345678901234567890123', ['bytes', 10, 40, '*'], 10, 2],
+            [false, '01234567890123456789'    , ['bytes', 11, 20, '*'], 10, 1],
+            [false, '012345678901234567890123', ['bytes', 11, 40, '*'], 10, 2],
+        ];
+    }
+
+    /**
+     * @dataProvider emitBodyRangeProvider
+     */
+    public function testEmitBodyRange($seekable, $contents, $range, $maxBufferLength, $expectedReads = 0)
+    {
+        list($unit, $first, $last, $length) = $range;
+
+        $position = $first;
+
+        $stream = $this->prophesize('Psr\Http\Message\StreamInterface');
+        $stream->getSize()->willReturn(strlen($contents));
+        $stream->isSeekable()->willReturn($seekable);
+        $stream->isReadable()->willReturn(true);
+        $stream->seek(Argument::type('integer'))->will(function ($args) use (&$position){
+            $position = $args[0];
+            return true;
+        });
+
+        $stream->eof()->will(function () use (&$contents, &$position){
+            return !isset($contents[$position]);
+        });
+
+        $stream->read(Argument::type('integer'))->will(function ($args) use (&$contents, &$position){
+            $data = substr($contents, $position, $args[0]);
+            $position += strlen($data);
+            return $data;
+        });
+
+        $response = (new Response())
+            ->withStatus(200)
+            ->withHeader('Content-Range', "$unit $first-$last/$length")
+            ->withBody($stream->reveal());
+
+        ob_start();
+        $this->emitter->emit($response, $maxBufferLength);
+        $stream->seek(Argument::type('integer'))->shouldBeCalledTimes($seekable ? 1 : 0);
+        $stream->read(Argument::type('integer'))->shouldBeCalledTimes($expectedReads);
+        $this->assertEquals(substr($contents, $first, $last - $first + 1), ob_get_clean());
+    }
+
     public function contentRangeProvider()
     {
         return [