Asynchronously handle streamed responses (#40)

* Asynchronously handle streamed responses

* Cleanup and add Content-Length

Author: andig

Bridges/HttpKernel.php

@@ -78,9 +78,9 @@ public function onRequest(ReactRequest $request, HttpResponse $response)
 
         $syRequest = $this->mapRequest($request);
 
-        //start buffering the output, so cgi is not sending any http headers
-        //this is necessary because it would break session handling since
-        //headers_sent() returns true if any unbuffered output reaches cgi stdout.
+        // start buffering the output, so cgi is not sending any http headers
+        // this is necessary because it would break session handling since
+        // headers_sent() returns true if any unbuffered output reaches cgi stdout.
         ob_start();
 
         try {
@@ -92,9 +92,15 @@ public function onRequest(ReactRequest $request, HttpResponse $response)
         } catch (\Exception $exception) {
             $response->writeHead(500); // internal server error
             $response->end();
+
+            // end buffering if we need to throw
+            @ob_end_clean();
             throw $exception;
         }
 
+        // should not receive output from application->handle()
+        @ob_end_clean();
+
         $this->mapResponse($response, $syResponse);
 
         if ($this->application instanceof TerminableInterface) {
@@ -170,15 +176,10 @@ protected function mapRequest(ReactRequest $reactRequest)
      */
     protected function mapResponse(HttpResponse $reactResponse, SymfonyResponse $syResponse)
     {
-        //end active session
+        // end active session
         if (PHP_SESSION_ACTIVE === session_status()) {
             session_write_close();
-            session_unset(); //reset $_SESSION
-        }
-
-        $content = $syResponse->getContent();
-        if ($syResponse instanceof SymfonyStreamedResponse) {
-            $syResponse->sendContent();
+            session_unset(); // reset $_SESSION
         }
 
         $nativeHeaders = [];
@@ -200,8 +201,8 @@ protected function mapResponse(HttpResponse $reactResponse, SymfonyResponse $syR
             }
         }
 
-        //after reading all headers we need to reset it, so next request
-        //operates on a clean header.
+        // after reading all headers we need to reset it, so next request
+        // operates on a clean header.
         header_remove();
 
         $headers = array_merge($nativeHeaders, $syResponse->headers->allPreserveCase());
@@ -238,14 +239,33 @@ protected function mapResponse(HttpResponse $reactResponse, SymfonyResponse $syR
             $headers['Set-Cookie'] = $cookies;
         }
 
-        $reactResponse->writeHead($syResponse->getStatusCode(), $headers);
+        if ($syResponse instanceof SymfonyStreamedResponse) {
+            $reactResponse->writeHead($syResponse->getStatusCode(), $headers);
+
+            // asynchronously get content
+            ob_start(function($buffer) use ($reactResponse) {
+                $reactResponse->write($buffer);
+                return '';
+            }, 4096);
+
+            $syResponse->sendContent();
 
-        $stdOut = '';
-        while ($buffer = @ob_get_clean()) {
-            $stdOut .= $buffer;
+            // flush remaining content
+            @ob_end_flush();
+            $reactResponse->end();
         }
+        else {
+            ob_start();
+            $content = $syResponse->getContent();
+            @ob_end_flush();
 
-        $reactResponse->end($stdOut . $content);
+            if (!isset($headers['Content-Length'])) {
+                $headers['Content-Length'] = strlen($content);
+            }
+
+            $reactResponse->writeHead($syResponse->getStatusCode(), $headers);
+            $reactResponse->end($content);
+        }
     }
 
     /**