From 2b0a985da9998b7ddb46c2da16ff7a75f243e3b4 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Sun, 11 Oct 2026 17:38:14 +0300 Subject: [PATCH] Send response headers from server --- composer.json | 2 +- packages/server/composer.json | 2 +- .../src/Server/BidirectionalStreamChannel.php | 11 +- .../server/src/Server/ClientStreamChannel.php | 11 +- .../server/src/Server/DecoratedStream.php | 2 + .../Internal/Http2/ConcurrentServerStream.php | 14 ++ .../Internal/Http2/ServerRequestHandler.php | 16 +- .../Server/Internal/Http2/StreamFactory.php | 3 + packages/server/src/Server/StreamInfo.php | 5 + packages/server/src/ServerStream.php | 5 + tests/MessageSizeTest.php | 4 +- tests/ResponseMetadataTest.php | 197 ++++++++++++++++++ 12 files changed, 263 insertions(+), 9 deletions(-) create mode 100644 tests/ResponseMetadataTest.php diff --git a/composer.json b/composer.json index 4f8b806..8bcb69d 100644 --- a/composer.json +++ b/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", diff --git a/packages/server/composer.json b/packages/server/composer.json index 98b0264..578c23c 100644 --- a/packages/server/composer.json +++ b/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", diff --git a/packages/server/src/Server/BidirectionalStreamChannel.php b/packages/server/src/Server/BidirectionalStreamChannel.php index 2660eb1..a6c1e00 100644 --- a/packages/server/src/Server/BidirectionalStreamChannel.php +++ b/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 $stream */ public function __construct( private ServerStream $stream, - ) {} + ) { + $this->responseHeaders = $stream->responseHeaders; + $this->trailers = $stream->trailers; + } /** * @no-named-arguments diff --git a/packages/server/src/Server/ClientStreamChannel.php b/packages/server/src/Server/ClientStreamChannel.php index 10ec029..e873120 100644 --- a/packages/server/src/Server/ClientStreamChannel.php +++ b/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 $stream */ public function __construct( private ServerStream $stream, - ) {} + ) { + $this->responseHeaders = $stream->responseHeaders; + $this->trailers = $stream->trailers; + } /** * @return TRequest diff --git a/packages/server/src/Server/DecoratedStream.php b/packages/server/src/Server/DecoratedStream.php index c899237..c3be266 100644 --- a/packages/server/src/Server/DecoratedStream.php +++ b/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; } diff --git a/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php b/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php index 65a9266..d74098d 100644 --- a/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php +++ b/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 $recv * @param Pipeline\Queue $send + * @param DeferredFuture $responseHeadersFuture * @param DeferredFuture>> $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); + } + } } diff --git a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php b/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php index bba2a1b..461d735 100644 --- a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php +++ b/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 $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; } diff --git a/packages/server/src/Server/Internal/Http2/StreamFactory.php b/packages/server/src/Server/Internal/Http2/StreamFactory.php index d3a865e..db1a093 100644 --- a/packages/server/src/Server/Internal/Http2/StreamFactory.php +++ b/packages/server/src/Server/Internal/Http2/StreamFactory.php @@ -42,6 +42,7 @@ public function __construct( /** * @param Handle<*> $handle + * @param DeferredFuture $responseHeaders completes when the response headers must be sent * @return ServerStream<*, *> */ public function create( @@ -49,6 +50,7 @@ public function create( Metadata $md, Request $request, Response $response, + DeferredFuture $responseHeaders, Cancellation $cancellation, ): ServerStream { /** @var Pipeline\Queue $send */ @@ -66,6 +68,7 @@ public function create( $md, $this->codec->decode($request->getBody(), $handle->type, $cancellation), $send, + $responseHeaders, $trailers, ); } diff --git a/packages/server/src/Server/StreamInfo.php b/packages/server/src/Server/StreamInfo.php index 2c4c5ca..73492ab 100644 --- a/packages/server/src/Server/StreamInfo.php +++ b/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, ) {} } diff --git a/packages/server/src/ServerStream.php b/packages/server/src/ServerStream.php index 183d232..89a3a19 100644 --- a/packages/server/src/ServerStream.php +++ b/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}). */ diff --git a/tests/MessageSizeTest.php b/tests/MessageSizeTest.php index 05d66ae..86b673b 100644 --- a/tests/MessageSizeTest.php +++ b/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); diff --git a/tests/ResponseMetadataTest.php b/tests/ResponseMetadataTest.php new file mode 100644 index 0000000..12fa226 --- /dev/null +++ b/tests/ResponseMetadataTest.php @@ -0,0 +1,197 @@ +server = new Server\Builder() + ->withServices( + new EchoServiceServerRegistry(new readonly class implements EchoServiceServer { + #[\Override] + public function echo(EchoRequest $request, Metadata $md, Cancellation $cancellation): EchoResponse + { + if ($request->sentence === 'error') { + throw new InvokeError(Code::FAILED_PRECONDITION, 'error'); + } + + return new EchoResponse($request->sentence); + } + }), + new TopicServiceServerRegistry(new readonly class implements TopicServiceServer { + #[\Override] + public function subscribe(SubscribeRequest $request, Metadata $md, Cancellation $cancellation): iterable + { + yield new Event($request->topic); + } + }), + new FileServiceServerRegistry(new readonly class implements FileServiceServer { + #[\Override] + public function upload(ClientStreamChannel $stream, Metadata $md, Cancellation $cancellation): FileInfo + { + ResponseMetadataTest::setMetadata($stream->responseHeaders, $stream->trailers); + + return new FileInfo(\count(iterator_to_array($stream, preserve_keys: false))); + } + }), + new MessengerServiceServerRegistry(new readonly class implements MessengerServiceServer { + #[\Override] + public function chat(BidirectionalStreamChannel $stream, Metadata $md, Cancellation $cancellation): void + { + ResponseMetadataTest::setMetadata($stream->responseHeaders, $stream->trailers); + + foreach ($stream as $message) { + $stream->send($message); + } + + $stream->close(); + } + }), + ) + ->withUnaryInterceptors(new CallableUnaryInterceptor(static function ( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + callable $next, + ): object { + self::setMetadata($info->responseHeaders, $info->trailers); + + return $next($request, $info, $md, $cancellation); + })) + ->withStreamInterceptors(new CallableStreamInterceptor(static function ( + ServerStream $stream, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + callable $next, + ): void { + if ($info->type === RpcType::ServerStream) { + self::setMetadata($info->responseHeaders, $info->trailers); + } + + $next($stream, $info, $md, $cancellation); + })) + ->build(); + + $this->server->start(); + } + + protected function tearDown(): void + { + $this->server->stop(); + } + + /** + * @param Invoke $invoke + */ + #[DataProvider('provideResponseMetadataCases')] + public function testResponseMetadata(Invoke $invoke, object $request, object $response): void + { + $stream = new Client\Builder()->build()->createStream($invoke); + $stream->send($request); + $stream->close(); + + self::assertEquals($response, $stream->receive()); + + $headers = $stream->headers(); + self::assertSame(['header'], $headers['x-header']); + self::assertSame(["\x00\xff"], $headers['x-header-bin']); + self::assertStringStartsWith('application/grpc', $headers->value('content-type') ?? ''); + self::assertSame(['trailer'], $stream->trailers()['x-trailer']); + } + + /** + * @return iterable, object, object}> + */ + public static function provideResponseMetadataCases(): iterable + { + yield 'unary' => [ + new Invoke('/echos.api.v1.EchoService/Echo', EchoResponse::class, RpcType::Unary), + new EchoRequest('ping'), + new EchoResponse('ping'), + ]; + + yield 'server stream' => [ + new Invoke('/topic.api.v1.TopicService/Subscribe', Event::class, RpcType::ServerStream), + new SubscribeRequest('payments'), + new Event('payments'), + ]; + + yield 'client stream' => [ + new Invoke('/file.api.v1.FileService/Upload', FileInfo::class, RpcType::ClientStream), + new Chunk('content'), + new FileInfo(1), + ]; + + yield 'bidirectional stream' => [ + new Invoke('/chat.api.v1.MessengerService/Chat', Message::class, RpcType::BidirectionalStream), + new Message('hello'), + new Message('hello'), + ]; + } + + public function testErrorBeforeFirstMessage(): void + { + $stream = new Client\Builder()->build()->createStream( + new Invoke('/echos.api.v1.EchoService/Echo', EchoResponse::class, RpcType::Unary), + ); + $stream->send(new EchoRequest('error')); + $stream->close(); + + try { + $stream->receive(); + self::fail('InvokeError expected.'); + } catch (InvokeError $e) { + self::assertSame(Code::FAILED_PRECONDITION, $e->statusCode); + } + + self::assertSame(['header'], $stream->headers()['x-header']); + self::assertSame(['trailer'], $stream->trailers()['x-trailer']); + } + + public static function setMetadata(Metadata $headers, Metadata $trailers): void + { + $headers->join(new Metadata([ + 'x-header' => 'header', + 'x-header-bin' => "\x00\xff", + 'content-type' => 'text/plain', + ])); + $trailers->join(new Metadata(['x-trailer' => 'trailer'])); + } +}