Feat: support RR[>=2023.3] streamed responses

README.md

@@ -70,6 +70,7 @@ baldinof_road_runner:
     kernel_reboot:
       strategy: on_exception
       allowed_exceptions:
+        - Spiral\RoadRunner\Http\Exception\StreamStoppedException
         - Symfony\Component\HttpKernel\Exception\HttpExceptionInterface
         - Symfony\Component\Serializer\Exception\ExceptionInterface
         - App\Exception\YourDomainException
@@ -112,6 +113,11 @@ baldinof_road_runner:
       max_jobs_dispersion: 0.2
 ```
 
+## StreamedResponse
+
+Replace your `StreamedResponse`s with `Baldinof\RoadRunnerBundle\Http\Response\StreamedResponse`. The only difference is
+that the `$callback` should be a `\Generator`. If you don't, the Symfony's Streamed Response will be loaded completely to memory
+before it's sent! 
 
 ## Events
 

src/Helpers/StreamedJsonResponseHelper.php

@@ -0,0 +1,131 @@
+<?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
+{
+    private static ?\Closure $streamedJsonResponseParameterExtractor = null;
+
+    public static function toGenerator(StreamedJsonResponse $response): \Generator
+    {
+        $placeholder = "__symfony_json__";
+        [$data, $encodingOptions] = self::getStreamedJsonResponseParameterExtractor()($response);
+
+        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 ? ']' : '}';
+    }
+
+    private static function getStreamedJsonResponseParameterExtractor(): \Closure
+    {
+        return self::$streamedJsonResponseParameterExtractor ?? (self::$streamedJsonResponseParameterExtractor = \Closure::bind(static fn(StreamedJsonResponse $binaryFileResponse) => [
+            $binaryFileResponse->data,
+            $binaryFileResponse->encodingOptions,
+        ], null, StreamedJsonResponse::class));
+    }
+}

src/Http/Response/StreamedResponse.php

@@ -0,0 +1,90 @@
+<?php
+
+namespace Baldinof\RoadRunnerBundle\Http\Response;
+
+use Symfony\Component\HttpFoundation\Response;
+
+/**
+ * Basically a copy of Symfony's StreamedResponse,
+ * but the callback needs to be a Generator
+ */
+class StreamedResponse extends Response
+{
+    protected ?\Closure $callback = null;
+    protected bool $streamed = false;
+
+    private bool $headersSent = false;
+
+    public function __construct(\Generator $callback = null, int $status = 200, array $headers = [])
+    {
+        parent::__construct(null, $status, $headers);
+
+        if (null !== $callback) {
+            $this->setCallback($callback);
+        }
+    }
+
+    public function setCallback(\Generator $callback): static
+    {
+        $this->callback = $callback(...);
+
+        return $this;
+    }
+
+    public function getCallback(): ?\Generator
+    {
+        if (!isset($this->callback)) {
+            return null;
+        }
+
+        return ($this->callback)(...);
+    }
+
+    public function sendHeaders(int $statusCode = null): static
+    {
+        if ($this->headersSent) {
+            return $this;
+        }
+
+        if ($statusCode < 100 || $statusCode >= 200) {
+            $this->headersSent = true;
+        }
+
+        return parent::sendHeaders($statusCode);
+    }
+
+    public function sendContent(): static
+    {
+        if ($this->streamed) {
+            return $this;
+        }
+
+        $this->streamed = true;
+
+        if (!isset($this->callback)) {
+            throw new \LogicException('The Response callback must be set.');
+        }
+
+        foreach (($this->callback)() as $value) {
+            echo $value;
+        }
+
+        return $this;
+    }
+
+    public function setContent(?string $content): static
+    {
+        if (null !== $content) {
+            throw new \LogicException('The content cannot be set on a StreamedResponse instance.');
+        }
+
+        $this->streamed = true;
+
+        return $this;
+    }
+
+    public function getContent(): string|false
+    {
+        return false;
+    }
+}