Proto payloads between RR and PHP Worker

composer.json

@@ -42,10 +42,11 @@
         "psr/http-factory": "^1.0.1",
         "psr/http-message": "^1.0.1 || ^2.0",
         "spiral/roadrunner": "^2023.3",
-        "spiral/roadrunner-worker": "^3.1.0"
+        "spiral/roadrunner-worker": "^3.1",
+        "roadrunner-php/roadrunner-api-dto": "^1.4",
+        "symfony/polyfill-php83": "^1.29"
     },
     "require-dev": {
-        "buggregator/trap": "^1.0",
         "jetbrains/phpstorm-attributes": "^1.0",
         "nyholm/psr7": "^1.3",
         "phpunit/phpunit": "^10.0",
@@ -72,7 +73,8 @@
         "analyze": "psalm"
     },
     "suggest": {
-        "spiral/roadrunner-cli": "Provides RoadRunner installation and management CLI tools"
+        "spiral/roadrunner-cli": "Provides RoadRunner installation and management CLI tools",
+        "ext-protobuf": "Provides Protocol Buffers support"
     },
     "config": {
         "sort-packages": true

src/HttpWorker.php

@@ -5,6 +5,10 @@
 namespace Spiral\RoadRunner\Http;
 
 use Generator;
+use RoadRunner\HTTP\DTO\V1BETA1\FileUpload;
+use RoadRunner\HTTP\DTO\V1BETA1\HeaderValue;
+use RoadRunner\HTTP\DTO\V1BETA1\Request as RequestProto;
+use RoadRunner\HTTP\DTO\V1BETA1\Response;
 use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
 use Spiral\RoadRunner\Message\Command\StreamStop;
 use Spiral\RoadRunner\Payload;
@@ -34,6 +38,8 @@
  */
 class HttpWorker implements HttpWorkerInterface
 {
+    private static ?bool $isProto = null;
+
     public function __construct(
         private readonly WorkerInterface $worker,
     ) {
@@ -56,13 +62,21 @@
             return null;
         }
 
+        if ($this->isProtoPayload($payload)) {
+            $message = new RequestProto();
+            $message->mergeFromString($payload->header);
+
+            return $this->requestFromProto($payload->body, $message);
+        }
+
         /** @var RequestContext $context */
         $context = \json_decode($payload->header, true, 512, \JSON_THROW_ON_ERROR);
 
-        return $this->createRequest($payload->body, $context);
+        return $this->arrayToRequest($payload->body, $context);
     }
 
     /**
+     * @param array<array-key, array<array-key, string>> $headers
      * @throws \JsonException
      */
     public function respond(int $status, string|Generator $body = '', array $headers = [], bool $endOfStream = true): void
@@ -76,21 +90,14 @@
             return;
         }
 
-        $head = \json_encode([
-            'status'  => $status,
-            'headers' => $headers ?: (object)[],
-        ], \JSON_THROW_ON_ERROR);
-
-        $this->worker->respond(new Payload($body, $head, $endOfStream));
+        $this->worker->respond($this->createRespondPayload($status, $body, $headers, $endOfStream));
     }
 
+    /**
+     * @param array<array-key, array<array-key, string>> $headers
+     */
     private function respondStream(int $status, Generator $body, array $headers = [], bool $endOfStream = true): void
     {
-        $head = \json_encode([
-            'status'  => $status,
-            'headers' => $headers ?: (object)[],
-        ], \JSON_THROW_ON_ERROR);
-
         $worker = $this->worker instanceof StreamWorkerInterface
             ? $this->worker->withStreamMode()
             : $this->worker;
@@ -103,7 +110,7 @@
                     // We don't need to send an empty frame if the stream is not ended
                     return;
                 }
-                $worker->respond(new Payload($content, $head, $endOfStream));
+                $worker->respond($this->createRespondPayload($status, $content, $headers, $endOfStream));
                 break;
             }
 
@@ -118,8 +125,7 @@
             }
 
             // Send a chunk of data
