[Swoole] Make StreamedResponse chunk size configurable

src/swoole/src/BridgeOptions.php

@@ -0,0 +1,47 @@
+<?php
+
+namespace Runtime\Swoole;
+
+/**
+ * Tunables for SymfonyHttpBridge::reflectSymfonyResponse(), grouped into one object so adding a
+ * new knob later never means adding another positional parameter to the bridge's signature.
+ *
+ * @author Piotr Kugla <piku235@gmail.com>
+ */
+class BridgeOptions
+{
+    private const DEFAULT_STREAMED_RESPONSE_CHUNK_SIZE = 4096;
+
+    /** @var int */
+    private $streamedResponseChunkSize;
+
+    public function __construct(int $streamedResponseChunkSize = self::DEFAULT_STREAMED_RESPONSE_CHUNK_SIZE)
+    {
+        $this->streamedResponseChunkSize = $streamedResponseChunkSize;
+    }
+
+    /**
+     * Builds options from a ServerFactory-style options array — the same shape ServerFactory
+     * itself accepts, so a single $options array can configure both.
+     */
+    public static function fromArray(array $options): self
+    {
+        return new self(
+            (int) ($options['streamed_response_chunk_size'] ?? self::DEFAULT_STREAMED_RESPONSE_CHUNK_SIZE)
+        );
+    }
+
+    /**
+     * Size, in bytes, at which a StreamedResponse's output buffer is flushed to the Swoole
+     * response. The default trades latency for fewer write() syscalls, which is fine for
+     * responses that fill it quickly (e.g. files) but starves a Server-Sent Events response,
+     * whose chunks are small and spaced out in time: nothing reaches the client until this many
+     * bytes accumulate or the response ends, which for a long-lived SSE connection reads as the
+     * connection hanging and can trip a reverse proxy's origin timeout. Use 1 for SSE (or any
+     * response where each write must reach the client immediately).
+     */
+    public function getStreamedResponseChunkSize(): int
+    {
+        return $this->streamedResponseChunkSize;
+    }
+}

src/swoole/src/LaravelRunner.php

@@ -19,11 +19,14 @@ class LaravelRunner implements RunnerInterface
     private $serverFactory;
     /** @var Kernel */
     private $application;
+    /** @var BridgeOptions */
+    private $bridgeOptions;
 
     public function __construct(ServerFactory $serverFactory, Kernel $application)
     {
         $this->serverFactory = $serverFactory;
         $this->application = $application;
+        $this->bridgeOptions = BridgeOptions::fromArray($serverFactory->getOptions());
     }
 
     public function run(): int
@@ -38,7 +41,7 @@ public function handle(Request $request, Response $response): void
         $sfRequest = LaravelRequest::createFromBase(SymfonyHttpBridge::convertSwooleRequest($request));
 
         $sfResponse = $this->application->handle($sfRequest);
-        SymfonyHttpBridge::reflectSymfonyResponse($sfResponse, $response);
+        SymfonyHttpBridge::reflectSymfonyResponse($sfResponse, $response, $this->bridgeOptions);
 
         $this->application->terminate($sfRequest, $sfResponse);
     }

src/swoole/src/ServerFactory.php

@@ -17,6 +17,7 @@ class ServerFactory
         'mode' => 2, // SWOOLE_PROCESS
         'sock_type' => 1, // SWOOLE_SOCK_TCP
         'settings' => [],
+        'streamed_response_chunk_size' => 4096,
     ];
 
     /** @var array */
@@ -33,6 +34,7 @@ public function __construct(array $options = [])
         $options['port'] = $options['port'] ?? $_SERVER['SWOOLE_PORT'] ?? $_ENV['SWOOLE_PORT'] ?? self::DEFAULT_OPTIONS['port'];
         $options['mode'] = $options['mode'] ?? $_SERVER['SWOOLE_MODE'] ?? $_ENV['SWOOLE_MODE'] ?? self::DEFAULT_OPTIONS['mode'];
         $options['sock_type'] = $options['sock_type'] ?? $_SERVER['SWOOLE_SOCK_TYPE'] ?? $_ENV['SWOOLE_SOCK_TYPE'] ?? self::DEFAULT_OPTIONS['sock_type'];
