Support for streamed output (#10)

Author: roxblnfk

composer.json

@@ -16,7 +16,7 @@
     "require": {
         "php": ">=7.4",
         "ext-json": "*",
-        "spiral/roadrunner-worker": "^2.0",
+        "spiral/roadrunner-worker": "^2.2.0",
         "psr/http-factory": "^1.0.1",
         "psr/http-message": "^1.0.1"
     },
@@ -30,15 +30,15 @@
         "phpstan/phpstan": "~0.12",
         "phpunit/phpunit": "~8.0",
         "jetbrains/phpstorm-attributes": "^1.0",
-        "vimeo/psalm": "^4.4",
+        "vimeo/psalm": "^4.22",
         "symfony/var-dumper": "^5.1"
     },
     "scripts": {
         "analyze": "psalm"
     },
     "extra": {
         "branch-alias": {
-            "dev-master": "2.1.x-dev"
+            "dev-master": "2.2.x-dev"
         }
     },
     "suggest": {

src/HttpWorker.php

@@ -11,8 +11,10 @@
 
 namespace Spiral\RoadRunner\Http;
 
+use Generator;
 use Spiral\RoadRunner\Payload;
 use Spiral\RoadRunner\WorkerInterface;
+use Stringable;
 
 /**
  * @psalm-import-type HeadersList from Request
@@ -83,12 +85,39 @@ public function waitRequest(): ?Request
      */
     public function respond(int $status, string $body, array $headers = []): void
     {
-        $headers = (string)\json_encode([
+        $head = (string)\json_encode([
             'status'  => $status,
             'headers' => $headers ?: (object)[],
         ], \JSON_THROW_ON_ERROR);
 
-        $this->worker->respond(new Payload($body, $headers));
+        $this->worker->respond(new Payload($body, $head));
+    }
+
+    /**
+     * Respond data using Streamed Output
+     *
+     * @param Generator<mixed, scalar|Stringable, mixed, Stringable|scalar|null> $body Body generator.
+     *        Each yielded value will be sent as a separated stream chunk.
+     *        Returned value will be sent as a last stream package.
+     */
+    public function respondStream(int $status, Generator $body, array $headers = []): void
+    {
+        $head = (string)\json_encode([
+            'status'  => $status,
+            'headers' => $headers ?: (object)[],
+        ], \JSON_THROW_ON_ERROR);
+
+        do {
+            if (!$body->valid()) {
+                $content = (string)$body->getReturn();
+                $this->worker->respond(new Payload($content, $head, true));
+                break;
+            }
+            $content = (string)$body->current();
+            $this->worker->respond(new Payload($content, $head, false));
+            $body->next();
+            $head = null;
+        } while (true);
     }
 
     /**
@@ -131,7 +160,7 @@ private function hydrateRequest(Request $request, array $context): void
     }
 
     /**
-     * @param array<mixed, mixed> $headers
+     * Remove all non-string and empty-string keys
      *
      * @return array<string, mixed>
      */
@@ -144,7 +173,7 @@ private function filterHeaders(array $headers): array
                 unset($headers[$key]);
             }
         }
-
+        /** @var array<string, mixed> $headers */
         return $headers;
     }
 }

src/PSR7Worker.php

@@ -11,13 +11,16 @@
 
 namespace Spiral\RoadRunner\Http;
 
+use Generator;
 use Psr\Http\Message\ResponseInterface;
 use Psr\Http\Message\ServerRequestFactoryInterface;
 use Psr\Http\Message\ServerRequestInterface;
 use Psr\Http\Message\StreamFactoryInterface;
+use Psr\Http\Message\StreamInterface;
 use Psr\Http\Message\UploadedFileFactoryInterface;
 use Psr\Http\Message\UploadedFileInterface;
 use Spiral\RoadRunner\WorkerInterface;
+use Stringable;
 
 /**
  * Manages PSR-7 request and response.
@@ -27,6 +30,8 @@
  */
 class PSR7Worker implements PSR7WorkerInterface
 {
+    public int $chunk_size = 8 * 1024;
+
     /**
      * @var HttpWorker
      */
@@ -108,13 +113,40 @@ public function waitRequest(): ?ServerRequestInterface
      */
     public function respond(ResponseInterface $response): void
     {
-        $this->httpWorker->respond(
+        $this->httpWorker->respondStream(
             $response->getStatusCode(),
-            (string)$response->getBody(),
+            $this->streamToGenerator($response->getBody()),
             $response->getHeaders()
         );
     }
 
+    /**
+     * @return Generator<mixed, scalar|Stringable, mixed, Stringable|scalar|null> Compatible
+     *         with {@see \Spiral\RoadRunner\Http\HttpWorker::respondStream()}.
+     */
+    private function streamToGenerator(StreamInterface $stream): Generator
+    {
+        $stream->rewind();
+        $size = $stream->getSize();
+        if ($size !== null && $size < $this->chunk_size) {
+            return (string)$stream;
+        }
+        $sum = 0;
+        while (!$stream->eof()) {
+            if ($size === null) {
+                $chunk = $stream->read($this->chunk_size);
+            } else {
+                $left = $size - $sum;
+                $chunk = $stream->read(\min($this->chunk_size, $left));
+                if ($left <= $this->chunk_size && \strlen($chunk) === $left) {
+                    return $chunk;
+                }
+            }
+            $sum += \strlen($chunk);
+            yield $chunk;
+        }
+    }
+
     /**
      * Returns altered copy of _SERVER variable. Sets ip-address,
      * request-time and other values.

src/Request.php

@@ -36,7 +36,6 @@ final class Request
     /**
      * @var string
      */
-
     public string $remoteAddr = '127.0.0.1';
 
     /**