Skip to content

Commit 49ced3d

Browse files
cretzcretz
authored
Update SDK and add context propagation sample (#68)
Fixes #14
1 parent 54bf6f0 commit 49ced3d

12 files changed

+413
-5
lines changed

‎Directory.Build.props‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,10 @@
1919
</PropertyGroup>
2020

2121
<ItemGroup>
22-
<PackageReference Include="Temporalio" Version="1.0.0" />
23-
<PackageReference Include="Temporalio.Extensions.DiagnosticSource" Version="1.0.0" />
24-
<PackageReference Include="Temporalio.Extensions.Hosting" Version="1.0.0" />
25-
<PackageReference Include="Temporalio.Extensions.OpenTelemetry" Version="1.0.0" />
22+
<PackageReference Include="Temporalio" Version="1.1.2" />
23+
<PackageReference Include="Temporalio.Extensions.DiagnosticSource" Version="1.1.2" />
24+
<PackageReference Include="Temporalio.Extensions.Hosting" Version="1.1.2" />
25+
<PackageReference Include="Temporalio.Extensions.OpenTelemetry" Version="1.1.2" />
2626
<!--
2727
Can also reference the SDK downloaded to a local directory:
2828
<ProjectReference Include="$(MSBuildThisFileDirectory)..\temporal-sdk-dotnet\src\Temporalio\Temporalio.csproj" />

‎README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ Prerequisites:
1717
* [ActivityWorker](src/ActivityWorker) - Use .NET activities from a workflow in another language.
1818
* [AspNet](src/AspNet) - Demonstration of a generic host worker and an ASP.NET workflow starter.
1919
* [ClientMtls](src/ClientMtls) - How to use client certificate authentication, e.g. for Temporal Cloud.
20+
* [ContextPropagation](src/ContextPropagation) - Context propagation via interceptors.
2021
* [DependencyInjection](src/DependencyInjection) - How to inject dependencies in activities and use generic hosts for workers
2122
* [Encryption](src/Encryption) - End-to-end encryption with Temporal payload codecs.
2223
* [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.

‎TemporalioSamples.sln‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "TemporalioSamples.Saga", "s
5353
EndProject
5454
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.WorkflowUpdate", "src\WorkflowUpdate\TemporalioSamples.WorkflowUpdate.csproj", "{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630}"
5555
EndProject
56+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.ContextPropagation", "src\ContextPropagation\TemporalioSamples.ContextPropagation.csproj", "{7B797D20-485F-441D-8E71-AF7E315FA9CF}"
57+
EndProject
5658
Global
5759
GlobalSection(SolutionConfigurationPlatforms) = preSolution
5860
Debug|Any CPU = Debug|Any CPU
@@ -143,6 +145,10 @@ Global
143145
{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630}.Debug|Any CPU.Build.0 = Debug|Any CPU
144146
{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630}.Release|Any CPU.ActiveCfg = Release|Any CPU
145147
{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630}.Release|Any CPU.Build.0 = Release|Any CPU
148+
{7B797D20-485F-441D-8E71-AF7E315FA9CF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
149+
{7B797D20-485F-441D-8E71-AF7E315FA9CF}.Debug|Any CPU.Build.0 = Debug|Any CPU
150+
{7B797D20-485F-441D-8E71-AF7E315FA9CF}.Release|Any CPU.ActiveCfg = Release|Any CPU
151+
{7B797D20-485F-441D-8E71-AF7E315FA9CF}.Release|Any CPU.Build.0 = Release|Any CPU
146152
EndGlobalSection
147153
GlobalSection(SolutionProperties) = preSolution
148154
HideSolutionNode = FALSE
@@ -171,6 +177,6 @@ Global
171177
{3168FB2D-D821-433A-A761-309E0474DE48} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
172178
{B79F07F7-3429-4C58-84C3-08587F748B2D} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
173179
{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
174-
180+
{7B797D20-485F-441D-8E71-AF7E315FA9CF} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
175181
EndGlobalSection
176182
EndGlobal

‎src/ContextPropagation/ContextPropagationInterceptor.cs‎

Lines changed: 169 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,169 @@
1+
namespace TemporalioSamples.ContextPropagation;
2+
3+
using System.Threading.Tasks;
4+
using Temporalio.Api.Common.V1;
5+
using Temporalio.Client;
6+
using Temporalio.Client.Interceptors;
7+
using Temporalio.Converters;
8+
using Temporalio.Worker.Interceptors;
9+
using Temporalio.Workflows;
10+
11+
/// <summary>
12+
/// General purpose interceptor that can be used to propagate async-local context through workflows
13+
/// and activities. This must be set on the client used for interacting with workflows and used for
14+
/// the worker.
15+
/// </summary>
16+
/// <typeparam name="T">Context data type.</typeparam>
17+
public class ContextPropagationInterceptor<T> : IClientInterceptor, IWorkerInterceptor
18+
{
19+
private readonly AsyncLocal<T> context;
20+
private readonly IPayloadConverter payloadConverter;
21+
private readonly string headerKey;
22+
23+
public ContextPropagationInterceptor(
24+
AsyncLocal<T> context,
25+
IPayloadConverter payloadConverter,
26+
string headerKey = "__my_context_key")
27+
{
28+
this.context = context;
29+
this.payloadConverter = payloadConverter;
30+
this.headerKey = headerKey;
31+
}
32+
33+
public ClientOutboundInterceptor InterceptClient(ClientOutboundInterceptor nextInterceptor) =>
34+
new ContextPropagationClientOutboundInterceptor(this, nextInterceptor);
35+
36+
public WorkflowInboundInterceptor InterceptWorkflow(WorkflowInboundInterceptor nextInterceptor) =>
37+
new ContextPropagationWorkflowInboundInterceptor(this, nextInterceptor);
38+
39+
public ActivityInboundInterceptor InterceptActivity(ActivityInboundInterceptor nextInterceptor) =>
40+
new ContextPropagationActivityInboundInterceptor(this, nextInterceptor);
41+
42+
private Dictionary<string, Payload> HeaderFromContext(IDictionary<string, Payload>? existing)
43+
{
44+
var ret = existing != null ?
45+
new Dictionary<string, Payload>(existing) : new Dictionary<string, Payload>(1);
46+
ret[headerKey] = payloadConverter.ToPayload(context.Value);
47+
return ret;
48+
}
49+
50+
private void WithHeadersApplied(IReadOnlyDictionary<string, Payload>? headers, Action func) =>
51+
WithHeadersApplied(
52+
headers,
53+
() =>
54+
{
55+
func();
56+
return (object?)null;
57+
});
58+
59+
private TResult WithHeadersApplied<TResult>(
60+
IReadOnlyDictionary<string, Payload>? headers, Func<TResult> func)
61+
{
62+
if (headers?.TryGetValue(headerKey, out var payload) == true && payload != null)
63+
{
64+
context.Value = payloadConverter.ToValue<T>(payload);
65+
}
66+
// These are async local, no need to unapply afterwards
67+
return func();
68+
}
69+
70+
private class ContextPropagationClientOutboundInterceptor : ClientOutboundInterceptor
71+
{
72+
private readonly ContextPropagationInterceptor<T> root;
73+
74+
public ContextPropagationClientOutboundInterceptor(
75+
ContextPropagationInterceptor<T> root, ClientOutboundInterceptor next)
76+
: base(next) => this.root = root;
77+
78+
public override Task<WorkflowHandle<TWorkflow, TResult>> StartWorkflowAsync<TWorkflow, TResult>(
79+
StartWorkflowInput input) =>
80+
base.StartWorkflowAsync<TWorkflow, TResult>(
81+
input with { Headers = root.HeaderFromContext(input.Headers) });
82+
83+
public override Task SignalWorkflowAsync(SignalWorkflowInput input) =>
84+
base.SignalWorkflowAsync(
85+
input with { Headers = root.HeaderFromContext(input.Headers) });
86+
87+
public override Task<TResult> QueryWorkflowAsync<TResult>(QueryWorkflowInput input) =>
88+
base.QueryWorkflowAsync<TResult>(
89+
input with { Headers = root.HeaderFromContext(input.Headers) });
90+
91+
public override Task<WorkflowUpdateHandle<TResult>> StartWorkflowUpdateAsync<TResult>(
92+
StartWorkflowUpdateInput input) =>
93+
base.StartWorkflowUpdateAsync<TResult>(
94+
input with { Headers = root.HeaderFromContext(input.Headers) });
95+
}
96+
97+
private class ContextPropagationWorkflowInboundInterceptor : WorkflowInboundInterceptor
98+
{
99+
private readonly ContextPropagationInterceptor<T> root;
100+
101+
public ContextPropagationWorkflowInboundInterceptor(
102+
ContextPropagationInterceptor<T> root, WorkflowInboundInterceptor next)
103+
: base(next) => this.root = root;
104+
105+
public override void Init(WorkflowOutboundInterceptor outbound) =>
106+
base.Init(new ContextPropagationWorkflowOutboundInterceptor(root, outbound));
107+
108+
public override Task<object?> ExecuteWorkflowAsync(ExecuteWorkflowInput input) =>
109+
root.WithHeadersApplied(Workflow.Info.Headers, () => Next.ExecuteWorkflowAsync(input));
110+
111+
public override Task HandleSignalAsync(HandleSignalInput input) =>
112+
root.WithHeadersApplied(input.Headers, () => Next.HandleSignalAsync(input));
113+
114+
public override object? HandleQuery(HandleQueryInput input) =>
115+
root.WithHeadersApplied(input.Headers, () => Next.HandleQuery(input));
116+
117+
public override void ValidateUpdate(HandleUpdateInput input) =>
118+
root.WithHeadersApplied(input.Headers, () => Next.ValidateUpdate(input));
119+
120+
public override Task<object?> HandleUpdateAsync(HandleUpdateInput input) =>
121+
root.WithHeadersApplied(input.Headers, () => Next.HandleUpdateAsync(input));
122+
}
123+
124+
private class ContextPropagationWorkflowOutboundInterceptor : WorkflowOutboundInterceptor
125+
{
126+
private readonly ContextPropagationInterceptor<T> root;
127+
128+
public ContextPropagationWorkflowOutboundInterceptor(
129+
ContextPropagationInterceptor<T> root, WorkflowOutboundInterceptor next)
130+
: base(next) => this.root = root;
131+
132+
public override Task<TResult> ScheduleActivityAsync<TResult>(
133+
ScheduleActivityInput input) =>
134+
Next.ScheduleActivityAsync<TResult>(
135+
input with { Headers = root.HeaderFromContext(input.Headers) });
136+
137+
public override Task<TResult> ScheduleLocalActivityAsync<TResult>(
138+
ScheduleLocalActivityInput input) =>
139+
Next.ScheduleLocalActivityAsync<TResult>(
140+
input with { Headers = root.HeaderFromContext(input.Headers) });
141+
142+
public override Task SignalChildWorkflowAsync(
143+
SignalChildWorkflowInput input) =>
144+
Next.SignalChildWorkflowAsync(
145+
input with { Headers = root.HeaderFromContext(input.Headers) });
146+
147+
public override Task SignalExternalWorkflowAsync(
148+
SignalExternalWorkflowInput input) =>
149+
Next.SignalExternalWorkflowAsync(
150+
input with { Headers = root.HeaderFromContext(input.Headers) });
151+
152+
public override Task<ChildWorkflowHandle<TWorkflow, TResult>> StartChildWorkflowAsync<TWorkflow, TResult>(
153+
StartChildWorkflowInput input) =>
154+
Next.StartChildWorkflowAsync<TWorkflow, TResult>(
155+
input with { Headers = root.HeaderFromContext(input.Headers) });
156+
}
157+
158+
private class ContextPropagationActivityInboundInterceptor : ActivityInboundInterceptor
159+
{
160+
private readonly ContextPropagationInterceptor<T> root;
161+
162+
public ContextPropagationActivityInboundInterceptor(
163+
ContextPropagationInterceptor<T> root, ActivityInboundInterceptor next)
164+
: base(next) => this.root = root;
165+
166+
public override Task<object?> ExecuteActivityAsync(ExecuteActivityInput input) =>
167+
root.WithHeadersApplied(input.Headers, () => Next.ExecuteActivityAsync(input));
168+
}
169+
}

‎src/ContextPropagation/MyContext.cs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
namespace TemporalioSamples.ContextPropagation;
2+
3+
public static class MyContext
4+
{
5+
public static readonly AsyncLocal<string> UserId = new();
6+
}

‎src/ContextPropagation/Program.cs‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
using Microsoft.Extensions.Logging;
2+
using Temporalio.Client;
3+
using Temporalio.Converters;
4+
using Temporalio.Worker;
5+
using TemporalioSamples.ContextPropagation;
6+
7+
using var loggerFactory = LoggerFactory.Create(builder =>
8+
builder.
9+
AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").
10+
SetMinimumLevel(LogLevel.Information));
11+
var logger = loggerFactory.CreateLogger<Program>();
12+
13+
// Create a client to localhost on default namespace
14+
var client = await TemporalClient.ConnectAsync(new("localhost:7233")
15+
{
16+
LoggerFactory = loggerFactory,
17+
// This is where we set the interceptor to propagate context
18+
Interceptors = new[]
19+
{
20+
new ContextPropagationInterceptor<string>(
21+
MyContext.UserId,
22+
DataConverter.Default.PayloadConverter),
23+
},
24+
});
25+
26+
async Task RunWorkerAsync()
27+
{
28+
// Cancellation token cancelled on ctrl+c
29+
using var tokenSource = new CancellationTokenSource();
30+
Console.CancelKeyPress += (_, eventArgs) =>
31+
{
32+
tokenSource.Cancel();
33+
eventArgs.Cancel = true;
34+
};
35+
36+
// Run worker until cancelled
37+
logger.LogInformation("Running worker");
38+
using var worker = new TemporalWorker(
39+
client,
40+
new TemporalWorkerOptions(taskQueue: "interceptors-sample").
41+
AddAllActivities<SayHelloActivities>(new()).
42+
AddWorkflow<SayHelloWorkflow>());
43+
try
44+
{
45+
await worker.ExecuteAsync(tokenSource.Token);
46+
}
47+
catch (OperationCanceledException)
48+
{
49+
logger.LogInformation("Worker cancelled");
50+
}
51+
}
52+
53+
async Task ExecuteWorkflowAsync()
54+
{
55+
// Set our user ID that can be accessed in the workflow and activity
56+
MyContext.UserId.Value = "some-user";
57+
58+
// Start workflow, send signal, wait for completion, issue query
59+
logger.LogInformation("Executing workflow");
60+
var handle = await client.StartWorkflowAsync(
61+
(SayHelloWorkflow wf) => wf.RunAsync("Temporal"),
62+
new(id: "interceptors-workflow-id", taskQueue: "interceptors-sample"));
63+
await handle.SignalAsync(wf => wf.SignalCompleteAsync());
64+
var result = await handle.GetResultAsync();
65+
_ = await handle.QueryAsync(wf => wf.IsComplete());
66+
logger.LogInformation("Workflow result: {Result}", result);
67+
}
68+
69+
switch (args.ElementAtOrDefault(0))
70+
{
71+
case "worker":
72+
await RunWorkerAsync();
73+
break;
74+
case "workflow":
75+
await ExecuteWorkflowAsync();
76+
break;
77+
default:
78+
throw new ArgumentException("Must pass 'worker' or 'workflow' as the single argument");
79+
}

‎src/ContextPropagation/README.md‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
# Interceptors
2+
3+
This sample demonstrates how to use interceptors to propagate contextual information from an `AsyncLocal` throughout the
4+
workflows and activities. While this demonstrates context propagation specifically, it can also be used to show how to
5+
create interceptors for any other purpose.
6+
7+
To run, first see [README.md](../../README.md) for prerequisites. Then, run the following from this directory in a
8+
separate terminal to start the worker:
9+
10+
dotnet run worker
11+
12+
Then in another terminal, run the workflow from this directory:
13+
14+
dotnet run workflow
15+
16+
The workflow terminal will show the completed workflow result and the worker terminal will show the contextual user ID
17+
is present in the workflow and activity.

‎src/ContextPropagation/SayHelloActivities.cs‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
namespace TemporalioSamples.ContextPropagation;
2+
3+
using Microsoft.Extensions.Logging;
4+
using Temporalio.Activities;
5+
6+
public class SayHelloActivities
7+
{
8+
[Activity]
9+
public string SayHello(string name)
10+
{
11+
ActivityExecutionContext.Current.Logger.LogInformation(
12+
"Activity called by user {UserId}",
13+
MyContext.UserId.Value);
14+
return $"Hello, {name}!";
15+
}
16+
}

‎src/ContextPropagation/SayHelloWorkflow.workflow.cs‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
namespace TemporalioSamples.ContextPropagation;
2+
3+
using Microsoft.Extensions.Logging;
4+
using Temporalio.Workflows;
5+
6+
[Workflow]
7+
public class SayHelloWorkflow
8+
{
9+
private bool complete;
10+
11+
[WorkflowRun]
12+
public async Task<string> RunAsync(string name)
13+
{
14+
Workflow.Logger.LogInformation(
15+
"Workflow called by user {UserId}",
16+
MyContext.UserId.Value);
17+
18+
// Wait for signal then run activity
19+
await Workflow.WaitConditionAsync(() => complete);
20+
return await Workflow.ExecuteActivityAsync(
21+
(SayHelloActivities act) => act.SayHello(name),
22+
new() { StartToCloseTimeout = TimeSpan.FromMinutes(5) });
23+
}
24+
25+
[WorkflowSignal]
26+
public async Task SignalCompleteAsync()
27+
{
28+
Workflow.Logger.LogInformation(
29+
"Signal called by user {UserId}",
30+
MyContext.UserId.Value);
31+
complete = true;
32+
}
33+
34+
[WorkflowQuery]
35+
public bool IsComplete()
36+
{
37+
Workflow.Logger.LogInformation(
38+
"Query called by user {UserId}",
39+
MyContext.UserId.Value);
40+
return complete;
41+
}
42+
}

0 commit comments

Comments
 (0)