feat: Implement streaming response functionality

.github/workflows/test.yml

@@ -16,10 +16,13 @@ jobs:
         php-version: '8.1'
 
     - name: Setup Docker
-      run: docker compose up -d --build
+      run: docker compose up -d --build --wait
 
-    - name: Wait for Server to be ready
-      run: sleep 10
+    - name: Run Unit Tests
+      run: docker compose exec web vendor/bin/phpunit --configuration phpunit.xml --testsuite unit
 
-    - name: Run Tests
-      run: docker compose exec web vendor/bin/phpunit --configuration phpunit.xml
+    - name: Run FPM E2E Tests
+      run: docker compose exec web vendor/bin/phpunit --configuration phpunit.xml --testsuite fpm
+
+    - name: Run Swoole E2E Tests
+      run: docker compose exec web vendor/bin/phpunit --configuration phpunit.xml --testsuite swoole

Dockerfile.swoole

@@ -0,0 +1,20 @@
+FROM phpswoole/swoole:php8.3-alpine
+
+ARG TESTING=true
+ENV TESTING=$TESTING
+
+WORKDIR /usr/src/code
+
+COPY composer.* /usr/src/code/
+
+RUN composer install --ignore-platform-reqs --optimize-autoloader \
+    --no-plugins --no-scripts --prefer-dist \
+    `if [ "$TESTING" != "true" ]; then echo "--no-dev"; fi`
+
+COPY ./src /usr/src/code/src
+COPY ./tests /usr/src/code/tests
+COPY ./phpunit.xml /usr/src/code/phpunit.xml
+
+EXPOSE 80
+
+CMD ["php", "/usr/src/code/tests/e2e/swoole-server.php"]

docker-compose.yml

@@ -5,4 +5,24 @@ services:
             - "9020:80"
         volumes:
             - ./src:/usr/share/nginx/html/src
-            - ./tests:/usr/share/nginx/html/tests
+            - ./tests:/usr/share/nginx/html/tests
+            - ./phpunit.xml:/usr/share/nginx/html/phpunit.xml
+        healthcheck:
+            test: ["CMD-SHELL", "curl -sf http://127.0.0.1/ || exit 1"]
+            interval: 2s
+            timeout: 5s
+            retries: 15
+    swoole-web:
+        build:
+            context: .
+            dockerfile: Dockerfile.swoole
+        ports:
+            - "9021:80"
+        volumes:
+            - ./src:/usr/src/code/src
+            - ./tests:/usr/src/code/tests
+        healthcheck:
+            test: ["CMD-SHELL", "curl -sf http://127.0.0.1/ || exit 1"]
+            interval: 2s
+            timeout: 5s
+            retries: 15

phpunit.xml

@@ -9,9 +9,19 @@
          stopOnFailure="false"
         >
     <testsuites>
-        <testsuite name="Application Test Suite">
+        <testsuite name="unit">
+            <directory suffix="Test.php">./tests/</directory>
+            <exclude>./tests/e2e</exclude>
+        </testsuite>
+        <testsuite name="fpm">
+            <file>./tests/e2e/Client.php</file>
+            <directory suffix="Test.php">./tests/e2e/</directory>
+            <exclude>./tests/e2e/SwooleResponseTest.php</exclude>
+        </testsuite>
+        <testsuite name="swoole">
             <file>./tests/e2e/Client.php</file>
-            <directory>./tests/</directory>
+            <file>./tests/e2e/ResponseTest.php</file>
+            <file>./tests/e2e/SwooleResponseTest.php</file>
         </testsuite>
     </testsuites>
-</phpunit>
+</phpunit>

src/Http/Adapter/Swoole/Request.php

@@ -374,6 +374,18 @@ protected function generateInput(): array
      */
     protected function generateHeaders(): array
     {
-        return $this->swoole->header;
+        $headers = $this->swoole->header ?? [];
+
+        // Swoole parses the Cookie header into $swoole->cookie and removes it
+        // from the header array. Reconstruct it so callers can access it.
+        if (!isset($headers['cookie']) && !empty($this->swoole->cookie)) {
+            $pairs = [];
+            foreach ($this->swoole->cookie as $k => $v) {
+                $pairs[] = $k . '=' . $v;
+            }
+            $headers['cookie'] = implode('; ', $pairs);
+        }
+
+        return $headers;
     }
 }

