Send response headers from server

composer.json

@@ -26,7 +26,7 @@
         "amphp/dns": "^2.4",
         "amphp/http": "^2.1",
         "amphp/http-client": "^5.3",
-        "amphp/http-server": "^3.4",
+        "amphp/http-server": "^3.4.7",
         "amphp/pipeline": "^1.2",
         "amphp/socket": "^2.3",
         "amphp/sync": "^2.3",

packages/server/composer.json

@@ -20,7 +20,7 @@
     "require": {
         "php": "^8.4",
         "amphp/http": "^2.1",
-        "amphp/http-server": "^3.4.5",
+        "amphp/http-server": "^3.4.7",
         "amphp/socket": "^2.3",
         "amphp/sync": "^2.3",
         "psr/log": "^3.0",

packages/server/src/Server/BidirectionalStreamChannel.php

@@ -4,6 +4,7 @@
 
 namespace Thesis\Grpc\Server;
 
+use Thesis\Grpc\Metadata;
 use Thesis\Grpc\ServerStream;
 
 /**
@@ -14,12 +15,20 @@
  */
 final readonly class BidirectionalStreamChannel implements \IteratorAggregate
 {
+    /** Can be modified before the first message is sent. */
+    public Metadata $responseHeaders;
+
+    public Metadata $trailers;
+
     /**
      * @param ServerStream<TRequest, TResponse> $stream
      */
     public function __construct(
         private ServerStream $stream,
-    ) {}
+    ) {
+        $this->responseHeaders = $stream->responseHeaders;
+        $this->trailers = $stream->trailers;
+    }
 
     /**
      * @no-named-arguments

packages/server/src/Server/ClientStreamChannel.php

@@ -4,6 +4,7 @@
 
 namespace Thesis\Grpc\Server;
 
+use Thesis\Grpc\Metadata;
 use Thesis\Grpc\ServerStream;
 
 /**
@@ -14,12 +15,20 @@
  */
 final readonly class ClientStreamChannel implements \IteratorAggregate
 {
+    /** Can be modified before the first message is sent. */
+    public Metadata $responseHeaders;
+
+    public Metadata $trailers;
+
     /**
      * @param ServerStream<TRequest, TResponse> $stream
      */
     public function __construct(
         private ServerStream $stream,
-    ) {}
+    ) {
+        $this->responseHeaders = $stream->responseHeaders;
+        $this->trailers = $stream->trailers;
+    }
 
     /**
      * @return TRequest

packages/server/src/Server/DecoratedStream.php

@@ -15,6 +15,8 @@
  */
 abstract class DecoratedStream implements ServerStream
 {
+    public Metadata $responseHeaders { get => $this->stream->responseHeaders; }
+
     public Metadata $trailers { get => $this->stream->trailers; }
 
     public Metadata $headers { get => $this->stream->headers; }

packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php

@@ -19,17 +19,21 @@
  */
 final class ConcurrentServerStream implements ServerStream
 {
+    public private(set) Metadata $responseHeaders { get => $this->responseHeaders ??= new Metadata(); }
+
     public private(set) Metadata $trailers { get => $this->trailers ??= new Metadata(); }
 
     /**
      * @param Pipeline\ConcurrentIterator<In> $recv
      * @param Pipeline\Queue<Out> $send
+     * @param DeferredFuture<Metadata> $responseHeadersFuture
      * @param DeferredFuture<array<non-empty-string, list<string>>> $trailersFuture
      */
     public function __construct(
         public readonly Metadata $headers,
         private readonly Pipeline\ConcurrentIterator $recv,
         private readonly Pipeline\Queue $send,
+        private readonly DeferredFuture $responseHeadersFuture,
         private readonly DeferredFuture $trailersFuture,
     ) {}
 
@@ -40,6 +44,8 @@ public function send(object $message): void
             throw new ServerStreamIsClosed();
         }
 
+        $this->sendHeaders();
+
         try {
             $this->send->push($message);
         } catch (Pipeline\DisposedException $e) {
@@ -64,6 +70,7 @@ public function close(): void
             return;
         }
 
+        $this->sendHeaders();
         $this->trailersFuture->complete(Http2\encodeMetadata($this->trailers));
         $this->send->complete();
     }
@@ -73,4 +80,11 @@ public function getIterator(): \Traversable
     {
         return $this->recv->getIterator();
     }
+
+    private function sendHeaders(): void
+    {
+        if (!$this->responseHeadersFuture->isComplete()) {
+            $this->responseHeadersFuture->complete($this->responseHeaders);
+        }
+    }
 }

