diff --git a/README.md b/README.md index e3c7ce1..bb5ec16 100644 --- a/README.md +++ b/README.md @@ -30,9 +30,11 @@ Prerequisites: * [NexusCancellation](src/NexusCancellation) - Demonstrates how to cancel a running Nexus operation from a caller workflow. * [NexusContextPropagation](src/NexusContextPropagation) - Context propagation through Nexus services. * [NexusDependencyInjection](src/NexusDependencyInjection) - How to configure a Nexus service handler for dependency injection using a generic host worker. -* [Nexus Messaging](src/NexusMessaging): Demonstrates how send signal, update and query messages through Nexus. +* [Nexus Messaging](src/NexusMessaging): Demonstrates how to send signal, update and query messages through Nexus. This contains two samples, one sending messages to an existing workflow and a second that creates a workflow through Nexus and sends messages to it. +* [Nexus Messaging Temporal Operation](src/NexusMessagingTemporalOperation): Demonstrates how to send signal, update and query messages through Nexus. + This version uses the `[TemporalOperation]` attribute to declare a Temporal-backed Nexus operation start handler directly on a method within a `[NexusServiceHandler]` class. * [NexusMultiArg](src/NexusMultiArg) - Nexus service implementation calling a workflow with multiple arguments. * [NexusSimple](src/NexusSimple) - Simple Nexus service implementation. * [NexusStandaloneActivity](src/NexusStandaloneActivity) - Nexus Operation backed by a Standalone Activity. diff --git a/TemporalioSamples.slnx b/TemporalioSamples.slnx index 92d2dd4..9d5ea2a 100644 --- a/TemporalioSamples.slnx +++ b/TemporalioSamples.slnx @@ -11,7 +11,6 @@ - @@ -60,6 +59,12 @@ + + + + + + diff --git a/src/NexusMessaging/CallerPattern/Handler/NexusGreetingService.cs b/src/NexusMessaging/CallerPattern/Handler/NexusGreetingService.cs index d0038ac..4c7aa4e 100644 --- a/src/NexusMessaging/CallerPattern/Handler/NexusGreetingService.cs +++ b/src/NexusMessaging/CallerPattern/Handler/NexusGreetingService.cs @@ -1,7 +1,6 @@ namespace TemporalioSamples.NexusMessaging.CallerPattern.Handler; using NexusRpc.Handlers; -using Temporalio.Client; using Temporalio.Nexus; using TemporalioSamples.NexusMessaging.CallerPattern; using TemporalioSamples.NexusMessaging.Common; @@ -12,63 +11,55 @@ namespace TemporalioSamples.NexusMessaging.CallerPattern.Handler; [NexusServiceHandler(typeof(INexusGreetingService))] public class NexusGreetingService { -#pragma warning disable VSTHRD200 // Names must match the INexusGreetingService operations, which can't take the Async suffix + // OperationHandler.Sync means the result is returned inline to the Nexus caller + // (as opposed to WorkflowRunOperationHandler, which returns an async operation token). + // The lambda may still be async internally. // Query: read-only, no state mutation — uses workflow query - [TemporalOperation] - public async Task> GetLanguages( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusGreetingService.GetLanguagesInput input) - { - // Access the Temporal client from the Nexus client passed to the handler - var handle = client.TemporalClient.GetWorkflowHandle( - WorkflowIdForUser(input.UserId)); - return TemporalOperationResult.SyncResult( - await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported))); - } + [NexusOperationHandler] + public IOperationHandler GetLanguages() => + OperationHandler.Sync( + async (ctx, input) => + { + // Access the Temporal client from the Nexus operation context + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(WorkflowIdForUser(input.UserId)); + return await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported)); + }); // Query: read-only — returns the workflow's current language - [TemporalOperation] - public async Task> GetLanguage( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusGreetingService.GetLanguageInput input) - { - var handle = client.TemporalClient.GetWorkflowHandle( - WorkflowIdForUser(input.UserId)); - return TemporalOperationResult.SyncResult( - await handle.QueryAsync(wf => wf.QueryLanguage())); - } + [NexusOperationHandler] + public IOperationHandler GetLanguage() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(WorkflowIdForUser(input.UserId)); + return await handle.QueryAsync(wf => wf.QueryLanguage()); + }); // Update: mutates state and returns the previous value — uses workflow update - [TemporalOperation] - public Task> SetLanguage( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusGreetingService.SetLanguageInput input) => - client.StartWorkflowUpdateAsync( - WorkflowIdForUser(input.UserId), - wf => wf.SetLanguageAsync(input.Language), - // An Update-backed Operation must wait for the Accepted stage. Any other stage is - // rejected with "nexus op workflow updates only support WorkflowUpdateStageAccepted - // for async updates". - new(WorkflowUpdateStage.Accepted)); + [NexusOperationHandler] + public IOperationHandler SetLanguage() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(WorkflowIdForUser(input.UserId)); + return await handle.ExecuteUpdateAsync(wf => wf.SetLanguageAsync(input.Language)); + }); // Signal: fire-and-forget, no return value needed — uses workflow signal - [TemporalOperation] - public async Task> Approve( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusGreetingService.ApproveInput input) - { - var handle = client.TemporalClient.GetWorkflowHandle( - WorkflowIdForUser(input.UserId)); - await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); - return TemporalOperationResult.SyncResult(default); - } - -#pragma warning restore VSTHRD200 + [NexusOperationHandler] + public IOperationHandler Approve() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(WorkflowIdForUser(input.UserId)); + await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); + return default; + }); private static string WorkflowIdForUser(string userId) => $"GreetingWorkflow_for_{userId}"; } diff --git a/src/NexusMessaging/CallerPattern/README.md b/src/NexusMessaging/CallerPattern/README.md index 54ff2a1..083e8ab 100644 --- a/src/NexusMessaging/CallerPattern/README.md +++ b/src/NexusMessaging/CallerPattern/README.md @@ -14,24 +14,19 @@ The caller Workflow: ### Running -This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download -the compatible binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). - -Start the Temporal dev server with the required namespaces pre-created and Workflow Update -callbacks enabled: +Start a Temporal server: ```bash -./temporal server start-dev \ - --dynamic-config-value history.enableUpdateCallbacks=true \ - --dynamic-config-value history.enableCHASMSignalBacklinks=true \ - --namespace nexus-messaging-handler-namespace \ - --namespace nexus-messaging-caller-namespace +temporal server start-dev ``` -Create the Nexus endpoint: +Create the namespaces and Nexus endpoint: ```bash -./temporal operator nexus endpoint create \ +temporal operator namespace create --namespace nexus-messaging-handler-namespace +temporal operator namespace create --namespace nexus-messaging-caller-namespace + +temporal operator nexus endpoint create \ --name nexus-messaging-caller-pattern-endpoint \ --target-namespace nexus-messaging-handler-namespace \ --target-task-queue nexus-messaging-handler-sample @@ -54,12 +49,3 @@ In a third terminal, run the following command to start the example: ```bash dotnet run --project src/NexusMessaging -- caller-workflow ``` - -Expected output: - -``` -Supported languages: Chinese, English -Current language: English -Set language from English to Chinese -Approved workflow -``` diff --git a/src/NexusMessaging/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs b/src/NexusMessaging/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs index 4e97f6e..39d8cd7 100644 --- a/src/NexusMessaging/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs +++ b/src/NexusMessaging/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs @@ -18,17 +18,6 @@ public async Task RunAsync() var userIdOne = "user-one"; var userIdTwo = "user-two"; - // Attach information before the Workflow exists. Since AttachApprovalContext is backed - // by Signal-with-Start on the handler, this call creates the Workflow and delivers the - // note to it. - await client.ExecuteNexusOperationAsync( - svc => svc.AttachApprovalContext(new INexusRemoteGreetingService.AttachApprovalContextInput( - "queued for localization review by the nightly batch", userIdOne))); - log.Add($"Attached approval context before the workflow existed: {userIdOne}"); - - // The Workflow for this user is already running due to AttachApprovalContext. The handler - // sets the conflict policy to UseExisting, so this call attaches the Operation's - // completion callback to the running execution. var handleOne = await client.StartNexusOperationAsync( svc => svc.RunFromRemote(new INexusRemoteGreetingService.RunFromRemoteInput(userIdOne))); log.Add($"Started remote workflow for user: {userIdOne}"); @@ -37,13 +26,6 @@ await client.ExecuteNexusOperationAsync( svc => svc.RunFromRemote(new INexusRemoteGreetingService.RunFromRemoteInput(userIdTwo))); log.Add($"Started remote workflow for user: {userIdTwo}"); - // This user's Workflow was created by RunFromRemote just above, so here Signal-with-Start - // skips the start and only delivers the Signal. - await client.ExecuteNexusOperationAsync( - svc => svc.AttachApprovalContext(new INexusRemoteGreetingService.AttachApprovalContextInput( - "translation approved by the localization team", userIdTwo))); - log.Add($"Attached approval context to the running workflow: {userIdTwo}"); - // Interact with workflow one: get languages, set language, approve var languagesOne = await client.ExecuteNexusOperationAsync( svc => svc.GetLanguages(new INexusRemoteGreetingService.GetLanguagesInput(false, userIdOne))); diff --git a/src/NexusMessaging/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs b/src/NexusMessaging/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs index e9c9c1d..f3042d7 100644 --- a/src/NexusMessaging/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs +++ b/src/NexusMessaging/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs @@ -17,7 +17,6 @@ public class GreetingWorkflow private Language currentLanguage = Language.English; private bool approved; private string approvedBy = string.Empty; - private string? approvalContext; [WorkflowRun] public async Task RunAsync(string userId) @@ -80,11 +79,4 @@ public Task ApproveAsync(string name) approvedBy = name; return Task.CompletedTask; } - - [WorkflowSignal] - public Task AttachApprovalContextAsync(string note) - { - approvalContext = note; - return Task.CompletedTask; - } } diff --git a/src/NexusMessaging/OnDemandPattern/Handler/NexusRemoteGreetingService.cs b/src/NexusMessaging/OnDemandPattern/Handler/NexusRemoteGreetingService.cs index f27e111..52b5c46 100644 --- a/src/NexusMessaging/OnDemandPattern/Handler/NexusRemoteGreetingService.cs +++ b/src/NexusMessaging/OnDemandPattern/Handler/NexusRemoteGreetingService.cs @@ -1,8 +1,6 @@ namespace TemporalioSamples.NexusMessaging.OnDemandPattern.Handler; using NexusRpc.Handlers; -using Temporalio.Api.Enums.V1; -using Temporalio.Client; using Temporalio.Nexus; using TemporalioSamples.NexusMessaging.Common; using TemporalioSamples.NexusMessaging.OnDemandPattern; @@ -13,100 +11,66 @@ namespace TemporalioSamples.NexusMessaging.OnDemandPattern.Handler; [NexusServiceHandler(typeof(INexusRemoteGreetingService))] public class NexusRemoteGreetingService { -#pragma warning disable VSTHRD200 // Names must match the INexusRemoteGreetingService operations, which can't take the Async suffix + // WorkflowRunOperationHandler starts a backing workflow and returns its handle to the + // Nexus infrastructure. The caller receives an async operation token and can poll for + // the workflow result later via GetResultAsync. + [NexusOperationHandler] + public IOperationHandler RunFromRemote() => + WorkflowRunOperationHandler.FromHandleFactory( + (WorkflowRunOperationContext context, INexusRemoteGreetingService.RunFromRemoteInput input) => + context.StartWorkflowAsync( + (GreetingWorkflow wf) => wf.RunAsync(input.UserId), + new() { Id = GetWorkflowId(input.UserId) })); - // Starts the GreetingWorkflow for the given user, or attaches to one already running. - // StartWorkflowAsync returns an async result, which attaches a completion callback, - // so the Operation completes when the Workflow returns. - [TemporalOperation] - public Task> RunFromRemote( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.RunFromRemoteInput input) => - client.StartWorkflowAsync( - (GreetingWorkflow wf) => wf.RunAsync(input.UserId), - new() - { - Id = GetWorkflowId(input.UserId), - - // By default, starting a Workflow whose ID is already running fails the Operation. - // Since AttachApprovalContext below can create the GreetingWorkflow first, this - // Operation needs to attach to the running execution rather than fail. - IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, - }); + // OperationHandler.Sync means the result is returned inline to the Nexus caller + // (as opposed to WorkflowRunOperationHandler, which returns an async operation token). + // The lambda may still be async internally. // Query: read-only, no state mutation — uses workflow query - [TemporalOperation] - public async Task> GetLanguages( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.GetLanguagesInput input) - { - // Access the Temporal client from the Nexus client passed to the handler - var handle = client.TemporalClient.GetWorkflowHandle( - GetWorkflowId(input.UserId)); - return TemporalOperationResult.SyncResult( - await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported))); - } + [NexusOperationHandler] + public IOperationHandler GetLanguages() => + OperationHandler.Sync( + async (ctx, input) => + { + // Access the Temporal client from the Nexus operation context + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(GetWorkflowId(input.UserId)); + return await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported)); + }); // Query: read-only — returns the workflow's current language - [TemporalOperation] - public async Task> GetLanguage( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.GetLanguageInput input) - { - var handle = client.TemporalClient.GetWorkflowHandle( - GetWorkflowId(input.UserId)); - return TemporalOperationResult.SyncResult( - await handle.QueryAsync(wf => wf.QueryLanguage())); - } + [NexusOperationHandler] + public IOperationHandler GetLanguage() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(GetWorkflowId(input.UserId)); + return await handle.QueryAsync(wf => wf.QueryLanguage()); + }); // Update: mutates state and returns the previous value — uses workflow update - [TemporalOperation] - public Task> SetLanguage( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.SetLanguageInput input) => - client.StartWorkflowUpdateAsync( - GetWorkflowId(input.UserId), - wf => wf.SetLanguageAsync(input.Language), - // An Update-backed Operation must wait for the Accepted stage. Any other stage is - // rejected with "nexus op workflow updates only support WorkflowUpdateStageAccepted - // for async updates". - new(WorkflowUpdateStage.Accepted)); + [NexusOperationHandler] + public IOperationHandler SetLanguage() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(GetWorkflowId(input.UserId)); + return await handle.ExecuteUpdateAsync(wf => wf.SetLanguageAsync(input.Language)); + }); // Signal: fire-and-forget, no return value needed — uses workflow signal - [TemporalOperation] - public async Task> Approve( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.ApproveInput input) - { - var handle = client.TemporalClient.GetWorkflowHandle( - GetWorkflowId(input.UserId)); - await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); - return TemporalOperationResult.SyncResult(default); - } - - // Signal-with-Start: starts the Workflow first if it is not already running. - // When the Workflow already exists, only the Signal is delivered. - [TemporalOperation] - public async Task> AttachApprovalContext( - TemporalOperationStartContext ctx, - ITemporalNexusClient client, - INexusRemoteGreetingService.AttachApprovalContextInput input) - { - var options = new WorkflowOptions( - id: GetWorkflowId(input.UserId), - taskQueue: NexusOperationExecutionContext.Current.Info.TaskQueue); - options.SignalWithStart((GreetingWorkflow wf) => wf.AttachApprovalContextAsync(input.Note)); - await client.TemporalClient.StartWorkflowAsync( - (GreetingWorkflow wf) => wf.RunAsync(input.UserId), options); - return TemporalOperationResult.SyncResult(default); - } - -#pragma warning restore VSTHRD200 + [NexusOperationHandler] + public IOperationHandler Approve() => + OperationHandler.Sync( + async (ctx, input) => + { + var client = NexusOperationExecutionContext.Current.TemporalClient; + var handle = client.GetWorkflowHandle(GetWorkflowId(input.UserId)); + await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); + return default; + }); private static string GetWorkflowId(string userId) => $"GreetingWorkflow_for_{userId}"; } diff --git a/src/NexusMessaging/OnDemandPattern/INexusRemoteGreetingService.cs b/src/NexusMessaging/OnDemandPattern/INexusRemoteGreetingService.cs index 5e3a6cf..d0b8121 100644 --- a/src/NexusMessaging/OnDemandPattern/INexusRemoteGreetingService.cs +++ b/src/NexusMessaging/OnDemandPattern/INexusRemoteGreetingService.cs @@ -21,9 +21,6 @@ public interface INexusRemoteGreetingService [NexusOperation] void Approve(ApproveInput input); - [NexusOperation] - void AttachApprovalContext(AttachApprovalContextInput input); - public record RunFromRemoteInput(string UserId); public record GetLanguagesInput(bool IncludeUnsupported, string UserId); @@ -35,6 +32,4 @@ public record GetLanguageInput(string UserId); public record SetLanguageInput(Language Language, string UserId); public record ApproveInput(string Name, string UserId); - - public record AttachApprovalContextInput(string Note, string UserId); } diff --git a/src/NexusMessaging/OnDemandPattern/README.md b/src/NexusMessaging/OnDemandPattern/README.md index 0643fb9..ef99357 100644 --- a/src/NexusMessaging/OnDemandPattern/README.md +++ b/src/NexusMessaging/OnDemandPattern/README.md @@ -1,41 +1,31 @@ ## On-demand pattern No Workflow is pre-started. The caller creates and controls Workflow instances through Nexus -operations. `NexusRemoteGreetingService` adds a `RunFromRemote` operation that starts a -`GreetingWorkflow`, and every operation includes a `UserId` so the handler can derive +operations. `NexusRemoteGreetingService` adds a `RunFromRemote` operation that starts a new +`GreetingWorkflow`, and every other operation includes a `UserId` so the handler can derive the target Workflow ID. The caller Workflow: -1. Attaches approval context for the first user via `AttachApprovalContext`, before anything has - started that user's Workflow -2. Starts two remote `GreetingWorkflow` instances via `RunFromRemote` (backed by a Workflow started - through `ITemporalNexusClient.StartWorkflowAsync`) -3. Attaches approval context for the second user, whose Workflow now already exists -4. Workflow one: queries supported languages, changes to Spanish, and approves -5. Workflow two: queries the current language, changes to French, and approves -6. Waits for each to complete and returns their results +1. Starts two remote `GreetingWorkflow` instances via `RunFromRemote` (backed by `WorkflowRunOperationHandler`) +2. Workflow one: queries supported languages, changes to Spanish, and approves +3. Workflow two: queries the current language, changes to French, and approves +4. Waits for each to complete and returns their results ### Running -This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download -the compatible binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). - -Start the Temporal dev server with the required namespaces pre-created, and Workflow Update -callbacks and signal-with-start from a Workflow enabled: +Start a Temporal server: ```bash -./temporal server start-dev \ - --dynamic-config-value history.enableUpdateCallbacks=true \ - --dynamic-config-value history.enableCHASMSignalBacklinks=true \ - --dynamic-config-value history.enableSignalWithStartFromWorkflow=true \ - --namespace nexus-messaging-handler-namespace \ - --namespace nexus-messaging-caller-namespace +temporal server start-dev ``` -Create the Nexus endpoint: +Create the namespaces and Nexus endpoint: ```bash -./temporal operator nexus endpoint create \ +temporal operator namespace create --namespace nexus-messaging-handler-namespace +temporal operator namespace create --namespace nexus-messaging-caller-namespace + +temporal operator nexus endpoint create \ --name nexus-messaging-on-demand-pattern-endpoint \ --target-namespace nexus-messaging-handler-namespace \ --target-task-queue nexus-messaging-handler-sample @@ -58,20 +48,3 @@ In a third terminal, run the following command to start the example: ```bash dotnet run --project src/NexusMessaging -- remote-caller-workflow ``` - -Expected output: - -``` -Attached approval context before the workflow existed: user-one -Started remote workflow for user: user-one -Started remote workflow for user: user-two -Attached approval context to the running workflow: user-two -[One] Supported languages: Chinese, English -[One] Set language from English to Spanish -[One] Approved -[Two] Current language: English -[Two] Set language from English to French -[Two] Approved -[One] Result: Hola, mundo (approved by CallerRemoteWorkflow) -[Two] Result: Bonjour, monde (approved by CallerRemoteWorkflow) -``` diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/Caller/CallerWorkflow.workflow.cs b/src/NexusMessagingTemporalOperation/CallerPattern/Caller/CallerWorkflow.workflow.cs new file mode 100644 index 0000000..e37c179 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/Caller/CallerWorkflow.workflow.cs @@ -0,0 +1,39 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Caller; + +using Temporalio.Workflows; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; + +[Workflow] +public class CallerWorkflow +{ + [WorkflowRun] + public async Task RunAsync(string userId) + { + var log = new List(); + var client = Workflow.CreateNexusWorkflowClient( + NexusEndpoints.GreetingService); + + // GetLanguages - query entity workflow for supported languages + var languagesOutput = await client.ExecuteNexusOperationAsync( + svc => svc.GetLanguages(new INexusGreetingService.GetLanguagesInput(false, userId))); + log.Add($"Supported languages: {string.Join(", ", languagesOutput.Languages)}"); + + // GetLanguage - query entity workflow for current language + var currentLanguage = await client.ExecuteNexusOperationAsync( + svc => svc.GetLanguage(new INexusGreetingService.GetLanguageInput(userId))); + log.Add($"Current language: {currentLanguage}"); + + // SetLanguage - update entity workflow to change language + var previousLanguage = await client.ExecuteNexusOperationAsync( + svc => svc.SetLanguage(new INexusGreetingService.SetLanguageInput(Language.Chinese, userId))); + log.Add($"Set language from {previousLanguage} to {Language.Chinese}"); + + // Approve - signal entity workflow to complete + await client.ExecuteNexusOperationAsync( + svc => svc.Approve(new INexusGreetingService.ApproveInput("CallerWorkflow", userId))); + log.Add("Approved workflow"); + + return log.ToArray(); + } +} diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/Caller/NexusEndpoints.cs b/src/NexusMessagingTemporalOperation/CallerPattern/Caller/NexusEndpoints.cs new file mode 100644 index 0000000..c8a3b13 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/Caller/NexusEndpoints.cs @@ -0,0 +1,6 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Caller; + +public static class NexusEndpoints +{ + public const string GreetingService = "nexus-messaging-caller-pattern-endpoint"; +} diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/Handler/GreetingWorkflow.workflow.cs b/src/NexusMessagingTemporalOperation/CallerPattern/Handler/GreetingWorkflow.workflow.cs new file mode 100644 index 0000000..01db39c --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/Handler/GreetingWorkflow.workflow.cs @@ -0,0 +1,82 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Handler; + +using Temporalio.Exceptions; +using Temporalio.Workflows; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; + +[Workflow] +public class GreetingWorkflow +{ + private readonly Dictionary greetings = new() + { + [Language.Chinese] = "你好，世界", + [Language.English] = "Hello, world", + }; + + private Language currentLanguage = Language.English; + private bool approved; + private string approvedBy = string.Empty; + + [WorkflowRun] + public async Task RunAsync(string userId) + { + // Wait for approve signal and all handlers to finish + await Workflow.WaitConditionAsync(() => approved && Workflow.AllHandlersFinished); + + var greeting = greetings.TryGetValue(currentLanguage, out var g) ? g : "Hello, world"; + return $"{greeting} (approved by {approvedBy})"; + } + + [WorkflowQuery] + public INexusGreetingService.GetLanguagesOutput QueryLanguages(bool includeUnsupported) + { + if (includeUnsupported) + { + return new(Enum.GetValues()); + } + + return new(greetings.Keys.ToArray()); + } + + [WorkflowQuery] + public Language QueryLanguage() => currentLanguage; + + [WorkflowUpdate] + public async Task SetLanguageAsync(Language newLanguage) + { + var prev = currentLanguage; + + // If language not yet in our greetings map, fetch from activity + if (!greetings.ContainsKey(newLanguage)) + { + var allGreetings = await Workflow.ExecuteActivityAsync( + (GreetingActivities a) => a.GetAllGreetingsAsync(), + new() { StartToCloseTimeout = TimeSpan.FromSeconds(10) }); + foreach (var kvp in allGreetings) + { + greetings[kvp.Key] = kvp.Value; + } + } + + currentLanguage = newLanguage; + return prev; + } + + [WorkflowUpdateValidator(nameof(SetLanguageAsync))] + public void ValidateSetLanguage(Language newLanguage) + { + if (!Enum.IsDefined(newLanguage)) + { + throw new ApplicationFailureException($"Unsupported language: {newLanguage}"); + } + } + + [WorkflowSignal] + public Task ApproveAsync(string name) + { + approved = true; + approvedBy = name; + return Task.CompletedTask; + } +} diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/Handler/NexusGreetingService.cs b/src/NexusMessagingTemporalOperation/CallerPattern/Handler/NexusGreetingService.cs new file mode 100644 index 0000000..504d41c --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/Handler/NexusGreetingService.cs @@ -0,0 +1,74 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Handler; + +using NexusRpc.Handlers; +using Temporalio.Client; +using Temporalio.Nexus; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; + +// Entity pattern: the handler worker pre-starts a GreetingWorkflow per user at boot time. +// This service routes each Nexus operation to that existing workflow by deriving the +// workflow ID from the caller-supplied UserId. +[NexusServiceHandler(typeof(INexusGreetingService))] +public class NexusGreetingService +{ +#pragma warning disable VSTHRD200 // Names must match the INexusGreetingService operations, which can't take the Async suffix + + // Query: read-only, no state mutation — uses workflow query + [TemporalOperation] + public async Task> GetLanguages( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusGreetingService.GetLanguagesInput input) + { + // Access the Temporal client from the Nexus client passed to the handler + var handle = client.TemporalClient.GetWorkflowHandle( + WorkflowIdForUser(input.UserId)); + return TemporalOperationResult.SyncResult( + await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported))); + } + + // Query: read-only — returns the workflow's current language + [TemporalOperation] + public async Task> GetLanguage( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusGreetingService.GetLanguageInput input) + { + var handle = client.TemporalClient.GetWorkflowHandle( + WorkflowIdForUser(input.UserId)); + return TemporalOperationResult.SyncResult( + await handle.QueryAsync(wf => wf.QueryLanguage())); + } + + // Update: mutates state and returns the previous value — uses workflow update + [TemporalOperation] + public Task> SetLanguage( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusGreetingService.SetLanguageInput input) => + client.StartWorkflowUpdateAsync( + WorkflowIdForUser(input.UserId), + wf => wf.SetLanguageAsync(input.Language), + // An Update-backed Operation must wait for the Accepted stage. Any other stage is + // rejected with "nexus op workflow updates only support WorkflowUpdateStageAccepted + // for async updates". + new(WorkflowUpdateStage.Accepted)); + + // Signal: fire-and-forget, no return value needed — uses workflow signal + [TemporalOperation] + public async Task> Approve( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusGreetingService.ApproveInput input) + { + var handle = client.TemporalClient.GetWorkflowHandle( + WorkflowIdForUser(input.UserId)); + await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); + return TemporalOperationResult.SyncResult(default); + } + +#pragma warning restore VSTHRD200 + + private static string WorkflowIdForUser(string userId) => $"GreetingWorkflow_for_{userId}"; +} diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/INexusGreetingService.cs b/src/NexusMessagingTemporalOperation/CallerPattern/INexusGreetingService.cs new file mode 100644 index 0000000..53db39b --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/INexusGreetingService.cs @@ -0,0 +1,30 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern; + +using NexusRpc; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; + +[NexusService] +public interface INexusGreetingService +{ + [NexusOperation] + GetLanguagesOutput GetLanguages(GetLanguagesInput input); + + [NexusOperation] + Language GetLanguage(GetLanguageInput input); + + [NexusOperation] + Language SetLanguage(SetLanguageInput input); + + [NexusOperation] + void Approve(ApproveInput input); + + public record GetLanguagesInput(bool IncludeUnsupported, string UserId); + + public record GetLanguagesOutput(IReadOnlyList Languages); + + public record GetLanguageInput(string UserId); + + public record SetLanguageInput(Language Language, string UserId); + + public record ApproveInput(string Name, string UserId); +} diff --git a/src/NexusMessagingTemporalOperation/CallerPattern/README.md b/src/NexusMessagingTemporalOperation/CallerPattern/README.md new file mode 100644 index 0000000..7549f18 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/CallerPattern/README.md @@ -0,0 +1,65 @@ +## Entity pattern + +The handler worker starts a `GreetingWorkflow` for a User ID at boot time. +`NexusGreetingService` routes every Nexus operation to that existing Workflow by deriving +the Workflow ID from the caller-supplied `UserId` (see the `WorkflowIdForUser` call). +The caller passes a `UserId`, not a Workflow ID -- the handler is responsible for the +ID mapping. + +The caller Workflow: +1. Queries for supported languages (`GetLanguages` -- backed by a Workflow query) +2. Queries the current language (`GetLanguage` -- backed by a Workflow query) +3. Changes the language to Chinese (`SetLanguage` -- backed by a Workflow update that calls an activity) +4. Approves the Workflow (`Approve` -- backed by a Workflow signal) + +### Running + +This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download +the compatible binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). + +Start the Temporal dev server with the required namespaces pre-created and Workflow Update +callbacks enabled: + +```bash +./temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true \ + --namespace nexus-messaging-handler-namespace \ + --namespace nexus-messaging-caller-namespace +``` + +Create the Nexus endpoint: + +```bash +./temporal operator nexus endpoint create \ + --name nexus-messaging-caller-pattern-endpoint \ + --target-namespace nexus-messaging-handler-namespace \ + --target-task-queue nexus-messaging-handler-sample +``` + +In one terminal, start the handler worker: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- handler-worker +``` + +In a second terminal, start the caller worker: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- caller-worker +``` + +In a third terminal, run the following command to start the example: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- caller-workflow +``` + +Expected output: + +``` +Supported languages: Chinese, English +Current language: English +Set language from English to Chinese +Approved workflow +``` diff --git a/src/NexusMessagingTemporalOperation/Common/GreetingActivities.cs b/src/NexusMessagingTemporalOperation/Common/GreetingActivities.cs new file mode 100644 index 0000000..d205ef1 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/Common/GreetingActivities.cs @@ -0,0 +1,19 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.Common; + +using Temporalio.Activities; + +public class GreetingActivities +{ + [Activity] + public Task> GetAllGreetingsAsync() => + Task.FromResult(new Dictionary + { + [Language.Arabic] = "مرحبا بالعالم", + [Language.Chinese] = "你好，世界", + [Language.English] = "Hello, world", + [Language.French] = "Bonjour, monde", + [Language.Hindi] = "नमस्ते दुनिया", + [Language.Portuguese] = "Olá, mundo", + [Language.Spanish] = "Hola, mundo", + }); +} diff --git a/src/NexusMessagingTemporalOperation/Common/Language.cs b/src/NexusMessagingTemporalOperation/Common/Language.cs new file mode 100644 index 0000000..6c8d1c4 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/Common/Language.cs @@ -0,0 +1,12 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.Common; + +public enum Language +{ + Arabic, + Chinese, + English, + French, + Hindi, + Portuguese, + Spanish, +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs b/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs new file mode 100644 index 0000000..9786758 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/CallerRemoteWorkflow.workflow.cs @@ -0,0 +1,82 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Caller; + +using Temporalio.Workflows; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern; + +[Workflow] +public class CallerRemoteWorkflow +{ + [WorkflowRun] + public async Task RunAsync() + { + var log = new List(); + var client = Workflow.CreateNexusWorkflowClient( + NexusEndpoints.RemoteGreetingService); + + // Start two remote greeting workflows on demand + var userIdOne = "user-one"; + var userIdTwo = "user-two"; + + // Attach information before the Workflow exists. Since AttachApprovalContext is backed + // by Signal-with-Start on the handler, this call creates the Workflow and delivers the + // note to it. + await client.ExecuteNexusOperationAsync( + svc => svc.AttachApprovalContext(new INexusRemoteGreetingService.AttachApprovalContextInput( + "queued for localization review by the nightly batch", userIdOne))); + log.Add($"Attached approval context before the workflow existed: {userIdOne}"); + + // The Workflow for this user is already running due to AttachApprovalContext. The handler + // sets the conflict policy to UseExisting, so this call attaches the Operation's + // completion callback to the running execution. + var handleOne = await client.StartNexusOperationAsync( + svc => svc.RunFromRemote(new INexusRemoteGreetingService.RunFromRemoteInput(userIdOne))); + log.Add($"Started remote workflow for user: {userIdOne}"); + + var handleTwo = await client.StartNexusOperationAsync( + svc => svc.RunFromRemote(new INexusRemoteGreetingService.RunFromRemoteInput(userIdTwo))); + log.Add($"Started remote workflow for user: {userIdTwo}"); + + // This user's Workflow was created by RunFromRemote just above, so here Signal-with-Start + // skips the start and only delivers the Signal. + await client.ExecuteNexusOperationAsync( + svc => svc.AttachApprovalContext(new INexusRemoteGreetingService.AttachApprovalContextInput( + "translation approved by the localization team", userIdTwo))); + log.Add($"Attached approval context to the running workflow: {userIdTwo}"); + + // Interact with workflow one: get languages, set language, approve + var languagesOne = await client.ExecuteNexusOperationAsync( + svc => svc.GetLanguages(new INexusRemoteGreetingService.GetLanguagesInput(false, userIdOne))); + log.Add($"[One] Supported languages: {string.Join(", ", languagesOne.Languages)}"); + + var prevLangOne = await client.ExecuteNexusOperationAsync( + svc => svc.SetLanguage(new INexusRemoteGreetingService.SetLanguageInput(Language.Spanish, userIdOne))); + log.Add($"[One] Set language from {prevLangOne} to {Language.Spanish}"); + + await client.ExecuteNexusOperationAsync( + svc => svc.Approve(new INexusRemoteGreetingService.ApproveInput("CallerRemoteWorkflow", userIdOne))); + log.Add("[One] Approved"); + + // Interact with workflow two: get language, set language, approve + var currentLangTwo = await client.ExecuteNexusOperationAsync( + svc => svc.GetLanguage(new INexusRemoteGreetingService.GetLanguageInput(userIdTwo))); + log.Add($"[Two] Current language: {currentLangTwo}"); + + var prevLangTwo = await client.ExecuteNexusOperationAsync( + svc => svc.SetLanguage(new INexusRemoteGreetingService.SetLanguageInput(Language.French, userIdTwo))); + log.Add($"[Two] Set language from {prevLangTwo} to {Language.French}"); + + await client.ExecuteNexusOperationAsync( + svc => svc.Approve(new INexusRemoteGreetingService.ApproveInput("CallerRemoteWorkflow", userIdTwo))); + log.Add("[Two] Approved"); + + // Wait for both remote workflows to complete + var resultOne = await handleOne.GetResultAsync(); + log.Add($"[One] Result: {resultOne}"); + + var resultTwo = await handleTwo.GetResultAsync(); + log.Add($"[Two] Result: {resultTwo}"); + + return log.ToArray(); + } +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/NexusEndpoints.cs b/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/NexusEndpoints.cs new file mode 100644 index 0000000..d98bfa5 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/Caller/NexusEndpoints.cs @@ -0,0 +1,6 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Caller; + +public static class NexusEndpoints +{ + public const string RemoteGreetingService = "nexus-messaging-on-demand-pattern-endpoint"; +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs b/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs new file mode 100644 index 0000000..45a51bd --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/GreetingWorkflow.workflow.cs @@ -0,0 +1,90 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Handler; + +using Temporalio.Exceptions; +using Temporalio.Workflows; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern; + +[Workflow] +public class GreetingWorkflow +{ + private readonly Dictionary greetings = new() + { + [Language.Chinese] = "你好，世界", + [Language.English] = "Hello, world", + }; + + private Language currentLanguage = Language.English; + private bool approved; + private string approvedBy = string.Empty; + private string? approvalContext; + + [WorkflowRun] + public async Task RunAsync(string userId) + { + // Wait for approve signal and all handlers to finish + await Workflow.WaitConditionAsync(() => approved && Workflow.AllHandlersFinished); + + var greeting = greetings.TryGetValue(currentLanguage, out var g) ? g : "Hello, world"; + return $"{greeting} (approved by {approvedBy})"; + } + + [WorkflowQuery] + public INexusRemoteGreetingService.GetLanguagesOutput QueryLanguages(bool includeUnsupported) + { + if (includeUnsupported) + { + return new(Enum.GetValues()); + } + + return new(greetings.Keys.ToArray()); + } + + [WorkflowQuery] + public Language QueryLanguage() => currentLanguage; + + [WorkflowUpdate] + public async Task SetLanguageAsync(Language newLanguage) + { + var prev = currentLanguage; + + // If language not yet in our greetings map, fetch from activity + if (!greetings.ContainsKey(newLanguage)) + { + var allGreetings = await Workflow.ExecuteActivityAsync( + (GreetingActivities a) => a.GetAllGreetingsAsync(), + new() { StartToCloseTimeout = TimeSpan.FromSeconds(10) }); + foreach (var kvp in allGreetings) + { + greetings[kvp.Key] = kvp.Value; + } + } + + currentLanguage = newLanguage; + return prev; + } + + [WorkflowUpdateValidator(nameof(SetLanguageAsync))] + public void ValidateSetLanguage(Language newLanguage) + { + if (!Enum.IsDefined(newLanguage)) + { + throw new ApplicationFailureException($"Unsupported language: {newLanguage}"); + } + } + + [WorkflowSignal] + public Task ApproveAsync(string name) + { + approved = true; + approvedBy = name; + return Task.CompletedTask; + } + + [WorkflowSignal] + public Task AttachApprovalContextAsync(string note) + { + approvalContext = note; + return Task.CompletedTask; + } +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/NexusRemoteGreetingService.cs b/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/NexusRemoteGreetingService.cs new file mode 100644 index 0000000..097eed9 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/Handler/NexusRemoteGreetingService.cs @@ -0,0 +1,112 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Handler; + +using NexusRpc.Handlers; +using Temporalio.Api.Enums.V1; +using Temporalio.Client; +using Temporalio.Nexus; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern; + +// On-demand pattern: no workflow is pre-started. The caller creates workflow instances +// through Nexus operations. Each operation includes a UserId so the handler can derive +// the target workflow ID. +[NexusServiceHandler(typeof(INexusRemoteGreetingService))] +public class NexusRemoteGreetingService +{ +#pragma warning disable VSTHRD200 // Names must match the INexusRemoteGreetingService operations, which can't take the Async suffix + + // Starts the GreetingWorkflow for the given user, or attaches to one already running. + // StartWorkflowAsync returns an async result, which attaches a completion callback, + // so the Operation completes when the Workflow returns. + [TemporalOperation] + public Task> RunFromRemote( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.RunFromRemoteInput input) => + client.StartWorkflowAsync( + (GreetingWorkflow wf) => wf.RunAsync(input.UserId), + new() + { + Id = GetWorkflowId(input.UserId), + + // By default, starting a Workflow whose ID is already running fails the Operation. + // Since AttachApprovalContext below can create the GreetingWorkflow first, this + // Operation needs to attach to the running execution rather than fail. + IdConflictPolicy = WorkflowIdConflictPolicy.UseExisting, + }); + + // Query: read-only, no state mutation — uses workflow query + [TemporalOperation] + public async Task> GetLanguages( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.GetLanguagesInput input) + { + // Access the Temporal client from the Nexus client passed to the handler + var handle = client.TemporalClient.GetWorkflowHandle( + GetWorkflowId(input.UserId)); + return TemporalOperationResult.SyncResult( + await handle.QueryAsync(wf => wf.QueryLanguages(input.IncludeUnsupported))); + } + + // Query: read-only — returns the workflow's current language + [TemporalOperation] + public async Task> GetLanguage( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.GetLanguageInput input) + { + var handle = client.TemporalClient.GetWorkflowHandle( + GetWorkflowId(input.UserId)); + return TemporalOperationResult.SyncResult( + await handle.QueryAsync(wf => wf.QueryLanguage())); + } + + // Update: mutates state and returns the previous value — uses workflow update + [TemporalOperation] + public Task> SetLanguage( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.SetLanguageInput input) => + client.StartWorkflowUpdateAsync( + GetWorkflowId(input.UserId), + wf => wf.SetLanguageAsync(input.Language), + // An Update-backed Operation must wait for the Accepted stage. Any other stage is + // rejected with "nexus op workflow updates only support WorkflowUpdateStageAccepted + // for async updates". + new(WorkflowUpdateStage.Accepted)); + + // Signal: fire-and-forget, no return value needed — uses workflow signal + [TemporalOperation] + public async Task> Approve( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.ApproveInput input) + { + var handle = client.TemporalClient.GetWorkflowHandle( + GetWorkflowId(input.UserId)); + await handle.SignalAsync(wf => wf.ApproveAsync(input.Name)); + return TemporalOperationResult.SyncResult(default); + } + + // Signal-with-Start: starts the Workflow first if it is not already running. + // When the Workflow already exists, only the Signal is delivered. + [TemporalOperation] + public async Task> AttachApprovalContext( + TemporalOperationStartContext ctx, + ITemporalNexusClient client, + INexusRemoteGreetingService.AttachApprovalContextInput input) + { + var options = new WorkflowOptions( + id: GetWorkflowId(input.UserId), + taskQueue: NexusOperationExecutionContext.Current.Info.TaskQueue); + options.SignalWithStart((GreetingWorkflow wf) => wf.AttachApprovalContextAsync(input.Note)); + await client.TemporalClient.StartWorkflowAsync( + (GreetingWorkflow wf) => wf.RunAsync(input.UserId), options); + return TemporalOperationResult.SyncResult(default); + } + +#pragma warning restore VSTHRD200 + + private static string GetWorkflowId(string userId) => $"GreetingWorkflow_for_{userId}"; +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/INexusRemoteGreetingService.cs b/src/NexusMessagingTemporalOperation/OnDemandPattern/INexusRemoteGreetingService.cs new file mode 100644 index 0000000..02cf3c6 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/INexusRemoteGreetingService.cs @@ -0,0 +1,40 @@ +namespace TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern; + +using NexusRpc; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; + +[NexusService] +public interface INexusRemoteGreetingService +{ + [NexusOperation] + string RunFromRemote(RunFromRemoteInput input); + + [NexusOperation] + GetLanguagesOutput GetLanguages(GetLanguagesInput input); + + [NexusOperation] + Language GetLanguage(GetLanguageInput input); + + [NexusOperation] + Language SetLanguage(SetLanguageInput input); + + [NexusOperation] + void Approve(ApproveInput input); + + [NexusOperation] + void AttachApprovalContext(AttachApprovalContextInput input); + + public record RunFromRemoteInput(string UserId); + + public record GetLanguagesInput(bool IncludeUnsupported, string UserId); + + public record GetLanguagesOutput(IReadOnlyList Languages); + + public record GetLanguageInput(string UserId); + + public record SetLanguageInput(Language Language, string UserId); + + public record ApproveInput(string Name, string UserId); + + public record AttachApprovalContextInput(string Note, string UserId); +} diff --git a/src/NexusMessagingTemporalOperation/OnDemandPattern/README.md b/src/NexusMessagingTemporalOperation/OnDemandPattern/README.md new file mode 100644 index 0000000..444c477 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/OnDemandPattern/README.md @@ -0,0 +1,77 @@ +## On-demand pattern + +No Workflow is pre-started. The caller creates and controls Workflow instances through Nexus +operations. `NexusRemoteGreetingService` adds a `RunFromRemote` operation that starts a +`GreetingWorkflow`, and every operation includes a `UserId` so the handler can derive +the target Workflow ID. + +The caller Workflow: +1. Attaches approval context for the first user via `AttachApprovalContext`, before anything has + started that user's Workflow +2. Starts two remote `GreetingWorkflow` instances via `RunFromRemote` (backed by a Workflow started + through `ITemporalNexusClient.StartWorkflowAsync`) +3. Attaches approval context for the second user, whose Workflow now already exists +4. Workflow one: queries supported languages, changes to Spanish, and approves +5. Workflow two: queries the current language, changes to French, and approves +6. Waits for each to complete and returns their results + +### Running + +This sample requires a Temporal dev server build that supports Workflow Update callbacks. Download +the compatible binary from the [Temporal CLI pre-release instructions](https://docs.temporal.io/standalone-nexus-operation#temporal-cli-support). + +Start the Temporal dev server with the required namespaces pre-created, and Workflow Update +callbacks and signal-with-start from a Workflow enabled: + +```bash +./temporal server start-dev \ + --dynamic-config-value history.enableUpdateCallbacks=true \ + --dynamic-config-value history.enableCHASMSignalBacklinks=true \ + --dynamic-config-value history.enableSignalWithStartFromWorkflow=true \ + --namespace nexus-messaging-handler-namespace \ + --namespace nexus-messaging-caller-namespace +``` + +Create the Nexus endpoint: + +```bash +./temporal operator nexus endpoint create \ + --name nexus-messaging-on-demand-pattern-endpoint \ + --target-namespace nexus-messaging-handler-namespace \ + --target-task-queue nexus-messaging-handler-sample +``` + +In one terminal, start the handler worker: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- remote-handler-worker +``` + +In a second terminal, start the caller worker: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- remote-caller-worker +``` + +In a third terminal, run the following command to start the example: + +```bash +dotnet run --project src/NexusMessagingTemporalOperation -- remote-caller-workflow +``` + +Expected output: + +``` +Attached approval context before the workflow existed: user-one +Started remote workflow for user: user-one +Started remote workflow for user: user-two +Attached approval context to the running workflow: user-two +[One] Supported languages: Chinese, English +[One] Set language from English to Spanish +[One] Approved +[Two] Current language: English +[Two] Set language from English to French +[Two] Approved +[One] Result: Hola, mundo (approved by CallerRemoteWorkflow) +[Two] Result: Bonjour, monde (approved by CallerRemoteWorkflow) +``` diff --git a/src/NexusMessagingTemporalOperation/Program.cs b/src/NexusMessagingTemporalOperation/Program.cs new file mode 100644 index 0000000..3932e3d --- /dev/null +++ b/src/NexusMessagingTemporalOperation/Program.cs @@ -0,0 +1,184 @@ +using Microsoft.Extensions.Logging; +using Temporalio.Client; +using Temporalio.Common.EnvConfig; +using Temporalio.Worker; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Caller; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Handler; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Caller; +using CallerGreetingWorkflow = TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Handler.GreetingWorkflow; +using OnDemandGreetingWorkflow = TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Handler.GreetingWorkflow; +using OnDemandNexusService = TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Handler.NexusRemoteGreetingService; + +using var loggerFactory = LoggerFactory.Create(builder => + builder. + AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] "). + SetMinimumLevel(LogLevel.Information)); +var logger = loggerFactory.CreateLogger(); + +// Cancellation token cancelled on ctrl+c +using var tokenSource = new CancellationTokenSource(); +Console.CancelKeyPress += (_, eventArgs) => +{ + tokenSource.Cancel(); + eventArgs.Cancel = true; +}; + +const string HandlerNamespace = "nexus-messaging-handler-namespace"; +const string CallerNamespace = "nexus-messaging-caller-namespace"; +const string HandlerTaskQueue = "nexus-messaging-handler-sample"; +const string CallerTaskQueue = "nexus-messaging-caller-sample"; + +Task ConnectClientAsync(string temporalNamespace) +{ + var connectOptions = ClientEnvConfig.LoadClientConnectOptions(); + connectOptions.TargetHost ??= "localhost:7233"; + connectOptions.Namespace = temporalNamespace; + connectOptions.LoggerFactory = loggerFactory; + return TemporalClient.ConnectAsync(connectOptions); +} + +async Task RunHandlerWorkerAsync() +{ + logger.LogInformation("Running handler worker"); + var client = await ConnectClientAsync(HandlerNamespace); + + // Start entity workflow with UseExisting policy (entity pattern: pre-start at boot) + var workflowId = "GreetingWorkflow_for_default-user"; + try + { + await client.StartWorkflowAsync( + (CallerGreetingWorkflow wf) => wf.RunAsync("default-user"), + new(id: workflowId, taskQueue: HandlerTaskQueue) + { + IdConflictPolicy = Temporalio.Api.Enums.V1.WorkflowIdConflictPolicy.UseExisting, + }); + logger.LogInformation("Started entity workflow {WorkflowId}", workflowId); + } + catch (Temporalio.Exceptions.RpcException ex) + { + logger.LogWarning(ex, "Could not start entity workflow"); + } + + using var worker = new TemporalWorker( + client, + new TemporalWorkerOptions(HandlerTaskQueue). + AddNexusService(new NexusGreetingService()). + AddWorkflow(). + AddAllActivities(new GreetingActivities())); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + logger.LogInformation("Handler worker cancelled"); + } +} + +async Task RunCallerWorkerAsync() +{ + logger.LogInformation("Running caller worker"); + using var worker = new TemporalWorker( + await ConnectClientAsync(CallerNamespace), + new TemporalWorkerOptions(CallerTaskQueue). + AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + logger.LogInformation("Caller worker cancelled"); + } +} + +async Task ExecuteCallerWorkflowAsync() +{ + logger.LogInformation("Executing caller workflow"); + var client = await ConnectClientAsync(CallerNamespace); + var result = await client.ExecuteWorkflowAsync( + (CallerWorkflow wf) => wf.RunAsync("default-user"), + new(id: "nexus-messaging-caller-workflow-id", taskQueue: CallerTaskQueue)); + logger.LogInformation("Caller workflow result:"); + foreach (var entry in result) + { + logger.LogInformation(" {Entry}", entry); + } +} + +async Task RunRemoteHandlerWorkerAsync() +{ + logger.LogInformation("Running remote handler worker (on-demand pattern)"); + using var worker = new TemporalWorker( + await ConnectClientAsync(HandlerNamespace), + new TemporalWorkerOptions(HandlerTaskQueue). + AddNexusService(new OnDemandNexusService()). + AddWorkflow(). + AddAllActivities(new GreetingActivities())); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + logger.LogInformation("Remote handler worker cancelled"); + } +} + +async Task RunRemoteCallerWorkerAsync() +{ + logger.LogInformation("Running remote caller worker (on-demand pattern)"); + using var worker = new TemporalWorker( + await ConnectClientAsync(CallerNamespace), + new TemporalWorkerOptions(CallerTaskQueue). + AddWorkflow()); + try + { + await worker.ExecuteAsync(tokenSource.Token); + } + catch (OperationCanceledException) + { + logger.LogInformation("Remote caller worker cancelled"); + } +} + +async Task ExecuteRemoteCallerWorkflowAsync() +{ + logger.LogInformation("Executing remote caller workflow (on-demand pattern)"); + var client = await ConnectClientAsync(CallerNamespace); + var result = await client.ExecuteWorkflowAsync( + (CallerRemoteWorkflow wf) => wf.RunAsync(), + new(id: "nexus-messaging-remote-caller-workflow-id", taskQueue: CallerTaskQueue)); + logger.LogInformation("Remote caller workflow result:"); + foreach (var entry in result) + { + logger.LogInformation(" {Entry}", entry); + } +} + +switch (args.ElementAtOrDefault(0)) +{ + case "handler-worker": + await RunHandlerWorkerAsync(); + break; + case "caller-worker": + await RunCallerWorkerAsync(); + break; + case "caller-workflow": + await ExecuteCallerWorkflowAsync(); + break; + case "remote-handler-worker": + await RunRemoteHandlerWorkerAsync(); + break; + case "remote-caller-worker": + await RunRemoteCallerWorkerAsync(); + break; + case "remote-caller-workflow": + await ExecuteRemoteCallerWorkflowAsync(); + break; + default: + throw new ArgumentException( + "Must pass 'handler-worker', 'caller-worker', 'caller-workflow', " + + "'remote-handler-worker', 'remote-caller-worker', or 'remote-caller-workflow' as the single argument"); +} diff --git a/src/NexusMessagingTemporalOperation/README.md b/src/NexusMessagingTemporalOperation/README.md new file mode 100644 index 0000000..aa2b3ea --- /dev/null +++ b/src/NexusMessagingTemporalOperation/README.md @@ -0,0 +1,22 @@ +This sample shows how to expose a long-running Workflow's queries, updates, and signals as Nexus operations. This +version uses the experimental `[TemporalOperation]` attribute to declare a Temporal-backed Nexus operation start handler +directly on a method within a `[NexusServiceHandler]` class, instead of returning an `IOperationHandler` from +`[NexusOperationHandler]`. + +There are two self-contained examples, each in its own directory: + +| | `CallerPattern/` | `OnDemandPattern/` | +|--------------------------------|------------------------------------------------|-----------------------------------------------------------------| +| **Pattern** | Query, update, and signal an existing Workflow | Create and run Workflows on demand, and send signals to them | +| **Who creates the Workflow?** | The handler worker starts it on boot | The caller starts it via a Nexus operation | +| **Who knows the Workflow ID?** | Only the handler | The caller passes a UserId; the handler derives the Workflow ID | +| **Nexus service** | `NexusGreetingService` | `NexusRemoteGreetingService` | + +The `GreetingActivities` and `Language` enum are shared between the two patterns (in the `Common/` +directory). Each pattern has its own `GreetingWorkflow`, Nexus service interface, and handler +implementation. This highlights that the same activity logic can be exposed through Nexus in different +ways depending on whether the caller needs lifecycle control. + +See each directory's README for running instructions: +- [CallerPattern README](CallerPattern/README.md) +- [OnDemandPattern README](OnDemandPattern/README.md) diff --git a/src/NexusMessagingTemporalOperation/TemporalioSamples.NexusMessagingTemporalOperation.csproj b/src/NexusMessagingTemporalOperation/TemporalioSamples.NexusMessagingTemporalOperation.csproj new file mode 100644 index 0000000..573e2c9 --- /dev/null +++ b/src/NexusMessagingTemporalOperation/TemporalioSamples.NexusMessagingTemporalOperation.csproj @@ -0,0 +1,11 @@ + + + + Exe + + + + + + + diff --git a/tests/NexusMessaging/CallerPatternTests.cs b/tests/NexusMessaging/CallerPatternTests.cs index 5ec1770..a9938b2 100644 --- a/tests/NexusMessaging/CallerPatternTests.cs +++ b/tests/NexusMessaging/CallerPatternTests.cs @@ -1,7 +1,6 @@ namespace TemporalioSamples.Tests.NexusMessaging; using Temporalio.Client; -using Temporalio.Testing; using Temporalio.Worker; using TemporalioSamples.NexusMessaging.CallerPattern.Caller; using TemporalioSamples.NexusMessaging.CallerPattern.Handler; @@ -9,40 +8,23 @@ namespace TemporalioSamples.Tests.NexusMessaging; using Xunit; using Xunit.Abstractions; -public class CallerPatternTests : TestBase +public class CallerPatternTests : WorkflowEnvironmentTestBase { - public CallerPatternTests(ITestOutputHelper output) - : base(output) + public CallerPatternTests(ITestOutputHelper output, WorkflowEnvironment env) + : base(output, env) { } [Fact] public async Task RunAsync_CallerWorkflow_Succeeds() { - // SetLanguage is backed by a Workflow Update, which needs a dev server build that supports - // update callbacks. - await using var env = await WorkflowEnvironment.StartLocalAsync(new() - { - DevServerOptions = new() - { - DownloadVersion = "v1.7.4-standalone-nexus-operations", - ExtraArgs = - [ - "--dynamic-config-value", - "history.enableUpdateCallbacks=true", - "--dynamic-config-value", - "history.enableCHASMSignalBacklinks=true", - ], - }, - }); - var handlerTaskQueue = $"tq-{Guid.NewGuid()}"; - await env.CreateNexusEndpointAsync(NexusEndpoints.GreetingService, handlerTaskQueue); + await Env.TestEnv.CreateNexusEndpointAsync(NexusEndpoints.GreetingService, handlerTaskQueue); var userId = $"user-{Guid.NewGuid()}"; var workflowId = $"GreetingWorkflow_for_{userId}"; // Start entity workflow - await env.Client.StartWorkflowAsync( + await Client.StartWorkflowAsync( (GreetingWorkflow wf) => wf.RunAsync(userId), new(id: workflowId, taskQueue: handlerTaskQueue) { @@ -51,7 +33,7 @@ await env.Client.StartWorkflowAsync( // Run handler worker using var handlerWorker = new TemporalWorker( - env.Client, + Client, new TemporalWorkerOptions(handlerTaskQueue). AddNexusService(new NexusGreetingService()). AddWorkflow(). @@ -60,12 +42,12 @@ await handlerWorker.ExecuteAsync(async () => { // Run caller worker using var callerWorker = new TemporalWorker( - env.Client, + Client, new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). AddWorkflow()); await callerWorker.ExecuteAsync(async () => { - var result = await env.Client.ExecuteWorkflowAsync( + var result = await Client.ExecuteWorkflowAsync( (CallerWorkflow wf) => wf.RunAsync(userId), new(id: $"wf-{Guid.NewGuid()}", taskQueue: callerWorker.Options.TaskQueue!)); diff --git a/tests/NexusMessaging/OnDemandPatternTests.cs b/tests/NexusMessaging/OnDemandPatternTests.cs index 4b7f7b3..61bc21c 100644 --- a/tests/NexusMessaging/OnDemandPatternTests.cs +++ b/tests/NexusMessaging/OnDemandPatternTests.cs @@ -1,7 +1,6 @@ namespace TemporalioSamples.Tests.NexusMessaging; using Temporalio.Client; -using Temporalio.Testing; using Temporalio.Worker; using TemporalioSamples.NexusMessaging.Common; using TemporalioSamples.NexusMessaging.OnDemandPattern.Caller; @@ -9,41 +8,22 @@ namespace TemporalioSamples.Tests.NexusMessaging; using Xunit; using Xunit.Abstractions; -public class OnDemandPatternTests : TestBase +public class OnDemandPatternTests : WorkflowEnvironmentTestBase { - public OnDemandPatternTests(ITestOutputHelper output) - : base(output) + public OnDemandPatternTests(ITestOutputHelper output, WorkflowEnvironment env) + : base(output, env) { } [Fact] public async Task RunAsync_CallerRemoteWorkflow_Succeeds() { - // SetLanguage is backed by a Workflow Update and AttachApprovalContext by - // Signal-with-Start, both of which need a dev server build that supports them. - await using var env = await WorkflowEnvironment.StartLocalAsync(new() - { - DevServerOptions = new() - { - DownloadVersion = "v1.7.4-standalone-nexus-operations", - ExtraArgs = - [ - "--dynamic-config-value", - "history.enableUpdateCallbacks=true", - "--dynamic-config-value", - "history.enableCHASMSignalBacklinks=true", - "--dynamic-config-value", - "history.enableSignalWithStartFromWorkflow=true", - ], - }, - }); - var handlerTaskQueue = $"tq-{Guid.NewGuid()}"; - await env.CreateNexusEndpointAsync(NexusEndpoints.RemoteGreetingService, handlerTaskQueue); + await Env.TestEnv.CreateNexusEndpointAsync(NexusEndpoints.RemoteGreetingService, handlerTaskQueue); // Run handler worker using var handlerWorker = new TemporalWorker( - env.Client, + Client, new TemporalWorkerOptions(handlerTaskQueue). AddNexusService(new NexusRemoteGreetingService()). AddWorkflow(). @@ -52,27 +32,25 @@ await handlerWorker.ExecuteAsync(async () => { // Run caller worker using var callerWorker = new TemporalWorker( - env.Client, + Client, new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). AddWorkflow()); await callerWorker.ExecuteAsync(async () => { - var result = await env.Client.ExecuteWorkflowAsync( + var result = await Client.ExecuteWorkflowAsync( (CallerRemoteWorkflow wf) => wf.RunAsync(), new(id: $"wf-{Guid.NewGuid()}", taskQueue: callerWorker.Options.TaskQueue!)); - Assert.Contains("Attached approval context before the workflow existed: user-one", result[0]); - Assert.Contains("Started remote workflow for user: user-one", result[1]); - Assert.Contains("Started remote workflow for user: user-two", result[2]); - Assert.Contains("Attached approval context to the running workflow: user-two", result[3]); - Assert.Contains("[One] Supported languages:", result[4]); - Assert.Contains($"[One] Set language from {Language.English} to {Language.Spanish}", result[5]); - Assert.Contains("[One] Approved", result[6]); - Assert.Contains("[Two] Current language:", result[7]); - Assert.Contains($"[Two] Set language from {Language.English} to {Language.French}", result[8]); - Assert.Contains("[Two] Approved", result[9]); - Assert.Contains("[One] Result:", result[10]); - Assert.Contains("[Two] Result:", result[11]); + Assert.Contains("Started remote workflow for user: user-one", result[0]); + Assert.Contains("Started remote workflow for user: user-two", result[1]); + Assert.Contains("[One] Supported languages:", result[2]); + Assert.Contains($"[One] Set language from {Language.English} to {Language.Spanish}", result[3]); + Assert.Contains("[One] Approved", result[4]); + Assert.Contains("[Two] Current language:", result[5]); + Assert.Contains($"[Two] Set language from {Language.English} to {Language.French}", result[6]); + Assert.Contains("[Two] Approved", result[7]); + Assert.Contains("[One] Result:", result[8]); + Assert.Contains("[Two] Result:", result[9]); }); }); } diff --git a/tests/NexusMessagingTemporalOperation/CallerPatternTests.cs b/tests/NexusMessagingTemporalOperation/CallerPatternTests.cs new file mode 100644 index 0000000..98a29b3 --- /dev/null +++ b/tests/NexusMessagingTemporalOperation/CallerPatternTests.cs @@ -0,0 +1,80 @@ +namespace TemporalioSamples.Tests.NexusMessagingTemporalOperation; + +using Temporalio.Client; +using Temporalio.Testing; +using Temporalio.Worker; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Caller; +using TemporalioSamples.NexusMessagingTemporalOperation.CallerPattern.Handler; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using Xunit; +using Xunit.Abstractions; + +public class CallerPatternTests : TestBase +{ + public CallerPatternTests(ITestOutputHelper output) + : base(output) + { + } + + [Fact] + public async Task RunAsync_CallerWorkflow_Succeeds() + { + // SetLanguage is backed by a Workflow Update, which needs a dev server build that supports + // update callbacks. + await using var env = await WorkflowEnvironment.StartLocalAsync(new() + { + DevServerOptions = new() + { + DownloadVersion = "v1.7.4-standalone-nexus-operations", + ExtraArgs = + [ + "--dynamic-config-value", + "history.enableUpdateCallbacks=true", + "--dynamic-config-value", + "history.enableCHASMSignalBacklinks=true", + ], + }, + }); + + var handlerTaskQueue = $"tq-{Guid.NewGuid()}"; + await env.CreateNexusEndpointAsync(NexusEndpoints.GreetingService, handlerTaskQueue); + var userId = $"user-{Guid.NewGuid()}"; + var workflowId = $"GreetingWorkflow_for_{userId}"; + + // Start entity workflow + await env.Client.StartWorkflowAsync( + (GreetingWorkflow wf) => wf.RunAsync(userId), + new(id: workflowId, taskQueue: handlerTaskQueue) + { + IdConflictPolicy = Temporalio.Api.Enums.V1.WorkflowIdConflictPolicy.UseExisting, + }); + + // Run handler worker + using var handlerWorker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions(handlerTaskQueue). + AddNexusService(new NexusGreetingService()). + AddWorkflow(). + AddAllActivities(new GreetingActivities())); + await handlerWorker.ExecuteAsync(async () => + { + // Run caller worker + using var callerWorker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddWorkflow()); + await callerWorker.ExecuteAsync(async () => + { + var result = await env.Client.ExecuteWorkflowAsync( + (CallerWorkflow wf) => wf.RunAsync(userId), + new(id: $"wf-{Guid.NewGuid()}", taskQueue: callerWorker.Options.TaskQueue!)); + + Assert.Equal(4, result.Length); + Assert.Contains("Supported languages:", result[0]); + Assert.Contains("Current language:", result[1]); + Assert.Contains($"Set language from {Language.English} to {Language.Chinese}", result[2]); + Assert.Equal("Approved workflow", result[3]); + }); + }); + } +} diff --git a/tests/NexusMessagingTemporalOperation/OnDemandPatternTests.cs b/tests/NexusMessagingTemporalOperation/OnDemandPatternTests.cs new file mode 100644 index 0000000..e467bc6 --- /dev/null +++ b/tests/NexusMessagingTemporalOperation/OnDemandPatternTests.cs @@ -0,0 +1,79 @@ +namespace TemporalioSamples.Tests.NexusMessagingTemporalOperation; + +using Temporalio.Client; +using Temporalio.Testing; +using Temporalio.Worker; +using TemporalioSamples.NexusMessagingTemporalOperation.Common; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Caller; +using TemporalioSamples.NexusMessagingTemporalOperation.OnDemandPattern.Handler; +using Xunit; +using Xunit.Abstractions; + +public class OnDemandPatternTests : TestBase +{ + public OnDemandPatternTests(ITestOutputHelper output) + : base(output) + { + } + + [Fact] + public async Task RunAsync_CallerRemoteWorkflow_Succeeds() + { + // SetLanguage is backed by a Workflow Update and AttachApprovalContext by + // Signal-with-Start, both of which need a dev server build that supports them. + await using var env = await WorkflowEnvironment.StartLocalAsync(new() + { + DevServerOptions = new() + { + DownloadVersion = "v1.7.4-standalone-nexus-operations", + ExtraArgs = + [ + "--dynamic-config-value", + "history.enableUpdateCallbacks=true", + "--dynamic-config-value", + "history.enableCHASMSignalBacklinks=true", + "--dynamic-config-value", + "history.enableSignalWithStartFromWorkflow=true", + ], + }, + }); + + var handlerTaskQueue = $"tq-{Guid.NewGuid()}"; + await env.CreateNexusEndpointAsync(NexusEndpoints.RemoteGreetingService, handlerTaskQueue); + + // Run handler worker + using var handlerWorker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions(handlerTaskQueue). + AddNexusService(new NexusRemoteGreetingService()). + AddWorkflow(). + AddAllActivities(new GreetingActivities())); + await handlerWorker.ExecuteAsync(async () => + { + // Run caller worker + using var callerWorker = new TemporalWorker( + env.Client, + new TemporalWorkerOptions($"tq-{Guid.NewGuid()}"). + AddWorkflow()); + await callerWorker.ExecuteAsync(async () => + { + var result = await env.Client.ExecuteWorkflowAsync( + (CallerRemoteWorkflow wf) => wf.RunAsync(), + new(id: $"wf-{Guid.NewGuid()}", taskQueue: callerWorker.Options.TaskQueue!)); + + Assert.Contains("Attached approval context before the workflow existed: user-one", result[0]); + Assert.Contains("Started remote workflow for user: user-one", result[1]); + Assert.Contains("Started remote workflow for user: user-two", result[2]); + Assert.Contains("Attached approval context to the running workflow: user-two", result[3]); + Assert.Contains("[One] Supported languages:", result[4]); + Assert.Contains($"[One] Set language from {Language.English} to {Language.Spanish}", result[5]); + Assert.Contains("[One] Approved", result[6]); + Assert.Contains("[Two] Current language:", result[7]); + Assert.Contains($"[Two] Set language from {Language.English} to {Language.French}", result[8]); + Assert.Contains("[Two] Approved", result[9]); + Assert.Contains("[One] Result:", result[10]); + Assert.Contains("[Two] Result:", result[11]); + }); + }); + } +} diff --git a/tests/TemporalioSamples.Tests.csproj b/tests/TemporalioSamples.Tests.csproj index 4ecadfd..bfbaa27 100644 --- a/tests/TemporalioSamples.Tests.csproj +++ b/tests/TemporalioSamples.Tests.csproj @@ -52,6 +52,7 @@ +