From 92d210f0176e58e9f60df7ef3e23df3e43f501cf Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Fri, 31 Jul 2026 21:07:42 +0300 Subject: [PATCH 1/4] Implement retries on gRPC stream level --- composer.json | 2 +- packages/client/src/Client/Builder.php | 15 ++- .../src/Client/Internal/AmphpHttpClient.php | 8 +- .../client/src/Client/Internal/Connection.php | 16 +++ .../Internal/Connection/DefaultConnection.php | 69 ++++++++++-- .../Internal/Connection/LazyConnection.php | 24 +++- .../Internal/Http2/ConcurrentClientStream.php | 5 +- .../src/Client/LoadBalancer/PickFirst.php | 10 +- .../src/Client/LoadBalancer/RoundRobin.php | 8 ++ packages/client/src/Client/PickContext.php | 7 ++ packages/client/src/Client/Retry.php | 102 +++++++++++++++++ packages/grpc/src/InvokeError.php | 14 ++- tests/Client/LoadBalancer/PickFirstTest.php | 28 +++++ tests/Client/LoadBalancer/RoundRobinTest.php | 27 +++++ tests/RetryFailoverTest.php | 106 ++++++++++++++++++ tests/genproto/Chat/Api/V1/Message.php | 2 +- .../Chat/Api/V1/MessengerServiceClient.php | 2 +- .../Chat/Api/V1/MessengerServiceServer.php | 2 +- .../Api/V1/MessengerServiceServerRegistry.php | 2 +- .../TestsProtosChatV1DescriptorRegistry.php | 2 +- .../Chat/Api/V1/autoload.metadata.php | 2 +- tests/genproto/Echos/Api/V1/EchoRequest.php | 2 +- tests/genproto/Echos/Api/V1/EchoResponse.php | 2 +- .../Echos/Api/V1/EchoServiceClient.php | 2 +- .../Echos/Api/V1/EchoServiceServer.php | 2 +- .../Api/V1/EchoServiceServerRegistry.php | 2 +- .../TestsProtosEchoV1DescriptorRegistry.php | 2 +- .../Echos/Api/V1/autoload.metadata.php | 2 +- tests/genproto/File/Api/V1/Chunk.php | 2 +- tests/genproto/File/Api/V1/FileInfo.php | 2 +- .../File/Api/V1/FileServiceClient.php | 2 +- .../File/Api/V1/FileServiceServer.php | 2 +- .../File/Api/V1/FileServiceServerRegistry.php | 2 +- .../TestsProtosFileV1DescriptorRegistry.php | 2 +- .../File/Api/V1/autoload.metadata.php | 2 +- tests/genproto/Topic/Api/V1/Event.php | 2 +- .../Topic/Api/V1/SubscribeRequest.php | 2 +- .../TestsProtosTopicV1DescriptorRegistry.php | 2 +- .../Topic/Api/V1/TopicServiceClient.php | 2 +- .../Topic/Api/V1/TopicServiceServer.php | 2 +- .../Api/V1/TopicServiceServerRegistry.php | 2 +- .../Topic/Api/V1/autoload.metadata.php | 2 +- 42 files changed, 439 insertions(+), 56 deletions(-) create mode 100644 packages/client/src/Client/Retry.php create mode 100644 tests/RetryFailoverTest.php diff --git a/composer.json b/composer.json index bf521e8..9536e91 100644 --- a/composer.json +++ b/composer.json @@ -37,7 +37,7 @@ "thesis/googleapis-rpc-types": "^0.1.6", "thesis/package-version": "^0.1.2", "thesis/protobuf": "^0.1.8", - "thesis/protobuf-known-types": "^0.1.5" + "thesis/protobuf-known-types": "^0.1.7" }, "require-dev": { "ext-bcmath": "*", diff --git a/packages/client/src/Client/Builder.php b/packages/client/src/Client/Builder.php index 0d95e0f..260fdc1 100644 --- a/packages/client/src/Client/Builder.php +++ b/packages/client/src/Client/Builder.php @@ -9,6 +9,7 @@ use Amp\Http\Client\DelegateHttpClient; use Amp\Http\Client\HttpClientBuilder; use Amp\Socket\ConnectContext; +use Amp\Socket\DnsSocketConnector; use Amp\Socket\SocketConnector; use Thesis\Grpc\Client; use Thesis\Grpc\Client\Internal\Connection; @@ -62,6 +63,8 @@ final class Builder private ?LoadBalancerFactory $loadBalancerFactory = null; + private ?Retry $retry = null; + /** @var \SplObjectStorage */ private \SplObjectStorage $endpointResolvers; @@ -186,6 +189,14 @@ public function withLoadBalancer(LoadBalancerFactory $factory): self return $builder; } + public function withRetry(Retry $policy): self + { + $builder = clone $this; + $builder->retry = $policy; + + return $builder; + } + public function withEndpointResolver(Scheme $scheme, EndpointResolver $resolver): self { $builder = clone $this; @@ -207,6 +218,7 @@ public function build(): Client $compressor = $this->compressor ?? IdentityCompressor::Compressor; $protobuf = $this->protobuf ?? Decoder\Builder::buildDefault(); $loadBalancerFactory = $this->loadBalancerFactory ?? new LoadBalancer\PickFirstFactory(); + $retry = $this->retry ?? Retry::disabled(); $tlsContext = $this->credentials?->createContext(); $uriFactory = new Http2\UriFactory($tlsContext !== null ? Internal\HttpScheme::Https : Internal\HttpScheme::Http); $transferTimeout = $this->transferTimeout; @@ -230,7 +242,7 @@ public function build(): Client ->usingPool(ConnectionLimitingPool::byAuthority( $this->connectionLimit, new DefaultConnectionFactory( - $this->connector, + $this->connector ?? new DnsSocketConnector(), new ConnectContext() ->withConnectTimeout($this->connectTimeout) ->withTlsContext($tlsContext), @@ -257,6 +269,7 @@ public function build(): Client encoder: $encoder, compressor: $compressor, ), + retry: $retry, ), ), ); diff --git a/packages/client/src/Client/Internal/AmphpHttpClient.php b/packages/client/src/Client/Internal/AmphpHttpClient.php index c56728a..b751c67 100644 --- a/packages/client/src/Client/Internal/AmphpHttpClient.php +++ b/packages/client/src/Client/Internal/AmphpHttpClient.php @@ -26,16 +26,12 @@ public function invoke( Metadata $md = new Metadata(), Cancellation $cancellation = new NullCancellation(), ): object { - $stream = $this->connection->createStream( + return $this->connection->invoke( + $request, $invoke, $md, $cancellation, ); - - $stream->send($request); - $stream->close(); - - return $stream->receive(); } #[\Override] diff --git a/packages/client/src/Client/Internal/Connection.php b/packages/client/src/Client/Internal/Connection.php index 884fb26..e516f8c 100644 --- a/packages/client/src/Client/Internal/Connection.php +++ b/packages/client/src/Client/Internal/Connection.php @@ -8,6 +8,7 @@ use Amp\NullCancellation; use Thesis\Grpc\Client\Invoke; use Thesis\Grpc\ClientStream; +use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; /** @@ -15,6 +16,21 @@ */ interface Connection { + /** + * @template In of object + * @template Out of object + * @param In $request + * @param Invoke $invoke + * @return Out + * @throws InvokeError + */ + public function invoke( + object $request, + Invoke $invoke, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): object; + /** * @template In of object * @template Out of object diff --git a/packages/client/src/Client/Internal/Connection/DefaultConnection.php b/packages/client/src/Client/Internal/Connection/DefaultConnection.php index 61da9fb..133fa9a 100644 --- a/packages/client/src/Client/Internal/Connection/DefaultConnection.php +++ b/packages/client/src/Client/Internal/Connection/DefaultConnection.php @@ -7,6 +7,7 @@ use Amp\Cancellation; use Amp\DeferredCancellation; use Amp\NullCancellation; +use Thesis\Grpc\Client\Endpoint; use Thesis\Grpc\Client\EndpointResolver; use Thesis\Grpc\Client\EndpointResolverListener; use Thesis\Grpc\Client\Internal\Connection; @@ -17,6 +18,7 @@ use Thesis\Grpc\Client\LoadBalancerFactory; use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\Client\Resolution; +use Thesis\Grpc\Client\Retry; use Thesis\Grpc\Client\Target; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; @@ -38,6 +40,7 @@ public function __construct( LoadBalancerFactory $loadBalancerFactory, private InterceptorComposer $interceptor, private StreamFactory $streams, + private Retry $retry, ) { $this->deferredCancellation = new DeferredCancellation(); @@ -51,26 +54,47 @@ public function __construct( } #[\Override] - public function createStream( + public function invoke( + object $request, Invoke $invoke, Metadata $md = new Metadata(), Cancellation $cancellation = new NullCancellation(), - ): ClientStream { - $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md)); + ): object { + /** @var list $excluded */ + $excluded = []; - return $this->interceptor->intercept( // @phpstan-ignore return.type - $invoke, - $md, - $cancellation, - fn(Invoke $invoke, Metadata $md, Cancellation $cancellation) => $this->streams->create( + return $this->retry->call( + function () use ( + &$excluded, + $request, $invoke, - $endpoint->address, $md, $cancellation, - ), + ): object { + $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md, $excluded)); + $excluded[] = $endpoint; + + $stream = $this->stream($endpoint, $invoke, $md, $cancellation); + $stream->send($request); + $stream->close(); + + return $stream->receive(); + }, + $cancellation, ); } + #[\Override] + public function createStream( + Invoke $invoke, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): ClientStream { + $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md)); + + return $this->stream($endpoint, $invoke, $md, $cancellation); + } + #[\Override] public function close(Cancellation $cancellation = new NullCancellation()): void { @@ -84,4 +108,29 @@ public function onResolve(Resolution|\Throwable $result): void $this->balancer->refresh($result->endpoints); } } + + /** + * @template In of object + * @template Out of object + * @param Invoke $invoke + * @return ClientStream + */ + private function stream( + Endpoint $endpoint, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + ): ClientStream { + return $this->interceptor->intercept( // @phpstan-ignore return.type + $invoke, + $md, + $cancellation, + fn(Invoke $invoke, Metadata $md, Cancellation $cancellation) => $this->streams->create( + $invoke, + $endpoint->address, + $md, + $cancellation, + ), + ); + } } diff --git a/packages/client/src/Client/Internal/Connection/LazyConnection.php b/packages/client/src/Client/Internal/Connection/LazyConnection.php index 5607c82..31d7bf2 100644 --- a/packages/client/src/Client/Internal/Connection/LazyConnection.php +++ b/packages/client/src/Client/Internal/Connection/LazyConnection.php @@ -28,16 +28,27 @@ public function __construct( private readonly \Closure $factory, ) {} + #[\Override] + public function invoke( + object $request, + Invoke $invoke, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): object { + return $this + ->createConnection($cancellation) + ->invoke($request, $invoke, $md, $cancellation); + } + #[\Override] public function createStream( Invoke $invoke, Metadata $md = new Metadata(), Cancellation $cancellation = new NullCancellation(), ): ClientStream { - $this->future ??= async($this->factory); - $connection = $this->future->await($cancellation); - - return $connection->createStream($invoke, $md, $cancellation); + return $this + ->createConnection($cancellation) + ->createStream($invoke, $md, $cancellation); } #[\Override] @@ -48,4 +59,9 @@ public function close(Cancellation $cancellation = new NullCancellation()): void $future?->await($cancellation)->close($cancellation); } + + private function createConnection(Cancellation $cancellation): Connection + { + return ($this->future ??= async($this->factory))->await($cancellation); + } } diff --git a/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php b/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php index 49da0ac..7b0b69e 100644 --- a/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php +++ b/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php @@ -40,7 +40,10 @@ public function __construct( private readonly \Closure $decode, private readonly ErrorHandler $errors, private readonly Future $complete, - ) {} + ) { + $responseFuture->ignore(); + $complete->ignore(); + } #[\Override] public function send(object $message): void diff --git a/packages/client/src/Client/LoadBalancer/PickFirst.php b/packages/client/src/Client/LoadBalancer/PickFirst.php index 7ff7591..51f52ef 100644 --- a/packages/client/src/Client/LoadBalancer/PickFirst.php +++ b/packages/client/src/Client/LoadBalancer/PickFirst.php @@ -20,7 +20,7 @@ final class PickFirst implements LoadBalancer * @param non-empty-list $endpoints */ public function __construct( - array $endpoints, + private array $endpoints, private readonly Randomizer $randomizer, ) { $this->current = $this->doPick($endpoints); @@ -29,12 +29,20 @@ public function __construct( #[\Override] public function refresh(array $endpoints): void { + $this->endpoints = $endpoints; $this->current = $this->doPick($endpoints, $this->current); } #[\Override] public function pick(PickContext $context): Endpoint { + foreach ([$this->current, ...$this->endpoints] as $endpoint) { + if (!$context->excluded($endpoint)) { + $this->current = $endpoint; + break; + } + } + return $this->current; } diff --git a/packages/client/src/Client/LoadBalancer/RoundRobin.php b/packages/client/src/Client/LoadBalancer/RoundRobin.php index c08dea0..b7c60bd 100644 --- a/packages/client/src/Client/LoadBalancer/RoundRobin.php +++ b/packages/client/src/Client/LoadBalancer/RoundRobin.php @@ -37,6 +37,14 @@ public function refresh(array $endpoints): void #[\Override] public function pick(PickContext $context): Endpoint { + for ($i = 0; $i < $this->count; ++$i) { + $endpoint = $this->endpoints[$this->cursor++ % $this->count]; // @phpstan-ignore offsetAccess.notFound + + if (!$context->excluded($endpoint)) { + return $endpoint; + } + } + return $this->endpoints[$this->cursor++ % $this->count]; // @phpstan-ignore offsetAccess.notFound } } diff --git a/packages/client/src/Client/PickContext.php b/packages/client/src/Client/PickContext.php index e6666d9..edb2405 100644 --- a/packages/client/src/Client/PickContext.php +++ b/packages/client/src/Client/PickContext.php @@ -13,9 +13,16 @@ { /** * @param non-empty-string $methodName + * @param list $excluded endpoints that already failed this call and should be skipped if possible */ public function __construct( public string $methodName, public Metadata $metadata, + private array $excluded = [], ) {} + + public function excluded(Endpoint $endpoint): bool + { + return array_any($this->excluded, $endpoint->equals(...)); + } } diff --git a/packages/client/src/Client/Retry.php b/packages/client/src/Client/Retry.php new file mode 100644 index 0000000..018a15d --- /dev/null +++ b/packages/client/src/Client/Retry.php @@ -0,0 +1,102 @@ + $retryableStatusCodes gRPC statuses that trigger a retry + */ + public function __construct( + private int $maxAttempts = 3, + private float $initialBackoff = 0.05, + private float $maxBackoff = 1.0, + private float $backoffMultiplier = 2.0, + private array $retryableStatusCodes = [ + Code::UNAVAILABLE, + ], + private Randomizer $randomizer = new Randomizer(), + ) {} + + public static function disabled(): self + { + return new self(maxAttempts: 1); + } + + /** + * @template T + * @param \Closure(): T $call + * @return T + * @throws GrpcException + */ + public function call(\Closure $call, Cancellation $cancellation): mixed + { + $error = null; + + for ($attempt = 1; $attempt <= $this->maxAttempts; ++$attempt) { + if ($attempt > 1) { + $this->backoff($attempt - 1, $cancellation); + } + + try { + return $call(); + } catch (\Throwable $e) { + if (!$this->shouldRetry($e)) { + throw self::createException($e); + } + + $error = $e; + } + } + + throw self::createException($error ?? new InvokeError(Code::UNAVAILABLE, 'no attempts were made')); + } + + private function shouldRetry(\Throwable $e): bool + { + if ($e instanceof InvokeError) { + return \in_array($e->statusCode, $this->retryableStatusCodes, true); + } + + return $e instanceof HttpException; + } + + /** + * @param positive-int $attempt + */ + private function backoff(int $attempt, Cancellation $cancellation): void + { + $ceiling = min($this->maxBackoff, $this->initialBackoff * $this->backoffMultiplier ** ($attempt - 1)); + + if ($ceiling > 0.0) { + delay($this->randomizer->getFloat(0.0, $ceiling), cancellation: $cancellation); + } + } + + private static function createException(\Throwable $e): GrpcException + { + if ($e instanceof InvokeError || $e instanceof ClientStreamIsClosed) { + return $e; + } + + return new InvokeError(Code::UNAVAILABLE, $e->getMessage(), previous: $e); + } +} diff --git a/packages/grpc/src/InvokeError.php b/packages/grpc/src/InvokeError.php index 73d6b25..67ac669 100644 --- a/packages/grpc/src/InvokeError.php +++ b/packages/grpc/src/InvokeError.php @@ -18,11 +18,15 @@ public function __construct( public readonly Code $statusCode, public readonly ?string $statusMessage = null, public readonly array $details = [], + ?\Throwable $previous = null, ) { - parent::__construct(\sprintf( - 'A grpc error with status code "%s" and message "%s" occurred', - $statusCode->name, - $statusMessage ?? '', - )); + parent::__construct( + \sprintf( + 'A grpc error with status code "%s" and message "%s" occurred', + $statusCode->name, + $statusMessage ?? '', + ), + previous: $previous, + ); } } diff --git a/tests/Client/LoadBalancer/PickFirstTest.php b/tests/Client/LoadBalancer/PickFirstTest.php index d1a48d4..c4fe524 100644 --- a/tests/Client/LoadBalancer/PickFirstTest.php +++ b/tests/Client/LoadBalancer/PickFirstTest.php @@ -118,6 +118,34 @@ public static function provideRefreshSwitchesPinnedEndpointCases(): iterable ]; } + public function testPickFailsOverToALiveEndpointWhenPinnedIsExcluded(): void + { + $a = new Endpoint(new Address('10.0.0.1:50051')); + $b = new Endpoint(new Address('10.0.0.2:50051')); + $c = new Endpoint(new Address('10.0.0.3:50051')); + + $balancer = new PickFirstFactory()->create([$a, $b, $c]); + $pinned = $balancer->pick(self::context()); + + $failedOver = $balancer->pick(new PickContext('/test.Service/Method', new Metadata(), [$pinned])); + + self::assertFalse($pinned->equals($failedOver)); + self::assertTrue($failedOver->equals($balancer->pick(self::context()))); + } + + public function testPickStaysPinnedWhenEveryEndpointIsExcluded(): void + { + $a = new Endpoint(new Address('10.0.0.1:50051')); + $b = new Endpoint(new Address('10.0.0.2:50051')); + + $balancer = new PickFirstFactory()->create([$a, $b]); + $pinned = $balancer->pick(self::context()); + + $picked = $balancer->pick(new PickContext('/test.Service/Method', new Metadata(), [$a, $b])); + + self::assertTrue($pinned->equals($picked)); + } + private static function context(): PickContext { return new PickContext('/test.Service/Method', new Metadata()); diff --git a/tests/Client/LoadBalancer/RoundRobinTest.php b/tests/Client/LoadBalancer/RoundRobinTest.php index 1c0fd4d..e8f0494 100644 --- a/tests/Client/LoadBalancer/RoundRobinTest.php +++ b/tests/Client/LoadBalancer/RoundRobinTest.php @@ -104,6 +104,33 @@ public static function provideRefreshCases(): iterable ]; } + public function testPickSkipsExcludedEndpoints(): void + { + $a = new Endpoint(new Address('10.0.0.1:50051')); + $b = new Endpoint(new Address('10.0.0.2:50051')); + $c = new Endpoint(new Address('10.0.0.3:50051')); + + $balancer = new RoundRobinFactory()->create([$a, $b, $c]); + + $context = new PickContext('/test.Service/Method', new Metadata(), [$a, $b]); + + for ($i = 0; $i < 5; ++$i) { + self::assertTrue($c->equals($balancer->pick($context))); + } + } + + public function testPickFallsBackWhenEveryEndpointIsExcluded(): void + { + $a = new Endpoint(new Address('10.0.0.1:50051')); + $b = new Endpoint(new Address('10.0.0.2:50051')); + + $balancer = new RoundRobinFactory()->create([$a, $b]); + + $picked = $balancer->pick(new PickContext('/test.Service/Method', new Metadata(), [$a, $b])); + + self::assertTrue($a->equals($picked) || $b->equals($picked)); + } + private static function context(): PickContext { return new PickContext('/test.Service/Method', new Metadata()); diff --git a/tests/RetryFailoverTest.php b/tests/RetryFailoverTest.php new file mode 100644 index 0000000..574c001 --- /dev/null +++ b/tests/RetryFailoverTest.php @@ -0,0 +1,106 @@ +server = new Server\Builder() + ->withAddresses('0.0.0.0:50052') + ->withServices(new EchoServiceServerRegistry(new SucceedingEchoServer())) + ->build(); + + $this->server->start(); + } + + protected function tearDown(): void + { + $this->server->stop(); + } + + public function testRetryFailsOverFromADeadEndpointToALiveOne(): void + { + $client = new EchoServiceClient( + new Client\Builder() + ->withHost(\sprintf('ipv4:%s,%s', self::DEAD, self::LIVE)) + ->withLoadBalancer(new RoundRobinFactory()) + ->withRetry(new Retry(maxAttempts: 3, initialBackoff: 0.0)) + ->build(), + ); + + for ($i = 0; $i < 10; ++$i) { + self::assertSame('ok', $client->echo(new EchoRequest('ping'))->sentence); + } + } + + public function testWithoutRetryADeadEndpointSurfacesAsUnavailable(): void + { + $client = new EchoServiceClient( + new Client\Builder() + ->withHost(\sprintf('ipv4:%s', self::DEAD)) + ->build(), + ); + + try { + $client->echo(new EchoRequest('ping')); + self::fail('Expected an InvokeError for the dead endpoint.'); + } catch (InvokeError $e) { + self::assertSame(Code::UNAVAILABLE, $e->statusCode); + } + } + + public function testRetryDoesNotHelpWhenEveryEndpointIsDead(): void + { + $client = new EchoServiceClient( + new Client\Builder() + ->withHost(\sprintf('ipv4:%s,127.0.0.1:50053', self::DEAD)) + ->withLoadBalancer(new RoundRobinFactory()) + ->withRetry(new Retry(maxAttempts: 3, initialBackoff: 0.0)) + ->build(), + ); + + $this->expectException(InvokeError::class); + $client->echo(new EchoRequest('ping')); + } +} + +/** + * @internal + */ +final readonly class SucceedingEchoServer implements EchoServiceServer +{ + #[\Override] + public function echo(EchoRequest $request, Metadata $md, Cancellation $cancellation): EchoResponse + { + return new EchoResponse('ok'); + } +} diff --git a/tests/genproto/Chat/Api/V1/Message.php b/tests/genproto/Chat/Api/V1/Message.php index c0c78d1..a87199e 100644 --- a/tests/genproto/Chat/Api/V1/Message.php +++ b/tests/genproto/Chat/Api/V1/Message.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceClient.php b/tests/genproto/Chat/Api/V1/MessengerServiceClient.php index 177dfe4..18fe027 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceClient.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceServer.php b/tests/genproto/Chat/Api/V1/MessengerServiceServer.php index 41cf805..f63e609 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceServer.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php b/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php index 3632285..76e6fe3 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php b/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php index cae9840..c085d87 100644 --- a/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php +++ b/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/autoload.metadata.php b/tests/genproto/Chat/Api/V1/autoload.metadata.php index 565556d..1972cd2 100644 --- a/tests/genproto/Chat/Api/V1/autoload.metadata.php +++ b/tests/genproto/Chat/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 */ diff --git a/tests/genproto/Echos/Api/V1/EchoRequest.php b/tests/genproto/Echos/Api/V1/EchoRequest.php index 50fac9b..784b54b 100644 --- a/tests/genproto/Echos/Api/V1/EchoRequest.php +++ b/tests/genproto/Echos/Api/V1/EchoRequest.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoResponse.php b/tests/genproto/Echos/Api/V1/EchoResponse.php index 712f9df..05773b7 100644 --- a/tests/genproto/Echos/Api/V1/EchoResponse.php +++ b/tests/genproto/Echos/Api/V1/EchoResponse.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceClient.php b/tests/genproto/Echos/Api/V1/EchoServiceClient.php index 6ab257a..1fbb5ff 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceClient.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceServer.php b/tests/genproto/Echos/Api/V1/EchoServiceServer.php index 132ecd6..53ba7c3 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceServer.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php b/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php index 2166e52..aa3cc35 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php b/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php index 23937c6..684402b 100644 --- a/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php +++ b/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/autoload.metadata.php b/tests/genproto/Echos/Api/V1/autoload.metadata.php index 042c19c..8dbdfe7 100644 --- a/tests/genproto/Echos/Api/V1/autoload.metadata.php +++ b/tests/genproto/Echos/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 */ diff --git a/tests/genproto/File/Api/V1/Chunk.php b/tests/genproto/File/Api/V1/Chunk.php index ad48142..5810e3a 100644 --- a/tests/genproto/File/Api/V1/Chunk.php +++ b/tests/genproto/File/Api/V1/Chunk.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileInfo.php b/tests/genproto/File/Api/V1/FileInfo.php index ddeaa38..bbfc773 100644 --- a/tests/genproto/File/Api/V1/FileInfo.php +++ b/tests/genproto/File/Api/V1/FileInfo.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceClient.php b/tests/genproto/File/Api/V1/FileServiceClient.php index 66c3543..e0ad74f 100644 --- a/tests/genproto/File/Api/V1/FileServiceClient.php +++ b/tests/genproto/File/Api/V1/FileServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceServer.php b/tests/genproto/File/Api/V1/FileServiceServer.php index 1fbaaf2..77b3eff 100644 --- a/tests/genproto/File/Api/V1/FileServiceServer.php +++ b/tests/genproto/File/Api/V1/FileServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceServerRegistry.php b/tests/genproto/File/Api/V1/FileServiceServerRegistry.php index e15a181..c2e2ce8 100644 --- a/tests/genproto/File/Api/V1/FileServiceServerRegistry.php +++ b/tests/genproto/File/Api/V1/FileServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php b/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php index fba3e1b..fe6fe75 100644 --- a/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php +++ b/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/autoload.metadata.php b/tests/genproto/File/Api/V1/autoload.metadata.php index d5ffaed..da6e84e 100644 --- a/tests/genproto/File/Api/V1/autoload.metadata.php +++ b/tests/genproto/File/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 */ diff --git a/tests/genproto/Topic/Api/V1/Event.php b/tests/genproto/Topic/Api/V1/Event.php index 9f7c100..b77df80 100644 --- a/tests/genproto/Topic/Api/V1/Event.php +++ b/tests/genproto/Topic/Api/V1/Event.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/SubscribeRequest.php b/tests/genproto/Topic/Api/V1/SubscribeRequest.php index 58fa66f..e3fabf6 100644 --- a/tests/genproto/Topic/Api/V1/SubscribeRequest.php +++ b/tests/genproto/Topic/Api/V1/SubscribeRequest.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php b/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php index 744320a..a9d884f 100644 --- a/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php +++ b/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceClient.php b/tests/genproto/Topic/Api/V1/TopicServiceClient.php index 511c6a1..184dd04 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceClient.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceServer.php b/tests/genproto/Topic/Api/V1/TopicServiceServer.php index 9306112..3e09a3d 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceServer.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php b/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php index 271f693..2693d39 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/autoload.metadata.php b/tests/genproto/Topic/Api/V1/autoload.metadata.php index 0a5a7df..85988a7 100644 --- a/tests/genproto/Topic/Api/V1/autoload.metadata.php +++ b/tests/genproto/Topic/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.21 + * thesis/protoc-plugin — v0.1.22 * protoc — v6.33.5 */ From 01b9c247a24d18f019c7ed02db73c40f91faa0aa Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Fri, 31 Jul 2026 21:24:39 +0300 Subject: [PATCH 2/4] chore: try to fix CI issue --- composer.json | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/composer.json b/composer.json index 9536e91..d7bcb0d 100644 --- a/composer.json +++ b/composer.json @@ -53,7 +53,12 @@ "type": "path", "url": "packages/*", "options": { - "symlink": true + "symlink": true, + "versions": { + "thesis/grpc-client": "0.1.x-dev", + "thesis/grpc-protocol": "0.1.x-dev", + "thesis/grpc-server": "0.1.x-dev" + } } } ], From c92f809ec92042171acdf8d95cf3564398301219 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Sun, 2 Aug 2026 21:43:39 +0300 Subject: [PATCH 3/4] gRPC interceptors refactoring --- packages/client/README.md | 28 ++++- packages/client/src/Client/Builder.php | 63 +++++++---- ...ptor.php => CallableStreamInterceptor.php} | 8 +- .../src/Client/CallableUnaryInterceptor.php | 40 +++++++ .../src/Client/Internal/AmphpHttpClient.php | 41 ++++++- .../client/src/Client/Internal/Connection.php | 22 +--- .../Internal/Connection/DefaultConnection.php | 76 ++----------- .../Internal/Connection/LazyConnection.php | 20 +--- .../AppendControlMetadataInterceptor.php | 31 +++-- ...oser.php => StreamInterceptorComposer.php} | 16 +-- .../Http2/UnaryInterceptorComposer.php | 59 ++++++++++ packages/client/src/Client/PickContext.php | 14 ++- packages/client/src/Client/Retry.php | 102 ----------------- ...{Interceptor.php => StreamInterceptor.php} | 8 +- .../client/src/Client/UnaryInterceptor.php | 30 +++++ packages/server/README.md | 26 ++++- .../src/Server/BidirectionalStreamHandler.php | 4 +- packages/server/src/Server/Builder.php | 30 ++++- ...ptor.php => CallableStreamInterceptor.php} | 4 +- .../src/Server/CallableUnaryInterceptor.php | 40 +++++++ .../server/src/Server/ClientStreamHandler.php | 4 +- .../Internal/Http2/ConcurrentServerStream.php | 4 + .../Internal/Http2/ServerRequestHandler.php | 92 +++++++++++---- .../Http2/StreamHandleInterceptor.php | 6 +- ...oser.php => StreamInterceptorComposer.php} | 10 +- .../Http2/UnaryInterceptorComposer.php | 62 ++++++++++ packages/server/src/Server/Rpc.php | 4 +- .../server/src/Server/ServerStreamHandler.php | 4 +- .../Server/{Handler.php => StreamHandler.php} | 2 +- ...{Interceptor.php => StreamInterceptor.php} | 4 +- packages/server/src/Server/UnaryHandler.php | 16 ++- .../server/src/Server/UnaryInterceptor.php | 33 ++++++ tests/ClientStreamTest.php | 4 +- tests/DecoratedStreamTest.php | 93 +++++++++++++++ tests/RetryFailoverTest.php | 106 ------------------ tests/Stub/AuthorizationClientInterceptor.php | 24 +++- tests/Stub/AuthorizationServerInterceptor.php | 31 ++++- tests/UnaryTest.php | 34 ++---- tests/genproto/Chat/Api/V1/Message.php | 2 +- .../Chat/Api/V1/MessengerServiceClient.php | 2 +- .../Chat/Api/V1/MessengerServiceServer.php | 2 +- .../Api/V1/MessengerServiceServerRegistry.php | 2 +- .../TestsProtosChatV1DescriptorRegistry.php | 2 +- .../Chat/Api/V1/autoload.metadata.php | 2 +- tests/genproto/Echos/Api/V1/EchoRequest.php | 2 +- tests/genproto/Echos/Api/V1/EchoResponse.php | 2 +- .../Echos/Api/V1/EchoServiceClient.php | 2 +- .../Echos/Api/V1/EchoServiceServer.php | 2 +- .../Api/V1/EchoServiceServerRegistry.php | 2 +- .../TestsProtosEchoV1DescriptorRegistry.php | 2 +- .../Echos/Api/V1/autoload.metadata.php | 2 +- tests/genproto/File/Api/V1/Chunk.php | 2 +- tests/genproto/File/Api/V1/FileInfo.php | 2 +- .../File/Api/V1/FileServiceClient.php | 2 +- .../File/Api/V1/FileServiceServer.php | 2 +- .../File/Api/V1/FileServiceServerRegistry.php | 2 +- .../TestsProtosFileV1DescriptorRegistry.php | 2 +- .../File/Api/V1/autoload.metadata.php | 2 +- tests/genproto/Topic/Api/V1/Event.php | 2 +- .../Topic/Api/V1/SubscribeRequest.php | 2 +- .../TestsProtosTopicV1DescriptorRegistry.php | 2 +- .../Topic/Api/V1/TopicServiceClient.php | 2 +- .../Topic/Api/V1/TopicServiceServer.php | 2 +- .../Api/V1/TopicServiceServerRegistry.php | 2 +- .../Topic/Api/V1/autoload.metadata.php | 2 +- 65 files changed, 759 insertions(+), 490 deletions(-) rename packages/client/src/Client/{CallableInterceptor.php => CallableStreamInterceptor.php} (81%) create mode 100644 packages/client/src/Client/CallableUnaryInterceptor.php rename packages/client/src/Client/Internal/Http2/{InterceptorComposer.php => StreamInterceptorComposer.php} (75%) create mode 100644 packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php delete mode 100644 packages/client/src/Client/Retry.php rename packages/client/src/Client/{Interceptor.php => StreamInterceptor.php} (79%) create mode 100644 packages/client/src/Client/UnaryInterceptor.php rename packages/server/src/Server/{CallableInterceptor.php => CallableStreamInterceptor.php} (87%) create mode 100644 packages/server/src/Server/CallableUnaryInterceptor.php rename packages/server/src/Server/Internal/Http2/{InterceptorComposer.php => StreamInterceptorComposer.php} (82%) create mode 100644 packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php rename packages/server/src/Server/{Handler.php => StreamHandler.php} (93%) rename packages/server/src/Server/{Interceptor.php => StreamInterceptor.php} (90%) create mode 100644 packages/server/src/Server/UnaryInterceptor.php create mode 100644 tests/DecoratedStreamTest.php delete mode 100644 tests/RetryFailoverTest.php diff --git a/packages/client/README.md b/packages/client/README.md index a8ba5e9..5ddd42f 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -236,23 +236,43 @@ $client = new Client\Builder() ## Interceptors -Interceptors let you add cross-cutting logic (auth, logging, tracing, retry, metadata enrichment) without changing service stubs. +Interceptors let you add cross-cutting logic (auth, logging, tracing, metadata enrichment) without changing service stubs. Mirroring gRPC, they come in two flavours, registered on separate chains: + +- `UnaryInterceptor` wraps a whole unary `request → response` call (via `withUnaryInterceptors()`). A retry policy, for instance, is naturally a unary interceptor — re-invoking `$invoker` reuses the call's pick context, so the transport fails over to another endpoint automatically. +- `StreamInterceptor` wraps the creation of a stream (via `withStreamInterceptors()`), and so also covers the underlying stream of every streaming RPC. + +A unary call does **not** run the stream chain and vice versa. An interceptor that must apply to both (e.g. auth) implements both interfaces and is registered on both chains. ```php use Amp\Cancellation; use Thesis\Grpc\Client; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\StreamInterceptor; +use Thesis\Grpc\Client\UnaryInterceptor; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; -final readonly class ClientAuthInterceptor implements Client\Interceptor +final readonly class ClientAuthInterceptor implements UnaryInterceptor, StreamInterceptor { #[\Override] - public function intercept(Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $next): ClientStream + public function interceptUnary(object $request, Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $invoker): object + { + return $invoker($request, $invoke, $md->with('Authorization', 'supertoken'), $cancellation); + } + + #[\Override] + public function interceptStream(Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $newStream): ClientStream { - return $next($invoke, $md->with('Authorization', 'supertoken'), $cancellation); + return $newStream($invoke, $md->with('Authorization', 'supertoken'), $cancellation); } } + +$auth = new ClientAuthInterceptor(); + +$client = new Client\Builder() + ->withUnaryInterceptors($auth) + ->withStreamInterceptors($auth) + ->build(); ``` ## Client streaming diff --git a/packages/client/src/Client/Builder.php b/packages/client/src/Client/Builder.php index 260fdc1..05a4574 100644 --- a/packages/client/src/Client/Builder.php +++ b/packages/client/src/Client/Builder.php @@ -38,8 +38,11 @@ final class Builder private ?DelegateHttpClient $httpclient = null; - /** @var list */ - private array $interceptors = []; + /** @var list */ + private array $unaryInterceptors = []; + + /** @var list */ + private array $streamInterceptors = []; private ?TransportCredentials $credentials = null; @@ -63,8 +66,6 @@ final class Builder private ?LoadBalancerFactory $loadBalancerFactory = null; - private ?Retry $retry = null; - /** @var \SplObjectStorage */ private \SplObjectStorage $endpointResolvers; @@ -119,11 +120,25 @@ public function withHost(string $host): self /** * @no-named-arguments */ - public function withInterceptors(Interceptor ...$interceptors): self + public function withUnaryInterceptors(UnaryInterceptor ...$interceptors): self + { + $builder = clone $this; + $builder->unaryInterceptors = [ + ...$builder->unaryInterceptors, + ...$interceptors, + ]; + + return $builder; + } + + /** + * @no-named-arguments + */ + public function withStreamInterceptors(StreamInterceptor ...$interceptors): self { $builder = clone $this; - $builder->interceptors = [ - ...$builder->interceptors, + $builder->streamInterceptors = [ + ...$builder->streamInterceptors, ...$interceptors, ]; @@ -189,14 +204,6 @@ public function withLoadBalancer(LoadBalancerFactory $factory): self return $builder; } - public function withRetry(Retry $policy): self - { - $builder = clone $this; - $builder->retry = $policy; - - return $builder; - } - public function withEndpointResolver(Scheme $scheme, EndpointResolver $resolver): self { $builder = clone $this; @@ -218,7 +225,6 @@ public function build(): Client $compressor = $this->compressor ?? IdentityCompressor::Compressor; $protobuf = $this->protobuf ?? Decoder\Builder::buildDefault(); $loadBalancerFactory = $this->loadBalancerFactory ?? new LoadBalancer\PickFirstFactory(); - $retry = $this->retry ?? Retry::disabled(); $tlsContext = $this->credentials?->createContext(); $uriFactory = new Http2\UriFactory($tlsContext !== null ? Internal\HttpScheme::Https : Internal\HttpScheme::Http); $transferTimeout = $this->transferTimeout; @@ -230,12 +236,21 @@ public function build(): Client Scheme::Ipv4, Scheme::Ipv6, Scheme::Unix => new EndpointResolver\StaticResolver(), }; - $interceptor = new Http2\InterceptorComposer([ - ...$this->interceptors, - new Http2\AppendControlMetadataInterceptor( - $encoder->name(), - $compressor->name(), - ), + $controlMetadata = new Http2\AppendControlMetadataInterceptor( + $encoder->name(), + $compressor->name(), + ); + + // Control metadata sits innermost (closest to the transport) so every user + // interceptor runs before the HTTP/2 headers are finalised. + $unary = new Http2\UnaryInterceptorComposer([ + ...$this->unaryInterceptors, + $controlMetadata, + ]); + + $stream = new Http2\StreamInterceptorComposer([ + ...$this->streamInterceptors, + $controlMetadata, ]); $httpclient = $this->httpclient ?? new HttpClientBuilder() @@ -259,7 +274,6 @@ public function build(): Client target: $target, resolver: $resolver, loadBalancerFactory: $loadBalancerFactory, - interceptor: $interceptor, streams: new Http2\StreamFactory( http: $httpclient, uri: $uriFactory, @@ -269,9 +283,10 @@ public function build(): Client encoder: $encoder, compressor: $compressor, ), - retry: $retry, ), ), + $unary, + $stream, ); } } diff --git a/packages/client/src/Client/CallableInterceptor.php b/packages/client/src/Client/CallableStreamInterceptor.php similarity index 81% rename from packages/client/src/Client/CallableInterceptor.php rename to packages/client/src/Client/CallableStreamInterceptor.php index 740f8a8..dc01c6b 100644 --- a/packages/client/src/Client/CallableInterceptor.php +++ b/packages/client/src/Client/CallableStreamInterceptor.php @@ -11,7 +11,7 @@ /** * @api */ -final readonly class CallableInterceptor implements Interceptor +final readonly class CallableStreamInterceptor implements StreamInterceptor { /** * @template In of object @@ -23,17 +23,17 @@ public function __construct( ) {} #[\Override] - public function intercept( + public function interceptStream( Invoke $invoke, Metadata $md, Cancellation $cancellation, - callable $next, + callable $newStream, ): ClientStream { return ($this->handler)( $invoke, $md, $cancellation, - $next, + $newStream, ); } } diff --git a/packages/client/src/Client/CallableUnaryInterceptor.php b/packages/client/src/Client/CallableUnaryInterceptor.php new file mode 100644 index 0000000..918b47e --- /dev/null +++ b/packages/client/src/Client/CallableUnaryInterceptor.php @@ -0,0 +1,40 @@ +, Metadata, Cancellation, callable(In, Invoke, Metadata, Cancellation): Out): Out $handler + */ + public function __construct( + private mixed $handler, + ) {} + + #[\Override] + public function interceptUnary( + object $request, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $invoker, + ): object { + return ($this->handler)( + $request, + $invoke, + $md, + $cancellation, + $invoker, + ); + } +} diff --git a/packages/client/src/Client/Internal/AmphpHttpClient.php b/packages/client/src/Client/Internal/AmphpHttpClient.php index b751c67..e0c7f2b 100644 --- a/packages/client/src/Client/Internal/AmphpHttpClient.php +++ b/packages/client/src/Client/Internal/AmphpHttpClient.php @@ -6,8 +6,14 @@ use Amp\Cancellation; use Amp\NullCancellation; +use Google\Rpc\Code; use Thesis\Grpc\Client; +use Thesis\Grpc\Client\Internal\Http2\StreamInterceptorComposer; +use Thesis\Grpc\Client\Internal\Http2\UnaryInterceptorComposer; +use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\ClientStream; +use Thesis\Grpc\GrpcException; +use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; /** @@ -17,6 +23,8 @@ { public function __construct( private Connection $connection, + private UnaryInterceptorComposer $unary, + private StreamInterceptorComposer $stream, ) {} #[\Override] @@ -26,11 +34,33 @@ public function invoke( Metadata $md = new Metadata(), Cancellation $cancellation = new NullCancellation(), ): object { - return $this->connection->invoke( + $pick = new PickContext($invoke->method, $md); + + return $this->unary->intercept( // @phpstan-ignore return.type $request, $invoke, $md, $cancellation, + function ( + object $request, + Client\Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + ) use ($pick): object { + try { + $stream = $this->connection->createStream($invoke, $md, $cancellation, $pick); + $stream->send($request); + $stream->close(); + + return $stream->receive(); + } catch (GrpcException $e) { + throw $e; + } catch (\Throwable $e) { + // Transport-level failures (e.g. a refused connection) map to UNAVAILABLE, + // so interceptors above see a gRPC status rather than a raw amphp exception. + throw new InvokeError(Code::UNAVAILABLE, $e->getMessage(), previous: $e); + } + }, ); } @@ -40,10 +70,17 @@ public function createStream( Metadata $md = new Metadata(), Cancellation $cancellation = new NullCancellation(), ): ClientStream { - return $this->connection->createStream( + $pick = new PickContext($invoke->method, $md); + + return $this->stream->intercept( // @phpstan-ignore return.type $invoke, $md, $cancellation, + fn( + Client\Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + ): ClientStream => $this->connection->createStream($invoke, $md, $cancellation, $pick), ); } diff --git a/packages/client/src/Client/Internal/Connection.php b/packages/client/src/Client/Internal/Connection.php index e516f8c..68a28a1 100644 --- a/packages/client/src/Client/Internal/Connection.php +++ b/packages/client/src/Client/Internal/Connection.php @@ -7,8 +7,8 @@ use Amp\Cancellation; use Amp\NullCancellation; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\ClientStream; -use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; /** @@ -16,21 +16,6 @@ */ interface Connection { - /** - * @template In of object - * @template Out of object - * @param In $request - * @param Invoke $invoke - * @return Out - * @throws InvokeError - */ - public function invoke( - object $request, - Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), - ): object; - /** * @template In of object * @template Out of object @@ -39,8 +24,9 @@ public function invoke( */ public function createStream( Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), + Metadata $md, + Cancellation $cancellation, + PickContext $pick, ): ClientStream; public function close(Cancellation $cancellation = new NullCancellation()): void; diff --git a/packages/client/src/Client/Internal/Connection/DefaultConnection.php b/packages/client/src/Client/Internal/Connection/DefaultConnection.php index 133fa9a..62c82f1 100644 --- a/packages/client/src/Client/Internal/Connection/DefaultConnection.php +++ b/packages/client/src/Client/Internal/Connection/DefaultConnection.php @@ -7,18 +7,15 @@ use Amp\Cancellation; use Amp\DeferredCancellation; use Amp\NullCancellation; -use Thesis\Grpc\Client\Endpoint; use Thesis\Grpc\Client\EndpointResolver; use Thesis\Grpc\Client\EndpointResolverListener; use Thesis\Grpc\Client\Internal\Connection; -use Thesis\Grpc\Client\Internal\Http2\InterceptorComposer; use Thesis\Grpc\Client\Internal\Http2\StreamFactory; use Thesis\Grpc\Client\Invoke; use Thesis\Grpc\Client\LoadBalancer; use Thesis\Grpc\Client\LoadBalancerFactory; use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\Client\Resolution; -use Thesis\Grpc\Client\Retry; use Thesis\Grpc\Client\Target; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; @@ -38,9 +35,7 @@ public function __construct( Target $target, EndpointResolver $resolver, LoadBalancerFactory $loadBalancerFactory, - private InterceptorComposer $interceptor, private StreamFactory $streams, - private Retry $retry, ) { $this->deferredCancellation = new DeferredCancellation(); @@ -53,46 +48,22 @@ public function __construct( $this->balancer = $loadBalancerFactory->create($resolution->endpoints); } - #[\Override] - public function invoke( - object $request, - Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), - ): object { - /** @var list $excluded */ - $excluded = []; - - return $this->retry->call( - function () use ( - &$excluded, - $request, - $invoke, - $md, - $cancellation, - ): object { - $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md, $excluded)); - $excluded[] = $endpoint; - - $stream = $this->stream($endpoint, $invoke, $md, $cancellation); - $stream->send($request); - $stream->close(); - - return $stream->receive(); - }, - $cancellation, - ); - } - #[\Override] public function createStream( Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), + Metadata $md, + Cancellation $cancellation, + PickContext $pick, ): ClientStream { - $endpoint = $this->balancer->pick(new PickContext($invoke->method, $md)); + $endpoint = $this->balancer->pick($pick); + $pick->exclude($endpoint); - return $this->stream($endpoint, $invoke, $md, $cancellation); + return $this->streams->create( + $invoke, + $endpoint->address, + $md, + $cancellation, + ); } #[\Override] @@ -108,29 +79,4 @@ public function onResolve(Resolution|\Throwable $result): void $this->balancer->refresh($result->endpoints); } } - - /** - * @template In of object - * @template Out of object - * @param Invoke $invoke - * @return ClientStream - */ - private function stream( - Endpoint $endpoint, - Invoke $invoke, - Metadata $md, - Cancellation $cancellation, - ): ClientStream { - return $this->interceptor->intercept( // @phpstan-ignore return.type - $invoke, - $md, - $cancellation, - fn(Invoke $invoke, Metadata $md, Cancellation $cancellation) => $this->streams->create( - $invoke, - $endpoint->address, - $md, - $cancellation, - ), - ); - } } diff --git a/packages/client/src/Client/Internal/Connection/LazyConnection.php b/packages/client/src/Client/Internal/Connection/LazyConnection.php index 31d7bf2..8428691 100644 --- a/packages/client/src/Client/Internal/Connection/LazyConnection.php +++ b/packages/client/src/Client/Internal/Connection/LazyConnection.php @@ -9,6 +9,7 @@ use Amp\NullCancellation; use Thesis\Grpc\Client\Internal\Connection; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; use function Amp\async; @@ -28,27 +29,16 @@ public function __construct( private readonly \Closure $factory, ) {} - #[\Override] - public function invoke( - object $request, - Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), - ): object { - return $this - ->createConnection($cancellation) - ->invoke($request, $invoke, $md, $cancellation); - } - #[\Override] public function createStream( Invoke $invoke, - Metadata $md = new Metadata(), - Cancellation $cancellation = new NullCancellation(), + Metadata $md, + Cancellation $cancellation, + PickContext $pick, ): ClientStream { return $this ->createConnection($cancellation) - ->createStream($invoke, $md, $cancellation); + ->createStream($invoke, $md, $cancellation, $pick); } #[\Override] diff --git a/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php b/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php index 543a8b4..a010547 100644 --- a/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php +++ b/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php @@ -5,15 +5,18 @@ namespace Thesis\Grpc\Client\Internal\Http2; use Amp\Cancellation; -use Thesis\Grpc\Client\Interceptor; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\StreamInterceptor; +use Thesis\Grpc\Client\UnaryInterceptor; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; /** * @internal */ -final readonly class AppendControlMetadataInterceptor implements Interceptor +final readonly class AppendControlMetadataInterceptor implements + UnaryInterceptor, + StreamInterceptor { /** * @param non-empty-string $encoding @@ -25,18 +28,32 @@ public function __construct( ) {} #[\Override] - public function intercept( + public function interceptUnary( + object $request, Invoke $invoke, Metadata $md, Cancellation $cancellation, - callable $next, + callable $invoker, + ): object { + return $invoker($request, $invoke, $this->decorate($md), $cancellation); + } + + #[\Override] + public function interceptStream( + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $newStream, ): ClientStream { - $md = $md + return $newStream($invoke, $this->decorate($md), $cancellation); + } + + private function decorate(Metadata $md): Metadata + { + return $md ->withKey(new Metadata\ContentType($this->encoding)) ->withKey(Metadata\UserAgent::Key) ->withKey(new Metadata\ContentEncoding($this->compression)) ->with('TE', 'trailers'); - - return $next($invoke, $md, $cancellation); } } diff --git a/packages/client/src/Client/Internal/Http2/InterceptorComposer.php b/packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php similarity index 75% rename from packages/client/src/Client/Internal/Http2/InterceptorComposer.php rename to packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php index c92b366..20a5cd5 100644 --- a/packages/client/src/Client/Internal/Http2/InterceptorComposer.php +++ b/packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php @@ -5,18 +5,18 @@ namespace Thesis\Grpc\Client\Internal\Http2; use Amp\Cancellation; -use Thesis\Grpc\Client\Interceptor; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\StreamInterceptor; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; /** * @internal */ -final readonly class InterceptorComposer +final readonly class StreamInterceptorComposer { /** - * @param list $interceptors + * @param list $interceptors */ public function __construct( private array $interceptors, @@ -26,28 +26,28 @@ public function __construct( * @template In of object * @template Out of object * @param Invoke $invoke - * @param callable(Invoke, Metadata, Cancellation): ClientStream $next + * @param callable(Invoke, Metadata, Cancellation): ClientStream $newStream * @return ClientStream */ public function intercept( Invoke $invoke, Metadata $md, Cancellation $cancellation, - callable $next, + callable $newStream, ): ClientStream { $handler = array_reduce( array_reverse($this->interceptors), - static fn(callable $stack, Interceptor $interceptor) => static fn( + static fn(callable $stack, StreamInterceptor $interceptor) => static fn( Invoke $invoke, Metadata $md, Cancellation $cancellation, - ) => $interceptor->intercept( + ) => $interceptor->interceptStream( $invoke, $md, $cancellation, $stack(...), // @phpstan-ignore argument.type ), - $next, + $newStream, ); /** @var ClientStream */ diff --git a/packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php b/packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php new file mode 100644 index 0000000..67796e6 --- /dev/null +++ b/packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php @@ -0,0 +1,59 @@ + $interceptors + */ + public function __construct( + private array $interceptors, + ) {} + + /** + * @template In of object + * @template Out of object + * @param In $request + * @param Invoke $invoke + * @param callable(In, Invoke, Metadata, Cancellation): Out $invoker + * @return Out + */ + public function intercept( + object $request, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $invoker, + ): object { + $handler = array_reduce( + array_reverse($this->interceptors), + static fn(callable $stack, UnaryInterceptor $interceptor) => static fn( + object $request, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + ) => $interceptor->interceptUnary( + $request, + $invoke, + $md, + $cancellation, + $stack(...), // @phpstan-ignore argument.type + ), + $invoker, + ); + + /** @var Out */ + return $handler($request, $invoke, $md, $cancellation); + } +} diff --git a/packages/client/src/Client/PickContext.php b/packages/client/src/Client/PickContext.php index edb2405..7e6ec36 100644 --- a/packages/client/src/Client/PickContext.php +++ b/packages/client/src/Client/PickContext.php @@ -9,15 +9,15 @@ /** * @api */ -final readonly class PickContext +final class PickContext { /** * @param non-empty-string $methodName * @param list $excluded endpoints that already failed this call and should be skipped if possible */ public function __construct( - public string $methodName, - public Metadata $metadata, + public readonly string $methodName, + public readonly Metadata $metadata, private array $excluded = [], ) {} @@ -25,4 +25,12 @@ public function excluded(Endpoint $endpoint): bool { return array_any($this->excluded, $endpoint->equals(...)); } + + /** + * Records the endpoint the transport just picked so a subsequent retry skips it. + */ + public function exclude(Endpoint $endpoint): void + { + $this->excluded[] = $endpoint; + } } diff --git a/packages/client/src/Client/Retry.php b/packages/client/src/Client/Retry.php deleted file mode 100644 index 018a15d..0000000 --- a/packages/client/src/Client/Retry.php +++ /dev/null @@ -1,102 +0,0 @@ - $retryableStatusCodes gRPC statuses that trigger a retry - */ - public function __construct( - private int $maxAttempts = 3, - private float $initialBackoff = 0.05, - private float $maxBackoff = 1.0, - private float $backoffMultiplier = 2.0, - private array $retryableStatusCodes = [ - Code::UNAVAILABLE, - ], - private Randomizer $randomizer = new Randomizer(), - ) {} - - public static function disabled(): self - { - return new self(maxAttempts: 1); - } - - /** - * @template T - * @param \Closure(): T $call - * @return T - * @throws GrpcException - */ - public function call(\Closure $call, Cancellation $cancellation): mixed - { - $error = null; - - for ($attempt = 1; $attempt <= $this->maxAttempts; ++$attempt) { - if ($attempt > 1) { - $this->backoff($attempt - 1, $cancellation); - } - - try { - return $call(); - } catch (\Throwable $e) { - if (!$this->shouldRetry($e)) { - throw self::createException($e); - } - - $error = $e; - } - } - - throw self::createException($error ?? new InvokeError(Code::UNAVAILABLE, 'no attempts were made')); - } - - private function shouldRetry(\Throwable $e): bool - { - if ($e instanceof InvokeError) { - return \in_array($e->statusCode, $this->retryableStatusCodes, true); - } - - return $e instanceof HttpException; - } - - /** - * @param positive-int $attempt - */ - private function backoff(int $attempt, Cancellation $cancellation): void - { - $ceiling = min($this->maxBackoff, $this->initialBackoff * $this->backoffMultiplier ** ($attempt - 1)); - - if ($ceiling > 0.0) { - delay($this->randomizer->getFloat(0.0, $ceiling), cancellation: $cancellation); - } - } - - private static function createException(\Throwable $e): GrpcException - { - if ($e instanceof InvokeError || $e instanceof ClientStreamIsClosed) { - return $e; - } - - return new InvokeError(Code::UNAVAILABLE, $e->getMessage(), previous: $e); - } -} diff --git a/packages/client/src/Client/Interceptor.php b/packages/client/src/Client/StreamInterceptor.php similarity index 79% rename from packages/client/src/Client/Interceptor.php rename to packages/client/src/Client/StreamInterceptor.php index d4db500..0df31ae 100644 --- a/packages/client/src/Client/Interceptor.php +++ b/packages/client/src/Client/StreamInterceptor.php @@ -11,19 +11,19 @@ /** * @api */ -interface Interceptor +interface StreamInterceptor { /** * @template In of object * @template Out of object * @param Invoke $invoke - * @param callable(Invoke, Metadata, Cancellation): ClientStream $next + * @param callable(Invoke, Metadata, Cancellation): ClientStream $newStream * @return ClientStream */ - public function intercept( + public function interceptStream( Invoke $invoke, Metadata $md, Cancellation $cancellation, - callable $next, + callable $newStream, ): ClientStream; } diff --git a/packages/client/src/Client/UnaryInterceptor.php b/packages/client/src/Client/UnaryInterceptor.php new file mode 100644 index 0000000..9fc2160 --- /dev/null +++ b/packages/client/src/Client/UnaryInterceptor.php @@ -0,0 +1,30 @@ + $invoke + * @param callable(In, Invoke, Metadata, Cancellation): Out $invoker + * @return Out + */ + public function interceptUnary( + object $request, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $invoker, + ): object; +} diff --git a/packages/server/README.md b/packages/server/README.md index fdd1941..c750853 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -113,22 +113,40 @@ $server = new Server\Builder() ## Interceptors -Interceptors let you apply cross-cutting server logic like auth, audit, tracing, and request validation around every RPC. +Interceptors let you apply cross-cutting server logic like auth, audit, tracing, and request validation around every RPC. Mirroring gRPC, they come in two flavours on separate chains: + +- `UnaryInterceptor` wraps a unary `request → response` call (via `withUnaryInterceptors()`), seeing the decoded request and response. +- `StreamInterceptor` wraps the `ServerStream` of the three streaming RPC types (via `withStreamInterceptors()`). + +A unary RPC runs only the unary chain, a streaming RPC only the stream chain. An interceptor that must guard both (e.g. auth) implements both interfaces and is registered on both chains. ```php use Amp\Cancellation; use Thesis\Grpc\Metadata; -use Thesis\Grpc\Server; use Thesis\Grpc\Server\StreamInfo; +use Thesis\Grpc\Server\StreamInterceptor; +use Thesis\Grpc\Server\UnaryInterceptor; use Thesis\Grpc\ServerStream; -final readonly class ServerAuthInterceptor implements Server\Interceptor +final readonly class ServerAuthInterceptor implements UnaryInterceptor, StreamInterceptor { - public function intercept(ServerStream $stream, StreamInfo $info, Metadata $md, Cancellation $cancellation, callable $next): void + public function interceptUnary(object $request, StreamInfo $info, Metadata $md, Cancellation $cancellation, callable $handler): object + { + return $handler($request, $info, $md, $cancellation); + } + + public function interceptStream(ServerStream $stream, StreamInfo $info, Metadata $md, Cancellation $cancellation, callable $next): void { $next($stream, $info, $md, $cancellation); } } + +$auth = new ServerAuthInterceptor(); + +$server = new Server\Builder() + ->withUnaryInterceptors($auth) + ->withStreamInterceptors($auth) + ->build(); ``` ## RPC types diff --git a/packages/server/src/Server/BidirectionalStreamHandler.php b/packages/server/src/Server/BidirectionalStreamHandler.php index 3667a72..3dc6cfa 100644 --- a/packages/server/src/Server/BidirectionalStreamHandler.php +++ b/packages/server/src/Server/BidirectionalStreamHandler.php @@ -12,9 +12,9 @@ * @api * @template TRequest of object * @template TResponse of object - * @template-implements Handler + * @template-implements StreamHandler */ -final readonly class BidirectionalStreamHandler implements Handler +final readonly class BidirectionalStreamHandler implements StreamHandler { /** * @param \Closure(BidirectionalStreamChannel, Metadata, Cancellation): void $handler diff --git a/packages/server/src/Server/Builder.php b/packages/server/src/Server/Builder.php index f1ae227..5e605aa 100644 --- a/packages/server/src/Server/Builder.php +++ b/packages/server/src/Server/Builder.php @@ -59,8 +59,11 @@ final class Builder /** @var list */ private array $middlewares = []; - /** @var list */ - private array $interceptors = []; + /** @var list */ + private array $unaryInterceptors = []; + + /** @var list */ + private array $streamInterceptors = []; /** @var list */ private array $services = []; @@ -159,11 +162,25 @@ public function withMiddlewares(Middleware ...$middlewares): self /** * @no-named-arguments */ - public function withInterceptors(Interceptor ...$interceptors): self + public function withUnaryInterceptors(UnaryInterceptor ...$interceptors): self + { + $builder = clone $this; + $builder->unaryInterceptors = [ + ...$builder->unaryInterceptors, + ...$interceptors, + ]; + + return $builder; + } + + /** + * @no-named-arguments + */ + public function withStreamInterceptors(StreamInterceptor ...$interceptors): self { $builder = clone $this; - $builder->interceptors = [ - ...$builder->interceptors, + $builder->streamInterceptors = [ + ...$builder->streamInterceptors, ...$interceptors, ]; @@ -379,7 +396,8 @@ public function build(): Server encoderFactory: new MessageEncoderFactory(array_values($this->encoders)), compressorFactory: new MessageCompressorFactory($compressors), protobuf: $this->protobuf ?? Protobuf\Encoder\Builder::buildDefault(), - interceptors: $this->interceptors, + unaryInterceptors: $this->unaryInterceptors, + streamInterceptors: $this->streamInterceptors, ), errorHandler: new ServerErrorHandler(), ); diff --git a/packages/server/src/Server/CallableInterceptor.php b/packages/server/src/Server/CallableStreamInterceptor.php similarity index 87% rename from packages/server/src/Server/CallableInterceptor.php rename to packages/server/src/Server/CallableStreamInterceptor.php index f343ebe..a144964 100644 --- a/packages/server/src/Server/CallableInterceptor.php +++ b/packages/server/src/Server/CallableStreamInterceptor.php @@ -11,7 +11,7 @@ /** * @api */ -final readonly class CallableInterceptor implements Interceptor +final readonly class CallableStreamInterceptor implements StreamInterceptor { /** * @template In of object @@ -23,7 +23,7 @@ public function __construct( ) {} #[\Override] - public function intercept( + public function interceptStream( ServerStream $stream, StreamInfo $info, Metadata $md, diff --git a/packages/server/src/Server/CallableUnaryInterceptor.php b/packages/server/src/Server/CallableUnaryInterceptor.php new file mode 100644 index 0000000..201fc87 --- /dev/null +++ b/packages/server/src/Server/CallableUnaryInterceptor.php @@ -0,0 +1,40 @@ +handler)( + $request, + $info, + $md, + $cancellation, + $handler, + ); + } +} diff --git a/packages/server/src/Server/ClientStreamHandler.php b/packages/server/src/Server/ClientStreamHandler.php index c1581c4..b8be952 100644 --- a/packages/server/src/Server/ClientStreamHandler.php +++ b/packages/server/src/Server/ClientStreamHandler.php @@ -12,9 +12,9 @@ * @api * @template TRequest of object * @template TResponse of object - * @template-implements Handler + * @template-implements StreamHandler */ -final readonly class ClientStreamHandler implements Handler +final readonly class ClientStreamHandler implements StreamHandler { /** * @param \Closure(ClientStreamChannel, Metadata, Cancellation): TResponse $handler diff --git a/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php b/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php index 8e8c472..bcdb409 100644 --- a/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php +++ b/packages/server/src/Server/Internal/Http2/ConcurrentServerStream.php @@ -35,6 +35,10 @@ public function __construct( #[\Override] public function send(object $message): void { + if ($this->send->isComplete()) { + throw new ServerStreamIsClosed(); + } + try { $this->send->push($message); } catch (Pipeline\DisposedException $e) { diff --git a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php b/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php index 9bad240..e8467bc 100644 --- a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php +++ b/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php @@ -16,11 +16,13 @@ use Amp\TimeoutCancellation; use Google\Rpc; use Thesis\Grpc\Metadata; -use Thesis\Grpc\Server\Interceptor; use Thesis\Grpc\Server\MessageCompressorFactory; use Thesis\Grpc\Server\MessageEncoderFactory; use Thesis\Grpc\Server\Service; use Thesis\Grpc\Server\StreamInfo; +use Thesis\Grpc\Server\StreamInterceptor; +use Thesis\Grpc\Server\UnaryHandler; +use Thesis\Grpc\Server\UnaryInterceptor; use Thesis\Grpc\ServerStream; use Thesis\Grpc\ServiceRegistrar; use Thesis\Grpc\UnimplementedException; @@ -37,26 +39,35 @@ final class ServerRequestHandler implements { private readonly Router $router; - private readonly InterceptorComposer $interceptor; + /** + * Always-on transport lifecycle: seeds the OK trailer, maps handler exceptions to a + * gRPC status, and closes the stream — for every RPC type. + */ + private readonly StreamHandleInterceptor $lifecycle; + + private readonly UnaryInterceptorComposer $unary; + + private readonly StreamInterceptorComposer $stream; /** @var \WeakMap, HandlerEntry> */ private \WeakMap $pending; /** - * @param list $interceptors + * @param list $unaryInterceptors + * @param list $streamInterceptors */ public function __construct( private readonly MessageEncoderFactory $encoderFactory, private readonly MessageCompressorFactory $compressorFactory, Protobuf\Encoder $protobuf, - array $interceptors, + array $unaryInterceptors, + array $streamInterceptors, ) { $this->pending = new \WeakMap(); $this->router = new Router(); - $this->interceptor = new InterceptorComposer([ - new StreamHandleInterceptor($protobuf), - ...$interceptors, - ]); + $this->lifecycle = new StreamHandleInterceptor($protobuf); + $this->unary = new UnaryInterceptorComposer($unaryInterceptors); + $this->stream = new StreamInterceptorComposer($streamInterceptors); } #[\Override] @@ -143,28 +154,61 @@ public function handleRequest(Request $request): Response $streamCancellation, ); - $handler = static fn( - ServerStream $stream, - StreamInfo $info, - Metadata $md, - Cancellation $cancellation, - ) => $rpc->handler->handle( - $stream, - $md, - $cancellation, - ); + $info = new StreamInfo($rpc->handle->method, $rpc->type); + $rpcHandler = $rpc->handler; + + $terminal = match (true) { + $rpcHandler instanceof UnaryHandler => function ( + ServerStream $stream, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + ) use ($rpcHandler): void { + $response = $this->unary->intercept( + $stream->receive(), + $info, + $md, + $cancellation, + static fn( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + ): object => $rpcHandler->invoke($request, $md, $cancellation), + ); + + $stream->send($response); + $stream->close(); + }, + default => function ( + ServerStream $stream, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + ) use ($rpcHandler): void { + $this->stream->intercept( + $stream, + $info, + $md, + $cancellation, + static fn( + ServerStream $stream, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + ) => $rpcHandler->handle($stream, $md, $cancellation), + ); + }, + }; /** @var Future $future */ $future = async( - $this->interceptor->intercept(...), + $this->lifecycle->interceptStream(...), $stream, - new StreamInfo( - $rpc->handle->method, - $rpc->type, - ), + $info, $md, $streamCancellation, - $handler, + $terminal, ); $future->ignore(); diff --git a/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php b/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php index 6e16441..23cf53d 100644 --- a/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php +++ b/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php @@ -9,8 +9,8 @@ use Google\Rpc; use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; -use Thesis\Grpc\Server\Interceptor; use Thesis\Grpc\Server\StreamInfo; +use Thesis\Grpc\Server\StreamInterceptor; use Thesis\Grpc\ServerStream; use Thesis\Grpc\Status; use Thesis\Protobuf; @@ -18,7 +18,7 @@ /** * @internal */ -final readonly class StreamHandleInterceptor implements Interceptor +final readonly class StreamHandleInterceptor implements StreamInterceptor { private Metadata $ok; @@ -29,7 +29,7 @@ public function __construct( } #[\Override] - public function intercept( + public function interceptStream( ServerStream $stream, StreamInfo $info, Metadata $md, diff --git a/packages/server/src/Server/Internal/Http2/InterceptorComposer.php b/packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php similarity index 82% rename from packages/server/src/Server/Internal/Http2/InterceptorComposer.php rename to packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php index d126f43..11c57fc 100644 --- a/packages/server/src/Server/Internal/Http2/InterceptorComposer.php +++ b/packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php @@ -8,17 +8,17 @@ use Amp\CancelledException; use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; -use Thesis\Grpc\Server\Interceptor; use Thesis\Grpc\Server\StreamInfo; +use Thesis\Grpc\Server\StreamInterceptor; use Thesis\Grpc\ServerStream; /** * @internal */ -final readonly class InterceptorComposer +final readonly class StreamInterceptorComposer { /** - * @param list $interceptors + * @param list $interceptors */ public function __construct( private array $interceptors, @@ -41,12 +41,12 @@ public function intercept( ): void { $handler = array_reduce( array_reverse($this->interceptors), - static fn(callable $stack, Interceptor $interceptor) => static fn( + static fn(callable $stack, StreamInterceptor $interceptor) => static fn( ServerStream $stream, StreamInfo $info, Metadata $md, Cancellation $cancellation, - ) => $interceptor->intercept( + ) => $interceptor->interceptStream( $stream, $info, $md, diff --git a/packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php b/packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php new file mode 100644 index 0000000..ee660ba --- /dev/null +++ b/packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php @@ -0,0 +1,62 @@ + $interceptors + */ + public function __construct( + private array $interceptors, + ) {} + + /** + * @template In of object + * @template Out of object + * @param In $request + * @param callable(In, StreamInfo, Metadata, Cancellation): Out $handler + * @return Out + * @throws InvokeError + * @throws CancelledException + */ + public function intercept( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + callable $handler, + ): object { + $stack = array_reduce( + array_reverse($this->interceptors), + static fn(callable $next, UnaryInterceptor $interceptor) => static fn( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + ) => $interceptor->interceptUnary( + $request, + $info, + $md, + $cancellation, + $next, // @phpstan-ignore argument.type + ), + $handler, + ); + + /** @var Out */ + return $stack($request, $info, $md, $cancellation); + } +} diff --git a/packages/server/src/Server/Rpc.php b/packages/server/src/Server/Rpc.php index 10e394c..ecd79d8 100644 --- a/packages/server/src/Server/Rpc.php +++ b/packages/server/src/Server/Rpc.php @@ -13,11 +13,11 @@ * @template In of object * @template Out of object * @param Handle $handle - * @param Handler $handler + * @param StreamHandler|UnaryHandler $handler a streaming handler, or a unary request → response handler */ public function __construct( public Handle $handle, - public Handler $handler, + public StreamHandler|UnaryHandler $handler, public RpcType $type, ) {} } diff --git a/packages/server/src/Server/ServerStreamHandler.php b/packages/server/src/Server/ServerStreamHandler.php index f4aaaf5..c88cb37 100644 --- a/packages/server/src/Server/ServerStreamHandler.php +++ b/packages/server/src/Server/ServerStreamHandler.php @@ -12,9 +12,9 @@ * @api * @template TRequest of object * @template TResponse of object - * @template-implements Handler + * @template-implements StreamHandler */ -final readonly class ServerStreamHandler implements Handler +final readonly class ServerStreamHandler implements StreamHandler { /** * @param \Closure(TRequest, Metadata, Cancellation): iterable $handler diff --git a/packages/server/src/Server/Handler.php b/packages/server/src/Server/StreamHandler.php similarity index 93% rename from packages/server/src/Server/Handler.php rename to packages/server/src/Server/StreamHandler.php index 489b1b5..73a0a68 100644 --- a/packages/server/src/Server/Handler.php +++ b/packages/server/src/Server/StreamHandler.php @@ -13,7 +13,7 @@ * @template In of object * @template Out of object */ -interface Handler +interface StreamHandler { /** * @param ServerStream $stream diff --git a/packages/server/src/Server/Interceptor.php b/packages/server/src/Server/StreamInterceptor.php similarity index 90% rename from packages/server/src/Server/Interceptor.php rename to packages/server/src/Server/StreamInterceptor.php index ae0ac03..48686a3 100644 --- a/packages/server/src/Server/Interceptor.php +++ b/packages/server/src/Server/StreamInterceptor.php @@ -13,7 +13,7 @@ /** * @api */ -interface Interceptor +interface StreamInterceptor { /** * @template In of object @@ -23,7 +23,7 @@ interface Interceptor * @throws InvokeError * @throws CancelledException */ - public function intercept( + public function interceptStream( ServerStream $stream, StreamInfo $info, Metadata $md, diff --git a/packages/server/src/Server/UnaryHandler.php b/packages/server/src/Server/UnaryHandler.php index 21ff983..3ceaadf 100644 --- a/packages/server/src/Server/UnaryHandler.php +++ b/packages/server/src/Server/UnaryHandler.php @@ -6,15 +6,13 @@ use Amp\Cancellation; use Thesis\Grpc\Metadata; -use Thesis\Grpc\ServerStream; /** * @api * @template TRequest of object * @template TResponse of object - * @template-implements Handler */ -final readonly class UnaryHandler implements Handler +final readonly class UnaryHandler { /** * @param \Closure(TRequest, Metadata, Cancellation): TResponse $handler @@ -23,12 +21,12 @@ public function __construct( private \Closure $handler, ) {} - #[\Override] - public function handle(ServerStream $stream, Metadata $md, Cancellation $cancellation): void + /** + * @param TRequest $request + * @return TResponse + */ + public function invoke(object $request, Metadata $md, Cancellation $cancellation): object { - $request = $stream->receive(); - $response = ($this->handler)($request, $md, $cancellation); - $stream->send($response); - $stream->close(); + return ($this->handler)($request, $md, $cancellation); } } diff --git a/packages/server/src/Server/UnaryInterceptor.php b/packages/server/src/Server/UnaryInterceptor.php new file mode 100644 index 0000000..2c64f06 --- /dev/null +++ b/packages/server/src/Server/UnaryInterceptor.php @@ -0,0 +1,33 @@ +server = new Server\Builder() ->withServices(new FileServiceServerRegistry(new ClientStreamServer())) - ->withInterceptors(new CallableInterceptor(static function ( + ->withStreamInterceptors(new CallableStreamInterceptor(static function ( ServerStream $stream, StreamInfo $info, Metadata $md, diff --git a/tests/DecoratedStreamTest.php b/tests/DecoratedStreamTest.php new file mode 100644 index 0000000..1f85b86 --- /dev/null +++ b/tests/DecoratedStreamTest.php @@ -0,0 +1,93 @@ +server = new Server\Builder() + ->withServices(new FileServiceServerRegistry(new SummingFileServer())) + ->build(); + + $this->server->start(); + } + + protected function tearDown(): void + { + $this->server->stop(); + } + + public function testStreamInterceptorWrapsTheClientStreamWithADecorator(): void + { + $client = new FileServiceClient( + new Client\Builder() + ->withStreamInterceptors(new Client\CallableStreamInterceptor( + static fn(Invoke $invoke, Metadata $md, Cancellation $cancellation, callable $next): ClientStream => new DoublingClientStream($next( + $invoke, + $md, + $cancellation, + )), + )) + ->build(), + ); + + $stream = $client->upload(); + + for ($i = 0; $i < 10; ++$i) { + $stream->send(new Chunk(random_bytes(10))); + } + + // Each 10-byte chunk is doubled on the way out by the decorator, so the server sees 200 bytes. + $info = $stream->close(); + self::assertSame(200, $info->size); + } +} + +final readonly class SummingFileServer implements FileServiceServer +{ + #[\Override] + public function upload(Server\ClientStreamChannel $stream, Metadata $md, Cancellation $cancellation): FileInfo + { + $size = 0; + + /** @var Chunk $chunk */ + foreach ($stream as $chunk) { + $size += \strlen($chunk->content); + } + + return new FileInfo($size); + } +} + +/** + * @template-extends Client\DecoratedStream + */ +final readonly class DoublingClientStream extends Client\DecoratedStream +{ + #[\Override] + public function send(object $message): void + { + parent::send(new Chunk("{$message->content}{$message->content}")); + } +} diff --git a/tests/RetryFailoverTest.php b/tests/RetryFailoverTest.php deleted file mode 100644 index 574c001..0000000 --- a/tests/RetryFailoverTest.php +++ /dev/null @@ -1,106 +0,0 @@ -server = new Server\Builder() - ->withAddresses('0.0.0.0:50052') - ->withServices(new EchoServiceServerRegistry(new SucceedingEchoServer())) - ->build(); - - $this->server->start(); - } - - protected function tearDown(): void - { - $this->server->stop(); - } - - public function testRetryFailsOverFromADeadEndpointToALiveOne(): void - { - $client = new EchoServiceClient( - new Client\Builder() - ->withHost(\sprintf('ipv4:%s,%s', self::DEAD, self::LIVE)) - ->withLoadBalancer(new RoundRobinFactory()) - ->withRetry(new Retry(maxAttempts: 3, initialBackoff: 0.0)) - ->build(), - ); - - for ($i = 0; $i < 10; ++$i) { - self::assertSame('ok', $client->echo(new EchoRequest('ping'))->sentence); - } - } - - public function testWithoutRetryADeadEndpointSurfacesAsUnavailable(): void - { - $client = new EchoServiceClient( - new Client\Builder() - ->withHost(\sprintf('ipv4:%s', self::DEAD)) - ->build(), - ); - - try { - $client->echo(new EchoRequest('ping')); - self::fail('Expected an InvokeError for the dead endpoint.'); - } catch (InvokeError $e) { - self::assertSame(Code::UNAVAILABLE, $e->statusCode); - } - } - - public function testRetryDoesNotHelpWhenEveryEndpointIsDead(): void - { - $client = new EchoServiceClient( - new Client\Builder() - ->withHost(\sprintf('ipv4:%s,127.0.0.1:50053', self::DEAD)) - ->withLoadBalancer(new RoundRobinFactory()) - ->withRetry(new Retry(maxAttempts: 3, initialBackoff: 0.0)) - ->build(), - ); - - $this->expectException(InvokeError::class); - $client->echo(new EchoRequest('ping')); - } -} - -/** - * @internal - */ -final readonly class SucceedingEchoServer implements EchoServiceServer -{ - #[\Override] - public function echo(EchoRequest $request, Metadata $md, Cancellation $cancellation): EchoResponse - { - return new EchoResponse('ok'); - } -} diff --git a/tests/Stub/AuthorizationClientInterceptor.php b/tests/Stub/AuthorizationClientInterceptor.php index 3568e15..b9cbee5 100644 --- a/tests/Stub/AuthorizationClientInterceptor.php +++ b/tests/Stub/AuthorizationClientInterceptor.php @@ -5,24 +5,38 @@ namespace Thesis\Grpc\Stub; use Amp\Cancellation; -use Thesis\Grpc\Client; use Thesis\Grpc\Client\Invoke; +use Thesis\Grpc\Client\StreamInterceptor; +use Thesis\Grpc\Client\UnaryInterceptor; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Metadata; -final readonly class AuthorizationClientInterceptor implements Client\Interceptor +final readonly class AuthorizationClientInterceptor implements + UnaryInterceptor, + StreamInterceptor { public function __construct( private string $password, ) {} #[\Override] - public function intercept( + public function interceptUnary( + object $request, Invoke $invoke, Metadata $md, Cancellation $cancellation, - callable $next, + callable $invoker, + ): object { + return $invoker($request, $invoke, $md->with('Authorization', $this->password), $cancellation); + } + + #[\Override] + public function interceptStream( + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $newStream, ): ClientStream { - return $next($invoke, $md->with('Authorization', $this->password), $cancellation); + return $newStream($invoke, $md->with('Authorization', $this->password), $cancellation); } } diff --git a/tests/Stub/AuthorizationServerInterceptor.php b/tests/Stub/AuthorizationServerInterceptor.php index f08c8d5..2a4cf15 100644 --- a/tests/Stub/AuthorizationServerInterceptor.php +++ b/tests/Stub/AuthorizationServerInterceptor.php @@ -8,28 +8,49 @@ use Google\Rpc\Code; use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; -use Thesis\Grpc\Server; use Thesis\Grpc\Server\StreamInfo; +use Thesis\Grpc\Server\StreamInterceptor; +use Thesis\Grpc\Server\UnaryInterceptor; use Thesis\Grpc\ServerStream; -final readonly class AuthorizationServerInterceptor implements Server\Interceptor +final readonly class AuthorizationServerInterceptor implements + UnaryInterceptor, + StreamInterceptor { public function __construct( private string $password, ) {} #[\Override] - public function intercept( + public function interceptUnary( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + callable $handler, + ): object { + $this->authorize($md); + + return $handler($request, $info, $md, $cancellation); + } + + #[\Override] + public function interceptStream( ServerStream $stream, StreamInfo $info, Metadata $md, Cancellation $cancellation, callable $next, ): void { + $this->authorize($md); + + $next($stream, $info, $md, $cancellation); + } + + private function authorize(Metadata $md): void + { if ($md->value('Authorization') !== $this->password) { throw new InvokeError(Code::UNAUTHENTICATED, 'Use authorization, Luke!'); } - - $next($stream, $info, $md, $cancellation); } } diff --git a/tests/UnaryTest.php b/tests/UnaryTest.php index 6f3acd1..cf82e4b 100644 --- a/tests/UnaryTest.php +++ b/tests/UnaryTest.php @@ -32,7 +32,7 @@ protected function setUp(): void { $this->server = new Server\Builder() ->withServices(new EchoServiceServerRegistry(new UnaryEchoServer())) - ->withInterceptors(new AuthorizationServerInterceptor('secret')) + ->withUnaryInterceptors(new AuthorizationServerInterceptor('secret')) ->build(); $this->server->start(); @@ -47,7 +47,7 @@ public function testAuthorizationInterceptor(): void { $client = new EchoServiceClient( new Client\Builder() - ->withInterceptors(new AuthorizationClientInterceptor('secret')) + ->withUnaryInterceptors(new AuthorizationClientInterceptor('secret')) ->build(), ); @@ -70,7 +70,7 @@ public function testPassMetadata(): void { $client = new EchoServiceClient( new Client\Builder() - ->withInterceptors(new AuthorizationClientInterceptor('secret')) + ->withUnaryInterceptors(new AuthorizationClientInterceptor('secret')) ->build(), ); @@ -78,17 +78,18 @@ public function testPassMetadata(): void self::assertSame($sentence, $response->sentence); } - public function testWrapClientStream(): void + public function testRewriteRequest(): void { $client = new EchoServiceClient( new Client\Builder() - ->withInterceptors( + ->withUnaryInterceptors( new AuthorizationClientInterceptor('secret'), - new Client\CallableInterceptor(static fn(Invoke $invoke, Metadata $metadata, Cancellation $cancellation, callable $next): ClientStream => new MitmClientStream($next( + new Client\CallableUnaryInterceptor(static fn(object $request, Invoke $invoke, Metadata $metadata, Cancellation $cancellation, callable $invoker): object => $invoker( + new EchoRequest('pong'), $invoke, $metadata, $cancellation, - ))), + )), ) ->build(), ); @@ -101,7 +102,7 @@ public function testServerHandlerException(): void { $client = new EchoServiceClient( new Client\Builder() - ->withInterceptors(new AuthorizationClientInterceptor('secret')) + ->withUnaryInterceptors(new AuthorizationClientInterceptor('secret')) ->build(), ); @@ -142,20 +143,3 @@ public function echo( return new EchoResponse($sentence); } } - -/** - * @template-extends Client\DecoratedStream - */ -final readonly class MitmClientStream extends Client\DecoratedStream -{ - public function __construct(ClientStream $stream) - { - parent::__construct($stream); - } - - #[\Override] - public function send(object $message): void - { - parent::send(new EchoRequest('pong')); - } -} diff --git a/tests/genproto/Chat/Api/V1/Message.php b/tests/genproto/Chat/Api/V1/Message.php index a87199e..4420c94 100644 --- a/tests/genproto/Chat/Api/V1/Message.php +++ b/tests/genproto/Chat/Api/V1/Message.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceClient.php b/tests/genproto/Chat/Api/V1/MessengerServiceClient.php index 18fe027..5ae62af 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceClient.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceServer.php b/tests/genproto/Chat/Api/V1/MessengerServiceServer.php index f63e609..000b53a 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceServer.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php b/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php index 76e6fe3..3bed1a0 100644 --- a/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php +++ b/tests/genproto/Chat/Api/V1/MessengerServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php b/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php index c085d87..bb566fa 100644 --- a/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php +++ b/tests/genproto/Chat/Api/V1/TestsProtosChatV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/chat_v1.proto */ diff --git a/tests/genproto/Chat/Api/V1/autoload.metadata.php b/tests/genproto/Chat/Api/V1/autoload.metadata.php index 1972cd2..d2f5e97 100644 --- a/tests/genproto/Chat/Api/V1/autoload.metadata.php +++ b/tests/genproto/Chat/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 */ diff --git a/tests/genproto/Echos/Api/V1/EchoRequest.php b/tests/genproto/Echos/Api/V1/EchoRequest.php index 784b54b..571d14f 100644 --- a/tests/genproto/Echos/Api/V1/EchoRequest.php +++ b/tests/genproto/Echos/Api/V1/EchoRequest.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoResponse.php b/tests/genproto/Echos/Api/V1/EchoResponse.php index 05773b7..f25fc76 100644 --- a/tests/genproto/Echos/Api/V1/EchoResponse.php +++ b/tests/genproto/Echos/Api/V1/EchoResponse.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceClient.php b/tests/genproto/Echos/Api/V1/EchoServiceClient.php index 1fbb5ff..29de9c1 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceClient.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceServer.php b/tests/genproto/Echos/Api/V1/EchoServiceServer.php index 53ba7c3..4004832 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceServer.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php b/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php index aa3cc35..c55aac2 100644 --- a/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php +++ b/tests/genproto/Echos/Api/V1/EchoServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php b/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php index 684402b..5a60d83 100644 --- a/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php +++ b/tests/genproto/Echos/Api/V1/TestsProtosEchoV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/echo_v1.proto */ diff --git a/tests/genproto/Echos/Api/V1/autoload.metadata.php b/tests/genproto/Echos/Api/V1/autoload.metadata.php index 8dbdfe7..ee2e6c0 100644 --- a/tests/genproto/Echos/Api/V1/autoload.metadata.php +++ b/tests/genproto/Echos/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 */ diff --git a/tests/genproto/File/Api/V1/Chunk.php b/tests/genproto/File/Api/V1/Chunk.php index 5810e3a..25dda34 100644 --- a/tests/genproto/File/Api/V1/Chunk.php +++ b/tests/genproto/File/Api/V1/Chunk.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileInfo.php b/tests/genproto/File/Api/V1/FileInfo.php index bbfc773..6385557 100644 --- a/tests/genproto/File/Api/V1/FileInfo.php +++ b/tests/genproto/File/Api/V1/FileInfo.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceClient.php b/tests/genproto/File/Api/V1/FileServiceClient.php index e0ad74f..8d46d98 100644 --- a/tests/genproto/File/Api/V1/FileServiceClient.php +++ b/tests/genproto/File/Api/V1/FileServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceServer.php b/tests/genproto/File/Api/V1/FileServiceServer.php index 77b3eff..4c6e0d2 100644 --- a/tests/genproto/File/Api/V1/FileServiceServer.php +++ b/tests/genproto/File/Api/V1/FileServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/FileServiceServerRegistry.php b/tests/genproto/File/Api/V1/FileServiceServerRegistry.php index c2e2ce8..c816b9d 100644 --- a/tests/genproto/File/Api/V1/FileServiceServerRegistry.php +++ b/tests/genproto/File/Api/V1/FileServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php b/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php index fe6fe75..4d2dd65 100644 --- a/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php +++ b/tests/genproto/File/Api/V1/TestsProtosFileV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/file_v1.proto */ diff --git a/tests/genproto/File/Api/V1/autoload.metadata.php b/tests/genproto/File/Api/V1/autoload.metadata.php index da6e84e..541a955 100644 --- a/tests/genproto/File/Api/V1/autoload.metadata.php +++ b/tests/genproto/File/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 */ diff --git a/tests/genproto/Topic/Api/V1/Event.php b/tests/genproto/Topic/Api/V1/Event.php index b77df80..78ba771 100644 --- a/tests/genproto/Topic/Api/V1/Event.php +++ b/tests/genproto/Topic/Api/V1/Event.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/SubscribeRequest.php b/tests/genproto/Topic/Api/V1/SubscribeRequest.php index e3fabf6..aff0917 100644 --- a/tests/genproto/Topic/Api/V1/SubscribeRequest.php +++ b/tests/genproto/Topic/Api/V1/SubscribeRequest.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php b/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php index a9d884f..c3f58af 100644 --- a/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php +++ b/tests/genproto/Topic/Api/V1/TestsProtosTopicV1DescriptorRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceClient.php b/tests/genproto/Topic/Api/V1/TopicServiceClient.php index 184dd04..895755d 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceClient.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceClient.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceServer.php b/tests/genproto/Topic/Api/V1/TopicServiceServer.php index 3e09a3d..9fed33c 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceServer.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceServer.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php b/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php index 2693d39..7198a66 100644 --- a/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php +++ b/tests/genproto/Topic/Api/V1/TopicServiceServerRegistry.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 * Source: tests/protos/topic_v1.proto */ diff --git a/tests/genproto/Topic/Api/V1/autoload.metadata.php b/tests/genproto/Topic/Api/V1/autoload.metadata.php index 85988a7..9c39487 100644 --- a/tests/genproto/Topic/Api/V1/autoload.metadata.php +++ b/tests/genproto/Topic/Api/V1/autoload.metadata.php @@ -3,7 +3,7 @@ /** * Code generated by thesis/protoc-plugin. DO NOT EDIT. * Versions: - * thesis/protoc-plugin — v0.1.22 + * thesis/protoc-plugin — v0.1.24 * protoc — v6.33.5 */ From 7c1fe76e07255787126104593507fce49b69b893 Mon Sep 17 00:00:00 2001 From: kafkiansky Date: Mon, 3 Aug 2026 10:37:04 +0300 Subject: [PATCH 4/4] change interceptor's path --- packages/client/src/Client/Builder.php | 6 +++--- packages/client/src/Client/Internal/AmphpHttpClient.php | 2 -- .../{Http2 => }/AppendControlMetadataInterceptor.php | 2 +- .../Internal/{Http2 => }/StreamInterceptorComposer.php | 2 +- .../Internal/{Http2 => }/UnaryInterceptorComposer.php | 2 +- .../src/Server/Internal/Http2/ServerRequestHandler.php | 3 +++ .../Server/Internal/{Http2 => }/StreamHandleInterceptor.php | 2 +- .../Internal/{Http2 => }/StreamInterceptorComposer.php | 2 +- .../Internal/{Http2 => }/UnaryInterceptorComposer.php | 2 +- 9 files changed, 12 insertions(+), 11 deletions(-) rename packages/client/src/Client/Internal/{Http2 => }/AppendControlMetadataInterceptor.php (96%) rename packages/client/src/Client/Internal/{Http2 => }/StreamInterceptorComposer.php (96%) rename packages/client/src/Client/Internal/{Http2 => }/UnaryInterceptorComposer.php (97%) rename packages/server/src/Server/Internal/{Http2 => }/StreamHandleInterceptor.php (97%) rename packages/server/src/Server/Internal/{Http2 => }/StreamInterceptorComposer.php (97%) rename packages/server/src/Server/Internal/{Http2 => }/UnaryInterceptorComposer.php (97%) diff --git a/packages/client/src/Client/Builder.php b/packages/client/src/Client/Builder.php index 05a4574..1ec6f47 100644 --- a/packages/client/src/Client/Builder.php +++ b/packages/client/src/Client/Builder.php @@ -236,19 +236,19 @@ public function build(): Client Scheme::Ipv4, Scheme::Ipv6, Scheme::Unix => new EndpointResolver\StaticResolver(), }; - $controlMetadata = new Http2\AppendControlMetadataInterceptor( + $controlMetadata = new Internal\AppendControlMetadataInterceptor( $encoder->name(), $compressor->name(), ); // Control metadata sits innermost (closest to the transport) so every user // interceptor runs before the HTTP/2 headers are finalised. - $unary = new Http2\UnaryInterceptorComposer([ + $unary = new Internal\UnaryInterceptorComposer([ ...$this->unaryInterceptors, $controlMetadata, ]); - $stream = new Http2\StreamInterceptorComposer([ + $stream = new Internal\StreamInterceptorComposer([ ...$this->streamInterceptors, $controlMetadata, ]); diff --git a/packages/client/src/Client/Internal/AmphpHttpClient.php b/packages/client/src/Client/Internal/AmphpHttpClient.php index e0c7f2b..c4d6025 100644 --- a/packages/client/src/Client/Internal/AmphpHttpClient.php +++ b/packages/client/src/Client/Internal/AmphpHttpClient.php @@ -8,8 +8,6 @@ use Amp\NullCancellation; use Google\Rpc\Code; use Thesis\Grpc\Client; -use Thesis\Grpc\Client\Internal\Http2\StreamInterceptorComposer; -use Thesis\Grpc\Client\Internal\Http2\UnaryInterceptorComposer; use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\ClientStream; use Thesis\Grpc\GrpcException; diff --git a/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php b/packages/client/src/Client/Internal/AppendControlMetadataInterceptor.php similarity index 96% rename from packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php rename to packages/client/src/Client/Internal/AppendControlMetadataInterceptor.php index a010547..aaade6c 100644 --- a/packages/client/src/Client/Internal/Http2/AppendControlMetadataInterceptor.php +++ b/packages/client/src/Client/Internal/AppendControlMetadataInterceptor.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Client\Internal\Http2; +namespace Thesis\Grpc\Client\Internal; use Amp\Cancellation; use Thesis\Grpc\Client\Invoke; diff --git a/packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php b/packages/client/src/Client/Internal/StreamInterceptorComposer.php similarity index 96% rename from packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php rename to packages/client/src/Client/Internal/StreamInterceptorComposer.php index 20a5cd5..5cc36fd 100644 --- a/packages/client/src/Client/Internal/Http2/StreamInterceptorComposer.php +++ b/packages/client/src/Client/Internal/StreamInterceptorComposer.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Client\Internal\Http2; +namespace Thesis\Grpc\Client\Internal; use Amp\Cancellation; use Thesis\Grpc\Client\Invoke; diff --git a/packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php b/packages/client/src/Client/Internal/UnaryInterceptorComposer.php similarity index 97% rename from packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php rename to packages/client/src/Client/Internal/UnaryInterceptorComposer.php index 67796e6..5625120 100644 --- a/packages/client/src/Client/Internal/Http2/UnaryInterceptorComposer.php +++ b/packages/client/src/Client/Internal/UnaryInterceptorComposer.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Client\Internal\Http2; +namespace Thesis\Grpc\Client\Internal; use Amp\Cancellation; use Thesis\Grpc\Client\Invoke; diff --git a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php b/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php index e8467bc..dc3a4bb 100644 --- a/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php +++ b/packages/server/src/Server/Internal/Http2/ServerRequestHandler.php @@ -16,6 +16,9 @@ use Amp\TimeoutCancellation; use Google\Rpc; use Thesis\Grpc\Metadata; +use Thesis\Grpc\Server\Internal\StreamHandleInterceptor; +use Thesis\Grpc\Server\Internal\StreamInterceptorComposer; +use Thesis\Grpc\Server\Internal\UnaryInterceptorComposer; use Thesis\Grpc\Server\MessageCompressorFactory; use Thesis\Grpc\Server\MessageEncoderFactory; use Thesis\Grpc\Server\Service; diff --git a/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php b/packages/server/src/Server/Internal/StreamHandleInterceptor.php similarity index 97% rename from packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php rename to packages/server/src/Server/Internal/StreamHandleInterceptor.php index 23cf53d..b24e15b 100644 --- a/packages/server/src/Server/Internal/Http2/StreamHandleInterceptor.php +++ b/packages/server/src/Server/Internal/StreamHandleInterceptor.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Server\Internal\Http2; +namespace Thesis\Grpc\Server\Internal; use Amp\Cancellation; use Amp\CancelledException; diff --git a/packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php b/packages/server/src/Server/Internal/StreamInterceptorComposer.php similarity index 97% rename from packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php rename to packages/server/src/Server/Internal/StreamInterceptorComposer.php index 11c57fc..3b3b289 100644 --- a/packages/server/src/Server/Internal/Http2/StreamInterceptorComposer.php +++ b/packages/server/src/Server/Internal/StreamInterceptorComposer.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Server\Internal\Http2; +namespace Thesis\Grpc\Server\Internal; use Amp\Cancellation; use Amp\CancelledException; diff --git a/packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php b/packages/server/src/Server/Internal/UnaryInterceptorComposer.php similarity index 97% rename from packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php rename to packages/server/src/Server/Internal/UnaryInterceptorComposer.php index ee660ba..5e3f0b3 100644 --- a/packages/server/src/Server/Internal/Http2/UnaryInterceptorComposer.php +++ b/packages/server/src/Server/Internal/UnaryInterceptorComposer.php @@ -2,7 +2,7 @@ declare(strict_types=1); -namespace Thesis\Grpc\Server\Internal\Http2; +namespace Thesis\Grpc\Server\Internal; use Amp\Cancellation; use Amp\CancelledException;