Nexus context propagation sample (temporalio#108)

Fixes temporalio#105

Author: cretz

Directory.Build.props

@@ -19,10 +19,10 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Temporalio" Version="1.7.0" />
-    <PackageReference Include="Temporalio.Extensions.DiagnosticSource" Version="1.6.0" />
-    <PackageReference Include="Temporalio.Extensions.Hosting" Version="1.6.0" />
-    <PackageReference Include="Temporalio.Extensions.OpenTelemetry" Version="1.6.0" />
+    <PackageReference Include="Temporalio" Version="1.9.0" />
+    <PackageReference Include="Temporalio.Extensions.DiagnosticSource" Version="1.9.0" />
+    <PackageReference Include="Temporalio.Extensions.Hosting" Version="1.9.0" />
+    <PackageReference Include="Temporalio.Extensions.OpenTelemetry" Version="1.9.0" />
     <!--
     Can also reference the SDK downloaded to a local directory:
     <ProjectReference Include="$(MSBuildThisFileDirectory)..\temporal-sdk-dotnet\src\Temporalio\Temporalio.csproj" />

README.md

@@ -23,6 +23,7 @@ Prerequisites:
 * [DependencyInjection](src/DependencyInjection) - How to inject dependencies in activities and use generic hosts for workers
 * [Encryption](src/Encryption) - End-to-end encryption with Temporal payload codecs.
 * [Mutex](src/Mutex) - How to implement a mutex as a workflow. Demonstrates how to avoid race conditions or parallel mutually exclusive operations on the same resource.
+* [NexusContextPropagation](src/NexusContextPropagation) - Context propagation through Nexus services.
 * [OpenTelemetry](src/OpenTelemetry) - Demonstrates how to set up OpenTelemetry tracing and metrics for both the client and worker, using both the .NET metrics API and internal forwarding from the Core SDK.
 * [Patching](src/Patching) - Alter workflows safely with Patch and DeprecatePatch.
 * [Polling](src/Polling) - Recommended implementation of an activity that needs to periodically poll an external resource waiting its successful completion.

TemporalioSamples.sln

@@ -2,5 +2,11 @@ namespace TemporalioSamples.ContextPropagation;
 
 public static class MyContext
 {
-    public static readonly AsyncLocal<string> UserId = new();
+    public static readonly AsyncLocal<string?> UserIdLocal = new();
+
+    public static string UserId
+    {
+        get => UserIdLocal.Value ?? "<unknown>";
+        set => UserIdLocal.Value = value;
+    }
 }

src/ContextPropagation/MyContext.cs

@@ -15,12 +15,12 @@
 {
     LoggerFactory = loggerFactory,
     // This is where we set the interceptor to propagate context
-    Interceptors = new[]
-    {
-        new ContextPropagationInterceptor<string>(
-            MyContext.UserId,
+    Interceptors =
+    [
+        new ContextPropagationInterceptor<string?>(
+            MyContext.UserIdLocal,
             DataConverter.Default.PayloadConverter),
-    },
+    ],
 });
 
 async Task RunWorkerAsync()
@@ -53,7 +53,7 @@ async Task RunWorkerAsync()
 async Task ExecuteWorkflowAsync()
 {
     // Set our user ID that can be accessed in the workflow and activity
-    MyContext.UserId.Value = "some-user";
+    MyContext.UserId = "some-user";
 
     // Start workflow, send signal, wait for completion, issue query
     logger.LogInformation("Executing workflow");

src/ContextPropagation/Program.cs

@@ -9,8 +9,7 @@ public class SayHelloActivities
     public string SayHello(string name)
     {
         ActivityExecutionContext.Current.Logger.LogInformation(
-            "Activity called by user {UserId}",
-            MyContext.UserId.Value);
+            "Activity called by user {UserId}", MyContext.UserId);
         return $"Hello, {name}!";
     }
 }

src/ContextPropagation/SayHelloActivities.cs

