diff --git a/.github/workflows/check.yaml b/.github/workflows/check.yaml index a7d7bb1..88a3ecc 100644 --- a/.github/workflows/check.yaml +++ b/.github/workflows/check.yaml @@ -41,6 +41,13 @@ jobs: - uses: actions/checkout@v6 - run: PHP_IMAGE_VERSION=${{ matrix.php }} make ${{ matrix.deps == 'lowest' && 'install-lowest' || '' }} test + interop: + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@v6 + - run: make interop + # infect: # runs-on: ubuntu-latest # steps: diff --git a/.php-cs-fixer.dist.php b/.php-cs-fixer.dist.php index a382c06..9a4b8f2 100644 --- a/.php-cs-fixer.dist.php +++ b/.php-cs-fixer.dist.php @@ -14,6 +14,8 @@ ->in(__DIR__ . '/packages/client/src') ->in(__DIR__ . '/packages/server/src') ->in(__DIR__ . '/tests') + ->in(__DIR__ . '/interop/src') + ->in(__DIR__ . '/interop/bin') ->exclude([ 'genproto', ]) diff --git a/Makefile b/Makefile index 90a318a..0c79691 100644 --- a/Makefile +++ b/Makefile @@ -107,6 +107,10 @@ test: var vendor compile-test-stub ## Run tests using PHPUnit $(RUN) vendor/bin/phpunit $(ARGS) --colors .PHONY: test +interop: var vendor ## Run gRPC interop tests against grpc-go + $(DOCKER_COMPOSE) --profile interop run --rm --build interop php interop/bin/matrix $(ARGS) +.PHONY: interop + infect: var vendor up ## Run mutation tests using Infection $(RUN) infection --show-mutations $(ARGS) .PHONY: infect @@ -130,7 +134,7 @@ composer-normalize-check: ## Check that composer.json is normalized fix: fixer rector composer-normalize ## Run all fixing recipes .PHONY: fix -check: fixer-check rector-check composer-validate composer-normalize-check deps-analyze phpstan test ## Run all project checks +check: fixer-check rector-check composer-validate composer-normalize-check deps-analyze phpstan test interop ## Run all project checks .PHONY: check compile-test-stub: @@ -143,6 +147,29 @@ compile-test-stub: --php-plugin_out=tests/genproto \ tests/protos/*.proto .PHONY: compile-test-stub + +compile-interop-stub: fetch-interop-protos + docker run --rm \ + --pull always \ + --user $(CONTAINER_USER) \ + -v $(PWD):/workspace \ + -w /workspace \ + ghcr.io/thesis-php/protoc-plugin:latest \ + -I interop/protos \ + --php-plugin_out=interop/genproto \ + interop/protos/src/proto/grpc/testing/*.proto +.PHONY: compile-interop-stub + +INTEROP_GRPC_COMMIT = 816d41355f831fdaec27e3d4cb4e5d3fe0d55637 + +fetch-interop-protos: ## Fetch interop protos and TLS test credentials from grpc/grpc at INTEROP_GRPC_COMMIT + for f in empty messages test; do \ + curl --fail --silent --location --output interop/protos/src/proto/grpc/testing/$$f.proto https://raw.githubusercontent.com/grpc/grpc/$(INTEROP_GRPC_COMMIT)/src/proto/grpc/testing/$$f.proto || exit 1; \ + done + for f in ca.pem server1.pem server1.key; do \ + curl --fail --silent --location --output interop/tls/$$f https://raw.githubusercontent.com/grpc/grpc/$(INTEROP_GRPC_COMMIT)/src/core/tsi/test_creds/$$f || exit 1; \ + done +.PHONY: fetch-interop-protos # ----------------------- help: diff --git a/compose.yaml b/compose.yaml index c059af9..4d813fe 100644 --- a/compose.yaml +++ b/compose.yaml @@ -13,3 +13,18 @@ services: - .:/workspace:cached working_dir: /workspace network_mode: host + + interop: + build: + context: interop + args: + PHP_IMAGE_VERSION: ${PHP_IMAGE_VERSION} + user: ${CONTAINER_USER:-} + environment: + XDEBUG_MODE: 'off' + init: true + volumes: + - .:/workspace:cached + working_dir: /workspace + network_mode: host + profiles: [interop] diff --git a/composer.json b/composer.json index 8bcb69d..a3de822 100644 --- a/composer.json +++ b/composer.json @@ -27,7 +27,7 @@ "amphp/http": "^2.1", "amphp/http-client": "^5.3", "amphp/http-server": "^3.4.7", - "amphp/pipeline": "^1.2", + "amphp/pipeline": "^1.2.1", "amphp/socket": "^2.3", "amphp/sync": "^2.3", "league/uri": "^7.8", @@ -41,6 +41,8 @@ }, "require-dev": { "ext-bcmath": "*", + "amphp/hpack": "^3.2", + "amphp/process": "^2.1", "phpunit/phpunit": "^12.4", "symfony/var-dumper": "^8.0", "thesis/grpc-client": "*@dev", @@ -82,14 +84,17 @@ "Chat\\Api\\V1\\": "tests/genproto/Chat/Api/V1/", "Echos\\Api\\V1\\": "tests/genproto/Echos/Api/V1/", "File\\Api\\V1\\": "tests/genproto/File/Api/V1/", + "Grpc\\Testing\\": "interop/genproto/Grpc/Testing/", "Thesis\\Grpc\\": "tests/", + "Thesis\\Grpc\\Interop\\": "interop/src/", "Topic\\Api\\V1\\": "tests/genproto/Topic/Api/V1/" }, "files": [ "tests/genproto/Echos/Api/V1/autoload.metadata.php", "tests/genproto/File/Api/V1/autoload.metadata.php", "tests/genproto/Topic/Api/V1/autoload.metadata.php", - "tests/genproto/Chat/Api/V1/autoload.metadata.php" + "tests/genproto/Chat/Api/V1/autoload.metadata.php", + "interop/genproto/Grpc/Testing/autoload.metadata.php" ] }, "config": { diff --git a/interop/Dockerfile b/interop/Dockerfile new file mode 100644 index 0000000..584a154 --- /dev/null +++ b/interop/Dockerfile @@ -0,0 +1,11 @@ +ARG PHP_IMAGE_VERSION + +FROM golang:1.25-alpine AS grpc-go +ARG GRPC_GO_VERSION=v1.84.0 +RUN CGO_ENABLED=0 go install \ + google.golang.org/grpc/interop/client@${GRPC_GO_VERSION} \ + google.golang.org/grpc/interop/server@${GRPC_GO_VERSION} + +FROM ghcr.io/phpyh/php:${PHP_IMAGE_VERSION} +COPY --from=grpc-go /go/bin/client /usr/local/bin/grpc-go-interop-client +COPY --from=grpc-go /go/bin/server /usr/local/bin/grpc-go-interop-server diff --git a/interop/bin/client b/interop/bin/client new file mode 100755 index 0000000..f8f3861 --- /dev/null +++ b/interop/bin/client @@ -0,0 +1,71 @@ +#!/usr/bin/env php + 'localhost', + 'server_host_override' => '', + 'server_port' => '10000', + 'test_case' => 'large_unary', + 'use_tls' => 'false', + 'use_test_ca' => 'false', + 'additional_metadata' => '', + 'soak_iterations' => '10', + 'soak_max_failures' => '0', + 'soak_per_iteration_max_acceptable_latency_ms' => '1000', + 'soak_overall_timeout_seconds' => '10', + 'soak_min_time_ms_between_rpcs' => '0', + 'soak_request_size' => '271828', + 'soak_response_size' => '314159', + 'soak_num_threads' => '1', +]); + +$host = $flags->string('server_host'); +$address = (str_contains($host, ':') ? "[{$host}]" : $host) . ":{$flags->int('server_port')}"; +$metadata = AdditionalMetadata::parse($flags->string('additional_metadata')); + +$builder = new Client\Builder() + ->withHost($address) + ->withInterceptors($metadata); + +if ($flags->bool('use_tls')) { + $credentials = new Client\TransportCredentials()->withPeerName($flags->string('server_host_override') !== '' ? $flags->string('server_host_override') : $host); + + if ($flags->bool('use_test_ca')) { + $credentials = $credentials->withCaCert(__DIR__ . '/../tls/ca.pem'); + } + + $builder = $builder->withTransportCredentials($credentials); +} + +$cases = new TestCases($builder->build(...), $flags, $address); + +match ($flags->string('test_case')) { + 'empty_unary' => $cases->emptyUnary(), + 'large_unary' => $cases->largeUnary(), + 'client_streaming' => $cases->clientStreaming(), + 'server_streaming' => $cases->serverStreaming(), + 'ping_pong' => $cases->pingPong(), + 'empty_stream' => $cases->emptyStream(), + 'status_code_and_message' => $cases->statusCodeAndMessage(), + 'special_status_message' => $cases->specialStatusMessage(), + 'unimplemented_method' => $cases->unimplementedMethod(), + 'unimplemented_service' => $cases->unimplementedService(), + 'cancel_after_begin' => $cases->cancelAfterBegin(), + 'cancel_after_first_response' => $cases->cancelAfterFirstResponse(), + 'timeout_on_sleeping_server' => $cases->timeoutOnSleepingServer(), + 'custom_metadata' => $cases->customMetadata(), + 'rpc_soak' => $cases->rpcSoak(), + 'channel_soak' => $cases->channelSoak(), + default => throw new InvalidArgumentException("Unsupported test case \"{$flags->string('test_case')}\"."), +}; + +echo "{$flags->string('test_case')}: passed\n"; diff --git a/interop/bin/matrix b/interop/bin/matrix new file mode 100755 index 0000000..bcd0f2e --- /dev/null +++ b/interop/bin/matrix @@ -0,0 +1,195 @@ +#!/usr/bin/env php + + */ +function server(string $server, bool $tls): array +{ + $flags = ['--port=' . PORT, '--use_tls=' . var_export($tls, true)]; + + return match ($server) { + 'go' => ['grpc-go-interop-server', ...$flags, '--tls_cert_file=' . TLS . '/server1.pem', '--tls_key_file=' . TLS . '/server1.key'], + default => [\PHP_BINARY, 'interop/bin/server', ...$flags], + }; +} + +/** + * @return list + */ +function client(string $client, bool $tls, string $case): array +{ + $flags = ['--server_host=localhost', '--server_port=' . PORT, "--test_case={$case}", '--use_tls=' . var_export($tls, true)]; + + if ($tls) { + $flags = [...$flags, '--use_test_ca=true', '--server_host_override=foo.test.google.fr']; + } + + return match ($client) { + 'go' => ['grpc-go-interop-client', ...$flags, ...($tls ? ['--ca_file=' . TLS . '/ca.pem'] : [])], + default => [\PHP_BINARY, 'interop/bin/client', ...$flags], + }; +} + +/** + * @return Future + */ +function output(Process $process): Future +{ + /** @var Future $stderr */ + $stderr = async(static fn() => buffer($process->getStderr())); + + /** @var Future */ + return async(static fn() => buffer($process->getStdout()) . $stderr->await()); +} + +function listening(): bool +{ + try { + connect('localhost:' . PORT)->close(); + + return true; + } catch (ConnectException) { + return false; + } +} + +/** + * @param Future $output + */ +function awaitServer(Process $process, Future $output): void +{ + for ($i = 0; $i < 100; ++$i) { + if (!$process->isRunning()) { + throw new RuntimeException("The server exited:\n" . tail($output->await())); + } + + if (listening()) { + return; + } + + delay(0.1); + } + + $process->kill(); + + throw new RuntimeException('The server did not start listening on port ' . PORT . ":\n" . tail($output->await())); +} + +function tail(string $output): string +{ + return implode("\n", array_slice(explode("\n", $output), -11)); +} + +// Runs every test case for each client → server pair in plaintext and TLS, prints the matrix and fails on any failed case. + +chdir(dirname(__DIR__, 2)); + +@mkdir(LOGS, recursive: true); + +$columns = []; +$matrix = []; +$failed = []; + +foreach (PAIRS as [$client, $server]) { + foreach (['plaintext' => false, 'tls' => true] as $security => $tls) { + $column = "{$client}→{$server} {$security}"; + $columns[] = $column; + + // Otherwise a foreign server would answer for ours while ours fails to bind. + if (listening()) { + throw new RuntimeException('Port ' . PORT . ' is already in use.'); + } + + $process = Process::start(server($server, $tls)); + $serverOutput = output($process); + awaitServer($process, $serverOutput); + + foreach (CASES as $case) { + $key = "{$client}-{$server}/{$security}/{$case}"; + $log = LOGS . '/' . str_replace('/', '-', $key) . '.log'; + + $start = hrtime(true); + $caseProcess = Process::start(client($client, $tls, $case)); + $output = output($caseProcess); + + try { + $exit = (string) $caseProcess->join(new TimeoutCancellation(CASE_TIMEOUT)); + } catch (CancelledException) { + $caseProcess->kill(); + $exit = 'timeout'; + } + + $elapsed = (hrtime(true) - $start) / 1_000_000_000; + file_put_contents($log, $output->await()); + $matrix[$case][$column] = $exit === '0' ? 'pass' : 'FAIL'; + + if ($exit !== '0') { + $failed[$key] = $log; + } + + fwrite(\STDERR, sprintf("%s: %s (exit %s, %.1fs)\n", $key, $matrix[$case][$column], $exit, $elapsed)); + } + + $process->signal(\SIGTERM); + $process->join(); + file_put_contents(LOGS . "/server-{$server}-{$security}.log", $serverOutput->await()); + } +} + +$width = max(array_map(strlen(...), CASES)); +printf("\n%-{$width}s %s\n", 'case', implode(' ', array_map(static fn(string $column) => mb_str_pad($column, 20), $columns))); + +foreach ($matrix as $case => $results) { + printf("%-{$width}s %s\n", $case, implode(' ', array_map(static fn(string $result) => str_pad($result, 20), $results))); +} + +foreach ($failed as $key => $log) { + printf("\n%s (%s):\n%s", $key, $log, tail((string) file_get_contents($log))); +} + +exit($failed === [] ? 0 : 1); diff --git a/interop/bin/server b/interop/bin/server new file mode 100755 index 0000000..bdd4924 --- /dev/null +++ b/interop/bin/server @@ -0,0 +1,38 @@ +#!/usr/bin/env php + '10000', + 'use_tls' => 'false', +]); + +$builder = new Server\Builder() + ->withAddresses("0.0.0.0:{$flags->int('port')}", "[::]:{$flags->int('port')}") + ->withServices(new TestServiceServerRegistry(new TestService())) + ->withUnaryInterceptors(new EchoMetadata()); + +if ($flags->bool('use_tls')) { + $builder = $builder->withTransportCredentials( + new Server\TransportCredentials() + ->withDefaultCertificate(new Certificate(__DIR__ . '/../tls/server1.pem', __DIR__ . '/../tls/server1.key')), + ); +} + +$server = $builder->build(); +$server->start(); + +trapSignal([\SIGINT, \SIGTERM]); + +$server->stop(); diff --git a/interop/genproto/Grpc/Testing/BoolValue.php b/interop/genproto/Grpc/Testing/BoolValue.php new file mode 100644 index 0000000..743cd99 --- /dev/null +++ b/interop/genproto/Grpc/Testing/BoolValue.php @@ -0,0 +1,33 @@ + $types The types of RPCs the client sends. + * @param list<\Grpc\Testing\ClientConfigureRequest\Metadata> $metadata The collection of custom metadata to be attached to RPCs sent by the client. + * @param int $timeoutSec The deadline to use, in seconds, for all RPCs. If unset or zero, the + * client will use the default from the command-line. + */ + public function __construct( + #[Reflection\Field(1, new Reflection\ListT(new Reflection\EnumT(\Grpc\Testing\ClientConfigureRequest\RpcType::class)))] + public array $types = [], + #[Reflection\Field(2, new Reflection\ListT(new Reflection\ObjectT(\Grpc\Testing\ClientConfigureRequest\Metadata::class)))] + public array $metadata = [], + #[Reflection\Field(3, Reflection\Int32T::T)] + public int $timeoutSec = 0, + ) {} +} diff --git a/interop/genproto/Grpc/Testing/ClientConfigureRequest/Metadata.php b/interop/genproto/Grpc/Testing/ClientConfigureRequest/Metadata.php new file mode 100644 index 0000000..18d74dc --- /dev/null +++ b/interop/genproto/Grpc/Testing/ClientConfigureRequest/Metadata.php @@ -0,0 +1,32 @@ +add(Registry\Descriptor::base64(self::EMPTY_DESCRIPTOR_BUFFER), new File( + name: 'src/proto/grpc/testing/empty.proto', + messages: [ + new File\MessageDescriptor('grpc.testing.Empty', \Grpc\Testing\Empty_::class), + ], + )); + + $pool->add(Registry\Descriptor::base64(self::MESSAGES_DESCRIPTOR_BUFFER), new File( + name: 'src/proto/grpc/testing/messages.proto', + messages: [ + new File\MessageDescriptor('grpc.testing.BoolValue', \Grpc\Testing\BoolValue::class), + new File\MessageDescriptor('grpc.testing.Payload', \Grpc\Testing\Payload::class), + new File\MessageDescriptor('grpc.testing.EchoStatus', \Grpc\Testing\EchoStatus::class), + new File\MessageDescriptor('grpc.testing.SimpleRequest', \Grpc\Testing\SimpleRequest::class), + new File\MessageDescriptor('grpc.testing.SimpleResponse', \Grpc\Testing\SimpleResponse::class), + new File\MessageDescriptor('grpc.testing.StreamingInputCallRequest', \Grpc\Testing\StreamingInputCallRequest::class), + new File\MessageDescriptor('grpc.testing.StreamingInputCallResponse', \Grpc\Testing\StreamingInputCallResponse::class), + new File\MessageDescriptor('grpc.testing.ResponseParameters', \Grpc\Testing\ResponseParameters::class), + new File\MessageDescriptor('grpc.testing.StreamingOutputCallRequest', \Grpc\Testing\StreamingOutputCallRequest::class), + new File\MessageDescriptor('grpc.testing.StreamingOutputCallResponse', \Grpc\Testing\StreamingOutputCallResponse::class), + new File\MessageDescriptor('grpc.testing.ReconnectParams', \Grpc\Testing\ReconnectParams::class), + new File\MessageDescriptor('grpc.testing.ReconnectInfo', \Grpc\Testing\ReconnectInfo::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsRequest', \Grpc\Testing\LoadBalancerStatsRequest::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsResponse', \Grpc\Testing\LoadBalancerStatsResponse::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsResponse.MetadataEntry', \Grpc\Testing\LoadBalancerStatsResponse\MetadataEntry::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsResponse.RpcMetadata', \Grpc\Testing\LoadBalancerStatsResponse\RpcMetadata::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsResponse.MetadataByPeer', \Grpc\Testing\LoadBalancerStatsResponse\MetadataByPeer::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerStatsResponse.RpcsByPeer', \Grpc\Testing\LoadBalancerStatsResponse\RpcsByPeer::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerAccumulatedStatsRequest', \Grpc\Testing\LoadBalancerAccumulatedStatsRequest::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerAccumulatedStatsResponse', \Grpc\Testing\LoadBalancerAccumulatedStatsResponse::class), + new File\MessageDescriptor('grpc.testing.LoadBalancerAccumulatedStatsResponse.MethodStats', \Grpc\Testing\LoadBalancerAccumulatedStatsResponse\MethodStats::class), + new File\MessageDescriptor('grpc.testing.ClientConfigureRequest', \Grpc\Testing\ClientConfigureRequest::class), + new File\MessageDescriptor('grpc.testing.ClientConfigureRequest.Metadata', \Grpc\Testing\ClientConfigureRequest\Metadata::class), + new File\MessageDescriptor('grpc.testing.ClientConfigureResponse', \Grpc\Testing\ClientConfigureResponse::class), + new File\MessageDescriptor('grpc.testing.MemorySize', \Grpc\Testing\MemorySize::class), + new File\MessageDescriptor('grpc.testing.TestOrcaReport', \Grpc\Testing\TestOrcaReport::class), + new File\MessageDescriptor('grpc.testing.SetReturnStatusRequest', \Grpc\Testing\SetReturnStatusRequest::class), + new File\MessageDescriptor('grpc.testing.HookRequest', \Grpc\Testing\HookRequest::class), + new File\MessageDescriptor('grpc.testing.HookResponse', \Grpc\Testing\HookResponse::class), + ], + enums: [ + new File\EnumDescriptor('grpc.testing.PayloadType', \Grpc\Testing\PayloadType::class), + new File\EnumDescriptor('grpc.testing.GrpclbRouteType', \Grpc\Testing\GrpclbRouteType::class), + new File\EnumDescriptor('grpc.testing.LoadBalancerStatsResponse.MetadataType', \Grpc\Testing\LoadBalancerStatsResponse\MetadataType::class), + new File\EnumDescriptor('grpc.testing.ClientConfigureRequest.RpcType', \Grpc\Testing\ClientConfigureRequest\RpcType::class), + new File\EnumDescriptor('grpc.testing.HookRequest.HookRequestCommand', \Grpc\Testing\HookRequest\HookRequestCommand::class), + ], + )); + + $pool->add(Registry\Descriptor::base64(self::TEST_DESCRIPTOR_BUFFER), new File( + name: 'src/proto/grpc/testing/test.proto', + dependencies: [ + 'src/proto/grpc/testing/empty.proto', + 'src/proto/grpc/testing/messages.proto', + ], + services: [ + new File\ServiceDescriptor( + name: 'grpc.testing.TestService', + methods: [ + new File\MethodDescriptor('EmptyCall', false, false), + new File\MethodDescriptor('UnaryCall', false, false), + new File\MethodDescriptor('CacheableUnaryCall', false, false), + new File\MethodDescriptor('StreamingOutputCall', false, true), + new File\MethodDescriptor('StreamingInputCall', true, false), + new File\MethodDescriptor('FullDuplexCall', true, true), + new File\MethodDescriptor('HalfDuplexCall', true, true), + new File\MethodDescriptor('UnimplementedCall', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.UnimplementedService', + methods: [ + new File\MethodDescriptor('UnimplementedCall', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.ReconnectService', + methods: [ + new File\MethodDescriptor('Start', false, false), + new File\MethodDescriptor('Stop', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.LoadBalancerStatsService', + methods: [ + new File\MethodDescriptor('GetClientStats', false, false), + new File\MethodDescriptor('GetClientAccumulatedStats', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.HookService', + methods: [ + new File\MethodDescriptor('Hook', false, false), + new File\MethodDescriptor('SetReturnStatus', false, false), + new File\MethodDescriptor('ClearReturnStatus', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.XdsUpdateHealthService', + methods: [ + new File\MethodDescriptor('SetServing', false, false), + new File\MethodDescriptor('SetNotServing', false, false), + new File\MethodDescriptor('SendHookRequest', false, false), + ], + ), + new File\ServiceDescriptor( + name: 'grpc.testing.XdsUpdateClientConfigureService', + methods: [ + new File\MethodDescriptor('Configure', false, false), + ], + ), + ], + )); + } +} diff --git a/interop/genproto/Grpc/Testing/EchoStatus.php b/interop/genproto/Grpc/Testing/EchoStatus.php new file mode 100644 index 0000000..64522e4 --- /dev/null +++ b/interop/genproto/Grpc/Testing/EchoStatus.php @@ -0,0 +1,31 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.HookService/Hook', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * Sets a return status for pending and upcoming calls to Hook + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function setReturnStatus( + \Grpc\Testing\SetReturnStatusRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\Empty_ { + /** @var Client\Invoke<\Grpc\Testing\SetReturnStatusRequest, \Grpc\Testing\Empty_> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.HookService/SetReturnStatus', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * Clears the return status. Incoming calls to Hook will "hang" + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function clearReturnStatus( + \Grpc\Testing\Empty_ $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\Empty_ { + /** @var Client\Invoke<\Grpc\Testing\Empty_, \Grpc\Testing\Empty_> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.HookService/ClearReturnStatus', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/HookServiceServer.php b/interop/genproto/Grpc/Testing/HookServiceServer.php new file mode 100644 index 0000000..d236b80 --- /dev/null +++ b/interop/genproto/Grpc/Testing/HookServiceServer.php @@ -0,0 +1,52 @@ +server->hook(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('SetReturnStatus', \Grpc\Testing\SetReturnStatusRequest::class), + handler: new Server\UnaryHandler($this->server->setReturnStatus(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('ClearReturnStatus', \Grpc\Testing\Empty_::class), + handler: new Server\UnaryHandler($this->server->clearReturnStatus(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsRequest.php b/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsRequest.php new file mode 100644 index 0000000..b3052c6 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsRequest.php @@ -0,0 +1,20 @@ + $numRpcsStartedByMethod The total number of RPCs have ever issued for each type. + * Deprecated: use stats_per_method.rpcs_started instead. + * @param Protobuf\Map $numRpcsSucceededByMethod The total number of RPCs have ever completed successfully for each type. + * Deprecated: use stats_per_method.result instead. + * @param Protobuf\Map $numRpcsFailedByMethod The total number of RPCs have ever failed for each type. + * Deprecated: use stats_per_method.result instead. + * @param Protobuf\Map $statsPerMethod Per-method RPC statistics. The key is the RpcType in string form; e.g. + * 'EMPTY_CALL' or 'UNARY_CALL' + */ + public function __construct( + /** @deprecated */ + #[Reflection\Field(1, new Reflection\MapT(Reflection\StringT::T, Reflection\Int32T::T))] + public Protobuf\Map $numRpcsStartedByMethod = new Protobuf\Map(), + /** @deprecated */ + #[Reflection\Field(2, new Reflection\MapT(Reflection\StringT::T, Reflection\Int32T::T))] + public Protobuf\Map $numRpcsSucceededByMethod = new Protobuf\Map(), + /** @deprecated */ + #[Reflection\Field(3, new Reflection\MapT(Reflection\StringT::T, Reflection\Int32T::T))] + public Protobuf\Map $numRpcsFailedByMethod = new Protobuf\Map(), + #[Reflection\Field( + 4, + new Reflection\MapT( + Reflection\StringT::T, + new Reflection\ObjectT(\Grpc\Testing\LoadBalancerAccumulatedStatsResponse\MethodStats::class), + ), + )] + public Protobuf\Map $statsPerMethod = new Protobuf\Map(), + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsResponse/MethodStats.php b/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsResponse/MethodStats.php new file mode 100644 index 0000000..610b5e0 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerAccumulatedStatsResponse/MethodStats.php @@ -0,0 +1,34 @@ + $result The number of RPCs that completed with each status for this method. The + * key is the integral value of a google.rpc.Code; the value is the count. + */ + public function __construct( + #[Reflection\Field(1, Reflection\Int32T::T)] + public int $rpcsStarted = 0, + #[Reflection\Field(2, new Reflection\MapT(Reflection\Int32T::T, Reflection\Int32T::T))] + public Protobuf\Map $result = new Protobuf\Map(), + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsRequest.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsRequest.php new file mode 100644 index 0000000..a19dae0 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsRequest.php @@ -0,0 +1,37 @@ + $metadataKeys Response header + trailer metadata entries we want the values of. + * Matching of the keys is case-insensitive as per rfc7540#section-8.1.2 + * * (asterisk) is a special value that will return all metadata entries + */ + public function __construct( + #[Reflection\Field(1, Reflection\Int32T::T)] + public int $numRpcs = 0, + #[Reflection\Field(2, Reflection\Int32T::T)] + public int $timeoutSec = 0, + #[Reflection\Field(3, new Reflection\ListT(Reflection\StringT::T))] + public array $metadataKeys = [], + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse.php new file mode 100644 index 0000000..ef78364 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse.php @@ -0,0 +1,51 @@ + $rpcsByPeer The number of completed RPCs for each peer. + * @param int $numFailures The number of RPCs that failed to record a remote peer. + * @param Protobuf\Map $rpcsByMethod + * @param Protobuf\Map $metadatasByPeer All the metadata of all RPCs for each peer. + */ + public function __construct( + #[Reflection\Field(1, new Reflection\MapT(Reflection\StringT::T, Reflection\Int32T::T))] + public Protobuf\Map $rpcsByPeer = new Protobuf\Map(), + #[Reflection\Field(2, Reflection\Int32T::T)] + public int $numFailures = 0, + #[Reflection\Field( + 3, + new Reflection\MapT( + Reflection\StringT::T, + new Reflection\ObjectT(\Grpc\Testing\LoadBalancerStatsResponse\RpcsByPeer::class), + ), + )] + public Protobuf\Map $rpcsByMethod = new Protobuf\Map(), + #[Reflection\Field( + 4, + new Reflection\MapT( + Reflection\StringT::T, + new Reflection\ObjectT(\Grpc\Testing\LoadBalancerStatsResponse\MetadataByPeer::class), + ), + )] + public Protobuf\Map $metadatasByPeer = new Protobuf\Map(), + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataByPeer.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataByPeer.php new file mode 100644 index 0000000..a23d31d --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataByPeer.php @@ -0,0 +1,29 @@ + $rpcMetadata List of RpcMetadata in for each RPC with a given peer + */ + public function __construct( + #[Reflection\Field(1, new Reflection\ListT(new Reflection\ObjectT(\Grpc\Testing\LoadBalancerStatsResponse\RpcMetadata::class)))] + public array $rpcMetadata = [], + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataEntry.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataEntry.php new file mode 100644 index 0000000..612cf90 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/MetadataEntry.php @@ -0,0 +1,36 @@ + $metadata metadata values for each rpc for the keys specified in + * LoadBalancerStatsRequest.metadata_keys. + */ + public function __construct( + #[Reflection\Field(1, new Reflection\ListT(new Reflection\ObjectT(\Grpc\Testing\LoadBalancerStatsResponse\MetadataEntry::class)))] + public array $metadata = [], + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/RpcsByPeer.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/RpcsByPeer.php new file mode 100644 index 0000000..4f6d343 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsResponse/RpcsByPeer.php @@ -0,0 +1,30 @@ + $rpcsByPeer The number of completed RPCs for each peer. + */ + public function __construct( + #[Reflection\Field(1, new Reflection\MapT(Reflection\StringT::T, Reflection\Int32T::T))] + public Protobuf\Map $rpcsByPeer = new Protobuf\Map(), + ) {} +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceClient.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceClient.php new file mode 100644 index 0000000..653482f --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceClient.php @@ -0,0 +1,83 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.LoadBalancerStatsService/GetClientStats', + output: \Grpc\Testing\LoadBalancerStatsResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * Gets the accumulated stats for RPCs sent by a test client. + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function getClientAccumulatedStats( + \Grpc\Testing\LoadBalancerAccumulatedStatsRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\LoadBalancerAccumulatedStatsResponse { + /** @var Client\Invoke<\Grpc\Testing\LoadBalancerAccumulatedStatsRequest, \Grpc\Testing\LoadBalancerAccumulatedStatsResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.LoadBalancerStatsService/GetClientAccumulatedStats', + output: \Grpc\Testing\LoadBalancerAccumulatedStatsResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceServer.php b/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceServer.php new file mode 100644 index 0000000..98264b0 --- /dev/null +++ b/interop/genproto/Grpc/Testing/LoadBalancerStatsServiceServer.php @@ -0,0 +1,42 @@ +server->getClientStats(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('GetClientAccumulatedStats', \Grpc\Testing\LoadBalancerAccumulatedStatsRequest::class), + handler: new Server\UnaryHandler($this->server->getClientAccumulatedStats(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/MemorySize.php b/interop/genproto/Grpc/Testing/MemorySize.php new file mode 100644 index 0000000..880698c --- /dev/null +++ b/interop/genproto/Grpc/Testing/MemorySize.php @@ -0,0 +1,26 @@ + $backoffMs + */ + public function __construct( + #[Reflection\Field(1, Reflection\BoolT::T)] + public bool $passed = false, + #[Reflection\Field(2, new Reflection\ListT(Reflection\Int32T::T))] + public array $backoffMs = [], + ) {} +} diff --git a/interop/genproto/Grpc/Testing/ReconnectParams.php b/interop/genproto/Grpc/Testing/ReconnectParams.php new file mode 100644 index 0000000..e72d417 --- /dev/null +++ b/interop/genproto/Grpc/Testing/ReconnectParams.php @@ -0,0 +1,29 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.ReconnectService/Start', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function stop( + \Grpc\Testing\Empty_ $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\ReconnectInfo { + /** @var Client\Invoke<\Grpc\Testing\Empty_, \Grpc\Testing\ReconnectInfo> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.ReconnectService/Stop', + output: \Grpc\Testing\ReconnectInfo::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/ReconnectServiceServer.php b/interop/genproto/Grpc/Testing/ReconnectServiceServer.php new file mode 100644 index 0000000..511c861 --- /dev/null +++ b/interop/genproto/Grpc/Testing/ReconnectServiceServer.php @@ -0,0 +1,36 @@ +server->start(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('Stop', \Grpc\Testing\Empty_::class), + handler: new Server\UnaryHandler($this->server->stop(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/ResponseParameters.php b/interop/genproto/Grpc/Testing/ResponseParameters.php new file mode 100644 index 0000000..8218d40 --- /dev/null +++ b/interop/genproto/Grpc/Testing/ResponseParameters.php @@ -0,0 +1,45 @@ + $responseParameters Configuration for each expected response message. + * @param ?\Grpc\Testing\Payload $payload Optional input payload sent along with the request. + * @param ?\Grpc\Testing\EchoStatus $responseStatus Whether server should return a given status + * @param ?\Grpc\Testing\TestOrcaReport $orcaOobReport If set the server should update this metrics report data at the OOB server. + */ + public function __construct( + #[Reflection\Field(1, new Reflection\EnumT(\Grpc\Testing\PayloadType::class))] + public \Grpc\Testing\PayloadType $responseType = \Grpc\Testing\PayloadType::COMPRESSABLE, + #[Reflection\Field(2, new Reflection\ListT(new Reflection\ObjectT(\Grpc\Testing\ResponseParameters::class)))] + public array $responseParameters = [], + #[Reflection\Field(3, new Reflection\ObjectT(\Grpc\Testing\Payload::class))] + public ?\Grpc\Testing\Payload $payload = null, + #[Reflection\Field(7, new Reflection\ObjectT(\Grpc\Testing\EchoStatus::class))] + public ?\Grpc\Testing\EchoStatus $responseStatus = null, + #[Reflection\Field(8, new Reflection\ObjectT(\Grpc\Testing\TestOrcaReport::class))] + public ?\Grpc\Testing\TestOrcaReport $orcaOobReport = null, + ) {} +} diff --git a/interop/genproto/Grpc/Testing/StreamingOutputCallResponse.php b/interop/genproto/Grpc/Testing/StreamingOutputCallResponse.php new file mode 100644 index 0000000..ad7e4e9 --- /dev/null +++ b/interop/genproto/Grpc/Testing/StreamingOutputCallResponse.php @@ -0,0 +1,34 @@ + $requestCost + * @param Protobuf\Map $utilization + */ + public function __construct( + #[Reflection\Field(1, Reflection\DoubleT::T)] + public float $cpuUtilization = 0, + #[Reflection\Field(2, Reflection\DoubleT::T)] + public float $memoryUtilization = 0, + #[Reflection\Field(3, new Reflection\MapT(Reflection\StringT::T, Reflection\DoubleT::T))] + public Protobuf\Map $requestCost = new Protobuf\Map(), + #[Reflection\Field(4, new Reflection\MapT(Reflection\StringT::T, Reflection\DoubleT::T))] + public Protobuf\Map $utilization = new Protobuf\Map(), + ) {} +} diff --git a/interop/genproto/Grpc/Testing/TestServiceClient.php b/interop/genproto/Grpc/Testing/TestServiceClient.php new file mode 100644 index 0000000..5a4da15 --- /dev/null +++ b/interop/genproto/Grpc/Testing/TestServiceClient.php @@ -0,0 +1,248 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/EmptyCall', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * One request followed by one response. + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function unaryCall( + \Grpc\Testing\SimpleRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\SimpleResponse { + /** @var Client\Invoke<\Grpc\Testing\SimpleRequest, \Grpc\Testing\SimpleResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/UnaryCall', + output: \Grpc\Testing\SimpleResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * One request followed by one response. Response has cache control + * headers set such that a caching HTTP proxy (such as GFE) can + * satisfy subsequent requests. + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function cacheableUnaryCall( + \Grpc\Testing\SimpleRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\SimpleResponse { + /** @var Client\Invoke<\Grpc\Testing\SimpleRequest, \Grpc\Testing\SimpleResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/CacheableUnaryCall', + output: \Grpc\Testing\SimpleResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * One request followed by a sequence of responses (streamed download). + * The server returns the payload with client desired type and sizes. + * + * @return Client\ServerStreamChannel<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> + */ + public function streamingOutputCall( + \Grpc\Testing\StreamingOutputCallRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): Client\ServerStreamChannel { + /** @var Client\Invoke<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/StreamingOutputCall', + output: \Grpc\Testing\StreamingOutputCallResponse::class, + type: Grpc\RpcType::ServerStream, + ); + + $stream = $this->client->createStream( + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + + $stream->send($request); + $stream->close(); + + return new Client\ServerStreamChannel($stream); + } + + /** + * A sequence of requests followed by one response (streamed upload). + * The server returns the aggregated size of client payload as the result. + * + * @return Client\ClientStreamChannel<\Grpc\Testing\StreamingInputCallRequest, \Grpc\Testing\StreamingInputCallResponse> + */ + public function streamingInputCall( + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): Client\ClientStreamChannel { + /** @var Client\Invoke<\Grpc\Testing\StreamingInputCallRequest, \Grpc\Testing\StreamingInputCallResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/StreamingInputCall', + output: \Grpc\Testing\StreamingInputCallResponse::class, + type: Grpc\RpcType::ClientStream, + ); + + $stream = $this->client->createStream( + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + + return new Client\ClientStreamChannel($stream); + } + + /** + * A sequence of requests with each request served by the server immediately. + * As one request could lead to multiple responses, this interface + * demonstrates the idea of full duplexing. + * + * @return Client\BidirectionalStreamChannel<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> + */ + public function fullDuplexCall( + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): Client\BidirectionalStreamChannel { + /** @var Client\Invoke<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/FullDuplexCall', + output: \Grpc\Testing\StreamingOutputCallResponse::class, + type: Grpc\RpcType::BidirectionalStream, + ); + + $stream = $this->client->createStream( + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + + return new Client\BidirectionalStreamChannel($stream); + } + + /** + * A sequence of requests followed by a sequence of responses. + * The server buffers all the client requests and then serves them in order. A + * stream of responses are returned to the client when the server starts with + * first request. + * + * @return Client\BidirectionalStreamChannel<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> + */ + public function halfDuplexCall( + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): Client\BidirectionalStreamChannel { + /** @var Client\Invoke<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/HalfDuplexCall', + output: \Grpc\Testing\StreamingOutputCallResponse::class, + type: Grpc\RpcType::BidirectionalStream, + ); + + $stream = $this->client->createStream( + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + + return new Client\BidirectionalStreamChannel($stream); + } + + /** + * The test server will not implement this method. It will be used + * to test the behavior when clients call unimplemented methods. + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function unimplementedCall( + \Grpc\Testing\Empty_ $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\Empty_ { + /** @var Client\Invoke<\Grpc\Testing\Empty_, \Grpc\Testing\Empty_> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.TestService/UnimplementedCall', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/TestServiceServer.php b/interop/genproto/Grpc/Testing/TestServiceServer.php new file mode 100644 index 0000000..4453d9d --- /dev/null +++ b/interop/genproto/Grpc/Testing/TestServiceServer.php @@ -0,0 +1,116 @@ + + */ + public function streamingOutputCall( + \Grpc\Testing\StreamingOutputCallRequest $request, + Metadata $md, + Cancellation $cancellation, + ): iterable; + + /** + * A sequence of requests followed by one response (streamed upload). + * The server returns the aggregated size of client payload as the result. + * + * @param Server\ClientStreamChannel<\Grpc\Testing\StreamingInputCallRequest, \Grpc\Testing\StreamingInputCallResponse> $stream + */ + public function streamingInputCall( + Server\ClientStreamChannel $stream, + Metadata $md, + Cancellation $cancellation, + ): \Grpc\Testing\StreamingInputCallResponse; + + /** + * A sequence of requests with each request served by the server immediately. + * As one request could lead to multiple responses, this interface + * demonstrates the idea of full duplexing. + * + * @param Server\BidirectionalStreamChannel<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> $stream + */ + public function fullDuplexCall( + Server\BidirectionalStreamChannel $stream, + Metadata $md, + Cancellation $cancellation, + ): void; + + /** + * A sequence of requests followed by a sequence of responses. + * The server buffers all the client requests and then serves them in order. A + * stream of responses are returned to the client when the server starts with + * first request. + * + * @param Server\BidirectionalStreamChannel<\Grpc\Testing\StreamingOutputCallRequest, \Grpc\Testing\StreamingOutputCallResponse> $stream + */ + public function halfDuplexCall( + Server\BidirectionalStreamChannel $stream, + Metadata $md, + Cancellation $cancellation, + ): void; + + /** + * The test server will not implement this method. It will be used + * to test the behavior when clients call unimplemented methods. + */ + public function unimplementedCall( + \Grpc\Testing\Empty_ $request, + Metadata $md, + Cancellation $cancellation, + ): \Grpc\Testing\Empty_; +} diff --git a/interop/genproto/Grpc/Testing/TestServiceServerRegistry.php b/interop/genproto/Grpc/Testing/TestServiceServerRegistry.php new file mode 100644 index 0000000..0b9d825 --- /dev/null +++ b/interop/genproto/Grpc/Testing/TestServiceServerRegistry.php @@ -0,0 +1,74 @@ +server->emptyCall(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('UnaryCall', \Grpc\Testing\SimpleRequest::class), + handler: new Server\UnaryHandler($this->server->unaryCall(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('CacheableUnaryCall', \Grpc\Testing\SimpleRequest::class), + handler: new Server\UnaryHandler($this->server->cacheableUnaryCall(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('StreamingOutputCall', \Grpc\Testing\StreamingOutputCallRequest::class), + handler: new Server\ServerStreamHandler($this->server->streamingOutputCall(...)), + type: Grpc\RpcType::ServerStream, + ), + new Server\Rpc( + handle: new Server\Handle('StreamingInputCall', \Grpc\Testing\StreamingInputCallRequest::class), + handler: new Server\ClientStreamHandler($this->server->streamingInputCall(...)), + type: Grpc\RpcType::ClientStream, + ), + new Server\Rpc( + handle: new Server\Handle('FullDuplexCall', \Grpc\Testing\StreamingOutputCallRequest::class), + handler: new Server\BidirectionalStreamHandler($this->server->fullDuplexCall(...)), + type: Grpc\RpcType::BidirectionalStream, + ), + new Server\Rpc( + handle: new Server\Handle('HalfDuplexCall', \Grpc\Testing\StreamingOutputCallRequest::class), + handler: new Server\BidirectionalStreamHandler($this->server->halfDuplexCall(...)), + type: Grpc\RpcType::BidirectionalStream, + ), + new Server\Rpc( + handle: new Server\Handle('UnimplementedCall', \Grpc\Testing\Empty_::class), + handler: new Server\UnaryHandler($this->server->unimplementedCall(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/UnimplementedServiceClient.php b/interop/genproto/Grpc/Testing/UnimplementedServiceClient.php new file mode 100644 index 0000000..d52c815 --- /dev/null +++ b/interop/genproto/Grpc/Testing/UnimplementedServiceClient.php @@ -0,0 +1,59 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.UnimplementedService/UnimplementedCall', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/UnimplementedServiceServer.php b/interop/genproto/Grpc/Testing/UnimplementedServiceServer.php new file mode 100644 index 0000000..aa01b7e --- /dev/null +++ b/interop/genproto/Grpc/Testing/UnimplementedServiceServer.php @@ -0,0 +1,34 @@ +server->unimplementedCall(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceClient.php b/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceClient.php new file mode 100644 index 0000000..ba97e27 --- /dev/null +++ b/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceClient.php @@ -0,0 +1,58 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.XdsUpdateClientConfigureService/Configure', + output: \Grpc\Testing\ClientConfigureResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceServer.php b/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceServer.php new file mode 100644 index 0000000..b47cb19 --- /dev/null +++ b/interop/genproto/Grpc/Testing/XdsUpdateClientConfigureServiceServer.php @@ -0,0 +1,33 @@ +server->configure(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceClient.php b/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceClient.php new file mode 100644 index 0000000..8da712b --- /dev/null +++ b/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceClient.php @@ -0,0 +1,105 @@ + $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.XdsUpdateHealthService/SetServing', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function setNotServing( + \Grpc\Testing\Empty_ $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\Empty_ { + /** @var Client\Invoke<\Grpc\Testing\Empty_, \Grpc\Testing\Empty_> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.XdsUpdateHealthService/SetNotServing', + output: \Grpc\Testing\Empty_::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } + + /** + * @throws ClientStreamIsClosed + * @throws InvokeError + */ + public function sendHookRequest( + \Grpc\Testing\HookRequest $request, + Metadata $md = new Metadata(), + Cancellation $cancellation = new NullCancellation(), + ): \Grpc\Testing\HookResponse { + /** @var Client\Invoke<\Grpc\Testing\HookRequest, \Grpc\Testing\HookResponse> $invoke */ + $invoke = new Client\Invoke( + method: '/grpc.testing.XdsUpdateHealthService/SendHookRequest', + output: \Grpc\Testing\HookResponse::class, + type: Grpc\RpcType::Unary, + ); + + return $this->client->invoke( + request: $request, + invoke: $invoke, + md: $md, + cancellation: $cancellation, + ); + } +} diff --git a/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceServer.php b/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceServer.php new file mode 100644 index 0000000..47342d2 --- /dev/null +++ b/interop/genproto/Grpc/Testing/XdsUpdateHealthServiceServer.php @@ -0,0 +1,42 @@ +server->setServing(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('SetNotServing', \Grpc\Testing\Empty_::class), + handler: new Server\UnaryHandler($this->server->setNotServing(...)), + type: Grpc\RpcType::Unary, + ), + new Server\Rpc( + handle: new Server\Handle('SendHookRequest', \Grpc\Testing\HookRequest::class), + handler: new Server\UnaryHandler($this->server->sendHookRequest(...)), + type: Grpc\RpcType::Unary, + ), + ]); + } +} diff --git a/interop/genproto/Grpc/Testing/autoload.metadata.php b/interop/genproto/Grpc/Testing/autoload.metadata.php new file mode 100644 index 0000000..be0d5d2 --- /dev/null +++ b/interop/genproto/Grpc/Testing/autoload.metadata.php @@ -0,0 +1,14 @@ +register( + new \Thesis\Protobuf\Registry\OnceRegistrar(new \Grpc\Testing\DescriptorRegistry()), +); diff --git a/interop/protos/src/proto/grpc/testing/empty.proto b/interop/protos/src/proto/grpc/testing/empty.proto new file mode 100644 index 0000000..6a0aa88 --- /dev/null +++ b/interop/protos/src/proto/grpc/testing/empty.proto @@ -0,0 +1,28 @@ + +// Copyright 2015 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +syntax = "proto3"; + +package grpc.testing; + +// An empty message that you can re-use to avoid defining duplicated empty +// messages in your project. A typical example is to use it as argument or the +// return value of a service API. For instance: +// +// service Foo { +// rpc Bar (grpc.testing.Empty) returns (grpc.testing.Empty) { }; +// }; +// +message Empty {} diff --git a/interop/protos/src/proto/grpc/testing/messages.proto b/interop/protos/src/proto/grpc/testing/messages.proto new file mode 100644 index 0000000..7312485 --- /dev/null +++ b/interop/protos/src/proto/grpc/testing/messages.proto @@ -0,0 +1,354 @@ +// Copyright 2015-2016 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Message definitions to be used by integration test service definitions. + +syntax = "proto3"; + +package grpc.testing; + +option java_package = "io.grpc.testing.integration"; + +// TODO(dgq): Go back to using well-known types once +// https://github.com/grpc/grpc/issues/6980 has been fixed. +// import "google/protobuf/wrappers.proto"; +message BoolValue { + // The bool value. + bool value = 1; +} + +// The type of payload that should be returned. +enum PayloadType { + // Compressable text format. + COMPRESSABLE = 0; +} + +// A block of data, to simply increase gRPC message size. +message Payload { + // The type of data in body. + PayloadType type = 1; + // Primary contents of payload. + bytes body = 2; +} + +// A protobuf representation for grpc status. This is used by test +// clients to specify a status that the server should attempt to return. +message EchoStatus { + int32 code = 1; + string message = 2; +} + +// The type of route that a client took to reach a server w.r.t. gRPCLB. +// The server must fill in "fallback" if it detects that the RPC reached +// the server via the "gRPCLB fallback" path, and "backend" if it detects +// that the RPC reached the server via "gRPCLB backend" path (i.e. if it got +// the address of this server from the gRPCLB server BalanceLoad RPC). Exactly +// how this detection is done is context and server dependent. +enum GrpclbRouteType { + // Server didn't detect the route that a client took to reach it. + GRPCLB_ROUTE_TYPE_UNKNOWN = 0; + // Indicates that a client reached a server via gRPCLB fallback. + GRPCLB_ROUTE_TYPE_FALLBACK = 1; + // Indicates that a client reached a server as a gRPCLB-given backend. + GRPCLB_ROUTE_TYPE_BACKEND = 2; +} + +// Unary request. +message SimpleRequest { + // Desired payload type in the response from the server. + // If response_type is RANDOM, server randomly chooses one from other formats. + PayloadType response_type = 1; + + // Desired payload size in the response from the server. + int32 response_size = 2; + + // Optional input payload sent along with the request. + Payload payload = 3; + + // Whether SimpleResponse should include username. + bool fill_username = 4; + + // Whether SimpleResponse should include OAuth scope. + bool fill_oauth_scope = 5; + + // Whether to request the server to compress the response. This field is + // "nullable" in order to interoperate seamlessly with clients not able to + // implement the full compression tests by introspecting the call to verify + // the response's compression status. + BoolValue response_compressed = 6; + + // Whether server should return a given status + EchoStatus response_status = 7; + + // Whether the server should expect this request to be compressed. + BoolValue expect_compressed = 8; + + // Whether SimpleResponse should include server_id. + bool fill_server_id = 9; + + // Whether SimpleResponse should include grpclb_route_type. + bool fill_grpclb_route_type = 10; + + // If set the server should record this metrics report data for the current RPC. + TestOrcaReport orca_per_query_report = 11; +} + +// Unary response, as configured by the request. +message SimpleResponse { + // Payload to increase message size. + Payload payload = 1; + // The user the request came from, for verifying authentication was + // successful when the client expected it. + string username = 2; + // OAuth scope. + string oauth_scope = 3; + + // Server ID. This must be unique among different server instances, + // but the same across all RPC's made to a particular server instance. + string server_id = 4; + // gRPCLB Path. + GrpclbRouteType grpclb_route_type = 5; + + // Server hostname. + string hostname = 6; +} + +// Client-streaming request. +message StreamingInputCallRequest { + // Optional input payload sent along with the request. + Payload payload = 1; + + // Whether the server should expect this request to be compressed. This field + // is "nullable" in order to interoperate seamlessly with servers not able to + // implement the full compression tests by introspecting the call to verify + // the request's compression status. + BoolValue expect_compressed = 2; + + // Not expecting any payload from the response. +} + +// Client-streaming response. +message StreamingInputCallResponse { + // Aggregated size of payloads received from the client. + int32 aggregated_payload_size = 1; +} + +// Configuration for a particular response. +message ResponseParameters { + // Desired payload sizes in responses from the server. + int32 size = 1; + + // Desired interval between consecutive responses in the response stream in + // microseconds. + int32 interval_us = 2; + + // Whether to request the server to compress the response. This field is + // "nullable" in order to interoperate seamlessly with clients not able to + // implement the full compression tests by introspecting the call to verify + // the response's compression status. + BoolValue compressed = 3; + + // Whether to request the server to send the requesting peer's socket + // address in the response. + BoolValue fill_peer_socket_address = 4; +} + +// Server-streaming request. +message StreamingOutputCallRequest { + // Desired payload type in the response from the server. + // If response_type is RANDOM, the payload from each response in the stream + // might be of different types. This is to simulate a mixed type of payload + // stream. + PayloadType response_type = 1; + + // Configuration for each expected response message. + repeated ResponseParameters response_parameters = 2; + + // Optional input payload sent along with the request. + Payload payload = 3; + + // Whether server should return a given status + EchoStatus response_status = 7; + + // If set the server should update this metrics report data at the OOB server. + TestOrcaReport orca_oob_report = 8; +} + +// Server-streaming response, as configured by the request and parameters. +message StreamingOutputCallResponse { + // Payload to increase response size. + Payload payload = 1; + + // The peer's socket address if requested. + string peer_socket_address = 2; +} + +// For reconnect interop test only. +// Client tells server what reconnection parameters it used. +message ReconnectParams { + int32 max_reconnect_backoff_ms = 1; +} + +// For reconnect interop test only. +// Server tells client whether its reconnects are following the spec and the +// reconnect backoffs it saw. +message ReconnectInfo { + bool passed = 1; + repeated int32 backoff_ms = 2; +} + +message LoadBalancerStatsRequest { + // Request stats for the next num_rpcs sent by client. + int32 num_rpcs = 1; + // If num_rpcs have not completed within timeout_sec, return partial results. + int32 timeout_sec = 2; + // Response header + trailer metadata entries we want the values of. + // Matching of the keys is case-insensitive as per rfc7540#section-8.1.2 + // * (asterisk) is a special value that will return all metadata entries + repeated string metadata_keys = 3; +} + +message LoadBalancerStatsResponse { + enum MetadataType { + UNKNOWN = 0; + INITIAL = 1; + TRAILING = 2; + } + message MetadataEntry { + // Key, exactly as received from the server. Case may be different from what + // was requested in the LoadBalancerStatsRequest) + string key = 1; + // Value, exactly as received from the server. + string value = 2; + // Metadata type + MetadataType type = 3; + } + message RpcMetadata { + // metadata values for each rpc for the keys specified in + // LoadBalancerStatsRequest.metadata_keys. + repeated MetadataEntry metadata = 1; + } + message MetadataByPeer { + // List of RpcMetadata in for each RPC with a given peer + repeated RpcMetadata rpc_metadata = 1; + } + message RpcsByPeer { + // The number of completed RPCs for each peer. + map rpcs_by_peer = 1; + } + // The number of completed RPCs for each peer. + map rpcs_by_peer = 1; + // The number of RPCs that failed to record a remote peer. + int32 num_failures = 2; + map rpcs_by_method = 3; + // All the metadata of all RPCs for each peer. + map metadatas_by_peer = 4; +} + +// Request for retrieving a test client's accumulated stats. +message LoadBalancerAccumulatedStatsRequest {} + +// Accumulated stats for RPCs sent by a test client. +message LoadBalancerAccumulatedStatsResponse { + // The total number of RPCs have ever issued for each type. + // Deprecated: use stats_per_method.rpcs_started instead. + map num_rpcs_started_by_method = 1 [deprecated = true]; + // The total number of RPCs have ever completed successfully for each type. + // Deprecated: use stats_per_method.result instead. + map num_rpcs_succeeded_by_method = 2 [deprecated = true]; + // The total number of RPCs have ever failed for each type. + // Deprecated: use stats_per_method.result instead. + map num_rpcs_failed_by_method = 3 [deprecated = true]; + + message MethodStats { + // The number of RPCs that were started for this method. + int32 rpcs_started = 1; + + // The number of RPCs that completed with each status for this method. The + // key is the integral value of a google.rpc.Code; the value is the count. + map result = 2; + } + + // Per-method RPC statistics. The key is the RpcType in string form; e.g. + // 'EMPTY_CALL' or 'UNARY_CALL' + map stats_per_method = 4; +} + +// Configurations for a test client. +message ClientConfigureRequest { + // Type of RPCs to send. + enum RpcType { + EMPTY_CALL = 0; + UNARY_CALL = 1; + } + + // Metadata to be attached for the given type of RPCs. + message Metadata { + RpcType type = 1; + string key = 2; + string value = 3; + } + + // The types of RPCs the client sends. + repeated RpcType types = 1; + // The collection of custom metadata to be attached to RPCs sent by the client. + repeated Metadata metadata = 2; + // The deadline to use, in seconds, for all RPCs. If unset or zero, the + // client will use the default from the command-line. + int32 timeout_sec = 3; +} + +// Response for updating a test client's configuration. +message ClientConfigureResponse {} + +message MemorySize { + int64 rss = 1; +} + +// Metrics data the server will update and send to the client. It mirrors orca load report +// https://github.com/cncf/xds/blob/eded343319d09f30032952beda9840bbd3dcf7ac/xds/data/orca/v3/orca_load_report.proto#L15, +// but avoids orca dependency. Used by both per-query and out-of-band reporting tests. +message TestOrcaReport { + double cpu_utilization = 1; + double memory_utilization = 2; + map request_cost = 3; + map utilization = 4; +} + +// Status that will be return to callers of the Hook method +message SetReturnStatusRequest { + int32 grpc_code_to_return = 1; + string grpc_status_description = 2; +} + +message HookRequest { + enum HookRequestCommand { + // Default value + UNSPECIFIED = 0; + // Start the HTTP endpoint + START = 1; + // Stop + STOP = 2; + // Return from HTTP GET/POST + RETURN = 3; + } + HookRequestCommand command = 1; + int32 grpc_code_to_return = 2; + string grpc_status_description = 3; + // Server port to listen to + int32 server_port = 4; +} + +message HookResponse { +} diff --git a/interop/protos/src/proto/grpc/testing/test.proto b/interop/protos/src/proto/grpc/testing/test.proto new file mode 100644 index 0000000..43ae0b5 --- /dev/null +++ b/interop/protos/src/proto/grpc/testing/test.proto @@ -0,0 +1,116 @@ + +// Copyright 2015-2016 gRPC authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// An integration test service that covers all the method signature permutations +// of unary/streaming requests/responses. + +syntax = "proto3"; + +import "src/proto/grpc/testing/empty.proto"; +import "src/proto/grpc/testing/messages.proto"; + +package grpc.testing; + +option java_package = "io.grpc.testing.integration"; + +// A simple service to test the various types of RPCs and experiment with +// performance with various types of payload. +service TestService { + // One empty request followed by one empty response. + rpc EmptyCall(grpc.testing.Empty) returns (grpc.testing.Empty); + + // One request followed by one response. + rpc UnaryCall(SimpleRequest) returns (SimpleResponse); + + // One request followed by one response. Response has cache control + // headers set such that a caching HTTP proxy (such as GFE) can + // satisfy subsequent requests. + rpc CacheableUnaryCall(SimpleRequest) returns (SimpleResponse); + + // One request followed by a sequence of responses (streamed download). + // The server returns the payload with client desired type and sizes. + rpc StreamingOutputCall(StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // A sequence of requests followed by one response (streamed upload). + // The server returns the aggregated size of client payload as the result. + rpc StreamingInputCall(stream StreamingInputCallRequest) + returns (StreamingInputCallResponse); + + // A sequence of requests with each request served by the server immediately. + // As one request could lead to multiple responses, this interface + // demonstrates the idea of full duplexing. + rpc FullDuplexCall(stream StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // A sequence of requests followed by a sequence of responses. + // The server buffers all the client requests and then serves them in order. A + // stream of responses are returned to the client when the server starts with + // first request. + rpc HalfDuplexCall(stream StreamingOutputCallRequest) + returns (stream StreamingOutputCallResponse); + + // The test server will not implement this method. It will be used + // to test the behavior when clients call unimplemented methods. + rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty); +} + +// A simple service NOT implemented at servers so clients can test for +// that case. +service UnimplementedService { + // A call that no server should implement + rpc UnimplementedCall(grpc.testing.Empty) returns (grpc.testing.Empty); +} + +// A service used to control reconnect server. +service ReconnectService { + rpc Start(grpc.testing.ReconnectParams) returns (grpc.testing.Empty); + rpc Stop(grpc.testing.Empty) returns (grpc.testing.ReconnectInfo); +} + +// A service used to obtain stats for verifying LB behavior. +service LoadBalancerStatsService { + // Gets the backend distribution for RPCs sent by a test client. + rpc GetClientStats(LoadBalancerStatsRequest) + returns (LoadBalancerStatsResponse) {} + + // Gets the accumulated stats for RPCs sent by a test client. + rpc GetClientAccumulatedStats(LoadBalancerAccumulatedStatsRequest) + returns (LoadBalancerAccumulatedStatsResponse) {} +} + +// Hook service. Used to keep Kubernetes from shutting the pod down. +service HookService { + // Sends a request that will "hang" until the return status is set by a call + // to a SetReturnStatus + rpc Hook(grpc.testing.Empty) returns (grpc.testing.Empty); + // Sets a return status for pending and upcoming calls to Hook + rpc SetReturnStatus(SetReturnStatusRequest) returns (grpc.testing.Empty); + // Clears the return status. Incoming calls to Hook will "hang" + rpc ClearReturnStatus(grpc.testing.Empty) returns (grpc.testing.Empty); +} + +// A service to remotely control health status of an xDS test server. +service XdsUpdateHealthService { + rpc SetServing(grpc.testing.Empty) returns (grpc.testing.Empty); + rpc SetNotServing(grpc.testing.Empty) returns (grpc.testing.Empty); + rpc SendHookRequest(HookRequest) returns (HookResponse); +} + +// A service to dynamically update the configuration of an xDS test client. +service XdsUpdateClientConfigureService { + // Update the tes client's configuration. + rpc Configure(ClientConfigureRequest) returns (ClientConfigureResponse); +} diff --git a/interop/src/AdditionalMetadata.php b/interop/src/AdditionalMetadata.php new file mode 100644 index 0000000..0d07bf9 --- /dev/null +++ b/interop/src/AdditionalMetadata.php @@ -0,0 +1,61 @@ +with($key, $value); + } + + return new self($md); + } + + #[\Override] + public function interceptUnary( + object $request, + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $invoker, + ): object { + return $invoker($request, $invoke, $md->merge($this->md), $cancellation); + } + + #[\Override] + public function interceptStream( + Invoke $invoke, + Metadata $md, + Cancellation $cancellation, + callable $newStream, + ): ClientStream { + return $newStream($invoke, $md->merge($this->md), $cancellation); + } +} diff --git a/interop/src/EchoMetadata.php b/interop/src/EchoMetadata.php new file mode 100644 index 0000000..b2b256f --- /dev/null +++ b/interop/src/EchoMetadata.php @@ -0,0 +1,35 @@ +join(new Metadata(array_intersect_key($md->kv, [self::INITIAL => true]))); + $trailers->join(new Metadata(array_intersect_key($md->kv, [self::TRAILING => true]))); + } + + #[\Override] + public function interceptUnary( + object $request, + StreamInfo $info, + Metadata $md, + Cancellation $cancellation, + callable $handler, + ): object { + self::echo($md, $info->responseHeaders, $info->trailers); + + return $handler($request, $info, $md, $cancellation); + } +} diff --git a/interop/src/Flags.php b/interop/src/Flags.php new file mode 100644 index 0000000..2497fd8 --- /dev/null +++ b/interop/src/Flags.php @@ -0,0 +1,60 @@ + $values + */ + private function __construct( + private array $values, + ) {} + + /** + * Accepts `--name=value` for every flag that has a default. + * + * @param array $defaults + */ + public static function parse(array $defaults): self + { + $flags = getopt('', array_map(static fn(string $name) => "{$name}:", array_keys($defaults))); + + return new self([ + ...$defaults, + ...array_filter($flags === false ? [] : $flags, \is_string(...)), + ]); + } + + /** + * @param non-empty-string $name + */ + public function string(string $name): string + { + return $this->values[$name] ?? throw new \InvalidArgumentException("Unknown flag --{$name}."); + } + + /** + * @param non-empty-string $name + */ + public function int(string $name): int + { + $value = $this->string($name); + + return is_numeric($value) ? (int) $value : throw new \InvalidArgumentException("Flag --{$name} must be an integer."); + } + + /** + * @param non-empty-string $name + */ + public function bool(string $name): bool + { + return match ($this->string($name)) { + 'true' => true, + 'false' => false, + default => throw new \InvalidArgumentException("Flag --{$name} must be true or false."), + }; + } +} diff --git a/interop/src/TestCases.php b/interop/src/TestCases.php new file mode 100644 index 0000000..1738e50 --- /dev/null +++ b/interop/src/TestCases.php @@ -0,0 +1,360 @@ +client = $connect(); + $this->test = new TestServiceClient($this->client); + } + + public function emptyUnary(): void + { + $this->test->emptyCall(new Empty_()); + } + + public function largeUnary(): void + { + $response = $this->test->unaryCall(new SimpleRequest(responseSize: 314_159, payload: self::payload(271_828))); + + Assert::assertSame(314_159, self::size($response->payload)); + } + + public function clientStreaming(): void + { + $stream = $this->test->streamingInputCall(); + $stream->send(...array_map( + static fn(int $size) => new StreamingInputCallRequest(self::payload($size)), + [27_182, 8, 1_828, 45_904], + )); + + Assert::assertSame(74_922, $stream->close()->aggregatedPayloadSize); + } + + public function serverStreaming(): void + { + $sizes = [31_415, 9, 2_653, 58_979]; + + $stream = $this->test->streamingOutputCall(new StreamingOutputCallRequest( + responseParameters: array_map(static fn(int $size) => new ResponseParameters($size), $sizes), + )); + + Assert::assertSame($sizes, self::sizes($stream)); + } + + public function pingPong(): void + { + $stream = $this->test->fullDuplexCall(); + + foreach ([[31_415, 27_182], [9, 8], [2_653, 1_828], [58_979, 45_904]] as [$responseSize, $requestSize]) { + $stream->send(new StreamingOutputCallRequest( + responseParameters: [new ResponseParameters($responseSize)], + payload: self::payload($requestSize), + )); + + Assert::assertSame($responseSize, self::size($stream->receive()->payload)); + } + + $stream->close(); + + Assert::assertSame([], self::sizes($stream)); + } + + public function emptyStream(): void + { + $stream = $this->test->fullDuplexCall(); + $stream->close(); + + Assert::assertSame([], self::sizes($stream)); + } + + public function statusCodeAndMessage(): void + { + $status = new EchoStatus(Code::UNKNOWN->value, self::STATUS_MESSAGE); + + $error = self::error(fn() => $this->test->unaryCall(new SimpleRequest(responseStatus: $status))); + Assert::assertSame([Code::UNKNOWN, self::STATUS_MESSAGE], [$error->statusCode, $error->statusMessage]); + + $error = self::error(function () use ($status): void { + $stream = $this->test->fullDuplexCall(); + $stream->send(new StreamingOutputCallRequest(responseStatus: $status)); + $stream->close(); + self::sizes($stream); + }); + Assert::assertSame([Code::UNKNOWN, self::STATUS_MESSAGE], [$error->statusCode, $error->statusMessage]); + } + + public function specialStatusMessage(): void + { + $error = self::error(fn() => $this->test->unaryCall(new SimpleRequest( + responseStatus: new EchoStatus(Code::UNKNOWN->value, self::SPECIAL_STATUS_MESSAGE), + ))); + + Assert::assertSame([Code::UNKNOWN, self::SPECIAL_STATUS_MESSAGE], [$error->statusCode, $error->statusMessage]); + } + + public function unimplementedMethod(): void + { + $error = self::error(fn() => $this->test->unimplementedCall(new Empty_())); + + Assert::assertSame(Code::UNIMPLEMENTED, $error->statusCode); + } + + public function unimplementedService(): void + { + $error = self::error(fn() => new UnimplementedServiceClient($this->client)->unimplementedCall(new Empty_())); + + Assert::assertSame(Code::UNIMPLEMENTED, $error->statusCode); + } + + public function cancelAfterBegin(): void + { + $cancellation = new DeferredCancellation(); + $stream = $this->test->streamingInputCall(cancellation: $cancellation->getCancellation()); + $cancellation->cancel(); + + Assert::assertSame(Code::CANCELLED, self::error($stream->close(...))->statusCode); + } + + public function cancelAfterFirstResponse(): void + { + $cancellation = new DeferredCancellation(); + $stream = $this->test->fullDuplexCall(cancellation: $cancellation->getCancellation()); + $stream->send(new StreamingOutputCallRequest( + responseParameters: [new ResponseParameters(31_415)], + payload: self::payload(27_182), + )); + $stream->receive(); + $cancellation->cancel(); + + Assert::assertSame(Code::CANCELLED, self::error($stream->receive(...))->statusCode); + } + + public function timeoutOnSleepingServer(): void + { + // Like grpc-go, the deadline may already expire while the call waits for a connection. + $error = self::error(function (): void { + $stream = $this->test->fullDuplexCall(new Metadata()->withKey(Metadata\Timeout::milliseconds(1))); + + try { + $stream->send(new StreamingOutputCallRequest(payload: self::payload(27_182))); + } catch (ClientStreamIsClosed) { + // Like io.EOF from Send() in grpc-go: the status comes from receive(). + } + + $stream->receive(); + }); + + Assert::assertSame(Code::DEADLINE_EXCEEDED, $error->statusCode); + } + + public function customMetadata(): void + { + $md = new Metadata([ + self::INITIAL_METADATA_KEY => self::INITIAL_METADATA_VALUE, + self::TRAILING_METADATA_KEY => self::TRAILING_METADATA_VALUE, + ]); + + // Client::invoke() does not expose the response metadata of a unary call. + $stream = $this->client->createStream( + new Client\Invoke('/grpc.testing.TestService/UnaryCall', SimpleResponse::class, RpcType::Unary), + $md, + ); + $stream->send(new SimpleRequest(responseSize: 314_159, payload: self::payload(271_828))); + $stream->close(); + + Assert::assertSame(314_159, self::size($stream->receive()->payload)); + self::assertEchoedMetadata($stream); + + $stream = $this->client->createStream( + new Client\Invoke('/grpc.testing.TestService/FullDuplexCall', StreamingOutputCallResponse::class, RpcType::BidirectionalStream), + $md, + ); + $stream->send(new StreamingOutputCallRequest( + responseParameters: [new ResponseParameters(314_159)], + payload: self::payload(271_828), + )); + $stream->close(); + + Assert::assertSame([314_159], self::sizes($stream)); + self::assertEchoedMetadata($stream); + } + + public function rpcSoak(): void + { + $this->soak(newChannel: false); + } + + public function channelSoak(): void + { + $this->soak(newChannel: true); + } + + private function soak(bool $newChannel): void + { + $iterations = $this->flags->int('soak_iterations'); + $threads = $this->flags->int('soak_num_threads'); + $maxLatency = $this->flags->int('soak_per_iteration_max_acceptable_latency_ms'); + $minInterval = $this->flags->int('soak_min_time_ms_between_rpcs'); + $requestSize = $this->flags->int('soak_request_size'); + $responseSize = $this->flags->int('soak_response_size'); + $deadline = hrtime(true) + $this->flags->int('soak_overall_timeout_seconds') * 1_000_000_000; + + Assert::assertSame(0, $iterations % $threads, 'soak_iterations must be divisible by soak_num_threads.'); + + $results = await(array_map(fn(int $thread) => async(function () use ( + $thread, + $iterations, + $threads, + $newChannel, + $maxLatency, + $minInterval, + $requestSize, + $responseSize, + $deadline, + ): array { + $failures = 0; + $i = 0; + + for (; $i < intdiv($iterations, $threads) && hrtime(true) < $deadline; ++$i) { + $start = hrtime(true); + $client = $newChannel ? ($this->connect)() : $this->client; + + try { + $response = new TestServiceClient($client)->unaryCall(new SimpleRequest( + responseSize: $responseSize, + payload: self::payload($requestSize), + )); + $error = self::size($response->payload) === $responseSize ? '' : 'unexpected response size'; + } catch (InvokeError $e) { + $error = $e->getMessage(); + } + + $elapsed = (int) ((hrtime(true) - $start) / 1_000_000); + + if ($newChannel) { + $client->close(); + } + + if ($error === '' && $elapsed > $maxLatency) { + $error = "exceeds max acceptable latency {$maxLatency}ms"; + } + + $failures += (int) ($error !== ''); + + fprintf( + \STDERR, + "thread_id: %d soak iteration: %d elapsed_ms: %d peer: %s server_uri: %s %s\n", + $thread, + $i, + $elapsed, + $this->serverUri, + $this->serverUri, + $error === '' ? 'succeeded' : "failed: {$error}", + ); + + delay(max($minInterval - $elapsed, 0) / 1_000); + } + + return [$i, $failures]; + }), range(0, $threads - 1))); + + Assert::assertSame($iterations, array_sum(array_column($results, 0)), 'Soak test did not complete all iterations in time.'); + Assert::assertLessThanOrEqual($this->flags->int('soak_max_failures'), array_sum(array_column($results, 1)), 'Soak test failures exceeded soak_max_failures.'); + } + + /** + * @param \Closure(): mixed $call + */ + private static function error(\Closure $call): InvokeError + { + try { + $call(); + } catch (InvokeError $e) { + return $e; + } + + Assert::fail('The call was expected to fail with a status.'); + } + + /** + * @param ClientStream<*, *> $stream + */ + private static function assertEchoedMetadata(ClientStream $stream): void + { + Assert::assertSame([self::INITIAL_METADATA_VALUE], $stream->headers()[self::INITIAL_METADATA_KEY]); + Assert::assertSame([self::TRAILING_METADATA_VALUE], $stream->trailers()[self::TRAILING_METADATA_KEY]); + } + + private static function payload(int $size): Payload + { + return new Payload(body: str_repeat("\0", $size)); + } + + private static function size(?Payload $payload): int + { + return \strlen($payload->body ?? ''); + } + + /** + * Reads the stream to the end, so a non-OK status is thrown. + * + * @param iterable $stream + * @return list + */ + private static function sizes(iterable $stream): array + { + return array_map( + static fn(StreamingOutputCallResponse $response) => self::size($response->payload), + iterator_to_array($stream, preserve_keys: false), + ); + } +} diff --git a/interop/src/TestService.php b/interop/src/TestService.php new file mode 100644 index 0000000..3582ca5 --- /dev/null +++ b/interop/src/TestService.php @@ -0,0 +1,110 @@ +responseStatus); + + return new SimpleResponse(new Payload(body: str_repeat("\0", $request->responseSize))); + } + + #[\Override] + public function cacheableUnaryCall(SimpleRequest $request, Metadata $md, Cancellation $cancellation): SimpleResponse + { + throw new InvokeError(Code::UNIMPLEMENTED); + } + + #[\Override] + public function streamingOutputCall(StreamingOutputCallRequest $request, Metadata $md, Cancellation $cancellation): iterable + { + yield from self::responses($request->responseParameters, $cancellation); + } + + #[\Override] + public function streamingInputCall(ClientStreamChannel $stream, Metadata $md, Cancellation $cancellation): StreamingInputCallResponse + { + $size = 0; + + foreach ($stream as $request) { + $size += \strlen($request->payload->body ?? ''); + } + + return new StreamingInputCallResponse($size); + } + + #[\Override] + public function fullDuplexCall(BidirectionalStreamChannel $stream, Metadata $md, Cancellation $cancellation): void + { + EchoMetadata::echo($md, $stream->responseHeaders, $stream->trailers); + + foreach ($stream as $request) { + self::echoStatus($request->responseStatus); + + foreach (self::responses($request->responseParameters, $cancellation) as $response) { + $stream->send($response); + } + } + } + + #[\Override] + public function halfDuplexCall(BidirectionalStreamChannel $stream, Metadata $md, Cancellation $cancellation): void + { + throw new InvokeError(Code::UNIMPLEMENTED); + } + + #[\Override] + public function unimplementedCall(Empty_ $request, Metadata $md, Cancellation $cancellation): Empty_ + { + throw new InvokeError(Code::UNIMPLEMENTED); + } + + private static function echoStatus(?EchoStatus $status): void + { + if ($status !== null && $status->code !== Code::OK->value) { + throw new InvokeError(Code::tryFrom($status->code) ?? Code::UNKNOWN, $status->message); + } + } + + /** + * @param list $parameters + * @return \Generator + */ + private static function responses(array $parameters, Cancellation $cancellation): \Generator + { + foreach ($parameters as $parameter) { + delay($parameter->intervalUs / 1_000_000, cancellation: $cancellation); + + yield new StreamingOutputCallResponse(new Payload(body: str_repeat("\0", $parameter->size))); + } + } +} diff --git a/interop/tls/ca.pem b/interop/tls/ca.pem new file mode 100644 index 0000000..49d39cd --- /dev/null +++ b/interop/tls/ca.pem @@ -0,0 +1,20 @@ +-----BEGIN CERTIFICATE----- +MIIDWjCCAkKgAwIBAgIUWrP0VvHcy+LP6UuYNtiL9gBhD5owDQYJKoZIhvcNAQEL +BQAwVjELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM +GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDEPMA0GA1UEAwwGdGVzdGNhMB4XDTIw +MDMxNzE4NTk1MVoXDTMwMDMxNTE4NTk1MVowVjELMAkGA1UEBhMCQVUxEzARBgNV +BAgMClNvbWUtU3RhdGUxITAfBgNVBAoMGEludGVybmV0IFdpZGdpdHMgUHR5IEx0 +ZDEPMA0GA1UEAwwGdGVzdGNhMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKC +AQEAsGL0oXflF0LzoM+Bh+qUU9yhqzw2w8OOX5mu/iNCyUOBrqaHi7mGHx73GD01 +diNzCzvlcQqdNIH6NQSL7DTpBjca66jYT9u73vZe2MDrr1nVbuLvfu9850cdxiUO +Inv5xf8+sTHG0C+a+VAvMhsLiRjsq+lXKRJyk5zkbbsETybqpxoJ+K7CoSy3yc/k +QIY3TipwEtwkKP4hzyo6KiGd/DPexie4nBUInN3bS1BUeNZ5zeaIC2eg3bkeeW7c +qT55b+Yen6CxY0TEkzBK6AKt/WUialKMgT0wbTxRZO7kUCH3Sq6e/wXeFdJ+HvdV +LPlAg5TnMaNpRdQih/8nRFpsdwIDAQABoyAwHjAMBgNVHRMEBTADAQH/MA4GA1Ud +DwEB/wQEAwICBDANBgkqhkiG9w0BAQsFAAOCAQEAkTrKZjBrJXHps/HrjNCFPb5a +THuGPCSsepe1wkKdSp1h4HGRpLoCgcLysCJ5hZhRpHkRihhef+rFHEe60UePQO3S +CVTtdJB4CYWpcNyXOdqefrbJW5QNljxgi6Fhvs7JJkBqdXIkWXtFk2eRgOIP2Eo9 +/OHQHlYnwZFrk6sp4wPyR+A95S0toZBcyDVz7u+hOW0pGK3wviOe9lvRgj/H3Pwt +bewb0l+MhRig0/DVHamyVxrDRbqInU1/GTNCwcZkXKYFWSf92U+kIcTth24Q1gcw +eZiLl5FfrWokUNytFElXob0V0a5/kbhiLc3yWmvWqHTpqCALbVyF+rKJo2f5Kw== +-----END CERTIFICATE----- diff --git a/interop/tls/server1.key b/interop/tls/server1.key new file mode 100644 index 0000000..0864629 --- /dev/null +++ b/interop/tls/server1.key @@ -0,0 +1,28 @@ +-----BEGIN PRIVATE KEY----- +MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDnE443EknxvxBq +6+hvn/t09hl8hx366EBYvZmVM/NC+7igXRAjiJiA/mIaCvL3MS0Iz5hBLxSGICU+ +WproA3GCIFITIwcf/ETyWj/5xpgZ4AKrLrjQmmX8mhwUajfF3UvwMJrCOVqPp67t +PtP+2kBXaqrXdvnvXR41FsIB8V7zIAuIZB6bHQhiGVlc1sgZYsE2EGG9WMmHtS86 +qkAOTjG2XyjmPTGAwhGDpYkYrpzp99IiDh4/Veai81hn0ssQkbry0XRD/Ig3jcHh +23WiriPNJ0JsbgXUSLKRPZObA9VgOLy2aXoN84IMaeK3yy+cwSYG/99w93fUZJte +MXwz4oYZAgMBAAECggEBAIVn2Ncai+4xbH0OLWckabwgyJ4IM9rDc0LIU368O1kU +koais8qP9dujAWgfoh3sGh/YGgKn96VnsZjKHlyMgF+r4TaDJn3k2rlAOWcurGlj +1qaVlsV4HiEzp7pxiDmHhWvp4672Bb6iBG+bsjCUOEk/n9o9KhZzIBluRhtxCmw5 +nw4Do7z00PTvN81260uPWSc04IrytvZUiAIx/5qxD72bij2xJ8t/I9GI8g4FtoVB +8pB6S/hJX1PZhh9VlU6Yk+TOfOVnbebG4W5138LkB835eqk3Zz0qsbc2euoi8Hxi +y1VGwQEmMQ63jXz4c6g+X55ifvUK9Jpn5E8pq+pMd7ECgYEA93lYq+Cr54K4ey5t +sWMa+ye5RqxjzgXj2Kqr55jb54VWG7wp2iGbg8FMlkQwzTJwebzDyCSatguEZLuB +gRGroRnsUOy9vBvhKPOch9bfKIl6qOgzMJB267fBVWx5ybnRbWN/I7RvMQf3k+9y +biCIVnxDLEEYyx7z85/5qxsXg/MCgYEA7wmWKtCTn032Hy9P8OL49T0X6Z8FlkDC +Rk42ygrc/MUbugq9RGUxcCxoImOG9JXUpEtUe31YDm2j+/nbvrjl6/bP2qWs0V7l +dTJl6dABP51pCw8+l4cWgBBX08Lkeen812AAFNrjmDCjX6rHjWHLJcpS18fnRRkP +V1d/AHWX7MMCgYEA6Gsw2guhp0Zf2GCcaNK5DlQab8OL4Hwrpttzo4kuTlwtqNKp +Q9H4al9qfF4Cr1TFya98+EVYf8yFRM3NLNjZpe3gwYf2EerlJj7VLcahw0KKzoN1 +QBENfwgPLRk5sDkx9VhSmcfl/diLroZdpAwtv3vo4nEoxeuGFbKTGx3Qkf0CgYEA +xyR+dcb05Ygm3w4klHQTowQ10s1H80iaUcZBgQuR1ghEtDbUPZHsoR5t1xCB02ys +DgAwLv1bChIvxvH/L6KM8ovZ2LekBX4AviWxoBxJnfz/EVau98B0b1auRN6eSC83 +FRuGldlSOW1z/nSh8ViizSYE5H5HX1qkXEippvFRE88CgYB3Bfu3YQY60ITWIShv +nNkdcbTT9eoP9suaRJjw92Ln+7ZpALYlQMKUZmJ/5uBmLs4RFwUTQruLOPL4yLTH +awADWUzs3IRr1fwn9E+zM8JVyKCnUEM3w4N5UZskGO2klashAd30hWO+knRv/y0r +uGIYs9Ek7YXlXIRVrzMwcsrt1w== +-----END PRIVATE KEY----- diff --git a/interop/tls/server1.pem b/interop/tls/server1.pem new file mode 100644 index 0000000..88244f8 --- /dev/null +++ b/interop/tls/server1.pem @@ -0,0 +1,22 @@ +-----BEGIN CERTIFICATE----- +MIIDtDCCApygAwIBAgIUbJfTREJ6k6/+oInWhV1O1j3ZT0IwDQYJKoZIhvcNAQEL +BQAwVjELMAkGA1UEBhMCQVUxEzARBgNVBAgMClNvbWUtU3RhdGUxITAfBgNVBAoM +GEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDEPMA0GA1UEAwwGdGVzdGNhMB4XDTIw +MDMxODAzMTA0MloXDTMwMDMxNjAzMTA0MlowZTELMAkGA1UEBhMCVVMxETAPBgNV +BAgMCElsbGlub2lzMRAwDgYDVQQHDAdDaGljYWdvMRUwEwYDVQQKDAxFeGFtcGxl +LCBDby4xGjAYBgNVBAMMESoudGVzdC5nb29nbGUuY29tMIIBIjANBgkqhkiG9w0B +AQEFAAOCAQ8AMIIBCgKCAQEA5xOONxJJ8b8Qauvob5/7dPYZfIcd+uhAWL2ZlTPz +Qvu4oF0QI4iYgP5iGgry9zEtCM+YQS8UhiAlPlqa6ANxgiBSEyMHH/xE8lo/+caY +GeACqy640Jpl/JocFGo3xd1L8DCawjlaj6eu7T7T/tpAV2qq13b5710eNRbCAfFe +8yALiGQemx0IYhlZXNbIGWLBNhBhvVjJh7UvOqpADk4xtl8o5j0xgMIRg6WJGK6c +6ffSIg4eP1XmovNYZ9LLEJG68tF0Q/yIN43B4dt1oq4jzSdCbG4F1EiykT2TmwPV +YDi8tml6DfOCDGnit8svnMEmBv/fcPd31GSbXjF8M+KGGQIDAQABo2swaTAJBgNV +HRMEAjAAMAsGA1UdDwQEAwIF4DBPBgNVHREESDBGghAqLnRlc3QuZ29vZ2xlLmZy +ghh3YXRlcnpvb2kudGVzdC5nb29nbGUuYmWCEioudGVzdC55b3V0dWJlLmNvbYcE +wKgBAzANBgkqhkiG9w0BAQsFAAOCAQEAS8hDQA8PSgipgAml7Q3/djwQ644ghWQv +C2Kb+r30RCY1EyKNhnQnIIh/OUbBZvh0M0iYsy6xqXgfDhCB93AA6j0i5cS8fkhH +Jl4RK0tSkGQ3YNY4NzXwQP/vmUgfkw8VBAZ4Y4GKxppdATjffIW+srbAmdDruIRM +wPeikgOoRrXf0LA1fi4TqxARzeRwenQpayNfGHTvVF9aJkl8HoaMunTAdG5pIVcr +9GKi/gEMpXUJbbVv3U5frX1Wo4CFo+rZWJ/LyCMeb0jciNLxSdMwj/E/ZuExlyeZ +gc9ctPjSMvgSyXEKv6Vwobleeg88V2ZgzenziORoWj4KszG/lbQZvg== +-----END CERTIFICATE----- diff --git a/packages/client/README.md b/packages/client/README.md index 0f15dd7..facaf74 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -114,13 +114,18 @@ Multi-endpoint target without DNS: ->withHost('ipv4:10.0.0.1:50051,10.0.0.2:50051,10.0.0.3:50051') ``` +The `:authority` of every call comes from the target, not from the resolved address, like in grpc-go: +the peer name of the transport credentials if set, `localhost` for `unix`, otherwise the target address, +or the first one when the target lists several. A client passed to `withHttpClient()` keeps receiving +the resolved address as the authority. + ## Load balancing When resolution returns multiple addresses, the load balancer picks endpoint per RPC. Built-in policies: -- `PickFirstFactory` (default): shuffles endpoint list, picks one, and keeps it pinned until it disappears after refresh. +- `PickFirstFactory` (default): picks the first endpoint in resolution order and keeps it pinned until it disappears after refresh or cannot be connected to, then moves to the next one. - `RoundRobinFactory`: cycles through all available endpoints. With DNS targets, resolver can re-resolve by TTL and call balancer `refresh()` with updated endpoints. Re-resolution interval is clamped between minimum and maximum bounds (default 30..300 seconds). diff --git a/packages/client/composer.json b/packages/client/composer.json index 7065d92..9c90b3e 100644 --- a/packages/client/composer.json +++ b/packages/client/composer.json @@ -21,6 +21,7 @@ "php": "^8.4", "amphp/cache": "^2.0", "amphp/dns": "^2.4", + "amphp/http": "^2.1", "amphp/http-client": "^5.3", "amphp/socket": "^2.3", "league/uri": "^7.8", diff --git a/packages/client/src/Client/Builder.php b/packages/client/src/Client/Builder.php index 7508a11..cff5888 100644 --- a/packages/client/src/Client/Builder.php +++ b/packages/client/src/Client/Builder.php @@ -8,6 +8,7 @@ use Amp\Http\Client\Connection\DefaultConnectionFactory; use Amp\Http\Client\DelegateHttpClient; use Amp\Http\Client\HttpClientBuilder; +use Amp\Socket\ClientTlsContext; use Amp\Socket\ConnectContext; use Amp\Socket\DnsSocketConnector; use Amp\Socket\SocketConnector; @@ -305,7 +306,9 @@ public function build(): Client $protobuf = $this->protobuf ?? Decoder\Builder::buildDefault(); $loadBalancerFactory = $this->loadBalancerFactory ?? new LoadBalancer\PickFirstFactory(); $tlsContext = $this->credentials?->createContext(); - $uriFactory = new Http2\UriFactory($tlsContext !== null ? Internal\HttpScheme::Https : Internal\HttpScheme::Http); + $scheme = $tlsContext !== null ? Internal\HttpScheme::Https : Internal\HttpScheme::Http; + // A custom HTTP client knows nothing about endpoints, so it gets the endpoint address in the URI as before. + $uriFactory = new Http2\UriFactory($scheme, $this->httpclient === null ? self::authority($target, $tlsContext) : null); $transferTimeout = $this->transferTimeout; $inactivityTimeout = $this->inactivityTimeout; $maxReceiveMessageSize = $this->maxReceiveMessageSize; @@ -348,16 +351,22 @@ public function build(): Client ); } + $sockets = new Internal\TrackingSocketConnector($connector); + + $connectContext = new ConnectContext() + ->withConnectTimeout($this->connectTimeout) + ->withTlsContext($tlsContext); + $connectionLimit = $this->connectionLimit; + $httpclient = $this->httpclient ?? new HttpClientBuilder() - ->usingPool(ConnectionLimitingPool::byAuthority( - $this->connectionLimit, - new DefaultConnectionFactory( - $connector, - new ConnectContext() - ->withConnectTimeout($this->connectTimeout) - ->withTlsContext($tlsContext), + ->usingPool(new Http2\EndpointConnectionPool( + static fn(Address $address) => ConnectionLimitingPool::byAuthority( + $connectionLimit, + new DefaultConnectionFactory(new Internal\AddressSocketConnector($sockets, $address), $connectContext), ), )) + // A gRPC request body is streamed once, so amphp cannot replay a request it was told was left unprocessed. + ->retry(0) ->skipDefaultUserAgent() ->skipAutomaticCompression() ->skipDefaultAcceptHeader() @@ -380,10 +389,28 @@ public function build(): Client maxReceiveMessageSize: $maxReceiveMessageSize, compressors: $compressors, ), + sockets: $sockets, ), ), $unary, $stream, ); } + + /** + * Like grpc-go: the server name of the transport credentials, "localhost" for a Unix socket, + * otherwise the target itself, or its first address when it lists several. + * + * @return non-empty-string + */ + private static function authority(Target $target, ?ClientTlsContext $tlsContext): string + { + $peerName = $tlsContext?->getPeerName() ?? ''; + + return match (true) { + $peerName !== '' => $peerName, + $target->scheme === Scheme::Unix->value => 'localhost', + default => (string) $target->addresses[0], + }; + } } diff --git a/packages/client/src/Client/EndpointResolver/DnsResolver.php b/packages/client/src/Client/EndpointResolver/DnsResolver.php index dcb0616..1ae02c1 100644 --- a/packages/client/src/Client/EndpointResolver/DnsResolver.php +++ b/packages/client/src/Client/EndpointResolver/DnsResolver.php @@ -77,7 +77,8 @@ private function resolveNow( $ttl = null; foreach ($target->addresses as $address) { - $records = $resolver->resolve($address->host, cancellation: $cancellation); + // An IPv6 literal keeps its brackets in the target; without them amphp returns it as is, like grpc-go. + $records = $resolver->resolve(trim($address->host, '[]'), cancellation: $cancellation); foreach ($records as $record) { $ip = $record->getValue(); diff --git a/packages/client/src/Client/Internal/AddressSocketConnector.php b/packages/client/src/Client/Internal/AddressSocketConnector.php new file mode 100644 index 0000000..95bc331 --- /dev/null +++ b/packages/client/src/Client/Internal/AddressSocketConnector.php @@ -0,0 +1,33 @@ +address->value, '/') ? "unix://{$this->address}" : "tcp://{$this->address}"; + + return $this->connector->connect($address, $context, $cancellation); + } +} diff --git a/packages/client/src/Client/Internal/AmphpHttpClient.php b/packages/client/src/Client/Internal/AmphpHttpClient.php index 6433ef8..36ef18c 100644 --- a/packages/client/src/Client/Internal/AmphpHttpClient.php +++ b/packages/client/src/Client/Internal/AmphpHttpClient.php @@ -5,14 +5,13 @@ namespace Thesis\Grpc\Client\Internal; use Amp\Cancellation; -use Amp\CancelledException; +use Amp\CompositeCancellation; use Amp\NullCancellation; -use Thesis\Google\Rpc\Code; +use Amp\TimeoutCancellation; use Thesis\Grpc\Client; use Thesis\Grpc\Client\PickContext; use Thesis\Grpc\ClientStream; use Thesis\Grpc\GrpcException; -use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; /** @@ -46,6 +45,8 @@ function ( Metadata $md, Cancellation $cancellation, ) use ($pick): object { + $cancellation = self::deadline($md, $cancellation); + try { $stream = $this->connection->createStream($invoke, $md, $cancellation, $pick); $stream->send($request); @@ -54,12 +55,8 @@ function ( return $stream->receive(); } catch (GrpcException $e) { throw $e; - } catch (CancelledException $e) { - throw CancellationError::from($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); + throw TransportError::from($e, $cancellation); } }, ); @@ -82,10 +79,12 @@ function ( Metadata $md, Cancellation $cancellation, ) use ($pick): ClientStream { + $cancellation = self::deadline($md, $cancellation); + try { return $this->connection->createStream($invoke, $md, $cancellation, $pick); - } catch (CancelledException $e) { - throw CancellationError::from($e); + } catch (\Throwable $e) { + throw TransportError::from($e, $cancellation); } }, ); @@ -96,4 +95,15 @@ public function close(Cancellation $cancellation = new NullCancellation()): void { $this->connection->close($cancellation); } + + /** + * Like grpc-go, the client enforces the deadline it sends in "grpc-timeout" itself, + * so an expired call ends with DEADLINE_EXCEEDED rather than whatever the server does on its side. + */ + private static function deadline(Metadata $md, Cancellation $cancellation): Cancellation + { + $timeout = Metadata\parseTimeout($md); + + return $timeout !== null ? new CompositeCancellation($cancellation, new TimeoutCancellation($timeout->toSeconds())) : $cancellation; + } } diff --git a/packages/client/src/Client/Internal/Connection/DefaultConnection.php b/packages/client/src/Client/Internal/Connection/DefaultConnection.php index 62c82f1..976309d 100644 --- a/packages/client/src/Client/Internal/Connection/DefaultConnection.php +++ b/packages/client/src/Client/Internal/Connection/DefaultConnection.php @@ -5,12 +5,16 @@ namespace Thesis\Grpc\Client\Internal\Connection; use Amp\Cancellation; +use Amp\CompositeCancellation; use Amp\DeferredCancellation; +use Amp\Http\Client\HttpException; use Amp\NullCancellation; +use Thesis\Google\Rpc\Code; use Thesis\Grpc\Client\EndpointResolver; use Thesis\Grpc\Client\EndpointResolverListener; use Thesis\Grpc\Client\Internal\Connection; use Thesis\Grpc\Client\Internal\Http2\StreamFactory; +use Thesis\Grpc\Client\Internal\TrackingSocketConnector; use Thesis\Grpc\Client\Invoke; use Thesis\Grpc\Client\LoadBalancer; use Thesis\Grpc\Client\LoadBalancerFactory; @@ -18,6 +22,7 @@ use Thesis\Grpc\Client\Resolution; use Thesis\Grpc\Client\Target; use Thesis\Grpc\ClientStream; +use Thesis\Grpc\InvokeError; use Thesis\Grpc\Metadata; /** @@ -36,6 +41,7 @@ public function __construct( EndpointResolver $resolver, LoadBalancerFactory $loadBalancerFactory, private StreamFactory $streams, + private TrackingSocketConnector $sockets, ) { $this->deferredCancellation = new DeferredCancellation(); @@ -55,21 +61,28 @@ public function createStream( Cancellation $cancellation, PickContext $pick, ): ClientStream { - $endpoint = $this->balancer->pick($pick); - $pick->exclude($endpoint); + $cancellation = new CompositeCancellation($cancellation, $this->deferredCancellation->getCancellation()); + $error = null; - return $this->streams->create( - $invoke, - $endpoint->address, - $md, - $cancellation, - ); + // Like pick_first in grpc-go, an endpoint that cannot be connected to is skipped in favour of the next one. + while (!$pick->excluded($endpoint = $this->balancer->pick($pick))) { + $pick->exclude($endpoint); + + try { + return $this->streams->create($invoke, $endpoint->address, $md, $cancellation); + } catch (HttpException $e) { + $error = $e; + } + } + + throw new InvokeError(Code::UNAVAILABLE, $error?->getMessage(), previous: $error); } #[\Override] public function close(Cancellation $cancellation = new NullCancellation()): void { $this->deferredCancellation->cancel(); + $this->sockets->close(); } #[\Override] diff --git a/packages/client/src/Client/Internal/Connection/LazyConnection.php b/packages/client/src/Client/Internal/Connection/LazyConnection.php index 7c89a4a..420e601 100644 --- a/packages/client/src/Client/Internal/Connection/LazyConnection.php +++ b/packages/client/src/Client/Internal/Connection/LazyConnection.php @@ -67,7 +67,9 @@ public function close(Cancellation $cancellation = new NullCancellation()): void private function createConnection(Cancellation $cancellation): Connection { - $future = $this->future ??= async($this->factory); + /** @var Future $future */ + $future = $this->future ?? async($this->factory); + $this->future = $future; try { return $future->await($cancellation); diff --git a/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php b/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php index 338f115..0967b45 100644 --- a/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php +++ b/packages/client/src/Client/Internal/Http2/ConcurrentClientStream.php @@ -5,13 +5,12 @@ namespace Thesis\Grpc\Client\Internal\Http2; use Amp\Cancellation; -use Amp\CancelledException; use Amp\Future; use Amp\Http\Client\Response; use Amp\NullCancellation; use Amp\Pipeline; use Thesis\Google\Rpc\Code; -use Thesis\Grpc\Client\Internal\CancellationError; +use Thesis\Grpc\Client\Internal\TransportError; use Thesis\Grpc\ClientStream; use Thesis\Grpc\Exception\ClientStreamIsClosed; use Thesis\Grpc\Internal\Http2; @@ -43,6 +42,7 @@ public function __construct( private readonly \Closure $decode, private readonly ErrorHandler $errors, private readonly Future $complete, + private readonly Cancellation $cancellation, ) { $responseFuture->ignore(); $complete->ignore(); @@ -63,12 +63,12 @@ public function receive(): object { try { if (!$this->recv->continue()) { - throw $this->errors->obtain($this) ?? new InvokeError(Code::UNKNOWN); + throw $this->errors->obtain($this->response) ?? new InvokeError(Code::UNKNOWN); } return $this->recv->getValue(); - } catch (CancelledException $e) { - throw CancellationError::from($e); + } catch (\Throwable $e) { + throw TransportError::from($e, $this->cancellation); } } @@ -77,15 +77,19 @@ public function headers(): Metadata { try { return Http2\decodeMetadata($this->response->getHeaders()); - } catch (CancelledException $e) { - throw CancellationError::from($e); + } catch (\Throwable $e) { + throw TransportError::from($e, $this->cancellation); } } #[\Override] public function trailers(Cancellation $cancellation = new NullCancellation()): Metadata { - return Http2\decodeMetadata($this->response->getTrailers()->await($cancellation)->getHeaders()); + try { + return Http2\decodeMetadata($this->response->getTrailers()->await($cancellation)->getHeaders()); + } catch (\Throwable $e) { + throw TransportError::from($e, $this->cancellation); + } } #[\Override] @@ -99,8 +103,8 @@ public function close(): void try { $this->complete->await(); - } catch (CancelledException $e) { - throw CancellationError::from($e); + } catch (\Throwable $e) { + throw TransportError::from($e, $this->cancellation); } } @@ -111,9 +115,9 @@ public function getIterator(): \Traversable yield from $this->recv; // A cancelled call ends the message stream quietly and surfaces here, while reading the trailers. - $error = $this->errors->obtain($this); - } catch (CancelledException $e) { - throw CancellationError::from($e); + $error = $this->errors->obtain($this->response); + } catch (\Throwable $e) { + throw TransportError::from($e, $this->cancellation); } if ($error !== null) { diff --git a/packages/client/src/Client/Internal/Http2/EndpointConnectionPool.php b/packages/client/src/Client/Internal/Http2/EndpointConnectionPool.php new file mode 100644 index 0000000..0736d1e --- /dev/null +++ b/packages/client/src/Client/Internal/Http2/EndpointConnectionPool.php @@ -0,0 +1,39 @@ + */ + private array $pools = []; + + /** + * @param \Closure(Address): ConnectionPool $createPool + */ + public function __construct( + private readonly \Closure $createPool, + ) {} + + #[\Override] + public function getStream(Request $request, Cancellation $cancellation): Stream + { + /** @var Address $address */ + $address = $request->getAttribute(Address::class); + + return ($this->pools[$address->value] ??= ($this->createPool)($address))->getStream($request, $cancellation); + } +} diff --git a/packages/client/src/Client/Internal/Http2/ErrorHandler.php b/packages/client/src/Client/Internal/Http2/ErrorHandler.php index 202d5f7..591169e 100644 --- a/packages/client/src/Client/Internal/Http2/ErrorHandler.php +++ b/packages/client/src/Client/Internal/Http2/ErrorHandler.php @@ -4,9 +4,11 @@ namespace Thesis\Grpc\Client\Internal\Http2; +use Amp\Http\Client\Response; use Thesis\Google\Rpc; -use Thesis\Grpc\ClientStream; +use Thesis\Grpc\Internal\Http2; use Thesis\Grpc\InvokeError; +use Thesis\Grpc\Metadata; use Thesis\Grpc\Status; use Thesis\Protobuf\Decoder; @@ -20,12 +22,28 @@ public function __construct( ) {} /** - * @param ClientStream<*, *> $stream * @throws Decoder\DecodingError */ - public function obtain(ClientStream $stream): ?InvokeError + public function obtain(Response $response): ?InvokeError { - $context = Status\deserializeContext($stream->headers()->merge($stream->trailers()), $this->decoder); + $md = Http2\decodeMetadata($response->getHeaders())->merge(Http2\decodeMetadata($response->getTrailers()->await()->getHeaders())); + + // A response without a gRPC status, e.g. from a proxy, is mapped by its HTTP status as PROTOCOL-HTTP2.md requires. + if ($md->value(Metadata\Status::STATUS_HEADER) === null && ($status = $response->getStatus()) !== 200) { + return new InvokeError( + match ($status) { + 400 => Rpc\Code::INTERNAL, + 401 => Rpc\Code::UNAUTHENTICATED, + 403 => Rpc\Code::PERMISSION_DENIED, + 404 => Rpc\Code::UNIMPLEMENTED, + 429, 502, 503, 504 => Rpc\Code::UNAVAILABLE, + default => Rpc\Code::UNKNOWN, + }, + "unexpected HTTP status code received from server: {$status}", + ); + } + + $context = Status\deserializeContext($md, $this->decoder); if ($context->code !== Rpc\Code::OK) { return new InvokeError( diff --git a/packages/client/src/Client/Internal/Http2/StreamFactory.php b/packages/client/src/Client/Internal/Http2/StreamFactory.php index d72f066..2a494cc 100644 --- a/packages/client/src/Client/Internal/Http2/StreamFactory.php +++ b/packages/client/src/Client/Internal/Http2/StreamFactory.php @@ -74,6 +74,14 @@ public function create( /** @var DeferredFuture $deferred */ $deferred = new DeferredFuture(); + /** @var DeferredFuture $connected */ + $connected = new DeferredFuture(); + $markConnected = static function () use ($connected): void { + if (!$connected->isComplete()) { + $connected->complete(); + } + }; + $request = new Request( uri: $this->uri->create($address, $invoke), method: 'POST', @@ -82,6 +90,7 @@ public function create( ), ); $request->setProtocolVersions(['2']); + $request->setAttribute(Address::class, $address); $request->setHeaders(Http2\encodeMetadata($md)); $request->setTransferTimeout($this->transferTimeout); $request->setInactivityTimeout($this->inactivityTimeout); @@ -95,16 +104,34 @@ public function create( $request->addEventListener( new RequestEventListener() ->onRequestBodyEnd($deferred->complete(...)) // @phpstan-ignore argument.type - ->onRequestFailed(static function (Request $request, \Throwable $e) use ($deferred): void { + ->onConnectionAcquired($markConnected) + // A custom HTTP client may respond without acquiring a connection. + ->onRequestEnd($markConnected) + ->onRequestFailed(static function (Request $request, \Throwable $e) use ($deferred, $connected): void { if (!$deferred->isComplete()) { $deferred->error($e); } + + if (!$connected->isComplete()) { + $connected->error($e); + } }), ); /** @var Future $response */ $response = async($this->http->request(...), $request, $cancellation); + // Like grpc-go, the call starts only on an established connection, so a failure to connect + // surfaces here, before anything is sent, and the caller can try another endpoint. + try { + $connected->getFuture()->await($cancellation); + } catch (\Throwable $e) { + $response->ignore(); + $deferred->getFuture()->ignore(); + + throw $e; + } + return new ConcurrentClientStream( responseFuture: $response, send: $send, @@ -122,6 +149,7 @@ public function create( }, errors: $this->errors, complete: $deferred->getFuture(), + cancellation: $cancellation, ); } } diff --git a/packages/client/src/Client/Internal/Http2/UriFactory.php b/packages/client/src/Client/Internal/Http2/UriFactory.php index 2623c7b..1306704 100644 --- a/packages/client/src/Client/Internal/Http2/UriFactory.php +++ b/packages/client/src/Client/Internal/Http2/UriFactory.php @@ -13,8 +13,12 @@ */ final readonly class UriFactory { + /** + * @param ?non-empty-string $authority the ":authority" of every call; the endpoint address when null + */ public function __construct( private HttpScheme $scheme, + private ?string $authority, ) {} /** @@ -22,6 +26,6 @@ public function __construct( */ public function create(Address $address, Invoke $invoke): string { - return \sprintf('%s://%s/%s', $this->scheme->value, $address->value, ltrim($invoke->method, '/')); + return \sprintf('%s://%s/%s', $this->scheme->value, $this->authority ?? $address->value, ltrim($invoke->method, '/')); } } diff --git a/packages/client/src/Client/Internal/TrackingSocketConnector.php b/packages/client/src/Client/Internal/TrackingSocketConnector.php new file mode 100644 index 0000000..0083cf7 --- /dev/null +++ b/packages/client/src/Client/Internal/TrackingSocketConnector.php @@ -0,0 +1,48 @@ + */ + private array $sockets = []; + + public function __construct( + private readonly SocketConnector $connector, + ) {} + + #[\Override] + public function connect(SocketAddress|string $uri, ?ConnectContext $context = null, ?Cancellation $cancellation = null): Socket + { + $socket = $this->connector->connect($uri, $context, $cancellation); + + $id = spl_object_id($socket); + $this->sockets[$id] = $socket; + $socket->onClose(function () use ($id): void { + unset($this->sockets[$id]); + }); + + return $socket; + } + + public function close(): void + { + foreach ($this->sockets as $socket) { + $socket->close(); + } + } +} diff --git a/packages/client/src/Client/Internal/TransportError.php b/packages/client/src/Client/Internal/TransportError.php new file mode 100644 index 0000000..8be1d4b --- /dev/null +++ b/packages/client/src/Client/Internal/TransportError.php @@ -0,0 +1,63 @@ +throwIfRequested(); + } catch (CancelledException $cancelled) { + return CancellationError::from($cancelled); + } + + if ($e instanceof CancelledException) { + return CancellationError::from($e); + } + + $code = Code::INTERNAL; + + for ($cause = $e; $cause !== null; $cause = $cause->getPrevious()) { + if ($cause instanceof Http2StreamException) { + $code = match ($cause->getCode()) { + Http2Parser::REFUSED_STREAM => Code::UNAVAILABLE, + Http2Parser::CANCEL => Code::CANCELLED, + Http2Parser::ENHANCE_YOUR_CALM => Code::RESOURCE_EXHAUSTED, + Http2Parser::INADEQUATE_SECURITY => Code::PERMISSION_DENIED, + default => Code::INTERNAL, + }; + + break; + } + + if ($cause instanceof SocketException || $cause instanceof Socket\SocketException) { + $code = Code::UNAVAILABLE; + } + } + + return new InvokeError($code, $e->getMessage(), previous: $e); + } +} diff --git a/packages/client/src/Client/LoadBalancer/PickFirst.php b/packages/client/src/Client/LoadBalancer/PickFirst.php index 51f52ef..c40bc3a 100644 --- a/packages/client/src/Client/LoadBalancer/PickFirst.php +++ b/packages/client/src/Client/LoadBalancer/PickFirst.php @@ -4,7 +4,6 @@ namespace Thesis\Grpc\Client\LoadBalancer; -use Random\Randomizer; use Thesis\Grpc\Client\Endpoint; use Thesis\Grpc\Client\LoadBalancer; use Thesis\Grpc\Client\PickContext; @@ -21,16 +20,18 @@ final class PickFirst implements LoadBalancer */ public function __construct( private array $endpoints, - private readonly Randomizer $randomizer, ) { - $this->current = $this->doPick($endpoints); + $this->current = $endpoints[0]; } #[\Override] public function refresh(array $endpoints): void { $this->endpoints = $endpoints; - $this->current = $this->doPick($endpoints, $this->current); + + if (!array_any($endpoints, $this->current->equals(...))) { + $this->current = $endpoints[0]; + } } #[\Override] @@ -45,19 +46,4 @@ public function pick(PickContext $context): Endpoint return $this->current; } - - /** - * @param non-empty-list $endpoints - */ - private function doPick(array $endpoints, ?Endpoint $current = null): Endpoint - { - /** @var non-empty-list $endpoints */ - $endpoints = $this->randomizer->shuffleArray($endpoints); - - if ($current === null || !array_any($endpoints, $current->equals(...))) { - $current = $endpoints[0]; - } - - return $current; - } } diff --git a/packages/client/src/Client/LoadBalancer/PickFirstFactory.php b/packages/client/src/Client/LoadBalancer/PickFirstFactory.php index 12bedc0..9f62447 100644 --- a/packages/client/src/Client/LoadBalancer/PickFirstFactory.php +++ b/packages/client/src/Client/LoadBalancer/PickFirstFactory.php @@ -4,7 +4,6 @@ namespace Thesis\Grpc\Client\LoadBalancer; -use Random\Randomizer; use Thesis\Grpc\Client\LoadBalancer; use Thesis\Grpc\Client\LoadBalancerFactory; @@ -13,10 +12,6 @@ */ final readonly class PickFirstFactory implements LoadBalancerFactory { - public function __construct( - private Randomizer $randomizer = new Randomizer(), - ) {} - #[\Override] public function name(): string { @@ -26,6 +21,6 @@ public function name(): string #[\Override] public function create(array $endpoints): LoadBalancer { - return new PickFirst($endpoints, $this->randomizer); + return new PickFirst($endpoints); } } diff --git a/packages/client/src/ClientStream.php b/packages/client/src/ClientStream.php index 37a3eb1..b621b00 100644 --- a/packages/client/src/ClientStream.php +++ b/packages/client/src/ClientStream.php @@ -31,6 +31,8 @@ public function receive(): object; /** * This method should be called after the ({@see send}) method has been called for the first time. * Otherwise, it will "block" the calling code. + * + * @throws InvokeError */ public function headers(): Metadata; @@ -38,6 +40,8 @@ public function headers(): Metadata; * This method should be called after the stream is closed ({@see close}) on either the client * or server side (when {@see receive} throws an exception). * Otherwise, it will "block" the calling code. + * + * @throws InvokeError */ public function trailers(Cancellation $cancellation = new NullCancellation()): Metadata; @@ -46,4 +50,11 @@ public function trailers(Cancellation $cancellation = new NullCancellation()): M * @throws InvokeError */ public function close(): void; + + /** + * @return \Traversable + * @throws InvokeError + */ + #[\Override] + public function getIterator(): \Traversable; } diff --git a/packages/grpc/composer.json b/packages/grpc/composer.json index 3907e5d..30cf608 100644 --- a/packages/grpc/composer.json +++ b/packages/grpc/composer.json @@ -22,7 +22,7 @@ "ext-ctype": "*", "amphp/amp": "^3.1", "amphp/byte-stream": "^2.1", - "amphp/pipeline": "^1.2", + "amphp/pipeline": "^1.2.1", "revolt/event-loop": "^1.0.8", "thesis/endian": "^0.3.3", "thesis/googleapis-rpc-types": "^0.3", diff --git a/phpstan.dist.neon b/phpstan.dist.neon index 1ceb813..4255561 100644 --- a/phpstan.dist.neon +++ b/phpstan.dist.neon @@ -7,6 +7,7 @@ parameters: level: max paths: - examples + - interop/bin - packages/grpc/src - packages/client/src - packages/server/src diff --git a/rector.php b/rector.php index 151451f..9ebc24e 100644 --- a/rector.php +++ b/rector.php @@ -8,6 +8,7 @@ __DIR__ . '/packages/client/src', __DIR__ . '/packages/server/src', __DIR__ . '/tests', + __DIR__ . '/interop/bin', ]) ->withParallel() ->withCache(__DIR__ . '/var/rector') diff --git a/tests/AuthorityTest.php b/tests/AuthorityTest.php new file mode 100644 index 0000000..fbaf881 --- /dev/null +++ b/tests/AuthorityTest.php @@ -0,0 +1,109 @@ +withAddresses(self::ADDRESS, new UnixAddress(self::SOCKET)) + ->withServices(new EchoServiceServerRegistry(new readonly class implements EchoServiceServer { + #[\Override] + public function echo(EchoRequest $request, Metadata $md, Cancellation $cancellation): EchoResponse + { + return new EchoResponse($request->sentence); + } + })) + ->withMiddlewares(new ClosureMiddleware(static function (Request $request, RequestHandler $next) use (&$authorities): Response { + $authorities[] = $request->getUri()->getAuthority(); + + return $next->handleRequest($request); + })); + + $client = new Client\Builder()->withHost($target); + + if ($tls) { + $builder = $builder->withTransportCredentials( + new Server\TransportCredentials()->withDefaultCertificate(new Certificate(self::tls('server.crt'), self::tls('server.key'))), + ); + $client = $client->withTransportCredentials( + new Client\TransportCredentials()->withCaCert(self::tls('ca.crt'))->withPeerName('localhost'), + ); + } + + $server = $builder->build(); + $server->start(); + + try { + new EchoServiceClient($client->build())->echo(new EchoRequest('ping')); + } finally { + $server->stop(); + } + + self::assertSame([$authority], $authorities); + } + + /** + * @return iterable + */ + public static function provideAuthorityCases(): iterable + { + yield 'host name, not the resolved address' => ['localhost:50075', 'localhost:50075']; + + yield 'dns scheme' => ['dns:///localhost:50075', 'localhost:50075']; + + yield 'ipv4' => ['ipv4:' . self::ADDRESS, self::ADDRESS]; + + yield 'first of several addresses, whichever the call goes to' => ['ipv4:' . self::DEAD_ADDRESS . ',' . self::ADDRESS, self::DEAD_ADDRESS]; + + yield 'unix socket' => ['unix://' . self::SOCKET, 'localhost']; + + // The peer name has no port: the amphp server fills in the port of the connection. + yield 'peer name of the transport credentials' => ['ipv4:' . self::ADDRESS, 'localhost:50075', true]; + } + + /** + * @param non-empty-string $filename + * @return non-empty-string + */ + private static function tls(string $filename): string + { + return __DIR__ . "/Stub/tls/{$filename}"; + } +} diff --git a/tests/Client/EndpointResolver/DnsResolverTest.php b/tests/Client/EndpointResolver/DnsResolverTest.php index 996e281..a9bf177 100644 --- a/tests/Client/EndpointResolver/DnsResolverTest.php +++ b/tests/Client/EndpointResolver/DnsResolverTest.php @@ -91,6 +91,36 @@ public static function provideResolveCases(): iterable ]; } + /** + * @param non-empty-string $target + */ + #[DataProvider('provideResolveIpLiteralCases')] + public function testResolveIpLiteral(string $target, Endpoint $endpoint): void + { + $deferredCancellation = new DeferredCancellation(); + + $resolution = new DnsResolver()->resolve( + Target::parse($target), + self::createStub(EndpointResolverListener::class), + $deferredCancellation->getCancellation(), + ); + $deferredCancellation->cancel(); + + self::assertEquals([$endpoint], $resolution->endpoints); + } + + /** + * @return iterable + */ + public static function provideResolveIpLiteralCases(): iterable + { + yield 'IPv4' => ['127.0.0.1:50051', new Endpoint(new Address('127.0.0.1:50051'))]; + + yield 'IPv6' => ['[::1]:50051', new Endpoint(new Address('[::1]:50051'))]; + + yield 'IPv6 with dns scheme' => ['dns:///[::1]:50051', new Endpoint(new Address('[::1]:50051'))]; + } + public function testResolveListener(): void { $deferredCancellation = new DeferredCancellation(); diff --git a/tests/Client/Internal/TransportErrorTest.php b/tests/Client/Internal/TransportErrorTest.php new file mode 100644 index 0000000..3e17805 --- /dev/null +++ b/tests/Client/Internal/TransportErrorTest.php @@ -0,0 +1,101 @@ +statusCode); + self::assertSame($e, $error->getPrevious()); + } + + /** + * @return iterable + */ + public static function provideTransportErrorCases(): iterable + { + foreach ([ + Http2Parser::GRACEFUL_SHUTDOWN => Code::INTERNAL, + Http2Parser::PROTOCOL_ERROR => Code::INTERNAL, + Http2Parser::INTERNAL_ERROR => Code::INTERNAL, + Http2Parser::FLOW_CONTROL_ERROR => Code::INTERNAL, + Http2Parser::SETTINGS_TIMEOUT => Code::INTERNAL, + Http2Parser::STREAM_CLOSED => Code::INTERNAL, + Http2Parser::FRAME_SIZE_ERROR => Code::INTERNAL, + Http2Parser::REFUSED_STREAM => Code::UNAVAILABLE, + Http2Parser::CANCEL => Code::CANCELLED, + Http2Parser::COMPRESSION_ERROR => Code::INTERNAL, + Http2Parser::CONNECT_ERROR => Code::INTERNAL, + Http2Parser::ENHANCE_YOUR_CALM => Code::RESOURCE_EXHAUSTED, + Http2Parser::INADEQUATE_SECURITY => Code::PERMISSION_DENIED, + Http2Parser::HTTP_1_1_REQUIRED => Code::INTERNAL, + 0xFF => Code::INTERNAL, + ] as $http2 => $code) { + yield "RST_STREAM {$http2}" => [self::reset($http2), $code]; + } + + yield 'reset in the body' => [new StreamException('HTTP response did not complete', previous: self::reset(Http2Parser::CANCEL)), Code::CANCELLED]; + + yield 'connection error' => [new SocketException('Connection error', Http2Parser::PROTOCOL_ERROR, new Http2ConnectionException('Protocol error', Http2Parser::PROTOCOL_ERROR)), Code::UNAVAILABLE]; + + yield 'connection lost' => [new SocketException('Connection closed unexpectedly', Http2Parser::INTERNAL_ERROR), Code::UNAVAILABLE]; + + yield 'connection lost in the body' => [new StreamException('HTTP response did not complete', previous: new SocketException('Connection closed unexpectedly')), Code::UNAVAILABLE]; + + yield 'connection refused' => [new SocketException('Connection refused'), Code::UNAVAILABLE]; + + yield 'connect failed' => [new ConnectException('Connection refused'), Code::UNAVAILABLE]; + + yield 'other http error' => [new HttpException('Invalid response'), Code::INTERNAL]; + + yield 'cancelled' => [new CancelledException(), Code::CANCELLED]; + + yield 'deadline' => [new CancelledException(new TimeoutException()), Code::DEADLINE_EXCEEDED]; + } + + public function testInvokeErrorIsKept(): void + { + $e = new InvokeError(Code::NOT_FOUND); + + self::assertSame($e, TransportError::from($e, new NullCancellation())); + } + + public function testRequestedCancellationWins(): void + { + $cancellation = new DeferredCancellation(); + $cancellation->cancel(new TimeoutException()); + + $error = TransportError::from(new SocketException('Connection closed unexpectedly'), $cancellation->getCancellation()); + + self::assertSame(Code::DEADLINE_EXCEEDED, $error->statusCode); + } + + private static function reset(int $code): SocketException + { + return new SocketException('Stream closed by server', $code, new Http2StreamException('Stream closed by server', 1, $code)); + } +} diff --git a/tests/Client/LoadBalancer/PickFirstTest.php b/tests/Client/LoadBalancer/PickFirstTest.php index c4fe524..96d272e 100644 --- a/tests/Client/LoadBalancer/PickFirstTest.php +++ b/tests/Client/LoadBalancer/PickFirstTest.php @@ -118,6 +118,19 @@ public static function provideRefreshSwitchesPinnedEndpointCases(): iterable ]; } + public function testPickFollowsResolutionOrder(): 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([$b, $a]); + self::assertTrue($b->equals($balancer->pick(self::context()))); + + $balancer->refresh([$c, $a]); + self::assertTrue($c->equals($balancer->pick(self::context()))); + } + public function testPickFailsOverToALiveEndpointWhenPinnedIsExcluded(): void { $a = new Endpoint(new Address('10.0.0.1:50051')); diff --git a/tests/ClientCancellationTest.php b/tests/ClientCancellationTest.php index 6d1ca3a..c257cab 100644 --- a/tests/ClientCancellationTest.php +++ b/tests/ClientCancellationTest.php @@ -8,6 +8,11 @@ use Amp\CancelledException; use Amp\DeferredCancellation; use Amp\TimeoutCancellation; +use Chat\Api\V1\Message; +use Chat\Api\V1\MessengerServiceClient; +use Chat\Api\V1\MessengerServiceServer; +use Chat\Api\V1\MessengerServiceServerRegistry; +use Composer\InstalledVersions; use Echos\Api\V1\EchoRequest; use Echos\Api\V1\EchoResponse; use Echos\Api\V1\EchoServiceClient; @@ -15,6 +20,7 @@ use Echos\Api\V1\EchoServiceServerRegistry; use File\Api\V1\FileServiceClient; use PHPUnit\Framework\Attributes\CoversClass; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; use Revolt\EventLoop; use Thesis\Google\Protobuf\Timestamp; @@ -28,12 +34,15 @@ use Thesis\Grpc\Client\Internal\Http2\ConcurrentClientStream; use Thesis\Grpc\Client\Resolution; use Thesis\Grpc\Client\Target; +use Thesis\Grpc\Server\BidirectionalStreamChannel; use Topic\Api\V1\Event; use Topic\Api\V1\SubscribeRequest; use Topic\Api\V1\TopicServiceClient; use Topic\Api\V1\TopicServiceServer; use Topic\Api\V1\TopicServiceServerRegistry; +use function Amp\async; use function Amp\delay; +use function Amp\Socket\listen; #[CoversClass(AmphpHttpClient::class)] #[CoversClass(ConcurrentClientStream::class)] @@ -41,6 +50,7 @@ final class ClientCancellationTest extends TestCase { private const string ADDRESS = '127.0.0.1:50072'; + private const string SILENT_ADDRESS = '127.0.0.1:50073'; private Server $server; @@ -51,6 +61,7 @@ protected function setUp(): void ->withServices( new EchoServiceServerRegistry(new SlowEchoServer()), new TopicServiceServerRegistry(new SlowTopicServer()), + new MessengerServiceServerRegistry(new SilentMessengerServer()), ) ->build(); @@ -128,6 +139,73 @@ public function testClientStreamCancellationBeforeCloseIsCancelled(): void self::assertSame(Code::CANCELLED, $error->statusCode); } + /** + * The server accepts the connection and never answers, so only the client can enforce the deadline it sends. + * + * @param \Closure(Client, Metadata, Cancellation): mixed $call + */ + #[DataProvider('provideDeadlineFromHeaderIsEnforcedByClientCases')] + public function testDeadlineFromHeaderIsEnforcedByClient(\Closure $call): void + { + $server = listen(self::SILENT_ADDRESS); + $sockets = []; + async(static function () use ($server, &$sockets): void { + while (($socket = $server->accept()) !== null) { + $sockets[] = $socket; + } + })->ignore(); + + $start = hrtime(true); + + try { + $error = self::catch(static fn() => $call( + new Client\Builder()->withHost(self::SILENT_ADDRESS)->build(), + new Metadata()->withKey(Metadata\Timeout::milliseconds(100)), + new TimeoutCancellation(5), + )); + } finally { + $server->close(); + } + + self::assertSame(Code::DEADLINE_EXCEEDED, $error->statusCode); + self::assertLessThan(1, (hrtime(true) - $start) / 1_000_000_000); + } + + /** + * @return iterable + */ + public static function provideDeadlineFromHeaderIsEnforcedByClientCases(): iterable + { + yield 'unary' => [ + static fn(Client $client, Metadata $md, Cancellation $cancellation) => new EchoServiceClient($client) + ->echo(new EchoRequest('ping'), $md, $cancellation), + ]; + + yield 'stream' => [ + static fn(Client $client, Metadata $md, Cancellation $cancellation) => new MessengerServiceClient($client) + ->chat($md, $cancellation) + ->receive(), + ]; + } + + public function testClientCloseCancelsActiveStream(): void + { + if (version_compare(InstalledVersions::getVersion('amphp/http-client') ?? '', '5.3.7', '<')) { + self::markTestSkipped('amphp/http-client < 5.3.7 throws UnhandledFutureError from Http2Stream::$requestBodyCompletion when a connection with an active stream is closed, fixed upstream in 423f023.'); + } + + $client = new Client\Builder()->withHost(self::ADDRESS)->build(); + $stream = new MessengerServiceClient($client)->chat(); + $stream->send(new Message('hello')); + + $receive = async($stream->receive(...)); + delay(0.1); + + $client->close(); + + self::assertSame(Code::CANCELLED, self::catch($receive->await(...))->statusCode); + } + public function testStreamIterationDeadlineIsDeadlineExceeded(): void { $stream = self::topicClient()->subscribe(new SubscribeRequest('events'), cancellation: new TimeoutCancellation(0.2)); @@ -187,6 +265,15 @@ public function subscribe(SubscribeRequest $request, Metadata $md, Cancellation } } +final readonly class SilentMessengerServer implements MessengerServiceServer +{ + #[\Override] + public function chat(BidirectionalStreamChannel $stream, Metadata $md, Cancellation $cancellation): void + { + foreach ($stream as $_); + } +} + final readonly class SlowResolver implements EndpointResolver { /** diff --git a/tests/ClientCloseTest.php b/tests/ClientCloseTest.php new file mode 100644 index 0000000..d8569cf --- /dev/null +++ b/tests/ClientCloseTest.php @@ -0,0 +1,65 @@ +withAddresses(self::ADDRESS) + ->withServices(new EchoServiceServerRegistry(new readonly class implements EchoServiceServer { + #[\Override] + public function echo(EchoRequest $request, Metadata $md, Cancellation $cancellation): EchoResponse + { + return new EchoResponse($request->sentence); + } + })) + ->build(); + + $server->start(); + + try { + $client = new Client\Builder()->withHost(self::ADDRESS)->build(); + new EchoServiceClient($client)->echo(new EchoRequest('ping')); + self::assertSame(1, self::clientConnections()); + + // Checked right away: the socket must be closed by the client, not after the server reacts to GOAWAY. + $client->close(); + self::assertSame(0, self::clientConnections()); + } finally { + $server->stop(); + } + } + + /** + * Established TCP connections to the server port, i.e. the client side only. + */ + private static function clientConnections(): int + { + $lines = file('/proc/net/tcp'); + self::assertNotFalse($lines); + + $pattern = \sprintf('/^\s*\d+: \S+ [0-9A-F]+:%04X 01 /', self::PORT); + + return \count(array_filter($lines, static fn(string $line) => preg_match($pattern, $line) === 1)); + } +} diff --git a/tests/GracefulShutdownTest.php b/tests/GracefulShutdownTest.php index 163810d..bbdce7b 100644 --- a/tests/GracefulShutdownTest.php +++ b/tests/GracefulShutdownTest.php @@ -14,6 +14,7 @@ use Echos\Api\V1\EchoServiceServerRegistry; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; +use Thesis\Google\Rpc\Code; use Thesis\Grpc\Server\Internal\AmphpHttpServer; use function Amp\async; use function Amp\delay; @@ -52,7 +53,12 @@ public function testGracefulShutdown(): void $this->server->stop(); $failFuture = async($client->echo(...), new EchoRequest('Hello, gRPC')); self::assertEquals(new EchoResponse('Hello, gRPC'), $successFuture->await()); - $this->expectExceptionMessage("Connection to '127.0.0.1:50051' failed"); - $failFuture->await(); + + try { + $failFuture->await(); + self::fail('Expected an InvokeError.'); + } catch (InvokeError $e) { + self::assertSame(Code::UNAVAILABLE, $e->statusCode); + } } } diff --git a/tests/GrpcTimeoutTest.php b/tests/GrpcTimeoutTest.php index 8d779f1..5657389 100644 --- a/tests/GrpcTimeoutTest.php +++ b/tests/GrpcTimeoutTest.php @@ -12,6 +12,7 @@ use Echos\Api\V1\EchoServiceServerRegistry; use PHPUnit\Framework\Attributes\CoversClass; use PHPUnit\Framework\TestCase; +use Thesis\Google\Rpc\Code; use Thesis\Grpc\Client\Internal\AmphpHttpClient; use Thesis\Grpc\Metadata\Timeout; use Thesis\Grpc\Server\Internal\AmphpHttpServer; @@ -56,7 +57,11 @@ public function testServerTimeout(): void { $client = new EchoServiceClient(new Client\Builder()->build()); - $this->expectExceptionMessage('A grpc error with status code "DEADLINE_EXCEEDED" and message "" occurred'); - $client->echo(new EchoRequest('Hello, gRPC'), new Metadata()->withKey(Timeout::milliseconds(200))); + try { + $client->echo(new EchoRequest('Hello, gRPC'), new Metadata()->withKey(Timeout::milliseconds(200))); + self::fail('InvokeError expected.'); + } catch (InvokeError $e) { + self::assertSame(Code::DEADLINE_EXCEEDED, $e->statusCode); + } } } diff --git a/tests/Internal/Http2/UriFactoryTest.php b/tests/Internal/Http2/UriFactoryTest.php index 84ff91c..0b42ef6 100644 --- a/tests/Internal/Http2/UriFactoryTest.php +++ b/tests/Internal/Http2/UriFactoryTest.php @@ -20,6 +20,7 @@ final class UriFactoryTest extends TestCase * @param non-empty-string $address * @param non-empty-string $method * @param non-empty-string $uri + * @param ?non-empty-string $authority */ #[TestWith([ 'http', @@ -45,13 +46,21 @@ final class UriFactoryTest extends TestCase 'test.api.v1.EchoController/Echo', 'http://192.168.0.1:50051/test.api.v1.EchoController/Echo', ])] + #[TestWith([ + 'https', + '192.168.0.1:50051', + 'test.api.v1.EchoController/Echo', + 'https://echo.example.com:50051/test.api.v1.EchoController/Echo', + 'echo.example.com:50051', + ])] public function testCreate( string $scheme, string $address, string $method, string $uri, + ?string $authority = null, ): void { - $factory = new UriFactory(HttpScheme::from($scheme)); + $factory = new UriFactory(HttpScheme::from($scheme), $authority); $invoke = new Invoke($method, \stdClass::class, RpcType::Unary); self::assertSame($uri, $factory->create(new Address($address), $invoke)); diff --git a/tests/LoadBalancingTest.php b/tests/LoadBalancingTest.php index e053523..46dc817 100644 --- a/tests/LoadBalancingTest.php +++ b/tests/LoadBalancingTest.php @@ -11,7 +11,9 @@ use Echos\Api\V1\EchoServiceServer; use Echos\Api\V1\EchoServiceServerRegistry; use PHPUnit\Framework\Attributes\CoversClass; +use PHPUnit\Framework\Attributes\DataProvider; use PHPUnit\Framework\TestCase; +use Thesis\Google\Rpc\Code; use Thesis\Grpc\Client\Internal\AmphpHttpClient; use Thesis\Grpc\Client\Internal\Connection\DefaultConnection; use Thesis\Grpc\Client\LoadBalancer\RoundRobin; @@ -22,6 +24,9 @@ #[CoversClass(RoundRobin::class)] final class LoadBalancingTest extends TestCase { + /** Nothing listens here. */ + private const string DEAD_ADDRESS = '127.0.0.1:50079'; + private Server $server1; private Server $server2; @@ -48,6 +53,41 @@ protected function tearDown(): void $this->server2->stop(); } + /** + * @param non-empty-string $target + */ + #[DataProvider('providePickFirstSkipsUnreachableEndpointCases')] + public function testPickFirstSkipsUnreachableEndpoint(string $target): void + { + $client = new EchoServiceClient(new Client\Builder()->withHost($target)->build()); + + for ($i = 0; $i < 3; ++$i) { + self::assertSame('[server-1] ping', $client->echo(new EchoRequest('ping'))->sentence); + } + } + + /** + * @return iterable + */ + public static function providePickFirstSkipsUnreachableEndpointCases(): iterable + { + yield 'unreachable first' => ['ipv4:' . self::DEAD_ADDRESS . ',127.0.0.1:50051']; + + yield 'unreachable last' => ['ipv4:127.0.0.1:50051,' . self::DEAD_ADDRESS]; + } + + public function testEveryEndpointUnreachableIsUnavailable(): void + { + $client = new EchoServiceClient(new Client\Builder()->withHost('ipv4:' . self::DEAD_ADDRESS)->build()); + + try { + $client->echo(new EchoRequest('ping')); + self::fail('InvokeError expected.'); + } catch (InvokeError $e) { + self::assertSame(Code::UNAVAILABLE, $e->statusCode); + } + } + public function testRoundRobinDistributesRequests(): void { $client = new EchoServiceClient( diff --git a/tests/TransportErrorTest.php b/tests/TransportErrorTest.php new file mode 100644 index 0000000..e0161d6 --- /dev/null +++ b/tests/TransportErrorTest.php @@ -0,0 +1,181 @@ +accept()) !== null) { + async(self::serve(...), $socket, $respond)->ignore(); + } + })->ignore(); + + $client = new Client\Builder()->withHost(self::ADDRESS)->build(); + + try { + $call($client); + self::fail('Expected an InvokeError.'); + } catch (InvokeError $e) { + self::assertSame($code, $e->statusCode, $e->getMessage()); + } finally { + $client->close(); + $server->close(); + } + } + + /** + * @return iterable + */ + public static function provideTransportErrorCases(): iterable + { + $calls = [ + 'unary' => static fn(Client $client) => new EchoServiceClient($client)->echo(new EchoRequest('ping')), + 'receive' => static fn(Client $client) => self::chat($client)->receive(), + 'foreach' => static function (Client $client): void { + foreach (self::chat($client) as $_); + }, + ]; + + $resets = [ + 'headers' => static fn(Client $client) => self::chat($client)->headers(), + 'trailers' => static fn(Client $client) => self::chat($client)->trailers(), + ]; + + $scenarios = [ + 'RST_STREAM(CANCEL)' => [self::reset(Http2Parser::CANCEL), Code::CANCELLED, [...$calls, ...$resets]], + 'RST_STREAM(ENHANCE_YOUR_CALM)' => [self::reset(Http2Parser::ENHANCE_YOUR_CALM), Code::RESOURCE_EXHAUSTED, [...$calls, ...$resets]], + 'GOAWAY' => [ + static fn(Socket $socket) => $socket->write(Http2Parser::compileFrame(pack('NN', 0, Http2Parser::GRACEFUL_SHUTDOWN), Http2Parser::GOAWAY, Http2Parser::NO_FLAG)), + Code::UNAVAILABLE, + $calls, + ], + 'connection closed' => [static fn(Socket $socket) => $socket->close(), Code::UNAVAILABLE, $calls], + 'HTTP 503' => [self::httpStatus(503), Code::UNAVAILABLE, $calls], + 'HTTP 404' => [self::httpStatus(404), Code::UNIMPLEMENTED, $calls], + ]; + + foreach ($scenarios as $scenario => [$respond, $code, $cases]) { + foreach ($cases as $case => $call) { + yield "{$scenario} on {$case}" => [$respond, $call, $code]; + } + } + } + + /** + * A bare HTTP/2 server: it acknowledges the client's settings and answers the first stream with $respond + * once the request is sent. An answer to an unfinished request trips amphp/http-client < 5.3.7, which leaves + * the failed request body future unhandled (fixed upstream in 423f023). + * + * @param \Closure(Socket): void $respond + */ + private static function serve(Socket $socket, \Closure $respond): void + { + $buffer = ''; + $read = static function (int $length) use ($socket, &$buffer): ?string { + while (\strlen($buffer) < $length) { + if (($chunk = $socket->read()) === null) { + return null; + } + + $buffer .= $chunk; + } + + $data = substr($buffer, 0, $length); + $buffer = substr($buffer, $length); + + return $data; + }; + + if ($read(\strlen(Http2Parser::PREFACE)) === null) { + return; + } + + $socket->write(Http2Parser::compileFrame('', Http2Parser::SETTINGS, Http2Parser::NO_FLAG)); + + while (($header = $read(9)) !== null) { + /** @var array{length: int, type: int, flags: int, stream: int} $frame */ + $frame = unpack('Nlength/Ctype/Cflags/Nstream', "\0{$header}"); + + if ($read($frame['length']) === null) { + return; + } + + if ($frame['type'] === Http2Parser::SETTINGS && ($frame['flags'] & Http2Parser::ACK) === 0) { + $socket->write(Http2Parser::compileFrame('', Http2Parser::SETTINGS, Http2Parser::ACK)); + } + + if ($frame['stream'] === 1 && ($frame['flags'] & Http2Parser::END_STREAM) !== 0) { + $respond($socket); + } + } + } + + /** + * @return \Closure(Socket): void + */ + private static function reset(int $code): \Closure + { + return static fn(Socket $socket) => $socket->write(Http2Parser::compileFrame(pack('N', $code), Http2Parser::RST_STREAM, Http2Parser::NO_FLAG, 1)); + } + + /** + * @return \Closure(Socket): void + */ + private static function httpStatus(int $status): \Closure + { + return static fn(Socket $socket) => $socket->write(Http2Parser::compileFrame( + new HPack()->encode([[':status', (string) $status]]), + Http2Parser::HEADERS, + Http2Parser::END_HEADERS | Http2Parser::END_STREAM, + 1, + )); + } + + /** + * @return ClientStream + */ + private static function chat(Client $client): ClientStream + { + /** @var Client\Invoke $invoke */ + $invoke = new Client\Invoke('/chat.api.v1.MessengerService/Chat', Message::class, RpcType::BidirectionalStream); + + $stream = $client->createStream($invoke); + $stream->send(new Message('hello')); + $stream->close(); + + return $stream; + } +}