Add withResponseBuffer() method to limit maximum response buffer size

Author: clue

README.md

@@ -66,6 +66,7 @@ mess with most of the low-level details.
         * [withRejectErrorResponse()](#withrejecterrorresponse)
         * [withBase()](#withbase)
         * [withProtocolVersion()](#withprotocolversion)
+        * [withResponseBuffer()](#withresponsebuffer)
         * [~~withOptions()~~](#withoptions)
         * [~~withoutBase()~~](#withoutbase)
     * [ResponseInterface](#responseinterface)
@@ -1196,6 +1197,43 @@ Notice that the [`Browser`](#browser) is an immutable object, i.e. this
 method actually returns a *new* [`Browser`](#browser) instance with the
 new protocol version applied.
 
+#### withResponseBuffer()
+
+The `withRespomseBuffer(int $maximumSize): Browser` method can be used to
+change the maximum size for buffering a response body.
+
+The preferred way to send an HTTP request is by using the above
+[request methods](#request-methods), for example the [`get()`](#get)
+method to send an HTTP `GET` request. Each of these methods will buffer
+the whole response body in memory by default. This is easy to get started
+and works reasonably well for smaller responses.
+
+By default, the response body buffer will be limited to 16 MiB. If the
+response body exceeds this maximum size, the request will be rejected.
+
+You can pass in the maximum number of bytes to buffer:
+
+```php
+$browser = $browser->withResponseBuffer(1024 * 1024);
+
+$browser->get($url)->then(function (Psr\Http\Message\ResponseInterface $response) {
+    // response body will not exceed 1 MiB
+    var_dump($response->getHeaders(), (string) $response->getBody());
+});
+```
+
+Note that the response body buffer has to be kept in memory for each
+pending request until its transfer is completed and it will only be freed
+after a pending request is fulfilled. As such, increasing this maximum
+buffer size to allow larger response bodies is usually not recommended.
+Instead, you can use the [`requestStreaming()` method](#requeststreaming)
+to receive responses with arbitrary sizes without buffering. Accordingly,
+this maximum buffer size setting has no effect on streaming responses.
+
+Notice that the [`Browser`](#browser) is an immutable object, i.e. this
+method actually returns a *new* [`Browser`](#browser) instance with the
+given setting applied.
+
 #### ~~withOptions()~~
 
 > Deprecated since v2.9.0, see [`withTimeout()`](#withtimeout), [`withFollowRedirects()`](#withfollowredirects)

composer.json

@@ -22,7 +22,7 @@
         "react/event-loop": "^1.0 || ^0.5",
         "react/http-client": "^0.5.10",
         "react/promise": "^2.2.1 || ^1.2.1",
-        "react/promise-stream": "^1.0 || ^0.1.1",
+        "react/promise-stream": "^1.0 || ^0.1.2",
         "react/socket": "^1.1",
         "react/stream": "^1.0 || ^0.7",
         "ringcentral/psr7": "^1.2"

src/Browser.php

@@ -744,6 +744,52 @@ public function withProtocolVersion($protocolVersion)
         return $browser;
     }
 
+    /**
+     * Changes the maximum size for buffering a response body.
+     *
+     * The preferred way to send an HTTP request is by using the above
+     * [request methods](#request-methods), for example the [`get()`](#get)
+     * method to send an HTTP `GET` request. Each of these methods will buffer
+     * the whole response body in memory by default. This is easy to get started
+     * and works reasonably well for smaller responses.
+     *
+     * By default, the response body buffer will be limited to 16 MiB. If the
+     * response body exceeds this maximum size, the request will be rejected.
+     *
+     * You can pass in the maximum number of bytes to buffer:
+     *
+     * ```php
+     * $browser = $browser->withResponseBuffer(1024 * 1024);
+     *
+     * $browser->get($url)->then(function (Psr\Http\Message\ResponseInterface $response) {
+     *     // response body will not exceed 1 MiB
+     *     var_dump($response->getHeaders(), (string) $response->getBody());
+     * });
+     * ```
+     *
+     * Note that the response body buffer has to be kept in memory for each
+     * pending request until its transfer is completed and it will only be freed
+     * after a pending request is fulfilled. As such, increasing this maximum
+     * buffer size to allow larger response bodies is usually not recommended.
+     * Instead, you can use the [`requestStreaming()` method](#requeststreaming)
+     * to receive responses with arbitrary sizes without buffering. Accordingly,
+     * this maximum buffer size setting has no effect on streaming responses.
+     *
+     * Notice that the [`Browser`](#browser) is an immutable object, i.e. this
+     * method actually returns a *new* [`Browser`](#browser) instance with the
+     * given setting applied.
+     *
+     * @param int $maximumSize
+     * @return self
+     * @see self::requestStreaming()
+     */
+    public function withResponseBuffer($maximumSize)
+    {
+        return $this->withOptions(array(
+            'maximumSize' => $maximumSize
+        ));
+    }
+
     /**
      * [Deprecated] Changes the [options](#options) to use:
      *

src/Io/Transaction.php

@@ -35,6 +35,8 @@ class Transaction
 
     private $streaming = false;
 
+    private $maximumSize = 16777216; // 16 MiB = 2^24 bytes
+
     public function __construct(Sender $sender, MessageFactory $messageFactory, LoopInterface $loop)
     {
         $this->sender = $sender;
@@ -169,21 +171,38 @@ public function bufferResponse(ResponseInterface $response, $deferred)
     {
         $stream = $response->getBody();
 
+        $size = $stream->getSize();
+        if ($size !== null && $size > $this->maximumSize) {
+            $stream->close();
+            return \React\Promise\reject(new \OverflowException(
+                'Response body size of ' . $size . ' bytes exceeds maximum of ' . $this->maximumSize . ' bytes',
+                \defined('SOCKET_EMSGSIZE') ? \SOCKET_EMSGSIZE : 0
+            ));
+        }
+
         // body is not streaming => already buffered
         if (!$stream instanceof ReadableStreamInterface) {
             return \React\Promise\resolve($response);
         }
 
         // buffer stream and resolve with buffered body
         $messageFactory = $this->messageFactory;
-        $promise = \React\Promise\Stream\buffer($stream)->then(
+        $maximumSize = $this->maximumSize;
+        $promise = \React\Promise\Stream\buffer($stream, $maximumSize)->then(
             function ($body) use ($response, $messageFactory) {
                 return $response->withBody($messageFactory->body($body));
             },
-            function ($e) use ($stream) {
+            function ($e) use ($stream, $maximumSize) {
                 // try to close stream if buffering fails (or is cancelled)
                 $stream->close();
 
+                if ($e instanceof \OverflowException) {
+                    $e = new \OverflowException(
+                        'Response body size exceeds maximum of ' . $this->maximumSize . ' bytes',
+                        \defined('SOCKET_EMSGSIZE') ? \SOCKET_EMSGSIZE : 0
+                    );
+                }
+
                 throw $e;
             }
         );

tests/BrowserTest.php

@@ -203,6 +203,13 @@ public function testWithRejectErrorResponseFalseSetsSenderOption()
         $this->browser->withRejectErrorResponse(false);
     }
 
+    public function testWithResponseBufferThousandSetsSenderOption()
+    {
+        $this->sender->expects($this->once())->method('withOptions')->with(array('maximumSize' => 1000))->willReturnSelf();
+
+        $this->browser->withResponseBuffer(1000);
+    }
+
     public function testWithBase()
     {
         $browser = $this->browser->withBase('http://example.com/root');

tests/FunctionalBrowserTest.php

@@ -315,6 +315,40 @@ public function testResponseStatus300WithoutLocationShouldResolveWithoutFollowin
         Block\await($this->browser->get($this->base . 'status/300'), $this->loop);
     }
 
+    /**
+     * @doesNotPerformAssertions
+     */
+    public function testGetRequestWithResponseBufferMatchedExactlyResolves()
+    {
+        $promise = $this->browser->withResponseBuffer(5)->get($this->base . 'get');
+
+        Block\await($promise, $this->loop);
+    }
+
+    public function testGetRequestWithResponseBufferExceededRejects()
+    {
+        $promise = $this->browser->withResponseBuffer(4)->get($this->base . 'get');
+
+        $this->setExpectedException(
+            'OverflowException',
+            'Response body size of 5 bytes exceeds maximum of 4 bytes',
+            defined('SOCKET_EMSGSIZE') ? SOCKET_EMSGSIZE : 0
+        );
+        Block\await($promise, $this->loop);
+    }
+
+    public function testGetRequestWithResponseBufferExceededDuringStreamingRejects()
+    {
+        $promise = $this->browser->withResponseBuffer(4)->get($this->base . 'stream/1');
+
+        $this->setExpectedException(
+            'OverflowException',
+            'Response body size exceeds maximum of 4 bytes',
+            defined('SOCKET_EMSGSIZE') ? SOCKET_EMSGSIZE : 0
+        );
+        Block\await($promise, $this->loop);
+    }
+
     /**
      * @group online
      * @doesNotPerformAssertions
@@ -595,4 +629,16 @@ public function testRequestStreamingGetReceivesStreamingResponseEvenWhenStreamin
         $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $response->getBody());
         $this->assertEquals('', (string)$response->getBody());
     }
+
+    public function testRequestStreamingGetReceivesStreamingResponseBodyEvenWhenResponseBufferExceeded()
+    {
+        $buffer = Block\await(
+            $this->browser->withResponseBuffer(4)->requestStreaming('GET', $this->base . 'get')->then(function (ResponseInterface $response) {
+                return Stream\buffer($response->getBody());
+            }),
+            $this->loop
+        );
+
+        $this->assertEquals('hello', $buffer);
+    }
 }

tests/Io/TransactionTest.php

@@ -433,6 +433,29 @@ public function testReceivingStreamingBodyWillResolveWithBufferedResponseByDefau
         $this->assertEquals('hello world', (string)$response->getBody());
     }
 
+    public function testReceivingStreamingBodyWithSizeExceedingMaximumResponseBufferWillRejectAndCloseResponseStream()
+    {
+        $messageFactory = new MessageFactory();
+        $loop = Factory::create();
+
+        $stream = new ThroughStream();
+        $stream->on('close', $this->expectCallableOnce());
+
+        $request = $this->getMockBuilder('Psr\Http\Message\RequestInterface')->getMock();
+
+        $response = $messageFactory->response(1.0, 200, 'OK', array('Content-Length' => '100000000'), $stream);
+
+        // mock sender to resolve promise with the given $response in response to the given $request
+        $sender = $this->makeSenderMock();
+        $sender->expects($this->once())->method('send')->with($this->equalTo($request))->willReturn(Promise\resolve($response));
+
+        $transaction = new Transaction($sender, $messageFactory, $loop);
+        $promise = $transaction->send($request);
+
+        $this->setExpectedException('OverflowException');
+        Block\await($promise, $loop, 0.001);
+    }
+
     public function testCancelBufferingResponseWillCloseStreamAndReject()
     {
         $messageFactory = new MessageFactory();