From d247b963215d91010f8d3f264f5f393b28fd9388 Mon Sep 17 00:00:00 2001 From: Roey Berman Date: Fri, 28 Feb 2025 14:55:35 -0800 Subject: [PATCH] Nexus SyncOperation context propagation --- nexus-context-propagation/README.md | 6 ++++-- .../caller/starter/main.go | 16 +++++++++++----- .../caller/worker/main.go | 1 + nexus-context-propagation/handler/app.go | 18 ++++++++++++++++++ .../handler/worker/main.go | 2 +- 5 files changed, 35 insertions(+), 8 deletions(-) diff --git a/nexus-context-propagation/README.md b/nexus-context-propagation/README.md index 7c036327..02ba59d3 100644 --- a/nexus-context-propagation/README.md +++ b/nexus-context-propagation/README.md @@ -39,6 +39,8 @@ go run ./starter \ which should result in: ``` -2025/02/27 12:57:40 Started workflow WorkflowID nexus_hello_caller_workflow_20240723195740 RunID c9789128-2fcd-4083-829d-95e43279f6d7 -2025/02/27 12:57:40 Workflow result: ¡Hola! Nexus, caller-id: samples-go 👋 +2025/02/28 14:54:29 Started workflow WorkflowID nexus_hello_caller_workflow_20250228145225 RunID 01954ec2-d1d2-72b7-a6cc-a83de5421754 +2025/02/28 14:54:30 Workflow result: Nexus Echo 👋, caller-id: samples-go +2025/02/28 14:54:30 Started workflow WorkflowID nexus_hello_caller_workflow_20250228145430 RunID 01954ec4-bc4d-7dac-a311-d4ceacacfa9a +2025/02/28 14:54:31 Workflow result: ¡Hola! Nexus, caller-id: samples-go 👋 ``` diff --git a/nexus-context-propagation/caller/starter/main.go b/nexus-context-propagation/caller/starter/main.go index 61445191..9f8150c4 100644 --- a/nexus-context-propagation/caller/starter/main.go +++ b/nexus-context-propagation/caller/starter/main.go @@ -27,16 +27,22 @@ func main() { } defer c.Close() ctx := context.Background() - workflowOptions := client.StartWorkflowOptions{ - ID: "nexus_hello_caller_workflow_" + time.Now().Format("20060102150405"), - TaskQueue: caller.TaskQueue, - } ctx = context.WithValue(ctx, ctxpropagation.PropagateKey, ctxpropagation.Values{ Key: "caller-id", Value: "samples-go", }) - wr, err := c.ExecuteWorkflow(ctx, workflowOptions, caller.HelloCallerWorkflow, "Nexus", service.ES) + + runWorkflow(ctx, c, caller.EchoCallerWorkflow, "Nexus Echo 👋") + runWorkflow(ctx, c, caller.HelloCallerWorkflow, "Nexus", service.ES) +} + +func runWorkflow(ctx context.Context, c client.Client, workflow interface{}, args ...interface{}) { + workflowOptions := client.StartWorkflowOptions{ + ID: "nexus_hello_caller_workflow_" + time.Now().Format("20060102150405"), + TaskQueue: caller.TaskQueue, + } + wr, err := c.ExecuteWorkflow(ctx, workflowOptions, workflow, args...) if err != nil { log.Fatalln("Unable to execute workflow", err) } diff --git a/nexus-context-propagation/caller/worker/main.go b/nexus-context-propagation/caller/worker/main.go index 9c632f56..c4415761 100644 --- a/nexus-context-propagation/caller/worker/main.go +++ b/nexus-context-propagation/caller/worker/main.go @@ -43,6 +43,7 @@ func main() { }) w.RegisterWorkflow(caller.HelloCallerWorkflow) + w.RegisterWorkflow(caller.EchoCallerWorkflow) err = w.Run(worker.InterruptCh()) if err != nil { diff --git a/nexus-context-propagation/handler/app.go b/nexus-context-propagation/handler/app.go index af44b515..7bd45cb0 100644 --- a/nexus-context-propagation/handler/app.go +++ b/nexus-context-propagation/handler/app.go @@ -14,9 +14,26 @@ import ( "github.com/temporalio/samples-go/nexus/service" ) +// NewSyncOperation is a meant for exposing simple RPC handlers. +var EchoOperation = nexus.NewSyncOperation(service.EchoOperationName, func(ctx context.Context, input service.EchoInput, options nexus.StartOperationOptions) (service.EchoOutput, error) { + // Values may be extracted from the context in the Operation handler body. + values, ok := ctx.Value(ctxpropagation.PropagateKey).(ctxpropagation.Values) + if ok { + input.Message += ", " + values.Key + ": " + values.Value + } + // Use temporalnexus.GetClient to get the client that the worker was initialized with to perform client calls + // such as signaling, querying, and listing workflows. Implementations are free to make arbitrary calls to other + // services or databases, or perform simple computations such as this one. + return service.EchoOutput(input), nil +}) + // Use the NewWorkflowRunOperation constructor, which is the easiest way to expose a workflow as an operation. // See alternatives at https://pkg.go.dev/go.temporal.io/sdk/temporalnexus. var HelloOperation = temporalnexus.NewWorkflowRunOperation(service.HelloOperationName, HelloHandlerWorkflow, func(ctx context.Context, input service.HelloInput, options nexus.StartOperationOptions) (client.StartWorkflowOptions, error) { + // Values may be extracted from the context in the Operation handler body if necessary, this sample propagates + // the context to the handler workflow. + // values, ok := ctx.Value(ctxpropagation.PropagateKey).(ctxpropagation.Values) + return client.StartWorkflowOptions{ // Workflow IDs should typically be business meaningful IDs and are used to dedupe workflow starts. // For this example, we're using the request ID allocated by Temporal when the caller workflow schedules @@ -27,6 +44,7 @@ var HelloOperation = temporalnexus.NewWorkflowRunOperation(service.HelloOperatio }) func HelloHandlerWorkflow(ctx workflow.Context, input service.HelloInput) (service.HelloOutput, error) { + // Values may be extracted from the handler workflow context. values, ok := ctx.Value(ctxpropagation.PropagateKey).(ctxpropagation.Values) if ok { input.Name += ", " + values.Key + ": " + values.Value diff --git a/nexus-context-propagation/handler/worker/main.go b/nexus-context-propagation/handler/worker/main.go index 40b0fdfb..dc55d1a4 100644 --- a/nexus-context-propagation/handler/worker/main.go +++ b/nexus-context-propagation/handler/worker/main.go @@ -47,7 +47,7 @@ func main() { }, }) service := nexus.NewService(service.HelloServiceName) - err = service.Register(handler.HelloOperation) + err = service.Register(handler.HelloOperation, handler.EchoOperation) if err != nil { log.Fatalln("Unable to register operations", err) }