Use StreamWorkerInterface::withStreamMode() for stream output if worker supports it

Author: roxblnfk

composer.json

@@ -42,13 +42,15 @@
         "psr/http-factory": "^1.0.1",
         "psr/http-message": "^1.0.1 || ^2.0",
         "spiral/roadrunner": "^2023.1",
-        "spiral/roadrunner-worker": "^3.0"
+        "spiral/roadrunner-worker": "dev-ping-pong-stream as 3.1.0"
     },
     "require-dev": {
+        "buggregator/trap": "^1.0",
         "jetbrains/phpstorm-attributes": "^1.0",
         "nyholm/psr7": "^1.3",
         "phpunit/phpunit": "^10.0",
         "symfony/process": "^6.2",
+        "symfony/var-dumper": "^6.3",
         "vimeo/psalm": "^5.9"
     },
     "autoload": {

src/HttpWorker.php

@@ -8,6 +8,7 @@
 use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
 use Spiral\RoadRunner\Message\Command\StreamStop;
 use Spiral\RoadRunner\Payload;
+use Spiral\RoadRunner\StreamWorkerInterface;
 use Spiral\RoadRunner\WorkerInterface;
 
 /**
@@ -86,18 +87,22 @@ private function respondStream(int $status, Generator $body, array $headers = []
             'headers' => $headers ?: (object)[],
         ], \JSON_THROW_ON_ERROR);
 
+        $worker = $this->worker instanceof StreamWorkerInterface
+            ? $this->worker->withStreamMode()
+            : $this->worker;
+
         do {
             if (!$body->valid()) {
                 $content = (string)$body->getReturn();
-                $this->worker->respond(new Payload($content, $head, true));
+                $worker->respond(new Payload($content, $head, true));
                 break;
             }
             $content = (string)$body->current();
-            if ($this->worker->getPayload(StreamStop::class) !== null) {
+            if ($worker->getPayload(StreamStop::class) !== null) {
                 $body->throw(new StreamStoppedException());
                 return;
             }
-            $this->worker->respond(new Payload($content, $head, false));
+            $worker->respond(new Payload($content, $head, false));
             $body->next();
             $head = null;
         } while (true);