Add proto to request converter; add the roadrunner-php/roadrunner-api-dto package

Author: roxblnfk

composer.json

@@ -42,13 +42,14 @@
         "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.3.0"
     },
     "require-dev": {
         "buggregator/trap": "^1.0",
         "jetbrains/phpstorm-attributes": "^1.0",
         "nyholm/psr7": "^1.3",
         "phpunit/phpunit": "^10.0",
+        "roadrunner-php/roadrunner-api-dto": "^1.4",
         "symfony/process": "^6.2",
         "symfony/var-dumper": "^6.3",
         "vimeo/psalm": "^5.9"
@@ -73,7 +74,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 Spiral\RoadRunner\Encoding;
 use Spiral\RoadRunner\Http\Exception\StreamStoppedException;
 use Spiral\RoadRunner\Message\Command\StreamStop;
 use Spiral\RoadRunner\Payload;
@@ -56,10 +60,17 @@ public function waitRequest(): ?Request
             return null;
         }
 
+        if ($payload->encoding === Encoding::Protobuf) {
+            $message = new RequestProto();
+            $message->mergeFromString($payload->body);
+
+            return $this->requestFromProto($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);
     }
 
     /**
@@ -134,7 +145,7 @@ private function respondStream(int $status, Generator $body, array $headers = []
     /**
      * @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 +165,47 @@ private function createRequest(string $body, array $context): Request
         );
     }
 
+    private function requestFromProto(RequestProto $message): Request
+    {
+        $headers = $this->headerValueToArray($message->getHeader());
+        $uploadedFiles = [];
+
+        /**
+         * @var non-empty-string $name
+         * @var FileUpload $uploads
+         */
+        foreach ($message->getUploads() as $name => $uploads) {
+            $uploadedFiles[$name] = [
+                'name' => $uploads->getName(),
+                'mime' => $uploads->getMime(),
+                'size' => $uploads->getSize(),
+                'error' => $uploads->getError(),
+                'tmpName' => $uploads->getTempFilename(),
+            ];
+        }
+
+        \parse_str($message->getRawQuery(), $query);
+        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,
+            // todo rawBody?
+            body: $message->getBody(),
+            parsed: $message->getParsed(),
+        );
+    }
+
     /**
      * Remove all non-string and empty-string keys
      *
@@ -164,7 +216,7 @@ private function filterHeaders(array $headers): array
     {
         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 +225,22 @@ private function filterHeaders(array $headers): array
         /** @var HeadersList $headers */
         return $headers;
     }
+
+    /**
+     * @param \Traversable<non-empty-string, HeaderValue> $message
+     * @return HeadersList
+     */
+    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;
+    }
 }