Merge pull request #367 from clue-labs/streaming

Add `StreamingRequestMiddleware` to stream incoming requests, mark `StreamingServer` as internal

Author: jsor

README.md

@@ -10,8 +10,6 @@
 
 $loop = Factory::create();
 
-// Note how this example still uses `Server` instead of `StreamingServer`.
-// The `StreamingServer` is only required for streaming *incoming* requests.
 $server = new Server(function (ServerRequestInterface $request) use ($loop) {
     if ($request->getMethod() !== 'GET' || $request->getUri()->getPath() !== '/') {
         return new Response(404);

examples/08-stream-response.php

@@ -13,8 +13,9 @@
 use React\Http\Middleware\LimitConcurrentRequestsMiddleware;
 use React\Http\Middleware\RequestBodyBufferMiddleware;
 use React\Http\Middleware\RequestBodyParserMiddleware;
+use React\Http\Middleware\StreamingRequestMiddleware;
 use React\Http\Response;
-use React\Http\StreamingServer;
+use React\Http\Server;
 
 require __DIR__ . '/../vendor/autoload.php';
 
@@ -121,9 +122,10 @@
     );
 };
 
-// Note how this example explicitly uses the advanced `StreamingServer` to apply
+// Note how this example explicitly uses the advanced `StreamingRequestMiddleware` to apply
 // custom request buffering limits below before running our request handler.
-$server = new StreamingServer(array(
+$server = new Server(array(
+    new StreamingRequestMiddleware(),
     new LimitConcurrentRequestsMiddleware(100), // 100 concurrent buffering handlers, queue otherwise
     new RequestBodyBufferMiddleware(8 * 1024 * 1024), // 8 MiB max, ignore body otherwise
     new RequestBodyParserMiddleware(100 * 1024, 1), // 1 file with 100 KiB max, reject upload otherwise

examples/12-upload.php

@@ -1,50 +1,52 @@
 <?php
 
-use Psr\Http\Message\ServerRequestInterface;
 use React\EventLoop\Factory;
-use React\Http\Response;
-use React\Http\StreamingServer;
-use React\Promise\Promise;
 
 require __DIR__ . '/../vendor/autoload.php';
 
 $loop = Factory::create();
 
-// Note how this example uses the advanced `StreamingServer` to allow streaming
+// Note how this example uses the advanced `StreamingRequestMiddleware` to allow streaming
 // the incoming HTTP request. This very simple example merely counts the size
 // of the streaming body, it does not otherwise buffer its contents in memory.
-$server = new StreamingServer(function (ServerRequestInterface $request) {
-    return new Promise(function ($resolve, $reject) use ($request) {
-        $contentLength = 0;
-        $requestBody = $request->getBody();
-        $requestBody->on('data', function ($data) use (&$contentLength) {
-            $contentLength += strlen($data);
+$server = new React\Http\Server(array(
+    new React\Http\Middleware\StreamingRequestMiddleware(),
+    function (Psr\Http\Message\ServerRequestInterface $request) {
+        $body = $request->getBody();
+        assert($body instanceof Psr\Http\Message\StreamInterface);
+        assert($body instanceof React\Stream\ReadableStreamInterface);
+
+        return new React\Promise\Promise(function ($resolve, $reject) use ($body) {
+            $bytes = 0;
+            $body->on('data', function ($data) use (&$bytes) {
+                $bytes += strlen($data);
+            });
+
+            $body->on('end', function () use ($resolve, &$bytes){
+                $resolve(new React\Http\Response(
+                    200,
+                    array(
+                        'Content-Type' => 'text/plain'
+                    ),
+                    "Received $bytes bytes\n"
+                ));
+            });
+
+            // an error occures e.g. on invalid chunked encoded data or an unexpected 'end' event
+            $body->on('error', function (\Exception $exception) use ($resolve, &$bytes) {
+                $resolve(new React\Http\Response(
+                    400,
+                    array(
+                        'Content-Type' => 'text/plain'
+                    ),
+                    "Encountered error after $bytes bytes: {$exception->getMessage()}\n"
+                ));
+            });
         });
+    }
+));
 
-        $requestBody->on('end', function () use ($resolve, &$contentLength){
-            $response = new Response(
-                200,
-                array(
-                    'Content-Type' => 'text/plain'
-                ),
-                "The length of the submitted request body is: " . $contentLength
-            );
-            $resolve($response);
-        });
-
-        // an error occures e.g. on invalid chunked encoded data or an unexpected 'end' event
-        $requestBody->on('error', function (\Exception $exception) use ($resolve, &$contentLength) {
-            $response = new Response(
-                400,
-                array(
-                    'Content-Type' => 'text/plain'
-                ),
-                "An error occured while reading at length: " . $contentLength
-            );
-            $resolve($response);
-        });
-    });
-});
+$server->on('error', 'printf');
 
 $socket = new \React\Socket\Server(isset($argv[1]) ? $argv[1] : '0.0.0.0:0', $loop);
 $server->listen($socket);

examples/13-stream-request.php

@@ -10,10 +10,10 @@
 
 $loop = Factory::create();
 
-// Note how this example uses the `Server` instead of `StreamingServer`.
+// Note how this example uses the `Server` without the `StreamingRequestMiddleware`.
 // This means that this proxy buffers the whole request before "processing" it.
 // As such, this is store-and-forward proxy. This could also use the advanced
-// `StreamingServer` to forward the incoming request as it comes in.
+// `StreamingRequestMiddleware` to forward the incoming request as it comes in.
 $server = new Server(function (RequestInterface $request) {
     if (strpos($request->getRequestTarget(), '://') === false) {
         return new Response(

examples/21-http-proxy.php

@@ -12,7 +12,7 @@
 $loop = Factory::create();
 $connector = new Connector($loop);
 
-// Note how this example uses the `Server` instead of `StreamingServer`.
+// Note how this example uses the `Server` without the `StreamingRequestMiddleware`.
 // Unlike the plain HTTP proxy, the CONNECT method does not contain a body
 // and we establish an end-to-end connection over the stream object, so this
 // doesn't have to store any payload data in memory at all.

examples/22-connect-proxy.php

@@ -27,7 +27,7 @@
 
 $loop = Factory::create();
 
-// Note how this example uses the `Server` instead of `StreamingServer`.
+// Note how this example uses the `Server` without the `StreamingRequestMiddleware`.
 // The initial incoming request does not contain a body and we upgrade to a
 // stream object below.
 $server = new Server(function (ServerRequestInterface $request) use ($loop) {

examples/31-upgrade-echo.php

@@ -35,7 +35,7 @@
 // this means that any Upgraded data will simply be sent back to the client
 $chat = new ThroughStream();
 
-// Note how this example uses the `Server` instead of `StreamingServer`.
+// Note how this example uses the `Server` without the `StreamingRequestMiddleware`.
 // The initial incoming request does not contain a body and we upgrade to a
 // stream object below.
 $server = new Server(function (ServerRequestInterface $request) use ($loop, $chat) {

examples/32-upgrade-chat.php

@@ -86,8 +86,6 @@ public function getSize()
     }
 }
 
-// Note how this example still uses `Server` instead of `StreamingServer`.
-// The `StreamingServer` is only required for streaming *incoming* requests.
 $server = new Server(function (ServerRequestInterface $request) use ($loop) {
     switch ($request->getUri()->getPath()) {
         case '/':

examples/99-benchmark-download.php

@@ -1,18 +1,11 @@
 <?php
 
-namespace React\Http;
+namespace React\Http\Io;
 
 use Evenement\EventEmitter;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\ServerRequestInterface;
-use React\Http\Io\ChunkedDecoder;
-use React\Http\Io\ChunkedEncoder;
-use React\Http\Io\CloseProtectionStream;
-use React\Http\Io\HttpBodyStream;
-use React\Http\Io\LengthLimitedStream;
-use React\Http\Io\MiddlewareRunner;
-use React\Http\Io\RequestHeaderParser;
-use React\Http\Io\ServerRequest;
+use React\Http\Response;
 use React\Promise;
 use React\Promise\CancellablePromiseInterface;
 use React\Promise\PromiseInterface;
@@ -22,7 +15,7 @@
 use React\Stream\WritableStreamInterface;
 
 /**
- * The advanced `StreamingServer` class is responsible for handling incoming connections and then
+ * The internal `StreamingServer` class is responsible for handling incoming connections and then
  * processing each incoming HTTP request.
  *
  * Unlike the [`Server`](#server) class, it does not buffer and parse the incoming
@@ -80,9 +73,11 @@
  * handler function may not be fully compatible with PSR-7. See also
  * [streaming request](#streaming-request) below for more details.
  *
+ * @see \React\Http\Server
  * @see Request
  * @see Response
  * @see self::listen()
+ * @internal
  */
 final class StreamingServer extends EventEmitter
 {
@@ -131,44 +126,8 @@ public function __construct($requestHandler)
     /**
      * Starts listening for HTTP requests on the given socket server instance
      *
-     * The server needs to be attached to an instance of
-     * `React\Socket\ServerInterface` which emits underlying streaming
-     * connections in order to then parse incoming data as HTTP.
-     * For each request, it executes the callback function passed to the
-     * constructor with the respective [request](#request) object and expects
-     * a respective [response](#response) object in return.
-     *
-     * You can attach this to a
-     * [`React\Socket\Server`](https://github.com/reactphp/socket#server)
-     * in order to start a plaintext HTTP server like this:
-     *
-     * ```php
-     * $server = new StreamingServer($handler);
-     *
-     * $socket = new React\Socket\Server(8080, $loop);
-     * $server->listen($socket);
-     * ```
-     *
-     * See also [example #1](examples) for more details.
-     *
-     * Similarly, you can also attach this to a
-     * [`React\Socket\SecureServer`](https://github.com/reactphp/socket#secureserver)
-     * in order to start a secure HTTPS server like this:
-     *
-     * ```php
-     * $server = new StreamingServer($handler);
-     *
-     * $socket = new React\Socket\Server(8080, $loop);
-     * $socket = new React\Socket\SecureServer($socket, $loop, array(
-     *     'local_cert' => __DIR__ . '/localhost.pem'
-     * ));
-     *
-     * $server->listen($socket);
-     * ```
-     *
-     * See also [example #11](examples) for more details.
-     *
      * @param ServerInterface $socket
+     * @see \React\Http\Server::listen()
      */
     public function listen(ServerInterface $socket)
     {