Support streaming=true option to resolve response without buffering body

Author: clue

README.md

@@ -35,6 +35,7 @@ mess with most of the low-level details.
     * [Methods](#methods)
     * [Promises](#promises)
     * [Blocking](#blocking)
+    * [Streaming](#streaming)
     * [submit()](#submit)
     * [send()](#send)
     * [withOptions()](#withoptions)
@@ -53,7 +54,6 @@ mess with most of the low-level details.
   * [SOCKS proxy](#socks-proxy)
   * [UNIX domain sockets](#unix-domain-sockets)
   * [Options](#options)
-  * [Streaming](#streaming)
 * [Install](#install)
 * [License](#license)
 
@@ -129,6 +129,18 @@ $browser->get($url)->then(
 
 If this looks strange to you, you can also use the more traditional [blocking API](#blocking).
 
+Keep in mind that resolving the Promise with the full response message means the
+whole response body has to be kept in memory.
+This is easy to get started and works reasonably well for smaller responses
+(such as common HTML pages or RESTful or JSON API requests).
+
+You may also want to look into the [streaming API](#streaming):
+
+* If you're dealing with lots of concurrent requests (100+) or
+* If you want to process individual data chunks as they happen (without having to wait for the full response body) or
+* If you're expecting a big response body size (1 MiB or more, for example when downloading binary files) or
+* If you're unsure about the response body size (better be safe than sorry when accessing arbitrary remote HTTP endpoints and the response body size is unknown in advance). 
+
 #### Blocking
 
 As stated above, this library provides you a powerful, async API by default.
@@ -167,6 +179,89 @@ $responses = Block\awaitAll($promises, $loop);
 
 Please refer to [clue/block-react](https://github.com/clue/php-block-react#readme) for more details.
 
+Keep in mind the above remark about buffering the whole response message in memory.
+As an alternative, you may also see the following chapter for the
+[streaming API](#streaming).
+
+#### Streaming
+
+All of the above examples assume you want to store the whole response body in memory.
+This is easy to get started and works reasonably well for smaller responses.
+
+However, there are several situations where it's usually a better idea to use a
+streaming approach, where only small chunks have to be kept in memory:
+
+* If you're dealing with lots of concurrent requests (100+) or
+* If you want to process individual data chunks as they happen (without having to wait for the full response body) or
+* If you're expecting a big response body size (1 MiB or more, for example when downloading binary files) or
+* If you're unsure about the response body size (better be safe than sorry when accessing arbitrary remote HTTP endpoints and the response body size is unknown in advance). 
+
+The streaming API uses the same HTTP message API, but does not buffer the response
+message body in memory.
+It only processes the response body in small chunks as data is received and
+forwards this data through [React's Stream API](https://github.com/reactphp/stream).
+This works for (any number of) responses of arbitrary sizes.
+
+This resolves with a normal [`ResponseInterface`](#responseinterface), which
+can be used access the response message parameters as usual.
+You can access the message body as usual, however it now
+implements React's [`ReadableStreamInterface`](https://github.com/reactphp/stream#readablestreaminterface)
+as well as parts of the PSR-7's [`StreamInterface`](http://www.php-fig.org/psr/psr-7/#3-4-psr-http-message-streaminterface).
+
+```php
+// turn on streaming responses (does no longer buffer response body)
+$streamingBrowser = $browser->withOptions(array('streaming' => true));
+
+// issue a normal GET request
+$streamingBrowser->get($url)->then(function (ResponseInterface $response) {
+    $body = $response->getBody();
+    /* @var $body \React\Stream\ReadableStreamInterface */
+    
+    $body->on('data', function ($chunk) {
+        echo $chunk;
+    });
+    
+    $body->on('error', function (Exception $error) {
+        echo 'Error: ' . $error->getMessage() . PHP_EOL;
+    });
+    
+    $body->on('close', function () {
+        echo '[DONE]' . PHP_EOL;
+    });
+});
+```
+
+See also the [stream bandwith example](examples/stream-bandwidth.php) and
+the [stream forwarding example](examples/stream-forwarding.php).
+
+You can invoke the following methods on the message body:
+
+```php
+$body->on($event, $callback);
+$body->eof();
+$body->isReadable();
+$body->close();
+$body->pause();
+$body->resume();
+```
+
+Because the message body is in a streaming state, invoking the following methods
+doesn't make much sense:
+
+```php
+$body->__toString(); // ''
+$body->detach(); // throws BadMethodCallException
+$body->getSize(); // null
+$body->tell(); // throws BadMethodCallException
+$body->isSeekable(); // false
+$body->seek(); // throws BadMethodCallException
+$body->rewind(); // throws BadMethodCallException
+$body->isWritable(); // false
+$body->write(); // throws BadMethodCallException
+$body->read(); // throws BadMethodCallException
+$body->getContents(); // throws BadMethodCallException
+```
+
 #### submit()
 
 The `submit($url, array $fields, $headers = array(), $method = 'POST')` method can be used to submit an array of field values similar to submitting a form (`application/x-www-form-urlencoded`).
@@ -433,31 +528,14 @@ can be controlled via the following API (and their defaults):
 $newBrowser = $browser->withOptions(array(
     'followRedirects' => true,
     'maxRedirects' => 10,
-    'obeySuccessCode' => true
+    'obeySuccessCode' => true,
+    'streaming' => false,
 ));
 ```
 
 Notice that the [`Browser`](#browser) is an immutable object, i.e. the `withOptions()` method
 actually returns a *new* [`Browser`](#browser) instance with the options applied.
 
-### Streaming
-
-Note: This API is subject to change.
-
-The [`Sender`](#sender) emits a `progress` event array on its `Promise` that can be used
-to intercept the underlying outgoing request stream (`React\HttpClient\Request` in the `requestStream` key)
-and the incoming response stream (`React\HttpClient\Response` in the `responseStream` key).
-
-```php
-$client->get('http://www.google.com/')->then($handler, null, function ($event) {
-    if (isset($event['responseStream'])) {
-        /* @var $stream React\HttpClient\Response */
-        $stream = $event['responseStream'];
-        $stream->on('data', function ($data) { });
-    }
-});
-```
-
 ## Install
 
 The recommended way to install this library is [through Composer](http://getcomposer.org).

examples/stream-bandwidth.php

@@ -0,0 +1,48 @@
+<?php
+
+use Clue\React\Buzz\Browser;
+use Psr\Http\Message\ResponseInterface;
+use React\Stream\ReadableStreamInterface;
+use RingCentral\Psr7;
+
+$url = isset($argv[1]) ? $argv[1] : 'http://google.com/';
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$loop = React\EventLoop\Factory::create();
+$client = new Browser($loop);
+
+echo 'Requesting ' . $url . '…' . PHP_EOL;
+
+$client->withOptions(array('streaming' => true))->get($url)->then(function (ResponseInterface $response) use ($loop) {
+    echo 'Headers received' . PHP_EOL;
+    echo Psr7\str($response);
+
+    $stream = $response->getBody();
+    if (!$stream instanceof ReadableStreamInterface) {
+        throw new UnexpectedValueException();
+    }
+
+    // count number of bytes received
+    $bytes = 0;
+    $stream->on('data', function ($chunk) use (&$bytes) {
+        $bytes += strlen($chunk);
+    });
+
+    // report progress every 0.1s
+    $timer = $loop->addPeriodicTimer(0.1, function () use (&$bytes) {
+        echo "\rDownloaded " . $bytes . " bytes…";
+    });
+
+    // report results once the stream closes
+    $time = microtime(true);
+    $stream->on('close', function() use (&$bytes, $timer, $loop, $time) {
+        $loop->cancelTimer($timer);
+
+        $time = microtime(true) - $time;
+
+        echo "\r" . 'Downloaded ' . $bytes . ' bytes in ' . round($time, 3) . 's => ' . round($bytes / $time / 1024 / 1024, 1) . ' MiB/s' . PHP_EOL;
+    });
+}, 'printf');
+
+$loop->run();

examples/stream-forwarding.php

@@ -0,0 +1,30 @@
+<?php
+
+use Clue\React\Buzz\Browser;
+use React\Stream\ReadableStreamInterface;
+use Psr\Http\Message\ResponseInterface;
+use React\Stream\Stream;
+use RingCentral\Psr7;
+
+$url = isset($argv[1]) ? $argv[1] : 'http://google.com/';
+
+require __DIR__ . '/../vendor/autoload.php';
+
+$loop = React\EventLoop\Factory::create();
+$client = new Browser($loop);
+
+$out = new Stream(STDOUT, $loop);
+$out->pause();
+
+$info = new Stream(STDERR, $loop);
+$info->pause();
+
+$info->write('Requesting ' . $url . '…' . PHP_EOL);
+
+$client->withOptions(array('streaming' => true))->get($url)->then(function (ResponseInterface $response) use ($info, $out) {
+    $info->write('Received' . PHP_EOL . Psr7\str($response));
+
+    $response->getBody()->pipe($out);
+}, 'printf');
+
+$loop->run();

src/Io/Transaction.php

@@ -35,6 +35,8 @@ class Transaction
     // context: http.ignore_errors
     private $obeySuccessCode = true;
 
+    private $streaming = false;
+
     public function __construct(RequestInterface $request, Sender $sender, array $options = array(), MessageFactory $messageFactory)
     {
         foreach ($options as $name => $value) {
@@ -62,7 +64,10 @@ protected function next(RequestInterface $request)
 
         return $this->sender->send($request, $this->messageFactory)->then(
             function (ResponseInterface $response) use ($that) {
-                return $that->bufferResponse($response);
+                if (!$that->streaming) {
+                    return $that->bufferResponse($response);
+                }
+                return $response;
             }
         )->then(
             function (ResponseInterface $response) use ($request, $that) {

tests/Io/TransactionTest.php

@@ -6,6 +6,8 @@
 use Clue\React\Buzz\Message\MessageFactory;
 use React\Promise;
 use Clue\React\Block;
+use React\EventLoop\Factory;
+use React\Stream\ReadableStream;
 
 class TransactionTest extends TestCase
 {
@@ -29,4 +31,51 @@ public function testReceivingErrorResponseWillRejectWithResponseException()
             $this->assertSame($response, $exception->getResponse());
         }
     }
+
+    public function testReceivingStreamingBodyWillResolveWithBufferedResponseByDefault()
+    {
+        $messageFactory = new MessageFactory();
+        $loop = Factory::create();
+
+        $stream = new ReadableStream();
+        $loop->addTimer(0.001, function () use ($stream) {
+            $stream->emit('data', array('hello world'));
+            $stream->close();
+        });
+
+        $request = $this->getMock('Psr\Http\Message\RequestInterface');
+        $response = $messageFactory->response(1.0, 200, 'OK', array(), $stream);
+
+        // mock sender to resolve promise with the given $response in response to the given $request
+        $sender = $this->getMockBuilder('Clue\React\Buzz\Io\Sender')->disableOriginalConstructor()->getMock();
+        $sender->expects($this->once())->method('send')->with($this->equalTo($request))->willReturn(Promise\resolve($response));
+
+        $transaction = new Transaction($request, $sender, array(), $messageFactory);
+        $promise = $transaction->send();
+
+        $response = Block\await($promise, $loop);
+
+        $this->assertEquals(200, $response->getStatusCode());
+        $this->assertEquals('hello world', (string)$response->getBody());
+    }
+
+    public function testReceivingStreamingBodyWillResolveWithStreamingResponseIfStreamingIsEnabled()
+    {
+        $messageFactory = new MessageFactory();
+
+        $request = $this->getMock('Psr\Http\Message\RequestInterface');
+        $response = $messageFactory->response(1.0, 200, 'OK', array(), $this->getMock('React\Stream\ReadableStreamInterface'));
+
+        // mock sender to resolve promise with the given $response in response to the given $request
+        $sender = $this->getMockBuilder('Clue\React\Buzz\Io\Sender')->disableOriginalConstructor()->getMock();
+        $sender->expects($this->once())->method('send')->with($this->equalTo($request))->willReturn(Promise\resolve($response));
+
+        $transaction = new Transaction($request, $sender, array('streaming' => true), $messageFactory);
+        $promise = $transaction->send();
+
+        $response = Block\await($promise, $this->getMock('React\EventLoop\LoopInterface'));
+
+        $this->assertEquals(200, $response->getStatusCode());
+        $this->assertEquals('', (string)$response->getBody());
+    }
 }