diff --git a/README.md b/README.md index ea71632bd..bbf988c69 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,18 @@ If you are using the SDK without integrations, the following sections of the doc - [How to connect a Temporal Client to a Temporal Service](https://docs.temporal.io/develop/php/temporal-clients#connect-to-a-dev-cluster) - [How to start a Workflow Execution](https://docs.temporal.io/develop/php/temporal-clients#start-workflow-execution) +Administrative operator APIs are available through [`Temporal\Client\GRPC\OperatorClient`](https://php.temporal.io/): + +```php +use Temporal\Api\Operatorservice\V1\DeleteNamespaceRequest; +use Temporal\Client\GRPC\OperatorClient; + +$operatorClient = OperatorClient::create('127.0.0.1:7233'); +$response = $operatorClient->DeleteNamespace( + (new DeleteNamespaceRequest())->setNamespace('example-namespace'), +); +``` + > [!NOTE] > Check out [the repository with examples](https://github.com/temporalio/samples-php) of using the PHP SDK. diff --git a/resources/scripts/generate-client.php b/resources/scripts/generate-client.php index 852a49a86..6f8adee5a 100644 --- a/resources/scripts/generate-client.php +++ b/resources/scripts/generate-client.php @@ -10,220 +10,326 @@ use Grpc\BaseStub; use Laminas\Code\Generator; use Laminas\Code\Generator\MethodGenerator; -use Temporal\Api\Workflowservice; +use Temporal\Api\Operatorservice\V1\OperatorServiceClient; +use Temporal\Api\Workflowservice\V1\WorkflowServiceClient; +use Temporal\Client\GRPC\GrpcClientInterface; use Temporal\Client\Common\ServerCapabilities; -use Temporal\Client\GRPC\Connection\ConnectionInterface; use Temporal\Client\GRPC\ContextInterface; +use Laminas\Code\DeclareStatement; require __DIR__ . '/../../vendor/autoload.php'; -echo "Compiling client...\n"; - -echo "reading client schema: "; - -$r = new ReflectionClass(Workflowservice\V1\WorkflowServiceClient::class); -$rBase = new ReflectionClass(BaseStub::class); +echo "Compiling clients...\n"; $ctxParam = Generator\ParameterGenerator::fromArray( [ - 'type' => \Temporal\Client\GRPC\ContextInterface::class, + 'type' => '?' . ContextInterface::class, 'name' => 'ctx', 'defaultValue' => null, - ] + ], ); +$addressParam = Generator\ParameterGenerator::fromArray(['type' => 'string', 'name' => 'address']); +$optionsParam = Generator\ParameterGenerator::fromArray(['type' => 'array', 'name' => 'options']); + +$buildMethodDocBlock = static function ( + ReflectionClass $serviceReflection, + string $method, + string $arg, + string $return, +): string { + $experimentalNotices = [ + 'Experimental. This API might significantly change or be removed in a future release.', + 'NOTE: Experimental API.', + ]; -$methodDocBlock = function (ReflectionClass $r, string $method, string $arg, string $return) { $block = []; - // copy from existing doc block - $orig = $r->getMethod($method)->getDocComment(); - foreach (explode("\n", $orig) as $line) { - $line = trim($line, "\n\r* "); + $original = $serviceReflection->getMethod($method)->getDocComment(); + foreach (\explode("\n", (string) $original) as $line) { + $line = \trim($line, "\n\r* "); if ($line === '/') { continue; } - if (substr($line, 0, 1) === '@') { + if (\str_starts_with($line, '@')) { break; } + if (\in_array($line, $experimentalNotices, true)) { + continue; + } + $block[] = $line; } + while ($block !== [] && $block[\array_key_last($block)] === '') { + \array_pop($block); + } + $block[] = ''; - $block[] = sprintf('@param \\%s $arg', $arg); - $block[] = sprintf('@param ContextInterface|null $ctx'); - $block[] = sprintf('@return \\%s', $return); - $block[] = sprintf('@throws ServiceClientException'); + $block[] = '@throws ServiceClientException'; - return join("\n", $block); + return \implode("\n", $block); }; -$methods = []; +$baseStubReflection = new ReflectionClass(BaseStub::class); +$buildRpcMap = static function (ReflectionClass $serviceReflection) use ($baseStubReflection): array { + $methods = []; -// fetching available methods + foreach ($serviceReflection->getMethods() as $method) { + if ($baseStubReflection->hasMethod($method->getName())) { + continue; + } -foreach ($r->getMethods() as $m) { - if ($rBase->hasMethod($m->getName())) { - continue; - } + $request = $method->getParameters()[0]->getType()?->getName(); + $response = $request === null ? null : \substr($request, 0, -7) . 'Response'; - $method = [ - 'request' => null, - 'response' => null, - ]; + \assert($request !== null); + \assert(\class_exists($request)); + \assert(\is_string($response) && \class_exists($response)); - // simple heuristics - $method['request'] = $m->getParameters()[0]->getType()->getName(); - $method['response'] = substr($method['request'], 0, -7) . 'Response'; + $methods[$method->getName()] = [ + 'request' => $request, + 'response' => $response, + ]; + } - assert(class_exists($method['request'])); - assert(class_exists($method['response'])); + return $methods; +}; - $methods[$m->getName()] = $method; -} +$buildCreateServiceClientMethod = static function ( + string $serviceClientFqcn, +) use ($addressParam, $optionsParam): MethodGenerator { + $method = new MethodGenerator( + 'createGrpcStub', + [$addressParam, $optionsParam], + MethodGenerator::FLAG_PROTECTED | MethodGenerator::FLAG_STATIC, + ); + $method->setReturnType('\Grpc\BaseStub'); + $method->setBody(\sprintf('return new \\%s($address, $options);', $serviceClientFqcn)); -echo "[OK]\n"; + return $method; +}; -echo "generating interface: "; +$buildGetServerCapabilitiesInterfaceMethod = static function (): MethodGenerator { + $method = new MethodGenerator('getServerCapabilities'); + $method->setReturnType('?\Temporal\Client\Common\ServerCapabilities'); -$interface = new Generator\InterfaceGenerator('ServiceClientInterface'); + return $method; +}; +$buildGetServerCapabilitiesImplementationMethod = static function (): MethodGenerator { + $method = new MethodGenerator('getServerCapabilities'); + $method->setReturnType('?\Temporal\Client\Common\ServerCapabilities'); + $method->setBody(<<<'PHP' +$connection = $this->getInternalConnection(); +if ($connection->getCapabilities() !== null) { + return $connection->getCapabilities(); +} -// getContext(): ContextInterface -$m = new MethodGenerator( - 'getContext', - [], - MethodGenerator::FLAG_PUBLIC, -); -$m->setReturnType(ContextInterface::class); -$interface->addMethodFromGenerator($m); -// withContext(ContextInterface $context): static -$m = new MethodGenerator( - 'withContext', - [Generator\ParameterGenerator::fromArray(['type' => ContextInterface::class, 'name' => 'context'])], - MethodGenerator::FLAG_PUBLIC, -); -$m->setReturnType('static'); -$interface->addMethodFromGenerator($m); -// withAuthKey(string $key): static -$m = new MethodGenerator( - 'withAuthKey', - [Generator\ParameterGenerator::fromArray(['type' => '\Stringable|string', 'name' => 'key'])], - MethodGenerator::FLAG_PUBLIC, -); -$m->setReturnType('static'); -$interface->addMethodFromGenerator($m); -// public function getConnection(): ConnectionInterface -$m = new MethodGenerator( - 'getConnection', - [], - MethodGenerator::FLAG_PUBLIC, -); -$m->setReturnType(ConnectionInterface::class); -$interface->addMethodFromGenerator($m); -// Add Capability methods -$m = new MethodGenerator( - 'getServerCapabilities', - [], - MethodGenerator::FLAG_PUBLIC, -); -$m->setReturnType('?' . ServerCapabilities::class); -$interface->addMethodFromGenerator($m); +try { + $systemInfo = $this->getSystemInfo(new V1\GetSystemInfoRequest()); + $capabilities = $systemInfo->getCapabilities(); -foreach ($methods as $method => $options) { - $m = new MethodGenerator($method); + if ($capabilities === null) { + return null; + } - $m->setDocBlock(($methodDocBlock)($r, $method, $options['request'], $options['response'])); - $m->setParameters( - [ - Generator\ParameterGenerator::fromArray(['type' => $options['request'], 'name' => 'arg']), - $ctxParam - ] + $serverCapabilities = new ServerCapabilities( + signalAndQueryHeader: $capabilities->getSignalAndQueryHeader(), + internalErrorDifferentiation: $capabilities->getInternalErrorDifferentiation(), + activityFailureIncludeHeartbeat: $capabilities->getActivityFailureIncludeHeartbeat(), + supportsSchedules: $capabilities->getSupportsSchedules(), + encodedFailureAttributes: $capabilities->getEncodedFailureAttributes(), + buildIdBasedVersioning: $capabilities->getBuildIdBasedVersioning(), + upsertMemo: $capabilities->getUpsertMemo(), + eagerWorkflowStart: $capabilities->getEagerWorkflowStart(), + sdkMetadata: $capabilities->getSdkMetadata(), + countGroupByExecutionStatus: $capabilities->getCountGroupByExecutionStatus(), + nexus: $capabilities->getNexus(), ); - $m->setReturnType($options['response']); - - $interface->addMethodFromGenerator($m); -} + $connection->setCapabilities($serverCapabilities); -$m = new MethodGenerator( - 'close', - [], - MethodGenerator::FLAG_PUBLIC, - null, - 'Close the communication channel associated with this stub.' -); -$m->setReturnType('void'); -$interface->addMethodFromGenerator($m); + return $serverCapabilities; +} catch (ServiceClientException $e) { + if ($e->getCode() === StatusCode::UNIMPLEMENTED) { + return null; + } -echo "[OK]\n"; -echo "writing interface: "; + throw $e; +} +PHP); -$file = new Generator\FileGenerator(); -$file->setNamespace('Temporal\\Client\\GRPC'); -$file->setClass($interface); -$file->setUses( - [ - 'Temporal\Api\Workflowservice\V1', - 'Temporal\Exception\Client\ServiceClientException', - ] -); + return $method; +}; -// write and shorten names -file_put_contents( - __DIR__ . '/../../src/Client/GRPC/ServiceClientInterface.php', - str_replace( - ['\\Temporal\\Api\\Workflowservice\\', '\\Temporal\\Client\\GRPC\\ContextInterface'], - ['', 'ContextInterface'], - $file->generate() - ) +$buildSetServerCapabilitiesMethod = static function (): MethodGenerator { + $method = new MethodGenerator( + 'setServerCapabilities', + [Generator\ParameterGenerator::fromArray(['type' => '\Temporal\Client\Common\ServerCapabilities', 'name' => 'capabilities'])], + ); + $method->setReturnType('void'); + $method->setBody(<<<'PHP' +\trigger_error( + 'Method ' . __METHOD__ . ' is deprecated and will be removed in the next major release.', + \E_USER_DEPRECATED, ); -echo "[OK]\n"; +$this->getInternalConnection()->setCapabilities($capabilities); +PHP); -echo "generating implementation: "; - -$impl = new Generator\ClassGenerator('ServiceClient'); -$impl->setExtendedClass('BaseClient'); + return $method; +}; -foreach ($methods as $method => $options) { - $m = new MethodGenerator($method); +$clients = [ + [ + 'label' => 'workflow', + 'serviceClass' => WorkflowServiceClient::class, + 'apiNamespace' => 'Temporal\\Api\\Workflowservice\\V1', + 'interfaceName' => 'ServiceClientInterface', + 'implementationName' => 'ServiceClient', + 'interfaceFile' => __DIR__ . '/../../src/Client/GRPC/ServiceClientInterface.php', + 'implementationFile' => __DIR__ . '/../../src/Client/GRPC/ServiceClient.php', + 'extraUses' => [ + 'interface' => [ + 'Temporal\Client\Common\ServerCapabilities', + ], + 'implementation' => [ + 'Temporal\Client\Common\ServerCapabilities', + ], + ], + 'extraInterfaceMethods' => static fn(): array => [$buildGetServerCapabilitiesInterfaceMethod()], + 'extraImplementationMethods' => static fn(): array => [ + $buildGetServerCapabilitiesImplementationMethod(), + $buildSetServerCapabilitiesMethod(), + $buildCreateServiceClientMethod(WorkflowServiceClient::class), + ], + ], + [ + 'label' => 'operator', + 'serviceClass' => OperatorServiceClient::class, + 'apiNamespace' => 'Temporal\\Api\\Operatorservice\\V1', + 'interfaceName' => 'OperatorClientInterface', + 'implementationName' => 'OperatorClient', + 'interfaceFile' => __DIR__ . '/../../src/Client/GRPC/OperatorClientInterface.php', + 'implementationFile' => __DIR__ . '/../../src/Client/GRPC/OperatorClient.php', + 'extraUses' => [ + 'interface' => [], + 'implementation' => [], + ], + 'extraInterfaceMethods' => static fn(): array => [], + 'extraImplementationMethods' => static fn(): array => [ + $buildCreateServiceClientMethod(OperatorServiceClient::class), + ], + ], +]; + +$generatedFiles = []; + +foreach ($clients as $client) { + echo "reading {$client['label']} schema: "; + $serviceReflection = new ReflectionClass($client['serviceClass']); + $methods = $buildRpcMap($serviceReflection); + echo "[OK]\n"; + + $apiAliasStrip = '\\' . \substr($client['apiNamespace'], 0, (int) \strrpos($client['apiNamespace'], '\\')) . '\\'; + + // Common uses for all generated files + $commonUses = ['Temporal\Exception\Client\ServiceClientException']; + + echo "generating {$client['interfaceName']}: "; + $interface = new Generator\InterfaceGenerator($client['interfaceName']); + $interface->setImplementedInterfaces([GrpcClientInterface::class]); + + foreach (($client['extraInterfaceMethods'])() as $method) { + $interface->addMethodFromGenerator($method); + } - $m->setDocBlock(($methodDocBlock)($r, $method, $options['request'], $options['response'])); - $m->setParameters( - [ + foreach ($methods as $name => $options) { + $method = new MethodGenerator($name); + $method->setDocBlock($buildMethodDocBlock($serviceReflection, $name, $options['request'], $options['response'])); + $method->setParameters([ Generator\ParameterGenerator::fromArray(['type' => $options['request'], 'name' => 'arg']), - $ctxParam - ] - ); - $m->setReturnType($options['response']); + $ctxParam, + ]); + $method->setReturnType($options['response']); + $interface->addMethodFromGenerator($method); + } - $m->setBody(sprintf('return $this->invoke("%s", $arg, $ctx);', $m->getName())); + echo "[OK]\n"; + + echo "writing {$client['interfaceName']}: "; + $file = new Generator\FileGenerator(); + $file->setNamespace('Temporal\\Client\\GRPC'); + $file->setDeclares([DeclareStatement::fromArray(['strict_types' => 1])]); + $file->setClass($interface); + $file->setUses([$client['apiNamespace'], ...$commonUses, ...($client['extraUses']['interface'] ?? [])]); + + \file_put_contents( + $client['interfaceFile'], + \str_replace( + [$apiAliasStrip, '\\Temporal\\Client\\GRPC\\', '\\Temporal\\Client\\Common\\'], + ['', '', ''], + $file->generate(), + ), + ); + $generatedFiles[] = $client['interfaceFile']; + echo "[OK]\n"; + + echo "generating {$client['implementationName']}: "; + $implementation = new Generator\ClassGenerator($client['implementationName']); + $implementation->setExtendedClass('BaseClient'); + $implementation->setImplementedInterfaces([$client['interfaceName']]); + + foreach ($methods as $name => $options) { + $method = new MethodGenerator($name); + $method->setDocBlock($buildMethodDocBlock($serviceReflection, $name, $options['request'], $options['response'])); + $method->setParameters([ + Generator\ParameterGenerator::fromArray(['type' => $options['request'], 'name' => 'arg']), + $ctxParam, + ]); + $method->setReturnType($options['response']); + $method->setBody(\sprintf('return $this->invoke("%s", $arg, $ctx);', $name)); + $implementation->addMethodFromGenerator($method); + } - $impl->addMethodFromGenerator($m); + foreach (($client['extraImplementationMethods'])() as $method) { + $implementation->addMethodFromGenerator($method); + } + echo "[OK]\n"; + + echo "writing {$client['implementationName']}: "; + $file = new Generator\FileGenerator(); + $file->setNamespace('Temporal\\Client\\GRPC'); + $file->setDeclares([DeclareStatement::fromArray(['strict_types' => 1])]); + $file->setClass($implementation); + $file->setUses([$client['apiNamespace'], ...$commonUses, ...($client['extraUses']['implementation'] ?? [])]); + + \file_put_contents( + $client['implementationFile'], + \str_replace( + [$apiAliasStrip, '\\Temporal\\Client\\GRPC\\', '\\Temporal\\Client\\Common\\'], + ['', '', ''], + $file->generate(), + ), + ); + $generatedFiles[] = $client['implementationFile']; + echo "[OK]\n"; } -echo "[OK]\n"; - -echo "writing implementation: "; - -$file = new Generator\FileGenerator(); -$file->setNamespace('Temporal\\Client\\GRPC'); -$file->setClass($impl); -$file->setUses( - [ - 'Temporal\Api\Workflowservice\V1', - 'Temporal\Exception\Client\ServiceClientException', - ] -); - -// write and shorten names -file_put_contents( - __DIR__ . '/../../src/Client/GRPC/ServiceClient.php', - str_replace( - ['\\Temporal\\Api\\Workflowservice\\', '\\Temporal\\Client\\GRPC\\ContextInterface'], - ['', 'ContextInterface'], - $file->generate() - ) -); +echo "formatting generated files: "; +$command = \implode(' ', [ + \escapeshellarg(PHP_BINARY), + \escapeshellarg(__DIR__ . '/../../vendor/bin/php-cs-fixer'), + 'fix', + '--config=' . \escapeshellarg(__DIR__ . '/../../.php-cs-fixer.dist.php'), + '--path-mode=intersection', + '--using-cache=no', + '--allow-unsupported-php-version=yes', + '--sequential', + ...\array_map(static fn(string $path): string => \escapeshellarg($path), $generatedFiles), +]); + +\passthru($command, $exitCode); +$exitCode === 0 or throw new RuntimeException('Failed to format generated files.'); echo "[OK]\n"; diff --git a/src/Client/GRPC/BaseClient.php b/src/Client/GRPC/BaseClient.php index 1f80cbbea..a86160945 100644 --- a/src/Client/GRPC/BaseClient.php +++ b/src/Client/GRPC/BaseClient.php @@ -12,12 +12,10 @@ namespace Temporal\Client\GRPC; use Carbon\CarbonInterval; +use Grpc\BaseStub; use Grpc\UnaryCall; -use Temporal\Api\Workflowservice\V1\GetSystemInfoRequest; -use Temporal\Api\Workflowservice\V1\WorkflowServiceClient; use Temporal\Client\Common\BackoffThrottler; use Temporal\Client\Common\RpcRetryOptions; -use Temporal\Client\Common\ServerCapabilities; use Temporal\Client\GRPC\Connection\Connection; use Temporal\Client\GRPC\Connection\ConnectionInterface; use Temporal\Exception\Client\CanceledException; @@ -26,7 +24,7 @@ use Temporal\Interceptor\GrpcClientInterceptor; use Temporal\Internal\Interceptor\Pipeline; -abstract class BaseClient implements ServiceClientInterface +abstract class BaseClient implements GrpcClientInterface { public const RETRYABLE_ERRORS = [ StatusCode::RESOURCE_EXHAUSTED, @@ -42,23 +40,23 @@ abstract class BaseClient implements ServiceClientInterface private \Stringable|string $apiKey = ''; /** - * @param WorkflowServiceClient|\Closure(): WorkflowServiceClient $workflowService Service Client or its factory + * @param BaseStub|\Closure(): BaseStub $serviceClient Service Client or its factory * * @private Use static factory methods instead * @see self::create() * @see self::createSSL() */ - final public function __construct(WorkflowServiceClient|\Closure $workflowService) + final public function __construct(BaseStub|\Closure $serviceClient) { - if ($workflowService instanceof WorkflowServiceClient) { + if ($serviceClient instanceof BaseStub) { \trigger_error( - 'Creating a ServiceClient instance via constructor is deprecated. Use static factory methods instead.', + 'Creating a gRPC client instance via constructor is deprecated. Use static factory methods instead.', \E_USER_DEPRECATED, ); - $workflowService = static fn(): WorkflowServiceClient => $workflowService; + $serviceClient = static fn(): BaseStub => $serviceClient; } - $this->connection = new Connection($workflowService); + $this->connection = new Connection($serviceClient); $this->context = Context::default(); } @@ -72,10 +70,12 @@ public static function create(string $address): static throw new \RuntimeException('The gRPC extension is required to use Temporal Client.'); } - return new static(static fn(): WorkflowServiceClient => new WorkflowServiceClient( - $address, - ['credentials' => \Grpc\ChannelCredentials::createInsecure()], - )); + return new static( + static fn(): BaseStub => static::createGrpcStub( + $address, + ['credentials' => \Grpc\ChannelCredentials::createInsecure()], + ), + ); } /** @@ -123,7 +123,7 @@ public static function createSSL( $options['grpc.ssl_target_name_override'] = $overrideServerName; } - return new static(static fn(): WorkflowServiceClient => new WorkflowServiceClient($address, $options)); + return new static(static fn(): BaseStub => static::createGrpcStub($address, $options)); } public function getContext(): ContextInterface @@ -175,57 +175,18 @@ final public function withInterceptorPipeline(?Pipeline $pipeline): static return $clone; } - public function getServerCapabilities(): ?ServerCapabilities - { - if ($this->connection->capabilities !== null) { - return $this->connection->capabilities; - } - - try { - $systemInfo = $this->getSystemInfo(new GetSystemInfoRequest()); - $capabilities = $systemInfo->getCapabilities(); - - if ($capabilities === null) { - return null; - } - - return $this->connection->capabilities = new ServerCapabilities( - signalAndQueryHeader: $capabilities->getSignalAndQueryHeader(), - internalErrorDifferentiation: $capabilities->getInternalErrorDifferentiation(), - activityFailureIncludeHeartbeat: $capabilities->getActivityFailureIncludeHeartbeat(), - supportsSchedules: $capabilities->getSupportsSchedules(), - encodedFailureAttributes: $capabilities->getEncodedFailureAttributes(), - buildIdBasedVersioning: $capabilities->getBuildIdBasedVersioning(), - upsertMemo: $capabilities->getUpsertMemo(), - eagerWorkflowStart: $capabilities->getEagerWorkflowStart(), - sdkMetadata: $capabilities->getSdkMetadata(), - countGroupByExecutionStatus: $capabilities->getCountGroupByExecutionStatus(), - nexus: $capabilities->getNexus(), - ); - } catch (ServiceClientException $e) { - if ($e->getCode() === StatusCode::UNIMPLEMENTED) { - return null; - } - - throw $e; - } - } - /** - * @deprecated + * Note: Experimental */ - public function setServerCapabilities(ServerCapabilities $capabilities): void + public function getConnection(): ConnectionInterface { - \trigger_error( - 'Method ' . __METHOD__ . ' is deprecated and will be removed in the next major release.', - \E_USER_DEPRECATED, - ); + return $this->connection; } /** - * Note: Experimental + * @internal */ - public function getConnection(): ConnectionInterface + final protected function getInternalConnection(): Connection { return $this->connection; } @@ -252,6 +213,12 @@ protected function invoke(string $method, object $arg, ?ContextInterface $ctx = : $this->call($method, $arg, $ctx); } + /** + * @param non-empty-string $address Temporal service address in format `host:port` + * @param array $options + */ + abstract protected static function createGrpcStub(string $address, array $options): BaseStub; + /** * Call a gRPC method. * Used in {@see withInterceptorPipeline()} @@ -273,11 +240,11 @@ private function call(string $method, object $arg, ContextInterface $ctx): objec $deadline = $ctx->getDeadline(); if ($deadline !== null) { $diff = (new \DateTime())->diff($deadline); - $options['timeout'] = CarbonInterval::instance($diff)->totalMicroseconds; + $options['timeout'] = \max(0, (int) CarbonInterval::instance($diff)->totalMicroseconds); } /** @var UnaryCall $call */ - $call = $this->connection->getWorkflowService()->{$method}($arg, $ctx->getMetadata(), $options); + $call = $this->connection->getClient()->{$method}($arg, $ctx->getMetadata(), $options); [$result, $status] = $call->wait(); if ($status->code !== 0) { diff --git a/src/Client/GRPC/Connection/Connection.php b/src/Client/GRPC/Connection/Connection.php index de6db09f3..cc82367ed 100644 --- a/src/Client/GRPC/Connection/Connection.php +++ b/src/Client/GRPC/Connection/Connection.php @@ -4,7 +4,7 @@ namespace Temporal\Client\GRPC\Connection; -use Temporal\Api\Workflowservice\V1\WorkflowServiceClient; +use Grpc\BaseStub; use Temporal\Client\Common\ServerCapabilities; /** @@ -12,8 +12,8 @@ */ final class Connection implements ConnectionInterface { - public ?ServerCapabilities $capabilities = null; - private WorkflowServiceClient $workflowService; + private ?ServerCapabilities $capabilities = null; + private BaseStub $client; /** * True if ServiceClient wasn't created yet @@ -21,17 +21,17 @@ final class Connection implements ConnectionInterface private bool $closed = true; /** - * @param \Closure(): WorkflowServiceClient $clientFactory Service Client factory + * @param \Closure(): BaseStub $clientFactory Service Client factory */ public function __construct( - public \Closure $clientFactory, + private readonly \Closure $clientFactory, ) { $this->initClient(); } public function isConnected(): bool { - return ConnectionState::from($this->workflowService->getConnectivityState(false)) === ConnectionState::Ready; + return ConnectionState::from($this->client->getConnectivityState(false)) === ConnectionState::Ready; } public function connect(float $timeout): void @@ -56,7 +56,7 @@ public function connect(float $timeout): void if ($isFiber) { \Fiber::suspend(); } else { - $this->workflowService->waitForReady(50); + $this->client->waitForReady(50); } $alive = \microtime(true) < $deadline; @@ -83,16 +83,26 @@ public function disconnect(): void $this->closed = true; $this->capabilities = null; - $this->workflowService->close(); + $this->client->close(); + } + + public function getCapabilities(): ?ServerCapabilities + { + return $this->capabilities; + } + + public function setCapabilities(?ServerCapabilities $capabilities): void + { + $this->capabilities = $capabilities; } /** - * @return WorkflowServiceClient Shouldn't be cached + * @return BaseStub Shouldn't be cached */ - public function getWorkflowService(): WorkflowServiceClient + public function getClient(): BaseStub { $this->initClient(); - return $this->workflowService; + return $this->client; } public function __destruct() @@ -102,7 +112,7 @@ public function __destruct() private function getState(bool $tryToConnect = false): ConnectionState { - return ConnectionState::from($this->workflowService->getConnectivityState($tryToConnect)); + return ConnectionState::from($this->client->getConnectivityState($tryToConnect)); } /** @@ -114,7 +124,7 @@ private function initClient(): void return; } - $this->workflowService = ($this->clientFactory)(); + $this->client = ($this->clientFactory)(); $this->closed = false; } @@ -129,6 +139,6 @@ private function initClient(): void private function waitForReady(float $timeout): bool { /** @psalm-suppress InvalidOperand */ - return $this->workflowService->waitForReady((int) ($timeout * 1_000_000)); + return $this->client->waitForReady((int) ($timeout * 1_000_000)); } } diff --git a/src/Client/GRPC/GrpcClientInterface.php b/src/Client/GRPC/GrpcClientInterface.php new file mode 100644 index 000000000..2534c3471 --- /dev/null +++ b/src/Client/GRPC/GrpcClientInterface.php @@ -0,0 +1,23 @@ +invoke("AddSearchAttributes", $arg, $ctx); + } + + /** + * RemoveSearchAttributes removes custom search attributes. + * + * Returns NOT_FOUND status code if a Search Attribute with any of the specified + * names is not registered + * + * @throws ServiceClientException + */ + public function RemoveSearchAttributes(V1\RemoveSearchAttributesRequest $arg, ?ContextInterface $ctx = null): V1\RemoveSearchAttributesResponse + { + return $this->invoke("RemoveSearchAttributes", $arg, $ctx); + } + + /** + * ListSearchAttributes returns comprehensive information about search attributes. + * + * @throws ServiceClientException + */ + public function ListSearchAttributes(V1\ListSearchAttributesRequest $arg, ?ContextInterface $ctx = null): V1\ListSearchAttributesResponse + { + return $this->invoke("ListSearchAttributes", $arg, $ctx); + } + + /** + * DeleteNamespace synchronously deletes a namespace and asynchronously reclaims + * all namespace resources. + * + * @throws ServiceClientException + */ + public function DeleteNamespace(V1\DeleteNamespaceRequest $arg, ?ContextInterface $ctx = null): V1\DeleteNamespaceResponse + { + return $this->invoke("DeleteNamespace", $arg, $ctx); + } + + /** + * AddOrUpdateRemoteCluster adds or updates remote cluster. + * + * @throws ServiceClientException + */ + public function AddOrUpdateRemoteCluster(V1\AddOrUpdateRemoteClusterRequest $arg, ?ContextInterface $ctx = null): V1\AddOrUpdateRemoteClusterResponse + { + return $this->invoke("AddOrUpdateRemoteCluster", $arg, $ctx); + } + + /** + * RemoveRemoteCluster removes remote cluster. + * + * @throws ServiceClientException + */ + public function RemoveRemoteCluster(V1\RemoveRemoteClusterRequest $arg, ?ContextInterface $ctx = null): V1\RemoveRemoteClusterResponse + { + return $this->invoke("RemoveRemoteCluster", $arg, $ctx); + } + + /** + * ListClusters returns information about Temporal clusters. + * + * @throws ServiceClientException + */ + public function ListClusters(V1\ListClustersRequest $arg, ?ContextInterface $ctx = null): V1\ListClustersResponse + { + return $this->invoke("ListClusters", $arg, $ctx); + } + + /** + * Get a registered Nexus endpoint by ID. The returned version can be used for + * optimistic updates. + * + * @throws ServiceClientException + */ + public function GetNexusEndpoint(V1\GetNexusEndpointRequest $arg, ?ContextInterface $ctx = null): V1\GetNexusEndpointResponse + { + return $this->invoke("GetNexusEndpoint", $arg, $ctx); + } + + /** + * Create a Nexus endpoint. This will fail if an endpoint with the same name is + * already registered with a status of + * ALREADY_EXISTS. + * Returns the created endpoint with its initial version. You may use this version + * for subsequent updates. + * + * @throws ServiceClientException + */ + public function CreateNexusEndpoint(V1\CreateNexusEndpointRequest $arg, ?ContextInterface $ctx = null): V1\CreateNexusEndpointResponse + { + return $this->invoke("CreateNexusEndpoint", $arg, $ctx); + } + + /** + * Optimistically update a Nexus endpoint based on provided version as obtained via + * the `GetNexusEndpoint` or + * `ListNexusEndpointResponse` APIs. This will fail with a status of + * FAILED_PRECONDITION if the version does not + * match. + * Returns the updated endpoint with its updated version. You may use this version + * for subsequent updates. You don't + * need to increment the version yourself. The server will increment the version + * for you after each update. + * + * @throws ServiceClientException + */ + public function UpdateNexusEndpoint(V1\UpdateNexusEndpointRequest $arg, ?ContextInterface $ctx = null): V1\UpdateNexusEndpointResponse + { + return $this->invoke("UpdateNexusEndpoint", $arg, $ctx); + } + + /** + * Delete an incoming Nexus service by ID. + * + * @throws ServiceClientException + */ + public function DeleteNexusEndpoint(V1\DeleteNexusEndpointRequest $arg, ?ContextInterface $ctx = null): V1\DeleteNexusEndpointResponse + { + return $this->invoke("DeleteNexusEndpoint", $arg, $ctx); + } + + /** + * List all Nexus endpoints for the cluster, sorted by ID in ascending order. Set + * page_token in the request to the + * next_page_token field of the previous response to get the next page of results. + * An empty next_page_token + * indicates that there are no more results. During pagination, a newly added + * service with an ID lexicographically + * earlier than the previous page's last endpoint's ID may be missed. + * + * @throws ServiceClientException + */ + public function ListNexusEndpoints(V1\ListNexusEndpointsRequest $arg, ?ContextInterface $ctx = null): V1\ListNexusEndpointsResponse + { + return $this->invoke("ListNexusEndpoints", $arg, $ctx); + } + + protected static function createGrpcStub(string $address, array $options): \Grpc\BaseStub + { + return new V1\OperatorServiceClient($address, $options); + } +} diff --git a/src/Client/GRPC/OperatorClientInterface.php b/src/Client/GRPC/OperatorClientInterface.php new file mode 100644 index 000000000..85051df11 --- /dev/null +++ b/src/Client/GRPC/OperatorClientInterface.php @@ -0,0 +1,124 @@ +invoke("UpdateWorkerDeploymentVersionMetadata", $arg, $ctx); } + /** + * Set/unset the ManagerIdentity of a Worker Deployment. + * + * @throws ServiceClientException + */ + public function SetWorkerDeploymentManager(V1\SetWorkerDeploymentManagerRequest $arg, ?ContextInterface $ctx = null): V1\SetWorkerDeploymentManagerResponse + { + return $this->invoke("SetWorkerDeploymentManager", $arg, $ctx); + } + /** * Invokes the specified Update function on user Workflow code. * @@ -1403,4 +1414,69 @@ public function UpdateWorkerConfig(V1\UpdateWorkerConfigRequest $arg, ?ContextIn { return $this->invoke("UpdateWorkerConfig", $arg, $ctx); } + + /** + * DescribeWorker returns information about the specified worker. + * + * @throws ServiceClientException + */ + public function DescribeWorker(V1\DescribeWorkerRequest $arg, ?ContextInterface $ctx = null): V1\DescribeWorkerResponse + { + return $this->invoke("DescribeWorker", $arg, $ctx); + } + + public function getServerCapabilities(): ?ServerCapabilities + { + $connection = $this->getInternalConnection(); + if ($connection->getCapabilities() !== null) { + return $connection->getCapabilities(); + } + + try { + $systemInfo = $this->getSystemInfo(new V1\GetSystemInfoRequest()); + $capabilities = $systemInfo->getCapabilities(); + + if ($capabilities === null) { + return null; + } + + $serverCapabilities = new ServerCapabilities( + signalAndQueryHeader: $capabilities->getSignalAndQueryHeader(), + internalErrorDifferentiation: $capabilities->getInternalErrorDifferentiation(), + activityFailureIncludeHeartbeat: $capabilities->getActivityFailureIncludeHeartbeat(), + supportsSchedules: $capabilities->getSupportsSchedules(), + encodedFailureAttributes: $capabilities->getEncodedFailureAttributes(), + buildIdBasedVersioning: $capabilities->getBuildIdBasedVersioning(), + upsertMemo: $capabilities->getUpsertMemo(), + eagerWorkflowStart: $capabilities->getEagerWorkflowStart(), + sdkMetadata: $capabilities->getSdkMetadata(), + countGroupByExecutionStatus: $capabilities->getCountGroupByExecutionStatus(), + nexus: $capabilities->getNexus(), + ); + $connection->setCapabilities($serverCapabilities); + + return $serverCapabilities; + } catch (ServiceClientException $e) { + if ($e->getCode() === StatusCode::UNIMPLEMENTED) { + return null; + } + + throw $e; + } + } + + public function setServerCapabilities(ServerCapabilities $capabilities): void + { + \trigger_error( + 'Method ' . __METHOD__ . ' is deprecated and will be removed in the next major release.', + \E_USER_DEPRECATED, + ); + + $this->getInternalConnection()->setCapabilities($capabilities); + } + + protected static function createGrpcStub(string $address, array $options): \Grpc\BaseStub + { + return new V1\WorkflowServiceClient($address, $options); + } } diff --git a/src/Client/GRPC/ServiceClientInterface.php b/src/Client/GRPC/ServiceClientInterface.php index 1f262bfa0..88eb24cd9 100644 --- a/src/Client/GRPC/ServiceClientInterface.php +++ b/src/Client/GRPC/ServiceClientInterface.php @@ -6,18 +6,11 @@ use Temporal\Api\Workflowservice\V1; use Temporal\Exception\Client\ServiceClientException; +use Temporal\Client\Common\ServerCapabilities; -interface ServiceClientInterface +interface ServiceClientInterface extends GrpcClientInterface { - public function getContext(): ContextInterface; - - public function withContext(ContextInterface $context): static; - - public function withAuthKey(\Stringable|string $key): static; - - public function getConnection(): \Temporal\Client\GRPC\Connection\ConnectionInterface; - - public function getServerCapabilities(): ?\Temporal\Client\Common\ServerCapabilities; + public function getServerCapabilities(): ?ServerCapabilities; /** * RegisterNamespace creates a new namespace which can be used as a container for @@ -99,8 +92,6 @@ public function StartWorkflowExecution(V1\StartWorkflowExecutionRequest $arg, ?C * Upon failure, it returns `MultiOperationExecutionFailure` where the status code * equals the status code of the *first* operation that failed to be started. * - * NOTE: Experimental API. - * * @throws ServiceClientException */ public function ExecuteMultiOperation(V1\ExecuteMultiOperationRequest $arg, ?ContextInterface $ctx = null): V1\ExecuteMultiOperationResponse; @@ -449,8 +440,10 @@ public function ListWorkflowExecutions(V1\ListWorkflowExecutionsRequest $arg, ?C public function ListArchivedWorkflowExecutions(V1\ListArchivedWorkflowExecutionsRequest $arg, ?ContextInterface $ctx = null): V1\ListArchivedWorkflowExecutionsResponse; /** - * ScanWorkflowExecutions is a visibility API to list large amount of workflow + * ScanWorkflowExecutions _was_ a visibility API to list large amount of workflow * executions in a specific namespace without order. + * It has since been deprecated in favor of `ListWorkflowExecutions` and rewritten + * to use `ListWorkflowExecutions` internally. * * Deprecated: Replaced with `ListWorkflowExecutions`. * (-- api-linter: core::0127::http-annotation=disabled @@ -875,6 +868,13 @@ public function ListWorkerDeployments(V1\ListWorkerDeploymentsRequest $arg, ?Con */ public function UpdateWorkerDeploymentVersionMetadata(V1\UpdateWorkerDeploymentVersionMetadataRequest $arg, ?ContextInterface $ctx = null): V1\UpdateWorkerDeploymentVersionMetadataResponse; + /** + * Set/unset the ManagerIdentity of a Worker Deployment. + * + * @throws ServiceClientException + */ + public function SetWorkerDeploymentManager(V1\SetWorkerDeploymentManagerRequest $arg, ?ContextInterface $ctx = null): V1\SetWorkerDeploymentManagerResponse; + /** * Invokes the specified Update function on user Workflow code. * @@ -1139,7 +1139,9 @@ public function FetchWorkerConfig(V1\FetchWorkerConfigRequest $arg, ?ContextInte public function UpdateWorkerConfig(V1\UpdateWorkerConfigRequest $arg, ?ContextInterface $ctx = null): V1\UpdateWorkerConfigResponse; /** - * Close the communication channel associated with this stub. + * DescribeWorker returns information about the specified worker. + * + * @throws ServiceClientException */ - public function close(): void; + public function DescribeWorker(V1\DescribeWorkerRequest $arg, ?ContextInterface $ctx = null): V1\DescribeWorkerResponse; } diff --git a/tests/Unit/Client/GRPC/OperatorClientTestCase.php b/tests/Unit/Client/GRPC/OperatorClientTestCase.php new file mode 100644 index 000000000..30c879af5 --- /dev/null +++ b/tests/Unit/Client/GRPC/OperatorClientTestCase.php @@ -0,0 +1,383 @@ + new class extends ApiOperatorServiceClient { + public function __construct() {} + + public function getConnectivityState($try_to_connect = false): int + { + return ConnectionState::Ready->value; + } + + public function close(): void {} + }))->withInterceptorPipeline( + Pipeline::prepare([new class($captured) implements GrpcClientInterceptor { + public function __construct( + private readonly object $captured, + ) {} + + public function interceptCall( + string $method, + object $arg, + ContextInterface $ctx, + callable $next, + ): object { + $this->captured->method = $method; + $this->captured->request = $arg; + $this->captured->context = $ctx; + + return (new DeleteNamespaceResponse())->setDeletedNamespace('temporal-system-deleted'); + } + }]), + )->withAuthKey('test-key'); + + $response = $client->DeleteNamespace((new DeleteNamespaceRequest())->setNamespace('test-namespace')); + + self::assertSame('DeleteNamespace', $captured->method); + self::assertSame('test-namespace', $captured->request?->getNamespace()); + self::assertSame(['Bearer test-key'], $captured->context?->getMetadata()['Authorization'] ?? null); + self::assertSame('temporal-system-deleted', $response->getDeletedNamespace()); + } + + #[Test] + public function addSearchAttributes(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new AddSearchAttributesResponse(), + ); + + $client->AddSearchAttributes(new AddSearchAttributesRequest()); + + self::assertSame('AddSearchAttributes', $captured->method); + self::assertInstanceOf(AddSearchAttributesRequest::class, $captured->arg); + } + + #[Test] + public function removeSearchAttributes(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new RemoveSearchAttributesResponse(), + ); + + $client->RemoveSearchAttributes(new RemoveSearchAttributesRequest()); + + self::assertSame('RemoveSearchAttributes', $captured->method); + self::assertInstanceOf(RemoveSearchAttributesRequest::class, $captured->arg); + } + + #[Test] + public function listSearchAttributes(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListSearchAttributesResponse(), + ); + + $client->ListSearchAttributes(new ListSearchAttributesRequest()); + + self::assertSame('ListSearchAttributes', $captured->method); + self::assertInstanceOf(ListSearchAttributesRequest::class, $captured->arg); + } + + #[Test] + public function addOrUpdateRemoteCluster(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new AddOrUpdateRemoteClusterResponse(), + ); + + $client->AddOrUpdateRemoteCluster(new AddOrUpdateRemoteClusterRequest()); + + self::assertSame('AddOrUpdateRemoteCluster', $captured->method); + self::assertInstanceOf(AddOrUpdateRemoteClusterRequest::class, $captured->arg); + } + + #[Test] + public function removeRemoteCluster(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new RemoveRemoteClusterResponse(), + ); + + $client->RemoveRemoteCluster(new RemoveRemoteClusterRequest()); + + self::assertSame('RemoveRemoteCluster', $captured->method); + self::assertInstanceOf(RemoveRemoteClusterRequest::class, $captured->arg); + } + + #[Test] + public function listClusters(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $client->ListClusters(new ListClustersRequest()); + + self::assertSame('ListClusters', $captured->method); + self::assertInstanceOf(ListClustersRequest::class, $captured->arg); + } + + #[Test] + public function getNexusEndpoint(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new GetNexusEndpointResponse(), + ); + + $client->GetNexusEndpoint(new GetNexusEndpointRequest()); + + self::assertSame('GetNexusEndpoint', $captured->method); + self::assertInstanceOf(GetNexusEndpointRequest::class, $captured->arg); + } + + #[Test] + public function createNexusEndpoint(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new CreateNexusEndpointResponse(), + ); + + $client->CreateNexusEndpoint(new CreateNexusEndpointRequest()); + + self::assertSame('CreateNexusEndpoint', $captured->method); + self::assertInstanceOf(CreateNexusEndpointRequest::class, $captured->arg); + } + + #[Test] + public function updateNexusEndpoint(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new UpdateNexusEndpointResponse(), + ); + + $client->UpdateNexusEndpoint(new UpdateNexusEndpointRequest()); + + self::assertSame('UpdateNexusEndpoint', $captured->method); + self::assertInstanceOf(UpdateNexusEndpointRequest::class, $captured->arg); + } + + #[Test] + public function deleteNexusEndpoint(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new DeleteNexusEndpointResponse(), + ); + + $client->DeleteNexusEndpoint(new DeleteNexusEndpointRequest()); + + self::assertSame('DeleteNexusEndpoint', $captured->method); + self::assertInstanceOf(DeleteNexusEndpointRequest::class, $captured->arg); + } + + #[Test] + public function listNexusEndpoints(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListNexusEndpointsResponse(), + ); + + $client->ListNexusEndpoints(new ListNexusEndpointsRequest()); + + self::assertSame('ListNexusEndpoints', $captured->method); + self::assertInstanceOf(ListNexusEndpointsRequest::class, $captured->arg); + } + + #[Test] + public function customContextIsPassedToInterceptor(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $ctx = $client->getContext()->withMetadata(['x-custom' => ['value']]); + $client->ListClusters(new ListClustersRequest(), $ctx); + + self::assertSame(['value'], $captured->ctx->getMetadata()['x-custom'] ?? null); + } + + #[Test] + public function authKeyIsInjectedIntoContext(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $client = $client->withAuthKey('operator-key'); + $client->ListClusters(new ListClustersRequest()); + + self::assertSame(['Bearer operator-key'], $captured->ctx->getMetadata()['Authorization'] ?? null); + } + + #[Test] + public function withoutAuthKeyNoAuthorizationHeader(): void + { + [$captured, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $client->ListClusters(new ListClustersRequest()); + + self::assertArrayNotHasKey('Authorization', $captured->ctx->getMetadata()); + } + + #[Test] + public function withContextReturnsNewImmutableInstance(): void + { + [, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $ctx = $client->getContext()->withMetadata(['foo' => ['bar']]); + $client2 = $client->withContext($ctx); + + self::assertNotSame($client, $client2); + self::assertSame($ctx, $client2->getContext()); + self::assertNotSame($ctx, $client->getContext()); + } + + #[Test] + public function withAuthKeyReturnsNewImmutableInstance(): void + { + [, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + $client2 = $client->withAuthKey('key'); + + self::assertNotSame($client, $client2); + } + + #[Test] + public function closeDisconnectsConnection(): void + { + $client = (new OperatorClient(static fn() => new class extends ApiOperatorServiceClient { + public function __construct() {} + + public function getConnectivityState($try_to_connect = false): int + { + return ConnectionState::TransientFailure->value; + } + + public function close(): void {} + })); + + $client->close(); + + self::assertFalse($client->getConnection()->isConnected()); + } + + #[Test] + public function implementsGrpcClientInterface(): void + { + [, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + self::assertInstanceOf(GrpcClientInterface::class, $client); + } + + #[Test] + public function implementsOperatorClientInterface(): void + { + [, $client] = $this->createInterceptedClient( + static fn() => new ListClustersResponse(), + ); + + self::assertInstanceOf(OperatorClientInterface::class, $client); + } + + /** + * @return array{object, OperatorClient} + */ + private function createInterceptedClient(\Closure $responseFactory): array + { + $captured = new class { + public ?string $method = null; + public ?object $arg = null; + public ?ContextInterface $ctx = null; + }; + + $client = (new OperatorClient(static fn() => new class extends ApiOperatorServiceClient { + public function __construct() {} + + public function getConnectivityState($try_to_connect = false): int + { + return ConnectionState::Ready->value; + } + + public function close(): void {} + }))->withInterceptorPipeline( + Pipeline::prepare([new class($captured, $responseFactory) implements GrpcClientInterceptor { + public function __construct( + private readonly object $captured, + private readonly \Closure $responseFactory, + ) {} + + public function interceptCall( + string $method, + object $arg, + ContextInterface $ctx, + callable $next, + ): object { + $this->captured->method = $method; + $this->captured->arg = $arg; + $this->captured->ctx = $ctx; + + return ($this->responseFactory)(); + } + }]), + ); + + return [$captured, $client]; + } +}