+        $options['streamed_response_chunk_size'] = $options['streamed_response_chunk_size'] ?? $_SERVER['SWOOLE_STREAMED_RESPONSE_CHUNK_SIZE'] ?? $_ENV['SWOOLE_STREAMED_RESPONSE_CHUNK_SIZE'] ?? self::DEFAULT_OPTIONS['streamed_response_chunk_size'];
 
         $this->options = array_replace_recursive(self::DEFAULT_OPTIONS, $options);
     }

src/swoole/src/SymfonyHttpBridge.php

@@ -35,8 +35,10 @@ public static function convertSwooleRequest(Request $request): SymfonyRequest
         return $sfRequest;
     }
 
-    public static function reflectSymfonyResponse(SymfonyResponse $sfResponse, Response $response): void
+    public static function reflectSymfonyResponse(SymfonyResponse $sfResponse, Response $response, ?BridgeOptions $options = null): void
     {
+        $options = $options ?? new BridgeOptions();
+
         foreach ($sfResponse->headers->all() as $name => $values) {
             foreach ((array) $values as $value) {
                 $response->header((string) $name, $value);
@@ -49,10 +51,17 @@ public static function reflectSymfonyResponse(SymfonyResponse $sfResponse, Respo
             case $sfResponse instanceof BinaryFileResponse && $sfResponse->headers->has('Content-Range'):
             case $sfResponse instanceof StreamedResponse:
                 ob_start(function ($buffer) use ($response) {
-                    $response->write($buffer);
+                    // At a small chunk size (e.g. 1, for SSE) every echo already flushes on its
+                    // own, so the buffer is typically empty by the time ob_end_clean() makes its
+                    // own final call to this handler. Swoole\Http\Response::write() itself checks
+                    // length === 0 and rejects it with a warning ("the data sent must not be
+                    // empty"), so this mirrors that same check to skip the call instead.
+                    if (strlen($buffer) > 0) {
+                        $response->write($buffer);
+                    }
 
                     return '';
-                }, 4096);
+                }, $options->getStreamedResponseChunkSize());
                 $sfResponse->sendContent();
                 ob_end_clean();
                 $response->end();

src/swoole/src/SymfonyRunner.php

@@ -19,11 +19,14 @@ class SymfonyRunner implements RunnerInterface
     private $serverFactory;
     /** @var HttpKernelInterface */
     private $application;
+    /** @var BridgeOptions */
+    private $bridgeOptions;
 
     public function __construct(ServerFactory $serverFactory, HttpKernelInterface $application)
     {
         $this->serverFactory = $serverFactory;
         $this->application = $application;
+        $this->bridgeOptions = BridgeOptions::fromArray($serverFactory->getOptions());
     }
 
     public function run(): int
@@ -38,7 +41,7 @@ public function handle(Request $request, Response $response): void
         $sfRequest = SymfonyHttpBridge::convertSwooleRequest($request);
 
         $sfResponse = $this->application->handle($sfRequest);
-        SymfonyHttpBridge::reflectSymfonyResponse($sfResponse, $response);
+        SymfonyHttpBridge::reflectSymfonyResponse($sfResponse, $response, $this->bridgeOptions);
 
         if ($this->application instanceof TerminableInterface) {
             $this->application->terminate($sfRequest, $sfResponse);

src/swoole/tests/Unit/BridgeOptionsTest.php

@@ -0,0 +1,39 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Runtime\Swoole\Tests\Unit;
+
+use PHPUnit\Framework\TestCase;
+use Runtime\Swoole\BridgeOptions;
+
+class BridgeOptionsTest extends TestCase
+{
+    public function testDefaultStreamedResponseChunkSize(): void
+    {
+        $options = new BridgeOptions();
+
+        self::assertSame(4096, $options->getStreamedResponseChunkSize());
+    }
+
+    public function testExplicitStreamedResponseChunkSize(): void
+    {
+        $options = new BridgeOptions(1);
+
+        self::assertSame(1, $options->getStreamedResponseChunkSize());
+    }
+
+    public function testFromArrayWithGivenChunkSize(): void
+    {
+        $options = BridgeOptions::fromArray(['streamed_response_chunk_size' => 1]);
+
+        self::assertSame(1, $options->getStreamedResponseChunkSize());
+    }
+
+    public function testFromArrayFallsBackToDefaultChunkSize(): void
+    {
+        $options = BridgeOptions::fromArray([]);
+
+        self::assertSame(4096, $options->getStreamedResponseChunkSize());
+    }
+}

src/swoole/tests/Unit/LaravelRunnerTest.php

@@ -12,6 +12,7 @@
 use Swoole\Http\Response;
 use Swoole\Http\Server;
 use Symfony\Component\HttpFoundation\Response as SymfonyResponse;
+use Symfony\Component\HttpFoundation\StreamedResponse;
 
 class LaravelRunnerTest extends TestCase
 {
@@ -46,4 +47,36 @@ public function testHandle(): void
         $runner = new LaravelRunner($factory, $application);
         $runner->handle($request, $response);
     }
+
+    public function testHandleHonoursTheServerFactorysStreamedResponseChunkSize(): void
+    {
+        // A chunk size of 1 forces a write() on every echo, regardless of its length — the setting
+        // this test exists to prove reaches SymfonyHttpBridge from ServerFactory's options.
+        $sfResponse = new StreamedResponse(static function () {
+            echo 'Foo';
+            echo 'Bar';
+        });
+
+        $application = $this->createMock(Kernel::class);
+        $application->expects(self::once())->method('handle')->willReturn($sfResponse);
+
+        $response = $this->createMock(Response::class);
+        $expectedWrites = ['Foo', 'Bar'];
+        $callCount = 0;
+        $response->expects(self::exactly(2))->method('write')
+            ->willReturnCallback(function ($string) use ($expectedWrites, &$callCount) {
+                $this->assertEquals($expectedWrites[$callCount], $string);
+                ++$callCount;
+
+                return true;
+            });
+        $response->expects(self::once())->method('end');
+
+        $request = $this->createMock(Request::class);
+        $factory = $this->createMock(ServerFactory::class);
+        $factory->method('getOptions')->willReturn(['streamed_response_chunk_size' => 1]);
+
+        $runner = new LaravelRunner($factory, $application);
+        $runner->handle($request, $response);
+    }
 }

src/swoole/tests/Unit/SymfonyHttpBridgeTest.php

@@ -107,10 +107,9 @@ public function testThatSymfonyStreamedResponseIsReflected(): void
         $expectedWrites = [
             "Foo\n",
             "Bar\n",
-            '',
         ];
         $callCount = 0;
-        $response->expects(self::exactly(3))->method('write')
+        $response->expects(self::exactly(2))->method('write')
             ->willReturnCallback(function ($string) use ($expectedWrites, &$callCount) {
                 $this->assertEquals($expectedWrites[$callCount], $string);
                 ++$callCount;

src/swoole/tests/Unit/SymfonyRunnerTest.php

@@ -11,6 +11,7 @@
 use Swoole\Http\Response;
 use Swoole\Http\Server;
 use Symfony\Component\HttpFoundation\Response as SymfonyResponse;
+use Symfony\Component\HttpFoundation\StreamedResponse;
 use Symfony\Component\HttpKernel\HttpKernelInterface;
 
 class SymfonyRunnerTest extends TestCase
@@ -46,4 +47,36 @@ public function testHandle(): void
         $runner = new SymfonyRunner($factory, $application);
         $runner->handle($request, $response);
     }
+
+    public function testHandleHonoursTheServerFactorysStreamedResponseChunkSize(): void
+    {
+        // A chunk size of 1 forces a write() on every echo, regardless of its length — the setting
+        // this test exists to prove reaches SymfonyHttpBridge from ServerFactory's options.
+        $sfResponse = new StreamedResponse(static function () {
+            echo 'Foo';
+            echo 'Bar';
+        });
+
+        $application = $this->createMock(HttpKernelInterface::class);
+        $application->expects(self::once())->method('handle')->willReturn($sfResponse);
+
+        $response = $this->createMock(Response::class);
+        $expectedWrites = ['Foo', 'Bar'];
+        $callCount = 0;
+        $response->expects(self::exactly(2))->method('write')
+            ->willReturnCallback(function ($string) use ($expectedWrites, &$callCount) {
+                $this->assertEquals($expectedWrites[$callCount], $string);
+                ++$callCount;
+
+                return true;
+            });
+        $response->expects(self::once())->method('end');
+
+        $request = $this->createMock(Request::class);
+        $factory = $this->createMock(ServerFactory::class);
+        $factory->method('getOptions')->willReturn(['streamed_response_chunk_size' => 1]);
+
+        $runner = new SymfonyRunner($factory, $application);
+        $runner->handle($request, $response);
+    }
 }