@@ -11,9 +11,7 @@ public class SayHelloWorkflow
     [WorkflowRun]
     public async Task<string> RunAsync(string name)
     {
-        Workflow.Logger.LogInformation(
-            "Workflow called by user {UserId}",
-            MyContext.UserId.Value);
+        Workflow.Logger.LogInformation("Workflow called by user {UserId}", MyContext.UserId);
 
         // Wait for signal then run activity
         await Workflow.WaitConditionAsync(() => complete);
@@ -25,18 +23,14 @@ public async Task<string> RunAsync(string name)
     [WorkflowSignal]
     public async Task SignalCompleteAsync()
     {
-        Workflow.Logger.LogInformation(
-            "Signal called by user {UserId}",
-            MyContext.UserId.Value);
+        Workflow.Logger.LogInformation("Signal called by user {UserId}", MyContext.UserId);
         complete = true;
     }
 
     [WorkflowQuery]
     public bool IsComplete()
     {
-        Workflow.Logger.LogInformation(
-            "Query called by user {UserId}",
-            MyContext.UserId.Value);
+        Workflow.Logger.LogInformation("Query called by user {UserId}", MyContext.UserId);
         return complete;
     }
 }

src/ContextPropagation/SayHelloWorkflow.workflow.cs

@@ -0,0 +1,18 @@
+namespace TemporalioSamples.NexusContextPropagation.Caller;
+
+using Microsoft.Extensions.Logging;
+using Temporalio.Workflows;
+using TemporalioSamples.ContextPropagation;
+
+[Workflow]
+public class HelloCallerWorkflow
+{
+    [WorkflowRun]
+    public async Task<string> RunAsync(string name, IHelloService.HelloLanguage language)
+    {
+        Workflow.Logger.LogInformation("Caller workflow called by user {UserId}", MyContext.UserId);
+        var output = await Workflow.CreateNexusClient<IHelloService>(IHelloService.EndpointName).
+            ExecuteNexusOperationAsync(svc => svc.SayHello(new(name, language)));
+        return output.Message;
+    }
+}

src/NexusContextPropagation/Caller/HelloCallerWorkflow.workflow.cs

@@ -0,0 +1,27 @@
+namespace TemporalioSamples.NexusContextPropagation.Handler;
+
+using Microsoft.Extensions.Logging;
+using Temporalio.Exceptions;
+using Temporalio.Workflows;
+using TemporalioSamples.ContextPropagation;
+
+[Workflow]
+public class HelloHandlerWorkflow
+{
+    [WorkflowRun]
+    public async Task<IHelloService.HelloOutput> RunAsync(IHelloService.HelloInput input)
+    {
+        Workflow.Logger.LogInformation("Handler workflow called by user {UserId}", MyContext.UserId);
+        var message = input.Language switch
+        {
+            IHelloService.HelloLanguage.En => $"Hello {input.Name} 👋",
+            IHelloService.HelloLanguage.Fr => $"Bonjour {input.Name} 👋",
+            IHelloService.HelloLanguage.De => $"Hallo {input.Name} 👋",
+            IHelloService.HelloLanguage.Es => $"¡Hola! {input.Name} 👋",
+            IHelloService.HelloLanguage.Tr => $"Merhaba {input.Name} 👋",
+            _ => throw new ApplicationFailureException(
+                $"Unsupported language: {input.Language}", errorType: "UNSUPPORTED_LANGUAGE"),
+        };
+        return new($"{message} (user id: {MyContext.UserId})");
+    }
+}

src/NexusContextPropagation/Handler/HelloHandlerWorkflow.workflow.cs

@@ -0,0 +1,27 @@
+namespace TemporalioSamples.NexusContextPropagation.Handler;
+
+using Microsoft.Extensions.Logging;
+using NexusRpc.Handlers;
+using Temporalio.Nexus;
+using TemporalioSamples.ContextPropagation;
+
+[NexusServiceHandler(typeof(IHelloService))]
+public class HelloService
+{
+    [NexusOperationHandler]
+    public IOperationHandler<IHelloService.HelloInput, IHelloService.HelloOutput> SayHello() =>
+        // This Nexus service operation is backed by a workflow run
+        WorkflowRunOperationHandler.FromHandleFactory(
+            (WorkflowRunOperationContext context, IHelloService.HelloInput input) =>
+            {
+                NexusOperationExecutionContext.Current.Logger.LogInformation(
+                    "Hello service called by user {UserId}", MyContext.UserId);
+                return context.StartWorkflowAsync(
+                    (HelloHandlerWorkflow wf) => wf.RunAsync(input),
+                    // 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 the operation,
+                    // this ID is guaranteed to be stable across retries of this operation.
+                    new() { Id = context.HandlerContext.RequestId });
+            });
+}