Feat: support RR[>=2023.3] streamed responses

README.md

@@ -112,6 +112,9 @@ baldinof_road_runner:
       max_jobs_dispersion: 0.2
 ```
 
+## StreamedResponse
+
+`$callback` should now return `\Generator` to be really streamed. Replace all `echo` (etc) with `yield`. Fallback simply loads all content to memory!
 
 ## Events
 

src/Helpers/StreamedJsonResponseHelper.php

@@ -0,0 +1,124 @@
+<?php
+
+namespace Baldinof\RoadRunnerBundle\Helpers;
+
+use Symfony\Component\HttpFoundation\StreamedJsonResponse;
+
+// Basically copy of Symfony\Component\HttpFoundation\StreamedJsonResponse
+// but adds `yield`ing, instead of `echo`s
+class StreamedJsonResponseHelper
+{
+    public static function toGenerator(StreamedJsonResponse $response): \Generator
+    {
+        $ref = new \ReflectionClass($response);
+
+        $encodingOptions = $ref->getProperty("encodingOptions")->getValue($response);
+        $data = $ref->getProperty("data")->getValue($response);
+        $placeholder = $ref->getConstant("PLACEHOLDER");
+
+        return self::stream($data, $encodingOptions, $placeholder);
+    }
+
+    private static function stream(iterable $data, int $encodingOptions, string $placeholder): \Generator
+    {
+        $jsonEncodingOptions = \JSON_THROW_ON_ERROR | $encodingOptions;
+        $keyEncodingOptions = $jsonEncodingOptions & ~\JSON_NUMERIC_CHECK;
+
+        return self::streamData($data, $jsonEncodingOptions, $keyEncodingOptions, $placeholder);
+    }
+
+    private static function streamData(mixed $data, int $jsonEncodingOptions, int $keyEncodingOptions, string $placeholder): \Generator
+    {
+        if (\is_array($data)) {
+            foreach (self::streamArray($data, $jsonEncodingOptions, $keyEncodingOptions, $placeholder) as $item) {
+                yield $item;
+            }
+
+            return;
+        }
+
+        if (is_iterable($data) && !$data instanceof \JsonSerializable) {
+            foreach (self::streamIterable($data, $jsonEncodingOptions, $keyEncodingOptions, $placeholder) as $item) {
+                yield $item;
+            }
+
+            return;
+        }
+
+        yield json_encode($data, $jsonEncodingOptions);
+    }
+
+    private static function streamArray(array $data, int $jsonEncodingOptions, int $keyEncodingOptions, string $placeholder): \Generator
+    {
+        $generators = [];
+
+        array_walk_recursive($data, function (&$item, $key) use (&$generators, $placeholder) {
+            if ($placeholder === $key) {
+                // if the placeholder is already in the structure it should be replaced with a new one that explode
+                // works like expected for the structure
+                $generators[] = $key;
+            }
+
+            // generators should be used but for better DX all kind of Traversable and objects are supported
+            if (\is_object($item)) {
+                $generators[] = $item;
+                $item = $placeholder;
+            } elseif ($placeholder === $item) {
+                // if the placeholder is already in the structure it should be replaced with a new one that explode
+                // works like expected for the structure
+                $generators[] = $item;
+            }
+        });
+
+        $jsonParts = explode('"' . $placeholder . '"', json_encode($data, $jsonEncodingOptions));
+
+        foreach ($generators as $index => $generator) {
+            // send first and between parts of the structure
+            yield $jsonParts[$index];
+
+            foreach (self::streamData($generator, $jsonEncodingOptions, $keyEncodingOptions, $placeholder) as $child) {
+                yield $child;
+            }
+        }
+
+        // send last part of the structure
+        yield $jsonParts[array_key_last($jsonParts)];
+    }
+
+    private static function streamIterable(iterable $iterable, int $jsonEncodingOptions, int $keyEncodingOptions, string $placeholder): \Generator
+    {
+        $isFirstItem = true;
+        $startTag = '[';
+
+        foreach ($iterable as $key => $item) {
+            if ($isFirstItem) {
+                $isFirstItem = false;
+                // depending on the first elements key the generator is detected as a list or map
+                // we can not check for a whole list or map because that would hurt the performance
+                // of the streamed response which is the main goal of this response class
+                if (0 !== $key) {
+                    $startTag = '{';
+                }
+
+                yield $startTag;
+            } else {
+                // if not first element of the generic, a separator is required between the elements
+                yield ',';
+            }
+
+            if ('{' === $startTag) {
+                yield json_encode((string)$key, $keyEncodingOptions) . ':';
+            }
+
+            foreach (self::streamData($item, $jsonEncodingOptions, $keyEncodingOptions, $placeholder) as $child) {
+                yield $child;
+            }
+        }
+
+        if ($isFirstItem) { // indicates that the generator was empty
+            yield '[';
+        }
+
+        yield '[' === $startTag ? ']' : '}';
+    }
+}

src/RoadRunnerBridge/HttpFoundationWorker.php

@@ -4,13 +4,18 @@
 
 namespace Baldinof\RoadRunnerBundle\RoadRunnerBridge;
 
+use Baldinof\RoadRunnerBundle\Helpers\StreamedJsonResponseHelper;
+use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
 use Spiral\RoadRunner\Http\HttpWorkerInterface;
 use Spiral\RoadRunner\Http\Request as RoadRunnerRequest;
 use Spiral\RoadRunner\WorkerInterface;
 use Symfony\Component\HttpFoundation\BinaryFileResponse;
 use Symfony\Component\HttpFoundation\File\UploadedFile;
+use Symfony\Component\HttpFoundation\Request;
 use Symfony\Component\HttpFoundation\Request as SymfonyRequest;
+use Symfony\Component\HttpFoundation\RequestStack;
 use Symfony\Component\HttpFoundation\Response as SymfonyResponse;
+use Symfony\Component\HttpFoundation\StreamedJsonResponse;
 use Symfony\Component\HttpFoundation\StreamedResponse;
 
 final class HttpFoundationWorker implements HttpFoundationWorkerInterface
@@ -35,32 +40,18 @@ public function waitRequest(): ?SymfonyRequest
         return $this->toSymfonyRequest($rrRequest);
     }
 
-    public function respond(SymfonyResponse $symfonyResponse): void
+    public function respond(SymfonyResponse $response): void
     {
-        if ($symfonyResponse instanceof BinaryFileResponse && !$symfonyResponse->headers->has('Content-Range')) {
-            $content = file_get_contents($symfonyResponse->getFile()->getPathname());
-            if ($content === false) {
-                throw new \RuntimeException(sprintf("Cannot read file '%s'", $symfonyResponse->getFile()->getPathname())); // TODO: custom error
-            }
-        } else {
-            if ($symfonyResponse instanceof StreamedResponse || $symfonyResponse instanceof BinaryFileResponse) {
-                $content = '';
-                ob_start(function ($buffer) use (&$content) {
-                    $content .= $buffer;
-
-                    return '';
-                });
-
-                $symfonyResponse->sendContent();
-                ob_end_clean();
-            } else {
-                $content = (string) $symfonyResponse->getContent();
-            }
-        }
+        $content = match (true) {
+            $response instanceof StreamedJsonResponse => $this->createStreamedJsonResponseGenerator($response),
+            $response instanceof StreamedResponse => $this->createStreamedResponseGenerator($response),
+            $response instanceof BinaryFileResponse => $this->createFileStreamGenerator($response),
+            default => $this->createDefaultContentGetter($response),
+        };
 
-        $headers = $this->stringifyHeaders($symfonyResponse->headers->all());
+        $headers = $this->stringifyHeaders($response->headers->all());
 
-        $this->httpWorker->respond($symfonyResponse->getStatusCode(), $content, $headers);
+        $this->httpWorker->respond($response->getStatusCode(), $content(), $headers);
     }
 
     public function getWorker(): WorkerInterface
@@ -112,7 +103,7 @@ private function configureServer(RoadRunnerRequest $request): array
         $server['REQUEST_URI'] = $components['path'] ?? '';
         if (isset($components['query']) && $components['query'] !== '') {
             $server['QUERY_STRING'] = $components['query'];
-            $server['REQUEST_URI'] .= '?'.$components['query'];
+            $server['REQUEST_URI'] .= '?' . $components['query'];
         }
 
         if (isset($components['scheme']) && $components['scheme'] === 'https') {
@@ -131,7 +122,7 @@ private function configureServer(RoadRunnerRequest $request): array
             if (\in_array($key, ['CONTENT_TYPE', 'CONTENT_LENGTH'])) {
                 $server[$key] = implode(', ', $value);
             } else {
-                $server['HTTP_'.$key] = implode(', ', $value);
+                $server['HTTP_' . $key] = implode(', ', $value);
             }
         }
 
@@ -188,7 +179,141 @@ private function timeFloat(): float
     private function stringifyHeaders(array $headers): array
     {
         return array_map(static function ($headerValues) {
-            return array_map(static fn ($val) => (string) $val, (array) $headerValues);
+            return array_map(static fn($val) => (string)$val, (array)$headerValues);
         }, $headers);
     }
+
+    /**
+     * Basically a copy of BinaryFileResponse->sendContent()
+     * @param BinaryFileResponse $response
+     * @return \Closure
+     */
+    private function createFileStreamGenerator(BinaryFileResponse $response): \Closure
+    {
+        return static function () use ($response) {
+            $ref = new \ReflectionClass($response);
+            $maxlen = $ref->getProperty("maxlen")->getValue($response);
+            $offset = $ref->getProperty("offset")->getValue($response);
+            $chunkSize = $ref->getProperty("chunkSize")->getValue($response);
+            $deleteFileAfterSend = $ref->getProperty("deleteFileAfterSend")->getValue($response);
+
+            try {
+                if (!$response->isSuccessful()) {
+                    return;
+                }
+
+                $file = fopen($response->getFile()->getPathname(), "r");
+
+                ignore_user_abort(true);
+
+                if ($maxlen === 0) {
+                    return;
+                }
+
+                if ($offset !== 0) {
+                    fseek($file, $offset);
+                }
+
+                $length = $maxlen;
+                while ($length && !feof($file)) {
+                    $read = $length > $chunkSize || 0 > $length ? $chunkSize : $length;
+
+                    if (false === $data = fread($file, $read)) {
+                        break;
+                    }
+
+                    while ('' !== $data) {
+                        try {
+                            yield $data;
+                        } catch (StreamStoppedException) {
+                            break 2;
+                        }
+
+                        if (0 < $length) {
+                            $length -= $read;
+                        }
+                        $data = substr($data, $read);
+                    }
+                }
+
+                fclose($file);
+            } finally {
+                if ($deleteFileAfterSend && is_file($response->getFile()->getPathname())) {
+                    unlink($response->getFile()->getPathname());
+                }
+            }
+        };
+    }
+
+    /**
+     * @param SymfonyResponse $response
+     * @return \Closure
+     */
+    private function createDefaultContentGetter(SymfonyResponse $response): \Closure
+    {
+        return static function () use ($response) {
+            ob_start();
+            $response->sendContent();
+            return ob_get_clean();
+        };
+    }
+
+    /**
+     * StreamedResponse callback can now use `yield` to be really streamed
+     * @param StreamedResponse $response
+     * @return \Closure
+     */
+    private function createStreamedResponseGenerator(StreamedResponse $response): \Closure
+    {
+        return function () use ($response): \Generator {
+            $kernelCallback = $response->getCallback();
+
+            $kernelCallbackRef = new \ReflectionFunction($kernelCallback);
+            $closureVars = $kernelCallbackRef->getClosureUsedVariables();
+
+            $ref = new \ReflectionFunction($closureVars["callback"]);
+            if ($ref->isGenerator()) {
+                $request = $closureVars["request"];
+                assert($request instanceof Request);
+
+                $requestStack = $closureVars["requestStack"];
+                assert($requestStack instanceof RequestStack);
+
+                try {
+                    $requestStack->push($request);
+
+                    foreach ($closureVars["callback"]() as $output) {
+                        try {
+                            yield $output;
+                        } catch (StreamStoppedException) {
+                            break;
+                        }
+                    }
+                } finally {
+                    $requestStack->pop();
+                }
+
+                return;
+            }
+
+            yield $this->createDefaultContentGetter($response)();
+        };
+    }
+
+    /**
+     * @param StreamedJsonResponse $response
+     * @return \Closure
+     */
+    private function createStreamedJsonResponseGenerator(StreamedJsonResponse $response): \Closure
+    {
+        return static function () use ($response): \Generator {
+            foreach (StreamedJsonResponseHelper::toGenerator($response) as $item) {
+                try {
+                    yield $item;
+                } catch (StreamStoppedException) {
+                    break;
+                }
+            }
+        };
+    }
 }