-            $worker->respond(new Payload($content, $head, false));
-            $head = null;
+            $worker->respond($this->createRespondPayload($status, $content, $headers, false));
 
             try {
                 $body->next();
@@ -134,7 +140,7 @@
     /**
      * @param RequestContext $context
      */
-    private function createRequest(string $body, array $context): Request
+    private function arrayToRequest(string $body, array $context): Request
     {
         \parse_str($context['rawQuery'], $query);
         return new Request(
@@ -154,6 +160,46 @@
         );
     }
 
+    private function requestFromProto(string $body, RequestProto $message): Request
+    {
+        $headers = $this->headerValueToArray($message->getHeader());
+        $uploadedFiles = [];
+
+        /**
+         * @var FileUpload $uploads
+         */
+        foreach ($message->getUploads()?->getList() ?? [] as $uploads) {
+            $uploadedFiles[$uploads->getName()] = [
+                'name' => $uploads->getName(),
+                'mime' => $uploads->getMime(),
+                'size' => (int) $uploads->getSize(),
+                'error' => (int) $uploads->getError(),
+                'tmpName' => $uploads->getTempFilename(),
+            ];
+        }
+
+        \parse_str($message->getRawQuery(), $query);
+        /** @psalm-suppress ArgumentTypeCoercion, MixedArgumentTypeCoercion */
+        return new Request(
+            remoteAddr: $message->getRemoteAddr(),
+            protocol: $message->getProtocol(),
+            method: $message->getMethod(),
+            uri: $message->getUri(),
+            headers: $this->filterHeaders($headers),
+            cookies: \array_map(
+                static fn(array $values) => \implode(',', $values),
+                $this->headerValueToArray($message->getCookies()),
+            ),
+            uploads: $uploadedFiles,
+            attributes: [
+                Request::PARSED_BODY_ATTRIBUTE_NAME => $message->getParsed(),
+            ] + \iterator_to_array($message->getAttributes()),
+            query: $query,
+            body: $body,
+            parsed: $message->getParsed(),
+        );
+    }
+
     /**
      * Remove all non-string and empty-string keys
      *
@@ -164,7 +210,7 @@
     {
         foreach ($headers as $key => $_) {
             if (!\is_string($key) || $key === '') {
-                // ignore invalid header names or values (otherwise, the worker will be crashed)
+                // ignore invalid header names or values (otherwise, the worker might be crashed)
                 // @see: <https://git.io/JzjgJ>
                 unset($headers[$key]);
             }
@@ -173,4 +219,61 @@
         /** @var HeadersList $headers */
         return $headers;
     }
+
+    /**
+     * @param \Traversable<non-empty-string, HeaderValue> $message
+     */
+    private function headerValueToArray(\Traversable $message): array
+    {
+        $result = [];
+        /**
+         * @var non-empty-string $key
+         * @var HeaderValue $value
+         */
+        foreach ($message as $key => $value) {
+            $result[$key] = \iterator_to_array($value->getValue());
+        }
+
+        return $result;
+    }
+
+    /**
+     * @param array<array-key, array<array-key, string>> $headers
+     * @return array<non-empty-string, HeaderValue>
+     */
+    private function arrayToHeaderValue(array $headers = []): array
+    {
+        $result = [];
+        /**
+         * @var non-empty-string $key
+         * @var array<array-key, string> $value
+         */
+        foreach ($headers as $key => $value) {
+            $result[$key] = new HeaderValue(['value' => $value]);
+        }
+
+        return $result;
+    }
+
+    /**
+     * @param array<array-key, array<array-key, string>> $headers
+     */
+    private function createRespondPayload(int $status, string $body, array $headers = [], bool $eos = true): Payload
+    {
+        $head = static::$isProto
+            ? (new Response(['status' => $status, 'headers' => $this->arrayToHeaderValue($headers)]))
+                ->serializeToString()
+            : \json_encode(['status' => $status, 'headers' => $headers ?: (object)[]], \JSON_THROW_ON_ERROR);
+
+        return new Payload(body: $body, header: $head, eos: $eos);
+    }
+
+    private function isProtoPayload(Payload $payload): bool
+    {
+        if (static::$isProto === null) {
+            static::$isProto = !json_validate($payload->header);
+        }
+
+        return static::$isProto;
+    }
 }

src/HttpWorkerInterface.php

@@ -26,8 +26,9 @@ public function waitRequest(): ?Request;
      *        If the body is a generator, then each yielded value will be sent as a separated stream chunk.
      *        Returned value will be sent as a last stream package.
      *        Note: Stream response is supported by RoadRunner since version 2023.3
-     * @param HeadersList|array $headers An associative array of the message's headers. Each key MUST be a header name,
-     *                                   and each value MUST be an array of strings for that header.
+     * @param HeadersList|array<array-key, array<array-key, string>> $headers $headers An associative array of the
+     *        message's headers. Each key MUST be a header name, and each value MUST be an array of strings for
+     *        that header.
      */
     public function respond(int $status, string|Generator $body, array $headers = []): void;
 }