packages/server/src/Server/Internal/Http2/ServerRequestHandler.php

@@ -7,6 +7,7 @@
 use Amp\Cancellation;
 use Amp\CompositeCancellation;
 use Amp\DeferredCancellation;
+use Amp\DeferredFuture;
 use Amp\Future;
 use Amp\Http\HttpStatus;
 use Amp\Http\Server\Request;
@@ -140,7 +141,10 @@ public function handleRequest(Request $request): Response
             $this->maxReceiveMessageSize,
         );
 
-        $response = new Response(status: HttpStatus::OK, headers: Http2\encodeMetadata($headers));
+        $response = new Response(status: HttpStatus::OK);
+
+        /** @var DeferredFuture<Metadata> $responseHeaders */
+        $responseHeaders = new DeferredFuture();
 
         $cancellation = new DeferredCancellation();
 
@@ -160,10 +164,11 @@ public function handleRequest(Request $request): Response
             $md,
             $request,
             $response,
+            $responseHeaders,
             $streamCancellation,
         );
 
-        $info = new StreamInfo($rpc->handle->method, $rpc->type);
+        $info = new StreamInfo($rpc->handle->method, $rpc->type, $stream->responseHeaders, $stream->trailers);
         $rpcHandler = $rpc->handler;
 
         $terminal = match (true) {
@@ -232,6 +237,13 @@ public function __construct(
             ) {}
         };
 
+        // Like grpc-go, the response headers are sent with the first message, so the handler can still set them.
+        // User metadata must not override the protocol headers.
+        $response->setHeaders(Http2\encodeMetadata(new Metadata([
+            ...$responseHeaders->getFuture()->await()->kv,
+            ...$headers->kv,
+        ])));
+
         return $response;
     }
 

packages/server/src/Server/Internal/Http2/StreamFactory.php

@@ -42,13 +42,15 @@ public function __construct(
 
     /**
      * @param Handle<*> $handle
+     * @param DeferredFuture<Metadata> $responseHeaders completes when the response headers must be sent
      * @return ServerStream<*, *>
      */
     public function create(
         Handle $handle,
         Metadata $md,
         Request $request,
         Response $response,
+        DeferredFuture $responseHeaders,
         Cancellation $cancellation,
     ): ServerStream {
         /** @var Pipeline\Queue<object> $send */
@@ -66,6 +68,7 @@ public function create(
             $md,
             $this->codec->decode($request->getBody(), $handle->type, $cancellation),
             $send,
+            $responseHeaders,
             $trailers,
         );
     }

packages/server/src/Server/StreamInfo.php

@@ -4,6 +4,7 @@
 
 namespace Thesis\Grpc\Server;
 
+use Thesis\Grpc\Metadata;
 use Thesis\Grpc\RpcType;
 
 /**
@@ -17,5 +18,9 @@
     public function __construct(
         public string $method,
         public RpcType $type,
+        /** Can be modified before the first message is sent. */
+        public Metadata $responseHeaders,
+        /** Can be modified before the stream is closed. */
+        public Metadata $trailers,
     ) {}
 }

packages/server/src/ServerStream.php

@@ -19,6 +19,11 @@ interface ServerStream extends \IteratorAggregate
      */
     public Metadata $headers { get; }
 
+    /**
+     * Outgoing response metadata. This property can be modified before the first message is sent ({@see send}) or the stream is closed ({@see close}).
+     */
+    public Metadata $responseHeaders { get; }
+
     /**
      * This property can be modified before the stream is closed ({@see close}).
      */

tests/MessageSizeTest.php

@@ -64,9 +64,7 @@ public function testClientStream(): void
 
     public function testServerStream(): void
     {
-        self::markTestSkipped('Corrupts and hangs until https://github.com/amphp/http-server/pull/396 is released.');
-
-        $server = self::server(new Server\Builder()); // @phpstan-ignore deadCode.unreachable (the test is skipped until the amphp fix is released)
+        $server = self::server(new Server\Builder());
 
         try {
             $events = iterator_to_array(new TopicServiceClient(new Client\Builder()->build())->subscribe(new SubscribeRequest('11')), preserve_keys: false);