src/Http/Adapter/Swoole/Response.php

@@ -3,6 +3,7 @@
 namespace Utopia\Http\Adapter\Swoole;
 
 use Swoole\Http\Response as SwooleResponse;
+use Swoole\Http\Server as SwooleServer;
 use Utopia\Http\Response as UtopiaResponse;
 
 class Response extends UtopiaResponse
@@ -14,6 +15,22 @@ class Response extends UtopiaResponse
      */
     protected SwooleResponse $swoole;
 
+    /**
+     * Swoole Server Object (needed for raw TCP streaming with Content-Length)
+     *
+     * @var SwooleServer|null
+     */
+    protected ?SwooleServer $server = null;
+
+    /**
+     * Whether to use detach() + raw TCP send for streaming.
+     * When true (and server is set): preserves Content-Length header so browsers show download progress.
+     * When false: uses Swoole's write() which forces chunked Transfer-Encoding.
+     *
+     * @var bool
+     */
+    protected bool $detach = true;
+
     /**
      * Response constructor.
      */
@@ -23,6 +40,37 @@ public function __construct(SwooleResponse $response)
         parent::__construct(\microtime(true));
     }
 
+    /**
+     * Set the Swoole server instance for raw TCP streaming.
+     *
+     * @param  SwooleServer  $server
+     * @return static
+     */
+    public function setServer(SwooleServer $server): static
+    {
+        $this->server = $server;
+
+        return $this;
+    }
+
+    /**
+     * Set whether to detach from Swoole's HTTP layer for streaming.
+     *
+     * When enabled (default): uses detach() + $server->send() to preserve
+     * Content-Length so browsers show download progress bars.
+     *
+     * When disabled: uses Swoole's write() which applies chunked Transfer-Encoding.
+     *
+     * @param  bool  $detach
+     * @return static
+     */
+    public function setDetach(bool $detach): static
+    {
+        $this->detach = $detach;
+
+        return $this;
+    }
+
     /**
      * Write
      *
@@ -45,6 +93,169 @@ public function end(?string $content = null): void
         $this->swoole->end($content);
     }
 
+    /**
+     * Stream response
+     *
+     * Uses detach() + $server->send() for raw TCP streaming that preserves
+     * Content-Length (so browsers show download progress). Falls back to
+     * the base class implementation if no server instance is available.
+     *
+     * @param  callable|\Generator  $source  Either a callable($offset, $length) or a Generator yielding string chunks
+     * @param  int  $totalSize  Total size of the content in bytes
+     * @return void
+     */
+    public function stream(callable|\Generator $source, int $totalSize): void
+    {
+        if ($this->sent) {
+            return;
+        }
+
+        $this->sent = true;
+
+        if ($this->disablePayload) {
+            $this->appendCookies();
+            $this->appendHeaders();
+            $this->swoole->end();
+            $this->disablePayload();
+
+            return;
+        }
+
+        // When detach is enabled and server is available, use raw TCP streaming
+        // to preserve Content-Length (browsers show download progress).
+        if ($this->detach && $this->server !== null) {
+            $this->streamDetached($source, $totalSize);
+
+            return;
+        }
+
+        // Non-detach path: use Swoole's write() (chunked Transfer-Encoding)
+        $this->addHeader('X-Debug-Speed', (string) (microtime(true) - $this->startTime), override: true);
+
+        if (!empty($this->contentType)) {
+            $this->addHeader('Content-Type', $this->contentType, override: true);
+        }
+
+        $this->appendCookies();
+        $this->appendHeaders();
+        $this->sendStatus($this->statusCode);
+
+        if ($source instanceof \Generator) {
+            foreach ($source as $chunk) {
+                if (!empty($chunk)) {
+                    $this->size += strlen($chunk);
+                    if ($this->swoole->write($chunk) === false) {
+                        break;
+                    }
+                }
+            }
+        } else {
+            $length = self::CHUNK_SIZE;
+            for ($offset = 0; $offset < $totalSize; $offset += $length) {
+                $chunk = $source($offset, min($length, $totalSize - $offset));
+                if (!empty($chunk)) {
+                    $this->size += strlen($chunk);
+                    if ($this->swoole->write($chunk) === false) {
+                        break;
+                    }
+                }
+            }
+        }
+
+        $this->swoole->end();
+        $this->disablePayload();
+    }
+
+    /**
+     * Stream using detach() + raw TCP send to preserve Content-Length.
+     *
+     * @param  callable|\Generator  $source
+     * @param  int  $totalSize
+     * @return void
+     */
+    protected function streamDetached(callable|\Generator $source, int $totalSize): void
+    {
+        $this->addHeader('Content-Length', (string) $totalSize, override: true);
+        $this->addHeader('Connection', 'close', override: true);
+        $this->addHeader('X-Debug-Speed', (string) (microtime(true) - $this->startTime), override: true);
+
+        if (!empty($this->contentType)) {
+            $this->addHeader('Content-Type', $this->contentType, override: true);
+        }
+
+        $statusReason = $this->getStatusCodeReason($this->statusCode);
+        $rawHeaders = "HTTP/1.1 {$this->statusCode} {$statusReason}\r\n";
+
+        foreach ($this->headers as $key => $value) {
+            if (\is_array($value)) {
+                foreach ($value as $v) {
+                    $rawHeaders .= "{$key}: {$v}\r\n";
+                }
+            } else {
+                $rawHeaders .= "{$key}: {$value}\r\n";
+            }
+        }
+
+        foreach ($this->cookies as $cookie) {
+            $cookieStr = \urlencode($cookie['name']) . '=' . \urlencode($cookie['value'] ?? '');
+            if (!empty($cookie['expire'])) {
+                $cookieStr .= '; Expires=' . \gmdate('D, d M Y H:i:s T', $cookie['expire']);
+            }
+            if (!empty($cookie['path'])) {
+                $cookieStr .= '; Path=' . $cookie['path'];
+            }
+            if (!empty($cookie['domain'])) {
+                $cookieStr .= '; Domain=' . $cookie['domain'];
+            }
+            if (!empty($cookie['secure'])) {
+                $cookieStr .= '; Secure';
+            }
+            if (!empty($cookie['httponly'])) {
+                $cookieStr .= '; HttpOnly';
+            }
+            if (!empty($cookie['samesite'])) {
+                $cookieStr .= '; SameSite=' . $cookie['samesite'];
+            }
+            $rawHeaders .= "Set-Cookie: {$cookieStr}\r\n";
+        }
+
+        $rawHeaders .= "\r\n";
+
+        $fd = $this->swoole->fd;
+        $this->swoole->detach();
+
+        if ($this->server->send($fd, $rawHeaders) === false) {
+            $this->disablePayload();
+
+            return;
+        }
+
+        if ($source instanceof \Generator) {
+            foreach ($source as $chunk) {
+                if (!empty($chunk)) {
+                    $this->size += strlen($chunk);
+                    if ($this->server->send($fd, $chunk) === false) {
+                        break;
+                    }
+                }
+            }
+        } else {
+            $length = self::CHUNK_SIZE;
+            for ($offset = 0; $offset < $totalSize; $offset += $length) {
+                $chunk = $source($offset, min($length, $totalSize - $offset));
+                if (!empty($chunk)) {
+                    $this->size += strlen($chunk);
+                    if ($this->server->send($fd, $chunk) === false) {
+                        break;
+                    }
+                }
+            }
+        }
+
+        $this->server->close($fd);
+        $this->disablePayload();
+    }
+
     /**
      * Get status code reason
      *

src/Http/Adapter/Swoole/Server.php

@@ -26,7 +26,9 @@ public function onRequest(callable $callback)
             Http::setResource('swooleRequest', fn () => $request);
             Http::setResource('swooleResponse', fn () => $response);
 
-            call_user_func($callback, new Request($request), new Response($response));
+            $utopiaResponse = new Response($response);
+            $utopiaResponse->setServer($this->server);
+            call_user_func($callback, new Request($request), $utopiaResponse);
         });
     }
 

src/Http/Http.php

@@ -885,6 +885,8 @@ public function run(Request $request, Response $response): static
      */
     private function runInternal(Request $request, Response $response): static
     {
+        $this->route = null;
+
         if ($this->compression) {
             $response->setAcceptEncoding($request->getHeader('accept-encoding', ''));
             $response->setCompressionMinSize($this->compressionMinSize);