diff --git a/.gitignore b/.gitignore index d8ebdcc..18849f0 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,5 @@ app/**/temporal.exe app/tests/**/*.log app/runtime app/vendor +app/**/temporal-test-server +.DS_Store diff --git a/app/composer.json b/app/composer.json index fc9b985..0aa9700 100644 --- a/app/composer.json +++ b/app/composer.json @@ -1,9 +1,15 @@ { "minimum-stability": "dev", "prefer-stable": true, + "repositories": [ + { + "type": "vcs", + "url": "https://github.com/temporalio/sdk-php" + } + ], "require": { "spiral/tokenizer": "^3.16.0", - "temporal/sdk": "^2.17", + "temporal/sdk": "dev-nexus-new as 2.18.0", "temporal/open-telemetry-interceptors": "^1.0", "open-telemetry/exporter-otlp": "^1.3", "open-telemetry/transport-grpc": "^1.1", @@ -12,7 +18,8 @@ "require-dev": { "buggregator/trap": "^1.15", "internal/dload": "^1.8.0", - "phpunit/phpunit": "^10.5" + "phpunit/phpunit": "^10.5", + "phpstan/phpstan": "^2.2" }, "autoload": { "psr-4": { @@ -33,6 +40,7 @@ }, "scripts": { "get:binaries": "dload get --no-interaction -vv", - "test:feat": "phpunit --testsuite=Feature --color=always --testdox" + "test:feat": "phpunit --testsuite=Feature --color=always --testdox", + "phpstan": "phpstan analyse --memory-limit=2G" } } diff --git a/app/composer.lock b/app/composer.lock index 5da7a68..fe68a97 100644 --- a/app/composer.lock +++ b/app/composer.lock @@ -4,7 +4,7 @@ "Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies", "This file is @generated automatically" ], - "content-hash": "e6e2a963ef95741e5a7bfe40d9478877", + "content-hash": "bb34ba5229d49452bfefff2f47439cc5", "packages": [ { "name": "brick/math", @@ -4510,28 +4510,29 @@ }, { "name": "temporal/sdk", - "version": "v2.17.1", + "version": "dev-nexus-new", "source": { "type": "git", "url": "https://github.com/temporalio/sdk-php.git", - "reference": "cdbf6874efd25b3eee50b52c5c2d54ae0d004be4" + "reference": "fec3ab3c9959be0100e2a5aef87e15933a7a81a2" }, "dist": { "type": "zip", - "url": "https://api.github.com/repos/temporalio/sdk-php/zipball/cdbf6874efd25b3eee50b52c5c2d54ae0d004be4", - "reference": "cdbf6874efd25b3eee50b52c5c2d54ae0d004be4", + "url": "https://api.github.com/repos/temporalio/sdk-php/zipball/fec3ab3c9959be0100e2a5aef87e15933a7a81a2", + "reference": "fec3ab3c9959be0100e2a5aef87e15933a7a81a2", "shasum": "" }, "require": { "ext-curl": "*", "ext-json": "*", "google/common-protos": "^4.9", - "google/protobuf": "^4.31.1", + "google/protobuf": "^4.33.6 || ^5.34.0", "grpc/grpc": "^1.57", "internal/destroy": "^1.0", "internal/promise": "^3.4", "nesbot/carbon": "^2.72.6 || ^3.8.4", "php": ">=8.1", + "psr/clock": "^1.0", "psr/log": "^2.0 || ^3.0.2", "ramsey/uuid": "^4.7.6", "roadrunner-php/roadrunner-api-dto": "^1.14.0", @@ -4542,7 +4543,7 @@ "spiral/roadrunner-kv": "^4.3.1", "spiral/roadrunner-worker": "^3.6.2", "symfony/filesystem": "^5.4.45 || ^6.4.13 || ^7.0 || ^8.0", - "symfony/http-client": "^5.4.49 || ^6.4.17 || ^7.0 || ^8.0", + "symfony/http-client": "^5.4.53 || ^6.4.17 || ^7.0 || ^8.0", "symfony/polyfill-php83": "^1.31.0", "symfony/process": "^5.4.51 || ^6.4.15 || ^7.0 || ^8.0" }, @@ -4553,7 +4554,7 @@ "dereuromark/composer-prefer-lowest": "^0.1.10", "doctrine/annotations": "^1.14.4 || ^2.0.2", "ext-simplexml": "*", - "internal/dload": "^1.2.0", + "internal/dload": "^1.10.0", "internal/toml": "^1.0.3", "jetbrains/phpstorm-attributes": "dev-master", "laminas/laminas-code": "^4.16", @@ -4574,23 +4575,84 @@ "extra": { "patches": { "phpunit/phpunit": { - "Make runTest protected": "patches/phpunit-run-test-protected.patch", - "Improve assertInstanceOf error message": "patches/phpunit-instance-of.patch" + "Improve assertInstanceOf error message": "patches/phpunit-instance-of.patch", + "Make runTest protected": "patches/phpunit-run-test-protected.patch" } } }, "autoload": { + "psr-4": { + "Temporal\\Api\\Testservice\\": "testing/api/testservice/Temporal/Api/Testservice", + "GPBMetadata\\Temporal\\Api\\Testservice\\": "testing/api/testservice/GPBMetadata/Temporal/Api/Testservice", + "Temporal\\Testing\\": "testing/src", + "Temporal\\": "src" + }, "files": [ "src/include.php" - ], + ] + }, + "autoload-dev": { "psr-4": { - "Temporal\\": "src", - "Temporal\\Testing\\": "testing/src", - "Temporal\\Api\\Testservice\\": "testing/api/testservice/Temporal/Api/Testservice", - "GPBMetadata\\Temporal\\Api\\Testservice\\": "testing/api/testservice/GPBMetadata/Temporal/Api/Testservice" - } + "Temporal\\Tests\\Acceptance\\App\\": "tests/Acceptance/App", + "Temporal\\Tests\\Interceptor\\": "tests/Fixtures/src/Interceptor", + "Temporal\\Tests\\Workflow\\": "tests/Fixtures/src/Workflow", + "Temporal\\Tests\\Activity\\": "tests/Fixtures/src/Activity", + "Temporal\\Tests\\DTO\\": "tests/Fixtures/src/DTO", + "Temporal\\Tests\\Proto\\": "tests/Fixtures/src/Proto", + "Temporal\\Tests\\": "tests" + } + }, + "scripts": { + "get:binaries": [ + "dload get --no-interaction -vv", + "Temporal\\Worker\\Transport\\RoadRunnerVersionChecker::postUpdate" + ], + "cs:diff": [ + "php-cs-fixer fix --dry-run -v --diff --show-progress dots" + ], + "cs:fix": [ + "php-cs-fixer fix -v" + ], + "psalm": [ + "psalm" + ], + "psalm:baseline": [ + "psalm --set-baseline=psalm-baseline.xml" + ], + "test:unit": [ + "tests/runner.php vendor/bin/phpunit --testsuite=Unit --color=always --testdox" + ], + "test:func": [ + "tests/runner.php vendor/bin/phpunit --testsuite=Functional --color=always --testdox" + ], + "test:func-timeskip": [ + "tests/runner.php vendor/bin/phpunit --testsuite=Functional-TimeSkipping --color=always --testdox" + ], + "test:arch": [ + "phpunit --testsuite=Arch --color=always --testdox" + ], + "test:accept": [ + "tests/runner.php vendor/bin/phpunit --testsuite=Acceptance --color=always --testdox" + ], + "test:accept-slow": [ + "tests/runner.php vendor/bin/phpunit --testsuite=\"Acceptance-Slow\" --color=always --testdox" + ], + "test:accept-fast": [ + "tests/runner.php vendor/bin/phpunit --testsuite=\"Acceptance-Fast\" --color=always --testdox" + ], + "transcripts:last": [ + "php tests/Acceptance/transcript-merge.php" + ], + "transcripts:list": [ + "php tests/Acceptance/transcript-merge.php --list" + ], + "transcripts:merge": [ + "php tests/Acceptance/transcript-merge.php" + ], + "transcripts:clean": [ + "rm -rf runtime/tests/transcripts/*" + ] }, - "notification-url": "https://packagist.org/downloads/", "license": [ "MIT" ], @@ -4607,12 +4669,12 @@ "workflow" ], "support": { - "docs": "https://docs.temporal.io", "forum": "https://community.temporal.io", + "docs": "https://docs.temporal.io", "issues": "https://github.com/temporalio/sdk-php/issues", "source": "https://github.com/temporalio/sdk-php" }, - "time": "2026-03-08T14:43:31+00:00" + "time": "2026-08-19T09:15:08+00:00" } ], "packages-dev": [ @@ -5358,6 +5420,70 @@ }, "time": "2024-10-02T11:34:13+00:00" }, + { + "name": "phpstan/phpstan", + "version": "2.2.8", + "dist": { + "type": "zip", + "url": "https://api.github.com/repos/phpstan/phpstan/zipball/e285254e60f33c21902efef4a926ca0987c06804", + "reference": "e285254e60f33c21902efef4a926ca0987c06804", + "shasum": "" + }, + "require": { + "php": "^7.4|^8.0" + }, + "conflict": { + "phpstan/phpstan-shim": "*" + }, + "bin": [ + "phpstan", + "phpstan.phar" + ], + "type": "library", + "autoload": { + "files": [ + "bootstrap.php" + ] + }, + "notification-url": "https://packagist.org/downloads/", + "license": [ + "MIT" + ], + "authors": [ + { + "name": "OndΕ™ej Mirtes" + }, + { + "name": "Markus Staab" + }, + { + "name": "Vincent Langlet" + } + ], + "description": "PHPStan - PHP Static Analysis Tool", + "keywords": [ + "dev", + "static analysis" + ], + "support": { + "docs": "https://phpstan.org/user-guide/getting-started", + "forum": "https://github.com/phpstan/phpstan/discussions", + "issues": "https://github.com/phpstan/phpstan/issues", + "security": "https://github.com/phpstan/phpstan/security/policy", + "source": "https://github.com/phpstan/phpstan-src" + }, + "funding": [ + { + "url": "https://github.com/ondrejmirtes", + "type": "github" + }, + { + "url": "https://github.com/phpstan", + "type": "github" + } + ], + "time": "2026-08-04T22:21:45+00:00" + }, { "name": "phpunit/php-code-coverage", "version": "10.1.16", @@ -7102,9 +7228,18 @@ "time": "2025-12-01T11:14:17+00:00" } ], - "aliases": [], + "aliases": [ + { + "package": "temporal/sdk", + "version": "dev-nexus-new", + "alias": "2.18.0", + "alias_normalized": "2.18.0.0" + } + ], "minimum-stability": "dev", - "stability-flags": {}, + "stability-flags": { + "temporal/sdk": 20 + }, "prefer-stable": true, "prefer-lowest": false, "platform": {}, diff --git a/app/phpstan.neon.dist b/app/phpstan.neon.dist new file mode 100644 index 0000000..ca8aa31 --- /dev/null +++ b/app/phpstan.neon.dist @@ -0,0 +1,10 @@ +parameters: + level: 6 + paths: + - src/Nexus + - src/NexusCancellation + - src/NexusContextPropagation + - src/NexusMultipleArguments + - tests/Feature + excludePaths: + - tests/Feature/worker.log diff --git a/app/phpunit.xml.dist b/app/phpunit.xml.dist index 0f5e939..84b37f3 100644 --- a/app/phpunit.xml.dist +++ b/app/phpunit.xml.dist @@ -18,7 +18,8 @@ - + + diff --git a/app/src/Interceptors/Interceptor/ExceptionLoggerInterceptor.php b/app/src/Interceptors/Interceptor/ExceptionLoggerInterceptor.php new file mode 100644 index 0000000..32c94d5 --- /dev/null +++ b/app/src/Interceptors/Interceptor/ExceptionLoggerInterceptor.php @@ -0,0 +1,34 @@ +logger->error($e->getMessage(), ['exception' => $e]); + throw $e; + } + } +} diff --git a/app/src/Nexus/.rr.caller.yaml b/app/src/Nexus/.rr.caller.yaml new file mode 100644 index 0000000..3d46805 --- /dev/null +++ b/app/src/Nexus/.rr.caller.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6102 + +server: + command: "php caller-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-caller-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/Nexus/.rr.handler.yaml b/app/src/Nexus/.rr.handler.yaml new file mode 100644 index 0000000..36ff019 --- /dev/null +++ b/app/src/Nexus/.rr.handler.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6101 + +server: + command: "php handler-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-target-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/Nexus/Caller/CallerWorker.php b/app/src/Nexus/Caller/CallerWorker.php new file mode 100644 index 0000000..7b3076b --- /dev/null +++ b/app/src/Nexus/Caller/CallerWorker.php @@ -0,0 +1,11 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function echo(string $message) + { + /** @var EchoOutput $output */ + $output = yield $this->sampleNexusService->echo(new EchoInput($message)); + return $output->message; + } +} diff --git a/app/src/Nexus/Caller/HelloCallerWorkflow.php b/app/src/Nexus/Caller/HelloCallerWorkflow.php new file mode 100644 index 0000000..0ff02e2 --- /dev/null +++ b/app/src/Nexus/Caller/HelloCallerWorkflow.php @@ -0,0 +1,16 @@ +sampleNexusService = Workflow::newUntypedNexusOperationStub( + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withService('SampleNexusService') + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function hello(string $message, Language $language) + { + /** @var NexusOperationHandle $handle */ + $handle = yield $this->sampleNexusService->start( + 'hello', + [new HelloInput($message, $language)], + HelloOutput::class, + ); + + Workflow::getLogger()->info( + 'Nexus operation started, token: ' . ($handle->getOperationToken() ?? ''), + ); + + /** @var HelloOutput $output */ + $output = yield $handle->getResult(); + + return $output->message; + } +} diff --git a/app/src/Nexus/ExecuteCommand.php b/app/src/Nexus/ExecuteCommand.php new file mode 100644 index 0000000..e8c9b3b --- /dev/null +++ b/app/src/Nexus/ExecuteCommand.php @@ -0,0 +1,69 @@ +withNamespace('my-caller-namespace'), + ); + + $output->writeln("Starting EchoCallerWorkflow..."); + $echoWorkflow = $client->newWorkflowStub( + EchoCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($echoWorkflow, 'Nexus Echo πŸ‘‹'); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + $output->writeln("\nStarting HelloCallerWorkflow..."); + $helloWorkflow = $client->newWorkflowStub( + HelloCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($helloWorkflow, 'Nexus', Language::ES); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + return self::SUCCESS; + } +} diff --git a/app/src/Nexus/Handler/EchoClient.php b/app/src/Nexus/Handler/EchoClient.php new file mode 100644 index 0000000..a40f63b --- /dev/null +++ b/app/src/Nexus/Handler/EchoClient.php @@ -0,0 +1,13 @@ +message); + } +} diff --git a/app/src/Nexus/Handler/HandlerWorker.php b/app/src/Nexus/Handler/HandlerWorker.php new file mode 100644 index 0000000..a76182c --- /dev/null +++ b/app/src/Nexus/Handler/HandlerWorker.php @@ -0,0 +1,10 @@ +language) { + case Language::EN: + return new HelloOutput("Hello {$input->name} πŸ‘‹"); + case Language::FR: + return new HelloOutput("Bonjour {$input->name} πŸ‘‹"); + case Language::DE: + return new HelloOutput("Hallo {$input->name} πŸ‘‹"); + case Language::ES: + return new HelloOutput("Β‘Hola! {$input->name} πŸ‘‹"); + case Language::TR: + return new HelloOutput("Merhaba {$input->name} πŸ‘‹"); + } + throw new ApplicationFailure( + "Unsupported language: {$input->language->value}", + 'UNSUPPORTED_LANGUAGE', + true, + ); + } +} diff --git a/app/src/Nexus/Handler/SampleNexusServiceImpl.php b/app/src/Nexus/Handler/SampleNexusServiceImpl.php new file mode 100644 index 0000000..07e839c --- /dev/null +++ b/app/src/Nexus/Handler/SampleNexusServiceImpl.php @@ -0,0 +1,37 @@ +echoClient = $echoClient ?? new EchoClientImpl(); + } + + public function echo(EchoInput $input): EchoOutput + { + return $this->echoClient->echo($input); + } + + public function hello(HelloInput $input): WorkflowHandle + { + return WorkflowHandle::fromWorkflowMethod( + HelloHandlerWorkflow::class, + WorkflowOptions::new()->withWorkflowId(Nexus::getStartDetails()->requestId), + $input, + ); + } +} diff --git a/app/src/Nexus/README.md b/app/src/Nexus/README.md new file mode 100644 index 0000000..7403634 --- /dev/null +++ b/app/src/Nexus/README.md @@ -0,0 +1,80 @@ +# Nexus sample + +A caller workflow in `my-caller-namespace` invokes operations on a Nexus +service hosted by a handler worker in `my-target-namespace`. Two operations: + +- `echo` β€” synchronous (`#[Operation]`) +- `hello` β€” async, `WorkflowRunOperation`, starts `HelloHandlerWorkflow` + server-side (`#[AsyncOperation(output: HelloOutput::class)]`) + +Two caller workflows: + +- `EchoCallerWorkflow` β€” typed stub, sync op. +- `HelloCallerWorkflow` β€” untyped stub via + `Workflow::newUntypedNexusOperationStub()`. `start()` resolves with a + `NexusOperationHandle` that exposes `operationToken` (string for async, + null for sync) before `getResult()` resolves with the typed result. + +## Prerequisites + +Beyond the usual (`./temporal`, `./rr`): + +```bash +./temporal operator namespace create --namespace my-target-namespace +./temporal operator namespace create --namespace my-caller-namespace + +./temporal operator nexus endpoint create \ + --name my-nexus-endpoint-name \ + --target-namespace my-target-namespace \ + --target-task-queue my-handler-task-queue \ + --description-file ./app/src/Nexus/Service/description.md +``` + +## Run + +Three terminals. + +**Handler worker** (`my-target-namespace`): + +```bash +cd app/src/Nexus +TEMPORAL_NAMESPACE=my-target-namespace ../../rr serve -c .rr.handler.yaml +``` + +**Caller worker** (`my-caller-namespace`): + +```bash +cd app/src/Nexus +TEMPORAL_NAMESPACE=my-caller-namespace ../../rr serve -c .rr.caller.yaml +``` + +**Starter**: + +```bash +php app/app.php nexus +``` + +Expected output: + +``` +Starting EchoCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Nexus Echo πŸ‘‹ + +Starting HelloCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Β‘Hola! Nexus πŸ‘‹ +``` + +## Constants + +| Name | Value | +|---|---| +| service | `SampleNexusService` | +| operations | `echo`, `hello` | +| endpoint | `my-nexus-endpoint-name` | +| handler task queue | `my-handler-task-queue` | +| caller task queue | `my-caller-workflow-task-queue` | +| target namespace | `my-target-namespace` | +| caller namespace | `my-caller-namespace` | +| `scheduleToCloseTimeout` | `10s` | diff --git a/app/src/Nexus/Service/EchoInput.php b/app/src/Nexus/Service/EchoInput.php new file mode 100644 index 0000000..eacb207 --- /dev/null +++ b/app/src/Nexus/Service/EchoInput.php @@ -0,0 +1,12 @@ +newWorker(CallerWorker::TASK_QUEUE) + ->registerWorkflowTypes( + EchoCallerWorkflowImpl::class, + HelloCallerWorkflowImpl::class, + ); + +$factory->run(); diff --git a/app/src/Nexus/handler-worker.php b/app/src/Nexus/handler-worker.php new file mode 100644 index 0000000..ed1c30b --- /dev/null +++ b/app/src/Nexus/handler-worker.php @@ -0,0 +1,39 @@ +withNamespace($namespace), +); + +$factory = WorkerFactory::create(client: $workflowClient); + +$worker = $factory->newWorker(HandlerWorker::TASK_QUEUE); +$worker->registerWorkflowTypes(HelloHandlerWorkflowImpl::class); +$worker->registerNexusServiceImplementation(new SampleNexusServiceImpl()); + +$factory->run(); diff --git a/app/src/NexusCancellation/.rr.caller.yaml b/app/src/NexusCancellation/.rr.caller.yaml new file mode 100644 index 0000000..ddbec40 --- /dev/null +++ b/app/src/NexusCancellation/.rr.caller.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6112 + +server: + command: "php caller-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-caller-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusCancellation/.rr.handler.yaml b/app/src/NexusCancellation/.rr.handler.yaml new file mode 100644 index 0000000..60d43d8 --- /dev/null +++ b/app/src/NexusCancellation/.rr.handler.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6111 + +server: + command: "php handler-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-target-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusCancellation/Caller/CallerWorker.php b/app/src/NexusCancellation/Caller/CallerWorker.php new file mode 100644 index 0000000..b2236cb --- /dev/null +++ b/app/src/NexusCancellation/Caller/CallerWorker.php @@ -0,0 +1,11 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)) + ->withCancellationType(NexusOperationCancellationType::WaitRequested), + ); + } + + public function hello(string $message) + { + $promises = []; + + $scope = Workflow::async(function () use ($message, &$promises): void { + foreach (Language::cases() as $language) { + $promises[] = $this->sampleNexusService->hello(new HelloInput($message, $language)); + } + }); + + /** @var HelloOutput $first */ + $first = yield Promise::any($promises); + + $scope->cancel(); + + foreach ($promises as $promise) { + try { + yield $promise; + } catch (CanceledFailure) { + } catch (NexusOperationFailure $e) { + if (!$e->getPrevious() instanceof CanceledFailure) { + throw $e; + } + } + } + + return $first->message; + } +} diff --git a/app/src/NexusCancellation/ExecuteCommand.php b/app/src/NexusCancellation/ExecuteCommand.php new file mode 100644 index 0000000..67fc57c --- /dev/null +++ b/app/src/NexusCancellation/ExecuteCommand.php @@ -0,0 +1,53 @@ +withNamespace('my-caller-namespace'), + ); + + $output->writeln("Starting HelloCallerWorkflow..."); + $workflow = $client->newWorkflowStub( + HelloCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($workflow, 'Nexus'); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + return self::SUCCESS; + } +} diff --git a/app/src/NexusCancellation/Handler/HandlerWorker.php b/app/src/NexusCancellation/Handler/HandlerWorker.php new file mode 100644 index 0000000..a1fce96 --- /dev/null +++ b/app/src/NexusCancellation/Handler/HandlerWorker.php @@ -0,0 +1,10 @@ + \random_int(0, 4)))); + + switch ($input->language) { + case Language::EN: + return new HelloOutput("Hello {$input->name} πŸ‘‹"); + case Language::FR: + return new HelloOutput("Bonjour {$input->name} πŸ‘‹"); + case Language::DE: + return new HelloOutput("Hallo {$input->name} πŸ‘‹"); + case Language::ES: + return new HelloOutput("Β‘Hola! {$input->name} πŸ‘‹"); + case Language::TR: + return new HelloOutput("Merhaba {$input->name} πŸ‘‹"); + } + throw new ApplicationFailure( + "Unsupported language: {$input->language->value}", + 'UNSUPPORTED_LANGUAGE', + true, + ); + } catch (CanceledFailure $e) { + yield Workflow::asyncDetached(function () { + yield Workflow::timer(CarbonInterval::seconds(yield Workflow::sideEffect(fn() => \random_int(0, 4)))); + }); + + Workflow::getLogger()->info('HelloHandlerWorkflow was cancelled successfully.'); + + throw $e; + } + } +} diff --git a/app/src/NexusCancellation/README.md b/app/src/NexusCancellation/README.md new file mode 100644 index 0000000..2fa27bf --- /dev/null +++ b/app/src/NexusCancellation/README.md @@ -0,0 +1,72 @@ +# Nexus Cancellation sample + +A caller workflow fans out the `hello` Nexus operation in 5 languages in +parallel. After the first reply arrives it cancels the cancellation scope β€” +the other 4 in-flight operations get cancellation requests, the handler +workflows clean up and rethrow `CanceledFailure`, and the caller drains the +promises (ignoring `CanceledFailure`) before returning the first result. + +Reuses the service contract and `SampleNexusServiceImpl` of the +[Nexus sample](../Nexus/README.md); the only sample-specific handler code is +`HelloHandlerWorkflowImpl`, which adds a random delay and a detached cleanup +scope. + +Caller passes `NexusOperationCancellationType::WaitRequested` so it returns +as soon as the handler acknowledges the cancellation request β€” it doesn't +wait for the detached cleanup scope to finish on the handler side. + +## Prerequisites + +Beyond the usual (`./temporal`, `./rr`), and the two namespaces the +[Nexus sample](../Nexus/README.md) creates: + +```bash +./temporal operator nexus endpoint create \ + --name my-cancellation-nexus-endpoint \ + --target-namespace my-target-namespace \ + --target-task-queue my-cancellation-handler-task-queue +``` + +This sample owns its endpoint, task queues and RoadRunner RPC ports, so it +can run alongside the other Nexus samples. + +## Run + +Three terminals. + +**Handler worker** (`my-target-namespace`): + +```bash +cd app/src/NexusCancellation +TEMPORAL_NAMESPACE=my-target-namespace ../../rr serve -c .rr.handler.yaml +``` + +**Caller worker** (`my-caller-namespace`): + +```bash +cd app/src/NexusCancellation +TEMPORAL_NAMESPACE=my-caller-namespace ../../rr serve -c .rr.caller.yaml +``` + +**Starter**: + +```bash +php app/app.php nexus-cancellation +``` + +Caller side prints the first reply (whichever language won the race): + +``` +Starting HelloCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Β‘Hola! Nexus πŸ‘‹ +``` + +Handler side logs each cancelled run: + +``` +HelloHandlerWorkflow was cancelled successfully. +HelloHandlerWorkflow was cancelled successfully. +HelloHandlerWorkflow was cancelled successfully. +HelloHandlerWorkflow was cancelled successfully. +``` diff --git a/app/src/NexusCancellation/caller-worker.php b/app/src/NexusCancellation/caller-worker.php new file mode 100644 index 0000000..997dbc9 --- /dev/null +++ b/app/src/NexusCancellation/caller-worker.php @@ -0,0 +1,24 @@ +newWorker(CallerWorker::TASK_QUEUE) + ->registerWorkflowTypes(HelloCallerWorkflowImpl::class); + +$factory->run(); diff --git a/app/src/NexusCancellation/handler-worker.php b/app/src/NexusCancellation/handler-worker.php new file mode 100644 index 0000000..ce400c4 --- /dev/null +++ b/app/src/NexusCancellation/handler-worker.php @@ -0,0 +1,39 @@ +withNamespace($namespace), +); + +$factory = WorkerFactory::create(client: $workflowClient); + +$worker = $factory->newWorker(HandlerWorker::TASK_QUEUE); +$worker->registerWorkflowTypes(HelloHandlerWorkflowImpl::class); +$worker->registerNexusServiceImplementation(new SampleNexusServiceImpl()); + +$factory->run(); diff --git a/app/src/NexusContextPropagation/.rr.caller.yaml b/app/src/NexusContextPropagation/.rr.caller.yaml new file mode 100644 index 0000000..148da33 --- /dev/null +++ b/app/src/NexusContextPropagation/.rr.caller.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6122 + +server: + command: "php caller-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-caller-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusContextPropagation/.rr.handler.yaml b/app/src/NexusContextPropagation/.rr.handler.yaml new file mode 100644 index 0000000..2b08d5e --- /dev/null +++ b/app/src/NexusContextPropagation/.rr.handler.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6121 + +server: + command: "php handler-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-target-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusContextPropagation/Caller/CallerWorker.php b/app/src/NexusContextPropagation/Caller/CallerWorker.php new file mode 100644 index 0000000..b7c9471 --- /dev/null +++ b/app/src/NexusContextPropagation/Caller/CallerWorker.php @@ -0,0 +1,11 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function echo(string $message) + { + MDC::put('x-nexus-caller-workflow-id', Workflow::getInfo()->execution->getID()); + + /** @var EchoOutput $output */ + $output = yield $this->sampleNexusService->echo(new EchoInput($message)); + return $output->message; + } +} diff --git a/app/src/NexusContextPropagation/Caller/HelloCallerWorkflowImpl.php b/app/src/NexusContextPropagation/Caller/HelloCallerWorkflowImpl.php new file mode 100644 index 0000000..eab4032 --- /dev/null +++ b/app/src/NexusContextPropagation/Caller/HelloCallerWorkflowImpl.php @@ -0,0 +1,40 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function hello(string $message, Language $language) + { + MDC::put('x-nexus-caller-workflow-id', Workflow::getInfo()->execution->getID()); + + /** @var HelloOutput $output */ + $output = yield $this->sampleNexusService->hello(new HelloInput($message, $language)); + return $output->message; + } +} diff --git a/app/src/NexusContextPropagation/ExecuteCommand.php b/app/src/NexusContextPropagation/ExecuteCommand.php new file mode 100644 index 0000000..b3cf669 --- /dev/null +++ b/app/src/NexusContextPropagation/ExecuteCommand.php @@ -0,0 +1,69 @@ +withNamespace('my-caller-namespace'), + ); + + $output->writeln("Starting EchoCallerWorkflow..."); + $echoWorkflow = $client->newWorkflowStub( + EchoCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($echoWorkflow, 'Nexus Echo πŸ‘‹'); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + $output->writeln("\nStarting HelloCallerWorkflow..."); + $helloWorkflow = $client->newWorkflowStub( + HelloCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($helloWorkflow, 'Nexus', Language::EN); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + return self::SUCCESS; + } +} diff --git a/app/src/NexusContextPropagation/Handler/HandlerWorker.php b/app/src/NexusContextPropagation/Handler/HandlerWorker.php new file mode 100644 index 0000000..226ef22 --- /dev/null +++ b/app/src/NexusContextPropagation/Handler/HandlerWorker.php @@ -0,0 +1,10 @@ +name + : $input->name . ', x-nexus-caller-workflow-id: ' . $callerWorkflowId; + + switch ($input->language) { + case Language::EN: + return new HelloOutput("Hello {$name} πŸ‘‹"); + case Language::FR: + return new HelloOutput("Bonjour {$name} πŸ‘‹"); + case Language::DE: + return new HelloOutput("Hallo {$name} πŸ‘‹"); + case Language::ES: + return new HelloOutput("Β‘Hola! {$name} πŸ‘‹"); + case Language::TR: + return new HelloOutput("Merhaba {$name} πŸ‘‹"); + } + throw new ApplicationFailure( + "Unsupported language: {$input->language->value}", + 'UNSUPPORTED_LANGUAGE', + true, + ); + } +} diff --git a/app/src/NexusContextPropagation/Handler/SampleNexusServiceImpl.php b/app/src/NexusContextPropagation/Handler/SampleNexusServiceImpl.php new file mode 100644 index 0000000..ec22e62 --- /dev/null +++ b/app/src/NexusContextPropagation/Handler/SampleNexusServiceImpl.php @@ -0,0 +1,49 @@ +echoClient = $echoClient ?? new EchoClientImpl(); + } + + public function echo(EchoInput $input): EchoOutput + { + $headers = Nexus::getCurrentOperationContext()->headers; + if (($id = $headers->get('x-nexus-caller-workflow-id')) !== null) { + \error_log("Echo called from a workflow with ID : {$id}"); + } + + return $this->echoClient->echo($input); + } + + public function hello(HelloInput $input): WorkflowHandle + { + $headers = Nexus::getCurrentOperationContext()->headers; + if (($id = $headers->get('x-nexus-caller-workflow-id')) !== null) { + \error_log("HelloHandlerWorkflow called from a workflow with ID : {$id}"); + } + + return WorkflowHandle::fromWorkflowMethod( + HelloHandlerWorkflow::class, + WorkflowOptions::new()->withWorkflowId(Nexus::getStartDetails()->requestId), + $input, + ); + } +} diff --git a/app/src/NexusContextPropagation/Propagation/MDC.php b/app/src/NexusContextPropagation/Propagation/MDC.php new file mode 100644 index 0000000..e8cc3c2 --- /dev/null +++ b/app/src/NexusContextPropagation/Propagation/MDC.php @@ -0,0 +1,39 @@ +> */ + private static array $contexts = []; + + public static function put(string $key, string $value): void + { + self::$contexts[self::scope()][$key] = $value; + } + + public static function get(string $key): ?string + { + return self::$contexts[self::scope()][$key] ?? null; + } + + /** @return array */ + public static function getAll(): array + { + return self::$contexts[self::scope()] ?? []; + } + + public static function clear(): void + { + unset(self::$contexts[self::scope()]); + } + + private static function scope(): string + { + return Workflow::getInfo()->execution->getID(); + } +} diff --git a/app/src/NexusContextPropagation/Propagation/NexusOutboundContextInterceptor.php b/app/src/NexusContextPropagation/Propagation/NexusOutboundContextInterceptor.php new file mode 100644 index 0000000..626897f --- /dev/null +++ b/app/src/NexusContextPropagation/Propagation/NexusOutboundContextInterceptor.php @@ -0,0 +1,29 @@ +nexusHeaders; + foreach (MDC::getAll() as $key => $value) { + if (\str_starts_with($key, self::NEXUS_HEADER_PREFIX)) { + $headers[$key] = $value; + } + } + + return $next($input->with(nexusHeaders: $headers)); + } +} diff --git a/app/src/NexusContextPropagation/Propagation/NexusStartContextInterceptor.php b/app/src/NexusContextPropagation/Propagation/NexusStartContextInterceptor.php new file mode 100644 index 0000000..d958268 --- /dev/null +++ b/app/src/NexusContextPropagation/Propagation/NexusStartContextInterceptor.php @@ -0,0 +1,50 @@ +with(header: Header::fromValues($propagated))); + } + + /** + * @return array + */ + private static function propagatedHeaders(): array + { + try { + $headers = Nexus::getCurrentOperationContext()->headers->all(); + } catch (\LogicException) { + return []; + } + + $propagated = []; + foreach ($headers as $name => $value) { + if (\str_starts_with($name, self::NEXUS_HEADER_PREFIX)) { + $propagated[$name] = $value; + } + } + + return $propagated; + } +} diff --git a/app/src/NexusContextPropagation/Propagation/WorkflowInboundContextInterceptor.php b/app/src/NexusContextPropagation/Propagation/WorkflowInboundContextInterceptor.php new file mode 100644 index 0000000..a3dc6d3 --- /dev/null +++ b/app/src/NexusContextPropagation/Propagation/WorkflowInboundContextInterceptor.php @@ -0,0 +1,30 @@ +header as $name => $_) { + $name = (string) $name; + if (\str_starts_with($name, self::NEXUS_HEADER_PREFIX)) { + MDC::put($name, (string) $input->header->getValue($name, 'string')); + } + } + + $next($input); + } +} diff --git a/app/src/NexusContextPropagation/README.md b/app/src/NexusContextPropagation/README.md new file mode 100644 index 0000000..609119d --- /dev/null +++ b/app/src/NexusContextPropagation/README.md @@ -0,0 +1,94 @@ +# Nexus Context Propagation sample + +Demonstrates carrying a value from a caller workflow all the way into the +workflow that backs an async Nexus operation. Four pieces, all in +`Propagation/`: + +| Piece | Role | +|---|---| +| `MDC` | per-workflow bag of `x-nexus-*` values | +| `NexusOutboundContextInterceptor` | caller side: MDC β†’ operation headers | +| `NexusStartContextInterceptor` | handler side: operation headers β†’ start header of the backing workflow | +| `WorkflowInboundContextInterceptor` | backing workflow: start header β†’ MDC | + +`NexusStartContextInterceptor` is a `WorkflowClientCallsInterceptor` registered +on the handler worker's `WorkflowClient`, which is the client the SDK uses to +start the backing workflow β€” the same shape as `ContextPropagator` in the Java +and Go samples. Nothing is propagated unless you register it: the SDK never +copies Nexus headers into a workflow by itself. + +The service contract, the caller workflow interfaces and the handler +`EchoClient` come from the [Nexus sample](../Nexus/README.md); only +`Handler/HelloHandlerWorkflow` stays local, so this sample's handler workflow +can be registered next to the base one in the shared feature-test worker. The +caller +implementations mirror it with one addition β€” the `MDC::put(...)` line. +`SampleNexusServiceImpl` logs the +propagated workflow ID, and `HelloHandlerWorkflowImpl` appends it to the +greeting, so the value is observable at both boundaries. + +> **⚠️ Do not put secrets into Nexus headers.** Nexus header values are +> plain strings on the wire. They bypass the workflow data-converter (the +> hook used by the encryption sample to encrypt payloads end-to-end) and +> are not routed through the gRPC proxy / codec server, so anything you +> stash here will land verbatim in handler-side logs and the Temporal Web +> UI. Use them only for trace IDs, tenant IDs, correlation IDs, and other +> non-sensitive metadata. + +## Prerequisites + +Beyond the usual (`./temporal`, `./rr`), and the two namespaces the +[Nexus sample](../Nexus/README.md) creates: + +```bash +./temporal operator nexus endpoint create \ + --name my-context-propagation-nexus-endpoint \ + --target-namespace my-target-namespace \ + --target-task-queue my-context-propagation-handler-task-queue +``` + +This sample owns its endpoint, task queues and RoadRunner RPC ports, so it +can run alongside the other Nexus samples. + +## Run + +Three terminals. + +**Handler worker** (`my-target-namespace`): + +```bash +cd app/src/NexusContextPropagation +TEMPORAL_NAMESPACE=my-target-namespace ../../rr serve -c .rr.handler.yaml +``` + +**Caller worker** (`my-caller-namespace`): + +```bash +cd app/src/NexusContextPropagation +TEMPORAL_NAMESPACE=my-caller-namespace ../../rr serve -c .rr.caller.yaml +``` + +**Starter**: + +```bash +php app/app.php nexus-context-propagation +``` + +Caller side: + +``` +Starting EchoCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Nexus Echo πŸ‘‹ + +Starting HelloCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Hello Nexus πŸ‘‹ +``` + +Handler side (in the rr stderr stream): + +``` +Echo called from a workflow with ID : +HelloHandlerWorkflow called from a workflow with ID : +``` diff --git a/app/src/NexusContextPropagation/caller-worker.php b/app/src/NexusContextPropagation/caller-worker.php new file mode 100644 index 0000000..2c382a5 --- /dev/null +++ b/app/src/NexusContextPropagation/caller-worker.php @@ -0,0 +1,34 @@ +newWorker( + CallerWorker::TASK_QUEUE, + interceptorProvider: new SimplePipelineProvider([ + new NexusOutboundContextInterceptor(), + ]), +)->registerWorkflowTypes( + EchoCallerWorkflowImpl::class, + HelloCallerWorkflowImpl::class, +); + +$factory->run(); diff --git a/app/src/NexusContextPropagation/handler-worker.php b/app/src/NexusContextPropagation/handler-worker.php new file mode 100644 index 0000000..c66104e --- /dev/null +++ b/app/src/NexusContextPropagation/handler-worker.php @@ -0,0 +1,46 @@ +withNamespace($namespace), + interceptorProvider: new SimplePipelineProvider([new NexusStartContextInterceptor()]), +); + +$factory = WorkerFactory::create(client: $workflowClient); + +$worker = $factory->newWorker( + HandlerWorker::TASK_QUEUE, + interceptorProvider: new SimplePipelineProvider([new WorkflowInboundContextInterceptor()]), +); +$worker->registerWorkflowTypes(HelloHandlerWorkflowImpl::class); +$worker->registerNexusServiceImplementation(new SampleNexusServiceImpl()); + +$factory->run(); diff --git a/app/src/NexusManualOperation/.rr.caller.yaml b/app/src/NexusManualOperation/.rr.caller.yaml new file mode 100644 index 0000000..bd2aa62 --- /dev/null +++ b/app/src/NexusManualOperation/.rr.caller.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6142 + +server: + command: "php caller-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-caller-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusManualOperation/.rr.handler.yaml b/app/src/NexusManualOperation/.rr.handler.yaml new file mode 100644 index 0000000..1319e79 --- /dev/null +++ b/app/src/NexusManualOperation/.rr.handler.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6141 + +server: + command: "php handler-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-target-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusManualOperation/Caller/CallerWorker.php b/app/src/NexusManualOperation/Caller/CallerWorker.php new file mode 100644 index 0000000..6be1a11 --- /dev/null +++ b/app/src/NexusManualOperation/Caller/CallerWorker.php @@ -0,0 +1,11 @@ +stub = Workflow::newUntypedNexusOperationStub( + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withService('SampleNexusService') + ->withScheduleToCloseTimeout(CarbonInterval::seconds(20)) + ->withCancellationType(NexusOperationCancellationType::TryCancel), + ); + } + + public function run(string $jobName) + { + /** @var NexusOperationHandle $handle */ + $handle = yield $this->stub->start( + 'startJob', + [new JobInput($jobName, instant: true)], + JobResult::class, + ); + + if ($handle->getOperationToken() !== null) { + throw new \LogicException('Sync result must not carry an operation token.'); + } + + /** @var JobResult $instantResult */ + $instantResult = yield $handle->getResult(); + + $handle2 = null; + $failure = null; + $scope = Workflow::async(function () use ($jobName, &$handle2, &$failure): \Generator { + try { + $handle2 = yield $this->stub->start( + 'startJob', + [new JobInput($jobName)], + JobResult::class, + ); + } catch (\Throwable $e) { + $failure = $e; + throw $e; + } + + yield $handle2->getResult(); + }); + + yield Workflow::await(function () use (&$handle2, &$failure): bool { + return $handle2 !== null || $failure !== null; + }); + + if ($failure !== null) { + throw $failure; + } + + $token2 = $handle2->getOperationToken(); + + yield Workflow::timer(1); + $scope->cancel(); + + $cancelled = false; + + try { + yield $scope; + } catch (CanceledFailure) { + $cancelled = true; + } catch (NexusOperationFailure $e) { + if (!$e->getPrevious() instanceof CanceledFailure) { + throw $e; + } + + $cancelled = true; + } + + if (!$cancelled) { + throw new \LogicException('Async operation did not end cancelled.'); + } + + return "[instant={$instantResult->message}] [token={$token2}] cancelled"; + } +} diff --git a/app/src/NexusManualOperation/ExecuteCommand.php b/app/src/NexusManualOperation/ExecuteCommand.php new file mode 100644 index 0000000..33c7322 --- /dev/null +++ b/app/src/NexusManualOperation/ExecuteCommand.php @@ -0,0 +1,53 @@ +withNamespace('my-caller-namespace'), + ); + + $output->writeln("Starting JobCallerWorkflow..."); + $workflow = $client->newWorkflowStub( + JobCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($workflow, 'Demo'); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + return self::SUCCESS; + } +} diff --git a/app/src/NexusManualOperation/Handler/ExternalJobClient.php b/app/src/NexusManualOperation/Handler/ExternalJobClient.php new file mode 100644 index 0000000..8186a04 --- /dev/null +++ b/app/src/NexusManualOperation/Handler/ExternalJobClient.php @@ -0,0 +1,18 @@ + */ +final class JobOperationHandler implements OperationHandlerInterface +{ + public function __construct( + private readonly ExternalJobClient $client, + ) {} + + public function start( + OperationContext $context, + OperationStartDetails $details, + mixed $param, + ): OperationStartResult { + if ($param->instant) { + return OperationStartResult::sync(new JobResult("done instantly: {$param->jobName}")); + } + + $jobId = $this->client->submit($param->jobName, $details->requestId); + + return OperationStartResult::async(new OperationInfo($jobId, OperationState::Running)); + } + + public function cancel( + OperationContext $context, + OperationCancelDetails $details, + ): void { + $this->client->abort($details->operationToken); + } +} diff --git a/app/src/NexusManualOperation/Handler/SampleNexusService.php b/app/src/NexusManualOperation/Handler/SampleNexusService.php new file mode 100644 index 0000000..33938ea --- /dev/null +++ b/app/src/NexusManualOperation/Handler/SampleNexusService.php @@ -0,0 +1,24 @@ +client); + } +} diff --git a/app/src/NexusManualOperation/README.md b/app/src/NexusManualOperation/README.md new file mode 100644 index 0000000..a2bdbd1 --- /dev/null +++ b/app/src/NexusManualOperation/README.md @@ -0,0 +1,96 @@ +# Nexus Manual Operation sample + +The third Nexus handler form: a **manual** operation backed by an +`OperationHandlerInterface` object. Unlike `#[Operation]` (sync method) and +the `WorkflowHandle`-returning `#[AsyncOperation]` (SDK-managed workflow run), +here one object owns the whole lifecycle: + +- `start()` decides **at runtime** whether to answer synchronously + (`OperationStartResult::sync(...)` β€” no token issued) or asynchronously + (`OperationStartResult::async(new OperationInfo($token, ...))` β€” the + handler mints its **own token**, here the external job id). +- `cancel()` receives that same token back via + `$details->operationToken` and aborts the external job. + +The `#[Service]` attribute sits on a self-contained **class** β€” no interface +is needed, because the operation's wire contract (input/output types) lives +on the `#[AsyncOperation(output: ..., input: ...)]` attribute of the +zero-parameter factory method, which runs once at worker registration. No +`WorkflowClient` is needed either: nothing starts a backing workflow. + +Completion of the async result is delivered by the external system via the +Nexus completion callback (`$details->callbackUrl`) and is out of scope here β€” +the sample demonstrates the sync fast-path, the custom token, and cancel. + +The caller (`JobCallerWorkflow`) uses the **untyped** stub +(`Workflow::newUntypedNexusOperationStub()`): call 1 with `instant: true` +returns immediately with a `null` token; call 2 runs async inside a +cancellable scope, the caller reads the handler-issued `job-...` token and +then cancels the scope, which routes to the handler's `cancel()`. + +## Prerequisites + +Beyond the usual (`./temporal`, `./rr`): + +```bash +./temporal operator namespace create --namespace my-target-namespace +./temporal operator namespace create --namespace my-caller-namespace + +./temporal operator nexus endpoint create \ + --name my-manual-nexus-endpoint \ + --target-namespace my-target-namespace \ + --target-task-queue my-manual-handler-task-queue +``` + +This sample owns its endpoint, task queues and RoadRunner RPC ports, so it +can run alongside the other Nexus samples. + +## Run + +Three terminals. + +**Handler worker** (`my-target-namespace`): + +```bash +cd app/src/NexusManualOperation +TEMPORAL_NAMESPACE=my-target-namespace ../../rr serve -c .rr.handler.yaml +``` + +**Caller worker** (`my-caller-namespace`): + +```bash +cd app/src/NexusManualOperation +TEMPORAL_NAMESPACE=my-caller-namespace ../../rr serve -c .rr.caller.yaml +``` + +**Starter**: + +```bash +php app/app.php nexus-manual-operation +``` + +Expected output: + +``` +Starting JobCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: [instant=done instantly: Demo] [token=job-...] cancelled +``` + +The `token=job-...` part is the handler-minted token (the fake external job +id built from the start request id); the handler worker logs +`ExternalJobClient: aborted job-...` when the cancel arrives. + +## Constants + +| Name | Value | +|---|---| +| service | `SampleNexusService` | +| operations | `startJob` | +| endpoint | `my-manual-nexus-endpoint` | +| handler task queue | `my-manual-handler-task-queue` | +| caller task queue | `my-manual-caller-task-queue` | +| target namespace | `my-target-namespace` | +| caller namespace | `my-caller-namespace` | +| `scheduleToCloseTimeout` | `20s` | +| `cancellationType` | `TryCancel` | diff --git a/app/src/NexusManualOperation/Service/JobInput.php b/app/src/NexusManualOperation/Service/JobInput.php new file mode 100644 index 0000000..d445a14 --- /dev/null +++ b/app/src/NexusManualOperation/Service/JobInput.php @@ -0,0 +1,13 @@ +newWorker(CallerWorker::TASK_QUEUE) + ->registerWorkflowTypes(JobCallerWorkflowImpl::class); + +$factory->run(); diff --git a/app/src/NexusManualOperation/handler-worker.php b/app/src/NexusManualOperation/handler-worker.php new file mode 100644 index 0000000..77daa93 --- /dev/null +++ b/app/src/NexusManualOperation/handler-worker.php @@ -0,0 +1,24 @@ +newWorker(HandlerWorker::TASK_QUEUE); +$worker->registerNexusServiceImplementation(new SampleNexusService()); + +$factory->run(); diff --git a/app/src/NexusMultipleArguments/.rr.caller.yaml b/app/src/NexusMultipleArguments/.rr.caller.yaml new file mode 100644 index 0000000..d82b5cc --- /dev/null +++ b/app/src/NexusMultipleArguments/.rr.caller.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6132 + +server: + command: "php caller-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-caller-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusMultipleArguments/.rr.handler.yaml b/app/src/NexusMultipleArguments/.rr.handler.yaml new file mode 100644 index 0000000..ac98e47 --- /dev/null +++ b/app/src/NexusMultipleArguments/.rr.handler.yaml @@ -0,0 +1,16 @@ +version: "3" + +rpc: + listen: tcp://127.0.0.1:6131 + +server: + command: "php handler-worker.php" + +temporal: + address: ${TEMPORAL_HOST:-localhost}:${TEMPORAL_PORT:-7233} + namespace: ${TEMPORAL_NAMESPACE:-my-target-namespace} + activities: + num_workers: 2 + +logs: + level: info diff --git a/app/src/NexusMultipleArguments/Caller/CallerWorker.php b/app/src/NexusMultipleArguments/Caller/CallerWorker.php new file mode 100644 index 0000000..0eccfb7 --- /dev/null +++ b/app/src/NexusMultipleArguments/Caller/CallerWorker.php @@ -0,0 +1,11 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function echo(string $message) + { + /** @var EchoOutput $output */ + $output = yield $this->sampleNexusService->echo(new EchoInput($message)); + return $output->message; + } +} diff --git a/app/src/NexusMultipleArguments/Caller/HelloCallerWorkflow.php b/app/src/NexusMultipleArguments/Caller/HelloCallerWorkflow.php new file mode 100644 index 0000000..bf47c5e --- /dev/null +++ b/app/src/NexusMultipleArguments/Caller/HelloCallerWorkflow.php @@ -0,0 +1,16 @@ +sampleNexusService = Workflow::newNexusServiceStub( + SampleNexusService::class, + NexusOperationOptions::new() + ->withEndpoint(CallerWorker::ENDPOINT_NAME) + ->withScheduleToCloseTimeout(CarbonInterval::seconds(10)), + ); + } + + public function hello(string $message, Language $language) + { + /** @var HelloOutput $output */ + $output = yield $this->sampleNexusService->hello(new HelloInput($message, $language)); + return $output->message; + } +} diff --git a/app/src/NexusMultipleArguments/ExecuteCommand.php b/app/src/NexusMultipleArguments/ExecuteCommand.php new file mode 100644 index 0000000..9cdd320 --- /dev/null +++ b/app/src/NexusMultipleArguments/ExecuteCommand.php @@ -0,0 +1,69 @@ +withNamespace('my-caller-namespace'), + ); + + $output->writeln("Starting EchoCallerWorkflow..."); + $echoWorkflow = $client->newWorkflowStub( + EchoCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($echoWorkflow, 'Nexus Echo πŸ‘‹'); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + $output->writeln("\nStarting HelloCallerWorkflow..."); + $helloWorkflow = $client->newWorkflowStub( + HelloCallerWorkflow::class, + WorkflowOptions::new()->withTaskQueue(CallerWorker::TASK_QUEUE), + ); + $run = $client->start($helloWorkflow, 'Nexus', Language::ES); + $execution = $run->getExecution(); + $output->writeln(\sprintf( + 'Started: WorkflowID=%s RunID=%s', + $execution->getID(), + $execution->getRunID(), + )); + $output->writeln(\sprintf("Result: %s", $run->getResult('string'))); + + return self::SUCCESS; + } +} diff --git a/app/src/NexusMultipleArguments/Handler/HandlerWorker.php b/app/src/NexusMultipleArguments/Handler/HandlerWorker.php new file mode 100644 index 0000000..e3e63c6 --- /dev/null +++ b/app/src/NexusMultipleArguments/Handler/HandlerWorker.php @@ -0,0 +1,10 @@ +value}", + 'UNSUPPORTED_LANGUAGE', + true, + ); + } +} diff --git a/app/src/NexusMultipleArguments/Handler/SampleNexusServiceImpl.php b/app/src/NexusMultipleArguments/Handler/SampleNexusServiceImpl.php new file mode 100644 index 0000000..d237de8 --- /dev/null +++ b/app/src/NexusMultipleArguments/Handler/SampleNexusServiceImpl.php @@ -0,0 +1,40 @@ +echoClient = $echoClient ?? new EchoClientImpl(); + } + + public function echo(EchoInput $input): EchoOutput + { + return $this->echoClient->echo($input); + } + + public function hello(HelloInput $input): WorkflowHandle + { + return WorkflowHandle::fromWorkflowMethod( + HelloHandlerWorkflow::class, + WorkflowOptions::new()->withWorkflowId(Nexus::getStartDetails()->requestId), + $input->name, + $input->language, + ); + } +} diff --git a/app/src/NexusMultipleArguments/README.md b/app/src/NexusMultipleArguments/README.md new file mode 100644 index 0000000..982fd4c --- /dev/null +++ b/app/src/NexusMultipleArguments/README.md @@ -0,0 +1,126 @@ +# Nexus Multiple Arguments sample + +Same Nexus contract as the [basic Nexus sample](../Nexus/README.md), but the +backing handler workflow takes **two positional arguments** instead of one +`HelloInput` DTO. The unpack happens in the service implementation: when the +operation receives a `HelloInput`, it passes `$input->name` and +`$input->language` as separate arguments to the workflow via +`WorkflowHandle::fromWorkflowMethod(class, options, ...$args)`. + +Useful when the Nexus contract DTO and the underlying workflow signature +have to evolve independently β€” e.g. the workflow already exists with a +multi-arg signature and is called from many places, and you want to wrap +it behind a Nexus operation without changing the workflow. + +The service contract is imported from the +[basic Nexus sample](../Nexus/README.md). Only `Handler/HelloHandlerWorkflow` +and `Handler/HelloHandlerWorkflowImpl` (multi-arg signature) and +`Handler/SampleNexusServiceImpl` (unpacks the input) are sample-specific. + +## Prerequisites + +Beyond the usual (`./temporal`, `./rr`), and the two namespaces the +[Nexus sample](../Nexus/README.md) creates: + +```bash +./temporal operator nexus endpoint create \ + --name my-multiple-arguments-nexus-endpoint \ + --target-namespace my-target-namespace \ + --target-task-queue my-multiple-arguments-handler-task-queue +``` + +This sample owns its endpoint, task queues and RoadRunner RPC ports, so it +can run alongside the other Nexus samples. + +## Run + +Three terminals. + +**Handler worker** (`my-target-namespace`): + +```bash +cd app/src/NexusMultipleArguments +TEMPORAL_NAMESPACE=my-target-namespace ../../rr serve -c .rr.handler.yaml +``` + +**Caller worker** (`my-caller-namespace`): + +```bash +cd app/src/NexusMultipleArguments +TEMPORAL_NAMESPACE=my-caller-namespace ../../rr serve -c .rr.caller.yaml +``` + +**Starter**: + +```bash +php app/app.php nexus-multiple-arguments +``` + +Expected output: + +``` +Starting EchoCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Nexus Echo πŸ‘‹ + +Starting HelloCallerWorkflow... +Started: WorkflowID=... RunID=... +Result: Β‘Hola! Nexus πŸ‘‹ +``` + +## What's different from the basic sample + +Three files: the handler workflow interface and implementation (multi-arg +signature) and the service implementation (unpacks the DTO). The rest is a +copy of the basic sample. + +`Handler/HelloHandlerWorkflow.php` β€” interface signature uses positional args: + +```php +#[WorkflowInterface] +interface HelloHandlerWorkflow +{ + #[WorkflowMethod] + public function hello(string $name, Language $language); +} +``` + +`Handler/SampleNexusServiceImpl.php::hello()` β€” unpacks `HelloInput` when +building the `WorkflowHandle`: + +```php +return WorkflowHandle::fromWorkflowMethod( + HelloHandlerWorkflow::class, + WorkflowOptions::new()->withWorkflowId(Nexus::getStartDetails()->requestId), + $input->name, // ← positional + $input->language, // ← positional +); +``` + +Compare with the basic sample, which passes the whole DTO: + +```php +return WorkflowHandle::fromWorkflowMethod( + HelloHandlerWorkflow::class, + WorkflowOptions::new()->withWorkflowId(Nexus::getStartDetails()->requestId), + $input, // ← whole DTO, single arg +); +``` + +The Nexus operation contract (`SampleNexusService::hello(HelloInput): WorkflowHandle`) +and the caller workflows are unchanged. + +## Constants + +Same as the [basic Nexus sample](../Nexus/README.md). Listed here for +quick reference: + +| Name | Value | +|---|---| +| service | `SampleNexusService` | +| operations | `echo`, `hello` | +| endpoint | `my-multiple-arguments-nexus-endpoint` | +| handler task queue | `my-multiple-arguments-handler-task-queue` | +| caller task queue | `my-multiple-arguments-caller-task-queue` | +| target namespace | `my-target-namespace` | +| caller namespace | `my-caller-namespace` | diff --git a/app/src/NexusMultipleArguments/caller-worker.php b/app/src/NexusMultipleArguments/caller-worker.php new file mode 100644 index 0000000..be655c6 --- /dev/null +++ b/app/src/NexusMultipleArguments/caller-worker.php @@ -0,0 +1,28 @@ +newWorker(CallerWorker::TASK_QUEUE) + ->registerWorkflowTypes( + EchoCallerWorkflowImpl::class, + HelloCallerWorkflowImpl::class, + ); + +$factory->run(); diff --git a/app/src/NexusMultipleArguments/handler-worker.php b/app/src/NexusMultipleArguments/handler-worker.php new file mode 100644 index 0000000..490ecc4 --- /dev/null +++ b/app/src/NexusMultipleArguments/handler-worker.php @@ -0,0 +1,39 @@ +withNamespace($namespace), +); + +$factory = WorkerFactory::create(client: $workflowClient); + +$worker = $factory->newWorker(HandlerWorker::TASK_QUEUE); +$worker->registerWorkflowTypes(HelloHandlerWorkflowImpl::class); +$worker->registerNexusServiceImplementation(new SampleNexusServiceImpl()); + +$factory->run(); diff --git a/app/tests/Feature/.rr.test.yaml b/app/tests/Feature/.rr.test.yaml index 0cbecd4..1c3c316 100644 --- a/app/tests/Feature/.rr.test.yaml +++ b/app/tests/Feature/.rr.test.yaml @@ -7,7 +7,7 @@ server: command: "php worker.php" temporal: - address: ${TEMPORAL_ADDRESS:-localhost:7236} + address: ${TEMPORAL_ADDRESS:-127.0.0.1:7236} activities: num_workers: 2 @@ -19,7 +19,6 @@ kv: logs: level: info - # mode: production channels: server: mode: production diff --git a/app/tests/Feature/Nexus/CallerWorkflowMockTest.php b/app/tests/Feature/Nexus/CallerWorkflowMockTest.php new file mode 100644 index 0000000..2a00b5b --- /dev/null +++ b/app/tests/Feature/Nexus/CallerWorkflowMockTest.php @@ -0,0 +1,41 @@ +newCaller(EchoCallerWorkflow::class); + + $result = $workflow->echo('ignored'); + + self::assertSame(MockEchoClient::CANNED, $result); + } + + public function testHelloUsesMockHandlerWorkflow(): void + { + $workflow = $this->newCaller(HelloCallerWorkflow::class); + + $result = $workflow->hello('World', Language::EN); + + self::assertSame(MockHelloHandlerWorkflowImpl::CANNED, $result); + } +} diff --git a/app/tests/Feature/Nexus/CallerWorkflowTest.php b/app/tests/Feature/Nexus/CallerWorkflowTest.php new file mode 100644 index 0000000..b47cf59 --- /dev/null +++ b/app/tests/Feature/Nexus/CallerWorkflowTest.php @@ -0,0 +1,49 @@ +newCaller(EchoCallerWorkflow::class); + + $result = $workflow->echo('Hello'); + + self::assertSame('Hello', $result); + } + + public function testHelloRoundTrip(): void + { + $workflow = $this->newCaller(HelloCallerWorkflow::class); + + $result = $workflow->hello('World', Language::EN); + + self::assertSame('Hello World πŸ‘‹', $result); + } + + public function testHelloRespectsLanguage(): void + { + $workflow = $this->newCaller(HelloCallerWorkflow::class); + + $result = $workflow->hello('Nexus', Language::ES); + + self::assertSame('Β‘Hola! Nexus πŸ‘‹', $result); + } +} diff --git a/app/tests/Feature/Nexus/CancellationTest.php b/app/tests/Feature/Nexus/CancellationTest.php new file mode 100644 index 0000000..66e880b --- /dev/null +++ b/app/tests/Feature/Nexus/CancellationTest.php @@ -0,0 +1,48 @@ +newCaller(HelloCallerWorkflow::class); + $run = $this->workflowClient->start($workflow, 'Nexus'); + + self::assertContains($run->getResult('string'), [ + 'Hello Nexus πŸ‘‹', + 'Bonjour Nexus πŸ‘‹', + 'Hallo Nexus πŸ‘‹', + 'Β‘Hola! Nexus πŸ‘‹', + 'Merhaba Nexus πŸ‘‹', + ]); + + self::assertSame(4, $this->countCancelRequests($run->getExecution())); + } + + private function countCancelRequests(\Temporal\Workflow\WorkflowExecution $execution): int + { + $count = 0; + foreach ($this->workflowClient->getWorkflowHistory($execution) as $event) { + \assert($event instanceof HistoryEvent); + if ($event->hasNexusOperationCancelRequestedEventAttributes()) { + ++$count; + } + } + + return $count; + } +} diff --git a/app/tests/Feature/Nexus/ContextPropagationTest.php b/app/tests/Feature/Nexus/ContextPropagationTest.php new file mode 100644 index 0000000..99cf4b7 --- /dev/null +++ b/app/tests/Feature/Nexus/ContextPropagationTest.php @@ -0,0 +1,49 @@ +newCaller( + EchoCallerWorkflow::class, + WorkflowOptions::new()->withWorkflowId($workflowId), + ); + + $result = $workflow->echo('ignored'); + + self::assertSame($workflowId, $result); + } + + public function testCallerWorkflowIdReachesTheBackingWorkflow(): void + { + $workflowId = 'ctx-prop-async-' . \bin2hex(\random_bytes(4)); + $workflow = $this->newCaller( + HelloCallerWorkflow::class, + WorkflowOptions::new()->withWorkflowId($workflowId), + ); + + $result = $workflow->hello('Nexus', Language::EN); + + self::assertStringContainsString($workflowId, $result); + } +} diff --git a/app/tests/Feature/Nexus/ManualOperationStartFailureTest.php b/app/tests/Feature/Nexus/ManualOperationStartFailureTest.php new file mode 100644 index 0000000..c672447 --- /dev/null +++ b/app/tests/Feature/Nexus/ManualOperationStartFailureTest.php @@ -0,0 +1,40 @@ +newCaller( + JobCallerWorkflow::class, + WorkflowOptions::new()->withWorkflowExecutionTimeout(CarbonInterval::seconds(15)), + ); + + try { + $workflow->run('Demo'); + self::fail('Expected the workflow to fail with the handler error.'); + } catch (WorkflowFailedException $e) { + self::assertStringContainsString( + FailingJobClient::FAILURE_MESSAGE, + (string) $e, + ); + } + } +} diff --git a/app/tests/Feature/Nexus/ManualOperationTest.php b/app/tests/Feature/Nexus/ManualOperationTest.php new file mode 100644 index 0000000..adfc0fa --- /dev/null +++ b/app/tests/Feature/Nexus/ManualOperationTest.php @@ -0,0 +1,31 @@ +newCaller(JobCallerWorkflow::class); + + $result = $workflow->run('Demo'); + + self::assertStringContainsString('done instantly: Demo', $result); + self::assertStringContainsString('[token=job-', $result); + self::assertStringEndsWith('cancelled', $result); + } +} diff --git a/app/tests/Feature/Nexus/Mock/FailingJobClient.php b/app/tests/Feature/Nexus/Mock/FailingJobClient.php new file mode 100644 index 0000000..dbc7f32 --- /dev/null +++ b/app/tests/Feature/Nexus/Mock/FailingJobClient.php @@ -0,0 +1,19 @@ +'; + + private readonly SampleNexusServiceImpl $inner; + + public function __construct() + { + $this->inner = new SampleNexusServiceImpl(); + } + + public function echo(EchoInput $input): EchoOutput + { + $headers = Nexus::getCurrentOperationContext()->headers; + + return new EchoOutput($headers->get('x-nexus-caller-workflow-id') ?? self::MISSING); + } + + public function hello(HelloInput $input): WorkflowHandle + { + return $this->inner->hello($input); + } +} diff --git a/app/tests/Feature/Nexus/Mock/MockEchoClient.php b/app/tests/Feature/Nexus/Mock/MockEchoClient.php new file mode 100644 index 0000000..b202a9e --- /dev/null +++ b/app/tests/Feature/Nexus/Mock/MockEchoClient.php @@ -0,0 +1,24 @@ +withWorkflowId(Nexus::getStartDetails()->requestId), + $input, + ); + } +} diff --git a/app/tests/Feature/Nexus/MultipleArgumentsTest.php b/app/tests/Feature/Nexus/MultipleArgumentsTest.php new file mode 100644 index 0000000..2391f69 --- /dev/null +++ b/app/tests/Feature/Nexus/MultipleArgumentsTest.php @@ -0,0 +1,28 @@ +newCaller(HelloCallerWorkflow::class); + + $result = $workflow->hello('World', Language::ES); + + self::assertSame('Β‘Hola! World πŸ‘‹', $result); + } +} diff --git a/app/tests/Feature/Nexus/NexusEndpointHelper.php b/app/tests/Feature/Nexus/NexusEndpointHelper.php new file mode 100644 index 0000000..d9e533c --- /dev/null +++ b/app/tests/Feature/Nexus/NexusEndpointHelper.php @@ -0,0 +1,132 @@ +operator = new OperatorServiceClient( + $temporalAddress, + ['credentials' => \Grpc\ChannelCredentials::createInsecure()], + ); + } + + /** + * Create a Nexus endpoint targeting `(namespace, taskQueue)` and return + * `[id, name]`. The id is used by HTTP routes, the name is what callers + * pass to {@see \Temporal\Workflow\NexusOperationOptions::withEndpoint()}. + * + * @return array{id: string, name: string} + */ + public function setupEndpoint(string $namespace, string $taskQueue, ?string $name = null): array + { + $name ??= 'samples-test-' . \bin2hex(\random_bytes(4)); + $this->deleteEndpointByName($name); + + $request = (new CreateNexusEndpointRequest()) + ->setSpec( + (new EndpointSpec()) + ->setName($name) + ->setTarget( + (new EndpointTarget())->setWorker( + (new WorkerTarget()) + ->setNamespace($namespace) + ->setTaskQueue($taskQueue), + ), + ), + ); + + [$response, $status] = $this->operator->CreateNexusEndpoint($request)->wait(); + + if ($status->code !== \Grpc\STATUS_OK) { + throw new \RuntimeException( + "CreateNexusEndpoint failed (gRPC code {$status->code}): {$status->details}", + ); + } + + $id = $response->getEndpoint()->getId(); + $this->awaitEndpointResolvable($id); + + return ['id' => $id, 'name' => $name]; + } + + private function awaitEndpointResolvable(string $endpointId, float $timeoutSeconds = 15.0): void + { + $deadline = \microtime(true) + $timeoutSeconds; + $request = (new GetNexusEndpointRequest())->setId($endpointId); + + do { + [, $status] = $this->operator->GetNexusEndpoint($request)->wait(); + + if ($status->code === \Grpc\STATUS_OK) { + return; + } + + \usleep(100_000); + } while (\microtime(true) < $deadline); + + throw new \RuntimeException( + "Nexus endpoint {$endpointId} did not become resolvable within {$timeoutSeconds}s.", + ); + } + + public function close(): void + { + $this->operator->close(); + } + + /** + * Endpoint names are global, and samples pin theirs by constant, so a + * previous run that died mid-test can leave one behind. + */ + private function deleteEndpointByName(string $name): void + { + [$response, $status] = $this->operator + ->ListNexusEndpoints((new ListNexusEndpointsRequest())->setName($name)) + ->wait(); + + if ($status->code !== \Grpc\STATUS_OK) { + return; + } + + foreach ($response->getEndpoints() as $endpoint) { + $this->deleteEndpoint($endpoint->getId(), $endpoint->getVersion()); + } + } + + public function deleteEndpoint(string $endpointId, int $expectedVersion = 1): void + { + $request = (new DeleteNexusEndpointRequest()) + ->setId($endpointId) + ->setVersion($expectedVersion); + + [, $status] = $this->operator->DeleteNexusEndpoint($request)->wait(); + + if ($status->code !== \Grpc\STATUS_OK) { + // Best-effort cleanup; surface, don't crash other tests. + \trigger_error( + "DeleteNexusEndpoint failed (gRPC code {$status->code}): {$status->details}", + E_USER_WARNING, + ); + } + } +} diff --git a/app/tests/Feature/Nexus/NexusServiceMockTest.php b/app/tests/Feature/Nexus/NexusServiceMockTest.php new file mode 100644 index 0000000..a5bc621 --- /dev/null +++ b/app/tests/Feature/Nexus/NexusServiceMockTest.php @@ -0,0 +1,40 @@ +newCaller(EchoCallerWorkflow::class); + + $result = $workflow->echo('ignored'); + + self::assertSame(MockSampleNexusServiceImpl::ECHO_CANNED, $result); + } + + public function testHelloFromMockedService(): void + { + $workflow = $this->newCaller(HelloCallerWorkflow::class); + + $result = $workflow->hello('World', Language::DE); + + self::assertSame(MockHelloHandlerWorkflowImpl::CANNED, $result); + } +} diff --git a/app/tests/Feature/Nexus/NexusTestCase.php b/app/tests/Feature/Nexus/NexusTestCase.php new file mode 100644 index 0000000..1ced641 --- /dev/null +++ b/app/tests/Feature/Nexus/NexusTestCase.php @@ -0,0 +1,82 @@ +setupEndpoint( + namespace: 'default', + taskQueue: static::TASK_QUEUE, + name: static::ENDPOINT_NAME === '' ? null : static::ENDPOINT_NAME, + ); + } + + public static function tearDownAfterClass(): void + { + if (self::$nexusHelper !== null) { + if (self::$sharedEndpoint !== null) { + self::$nexusHelper->deleteEndpoint(self::$sharedEndpoint['id']); + } + + self::$nexusHelper->close(); + } + + self::$nexusHelper = null; + self::$sharedEndpoint = null; + + parent::tearDownAfterClass(); + } + + protected function setUp(): void + { + parent::setUp(); + + $this->endpoint = self::$sharedEndpoint; + } + + /** + * @template T of object + * @param class-string $workflowClass + * @return T + */ + protected function newCaller(string $workflowClass, ?WorkflowOptions $options = null): object + { + return $this->workflowClient->newWorkflowStub( + $workflowClass, + ($options ?? WorkflowOptions::new()->withWorkflowExecutionTimeout(CarbonInterval::seconds(60))) + ->withTaskQueue(static::TASK_QUEUE), + ); + } +} diff --git a/app/tests/Feature/TestCase.php b/app/tests/Feature/TestCase.php index 2418fea..4b63c69 100644 --- a/app/tests/Feature/TestCase.php +++ b/app/tests/Feature/TestCase.php @@ -15,7 +15,7 @@ abstract class TestCase extends \PHPUnit\Framework\TestCase protected function setUp(): void { - $this->workflowClient = new WorkflowClient( + $this->workflowClient = WorkflowClient::create( ServiceClient::create(\getenv('TEMPORAL_ADDRESS')), ); diff --git a/app/tests/Feature/bootstrap.php b/app/tests/Feature/bootstrap.php index 4dd680a..62dfea9 100644 --- a/app/tests/Feature/bootstrap.php +++ b/app/tests/Feature/bootstrap.php @@ -12,10 +12,23 @@ $sysInfo = \Temporal\Testing\SystemInfo::detect(); -$environment->startTemporalTestServer(); +// Nexus needs the full Temporal server; the time-skipping test server +// shipped via `startTemporalTestServer()` doesn't expose the Nexus APIs. +// Mirrors what sdk-php's acceptance harness does in TemporalStarter. +$httpPort = getenv('TEMPORAL_HTTP_PORT') ?: '7246'; + +$environment->startTemporalServer( + parameters: [ + '--http-port', $httpPort, + ], +); +// rr's `-c` is resolved relative to its `-w` workdir, while +// Environment::startRoadRunner's internal readiness check (`rr workers -c …`) +// runs from PHP's cwd. They need different paths. $environment->startRoadRunner( - rrCommand: sprintf('%s serve -c .rr.test.yaml -w tests/Feature', $sysInfo->rrExecutable), - commandTimeout: 5 + rrCommand: [$sysInfo->rrExecutable, 'serve', '-c', '.rr.test.yaml', '-w', 'tests/Feature'], + commandTimeout: 5, + configFile: 'tests/Feature/.rr.test.yaml', ); register_shutdown_function(fn() => $environment->stop()); diff --git a/app/tests/Feature/worker.php b/app/tests/Feature/worker.php index 536ed35..602a019 100644 --- a/app/tests/Feature/worker.php +++ b/app/tests/Feature/worker.php @@ -2,6 +2,27 @@ declare(strict_types=1); +use App\Tests\Feature\Nexus\CallerWorkflowMockTest; +use App\Tests\Feature\Nexus\CallerWorkflowTest; +use App\Tests\Feature\Nexus\CancellationTest; +use App\Tests\Feature\Nexus\ContextPropagationTest; +use App\Tests\Feature\Nexus\ManualOperationStartFailureTest; +use App\Tests\Feature\Nexus\ManualOperationTest; +use App\Tests\Feature\Nexus\Mock\FailingJobClient; +use App\Tests\Feature\Nexus\Mock\HeaderEchoNexusServiceImpl; +use App\Tests\Feature\Nexus\Mock\MockEchoClient; +use App\Tests\Feature\Nexus\Mock\MockHelloHandlerWorkflowImpl; +use App\Tests\Feature\Nexus\Mock\MockSampleNexusServiceImpl; +use App\Tests\Feature\Nexus\MultipleArgumentsTest; +use App\Tests\Feature\Nexus\NexusServiceMockTest; +use Temporal\Client\GRPC\ServiceClient; +use Temporal\Client\WorkflowClient; +use Temporal\Interceptor\SimplePipelineProvider; +use Temporal\Samples\Nexus\Handler\HelloHandlerWorkflowImpl; +use Temporal\Samples\Nexus\Handler\SampleNexusServiceImpl; +use Temporal\Samples\NexusContextPropagation\Propagation\NexusOutboundContextInterceptor; +use Temporal\Samples\NexusContextPropagation\Propagation\NexusStartContextInterceptor; +use Temporal\Samples\NexusContextPropagation\Propagation\WorkflowInboundContextInterceptor; use Temporal\Testing\WorkerFactory; ini_set('display_errors', 'stderr'); @@ -9,15 +30,109 @@ chdir(__DIR__ . '/../..'); require_once 'vendor/autoload.php'; -$workerFactory = WorkerFactory::create(); +// Async Nexus operations (WorkflowRunOperation) start a backing workflow, so +// the worker needs a WorkflowClient threaded through to the operation context. +$workflowClient = WorkflowClient::create( + ServiceClient::create(\getenv('TEMPORAL_ADDRESS') ?: 'localhost:7236'), + interceptorProvider: new SimplePipelineProvider([new NexusStartContextInterceptor()]), +); -$worker = $workerFactory->newWorker(taskQueue: 'tests'); +$workerFactory = WorkerFactory::create(client: $workflowClient); -// make sure to register concrete workflow implementations -$worker->registerWorkflowTypes(\Temporal\Samples\SimpleActivity\GreetingWorkflow::class); -$worker->registerActivity( - \Temporal\Samples\SimpleActivity\GreetingActivity::class, - fn() => new \Temporal\Samples\SimpleActivity\GreetingActivity(), -); +// Existing β€” SimpleActivity feature test. +$workerFactory->newWorker(taskQueue: 'tests') + ->registerWorkflowTypes(\Temporal\Samples\SimpleActivity\GreetingWorkflow::class) + ->registerActivity( + \Temporal\Samples\SimpleActivity\GreetingActivity::class, + fn() => new \Temporal\Samples\SimpleActivity\GreetingActivity(), + ); + +// === Nexus Feature tests === +// +// One task queue per scenario, all registered together in this single rr +// process. Each test creates its own Nexus endpoint (via gRPC OperatorService +// in setUp) targeting the matching queue, runs the test caller workflow on +// the same queue, then drops the endpoint in tearDown. + +// Scenario 1: real production handler. Mirrors `php app/app.php nexus`. +$workerFactory->newWorker(taskQueue: CallerWorkflowTest::TASK_QUEUE) + ->registerWorkflowTypes( + \Temporal\Samples\Nexus\Caller\EchoCallerWorkflowImpl::class, + \Temporal\Samples\Nexus\Caller\HelloCallerWorkflowImpl::class, + HelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new SampleNexusServiceImpl()); + +// Scenario 2: production service-impl with the EchoClient dependency mocked, +// and a stand-in handler workflow class that returns canned data. Demonstrates +// the DI-mock + workflow-replacement test style. +$workerFactory->newWorker(taskQueue: CallerWorkflowMockTest::TASK_QUEUE) + ->registerWorkflowTypes( + \Temporal\Samples\Nexus\Caller\EchoCallerWorkflowImpl::class, + \Temporal\Samples\Nexus\Caller\HelloCallerWorkflowImpl::class, + MockHelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new SampleNexusServiceImpl(new MockEchoClient())); + +// Scenario 3: the entire Nexus service implementation is replaced. Useful +// when the production impl isn't reachable from the test (e.g. it lives in +// another package). The mock service still routes async ops through a real +// workflow start so the wire-level state machine is exercised end-to-end. +$workerFactory->newWorker(taskQueue: NexusServiceMockTest::TASK_QUEUE) + ->registerWorkflowTypes( + \Temporal\Samples\Nexus\Caller\EchoCallerWorkflowImpl::class, + \Temporal\Samples\Nexus\Caller\HelloCallerWorkflowImpl::class, + MockHelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new MockSampleNexusServiceImpl()); + +// Scenario 4: the NexusCancellation sample β€” fan-out callers + a handler +// workflow that sleeps and honours cancellation. +$workerFactory->newWorker(taskQueue: CancellationTest::TASK_QUEUE) + ->registerWorkflowTypes( + \Temporal\Samples\NexusCancellation\Caller\HelloCallerWorkflowImpl::class, + \Temporal\Samples\NexusCancellation\Handler\HelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new \Temporal\Samples\Nexus\Handler\SampleNexusServiceImpl()); + +// Scenario 5: the NexusContextPropagation sample β€” the production outbound +// interceptor on the caller side, a header-echoing service double on the +// handler side so the test can observe the propagated value. +$workerFactory->newWorker( + taskQueue: ContextPropagationTest::TASK_QUEUE, + interceptorProvider: new SimplePipelineProvider([ + new NexusOutboundContextInterceptor(), + new WorkflowInboundContextInterceptor(), + ]), +) + ->registerWorkflowTypes( + \Temporal\Samples\NexusContextPropagation\Caller\EchoCallerWorkflowImpl::class, + \Temporal\Samples\NexusContextPropagation\Caller\HelloCallerWorkflowImpl::class, + \Temporal\Samples\NexusContextPropagation\Handler\HelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new HeaderEchoNexusServiceImpl()); + +// Scenario 6: the NexusMultipleArguments sample β€” single-DTO contract backed +// by a multi-argument handler workflow. +$workerFactory->newWorker(taskQueue: MultipleArgumentsTest::TASK_QUEUE) + ->registerWorkflowTypes( + \Temporal\Samples\NexusMultipleArguments\Caller\HelloCallerWorkflowImpl::class, + \Temporal\Samples\NexusMultipleArguments\Handler\HelloHandlerWorkflowImpl::class, + ) + ->registerNexusServiceImplementation(new \Temporal\Samples\NexusMultipleArguments\Handler\SampleNexusServiceImpl()); + +// Scenario 7: the NexusManualOperation sample β€” a manual handler object that +// owns start (sync fast-path or async with its own token) and cancel. +$workerFactory->newWorker(taskQueue: ManualOperationTest::TASK_QUEUE) + ->registerWorkflowTypes(\Temporal\Samples\NexusManualOperation\Caller\JobCallerWorkflowImpl::class) + ->registerNexusServiceImplementation(new \Temporal\Samples\NexusManualOperation\Handler\SampleNexusService()); + +// Scenario 8: same manual-operation caller, but the external job backend +// rejects the async submit so the caller must surface the start failure. +$workerFactory->newWorker(taskQueue: ManualOperationStartFailureTest::TASK_QUEUE) + ->registerWorkflowTypes(\Temporal\Samples\NexusManualOperation\Caller\JobCallerWorkflowImpl::class) + ->registerNexusServiceImplementation( + new \Temporal\Samples\NexusManualOperation\Handler\SampleNexusService(new FailingJobClient()), + ); $workerFactory->run(); diff --git a/app/tests/bootstrap.php b/app/tests/bootstrap.php index 46a57c1..5a185b7 100644 --- a/app/tests/bootstrap.php +++ b/app/tests/bootstrap.php @@ -17,7 +17,7 @@ # Check --filter parameter if (\preg_match('/--filter(?:=|\s++)([^"\']\S++|\'[^\']*+\'|"[^\']*+")/', $string, $matches)) { $filter = str_replace('\\\\', '\\', \trim($matches[1], '\'"')); - if (\preg_match('/Temporal\\\\Tests\\\\(\\w+)\\\\/', $filter, $matches)) { + if (\preg_match('/(?:Temporal|App)\\\\Tests\\\\(\\w+)\\\\/', $filter, $matches)) { return $matches[1]; } }