Skip to content

Commit 0cacb5d

Browse files
rrossrross
authored
Added Counter Interceptor Sample (#77)
1 parent 175feb5 commit 0cacb5d

File tree

12 files changed

+445
-0
lines changed

12 files changed

+445
-0
lines changed

‎README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ Prerequisites:
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.
2020
* [ContextPropagation](src/ContextPropagation) - Context propagation via interceptors.
21+
* [CounterInterceptor](src/CounterInterceptor/) - Simple Workflow and Client Interceptors example.
2122
* [DependencyInjection](src/DependencyInjection) - How to inject dependencies in activities and use generic hosts for workers
2223
* [Encryption](src/Encryption) - End-to-end encryption with Temporal payload codecs.
2324
* [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 & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.ContextPr
5757
EndProject
5858
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.SafeMessageHandlers", "src\SafeMessageHandlers\TemporalioSamples.SafeMessageHandlers.csproj", "{FAF5984A-5B1C-4686-B056-A4F2AC0E8EB7}"
5959
EndProject
60+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.CounterInterceptor", "src\CounterInterceptor\TemporalioSamples.CounterInterceptor.csproj", "{F9C44936-8BF9-4919-BB66-8F1888E22AEB}"
61+
EndProject
6062
Global
6163
GlobalSection(SolutionConfigurationPlatforms) = preSolution
6264
Debug|Any CPU = Debug|Any CPU
@@ -155,6 +157,10 @@ Global
155157
{FAF5984A-5B1C-4686-B056-A4F2AC0E8EB7}.Debug|Any CPU.Build.0 = Debug|Any CPU
156158
{FAF5984A-5B1C-4686-B056-A4F2AC0E8EB7}.Release|Any CPU.ActiveCfg = Release|Any CPU
157159
{FAF5984A-5B1C-4686-B056-A4F2AC0E8EB7}.Release|Any CPU.Build.0 = Release|Any CPU
160+
{F9C44936-8BF9-4919-BB66-8F1888E22AEB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
161+
{F9C44936-8BF9-4919-BB66-8F1888E22AEB}.Debug|Any CPU.Build.0 = Debug|Any CPU
162+
{F9C44936-8BF9-4919-BB66-8F1888E22AEB}.Release|Any CPU.ActiveCfg = Release|Any CPU
163+
{F9C44936-8BF9-4919-BB66-8F1888E22AEB}.Release|Any CPU.Build.0 = Release|Any CPU
158164
EndGlobalSection
159165
GlobalSection(SolutionProperties) = preSolution
160166
HideSolutionNode = FALSE
@@ -185,5 +191,6 @@ Global
185191
{B3DB7B8C-7BD3-4A53-A809-AB6279B1A630} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
186192
{7B797D20-485F-441D-8E71-AF7E315FA9CF} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
187193
{FAF5984A-5B1C-4686-B056-A4F2AC0E8EB7} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
194+
{F9C44936-8BF9-4919-BB66-8F1888E22AEB} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
188195
EndGlobalSection
189196
EndGlobal

‎src/CounterInterceptor/Counts.cs‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
public class Counts
3+
{
4+
private uint clientExecutions;
5+
private uint clientQueries;
6+
private uint clientSignals;
7+
private uint workflowReplays;
8+
private uint workflowSignals;
9+
private uint workflowQueries;
10+
private uint workflowChildExecutions;
11+
private uint workflowActivityExecutions;
12+
13+
public ref uint ClientExecutions => ref clientExecutions;
14+
15+
public ref uint ClientSignals => ref clientSignals;
16+
17+
public ref uint ClientQueries => ref clientQueries;
18+
19+
public string ClientInfo() =>
20+
$"\n\tTotal Number of Workflow Exec: {ClientExecutions}\n\t" +
21+
$"Total Number of Signals: {ClientSignals}\n\t" +
22+
$"Total Number of Queries: {ClientQueries}";
23+
24+
public ref uint WorkflowReplays => ref workflowReplays;
25+
26+
public ref uint WorkflowSignals => ref workflowSignals;
27+
28+
public ref uint WorkflowQueries => ref workflowQueries;
29+
30+
public ref uint WorkflowChildExecutions => ref workflowChildExecutions;
31+
32+
public ref uint WorkflowActivityExecutions => ref workflowActivityExecutions;
33+
34+
public string WorkflowInfo() =>
35+
$"\n\tTotal Number of Workflow Replays: {WorkflowReplays}\n\t" +
36+
$"Total Number of Child Workflow Exec: {WorkflowChildExecutions}\n\t" +
37+
$"Total Number of Activity Exec: {WorkflowActivityExecutions}\n\t" +
38+
$"Total Number of Signals: {WorkflowSignals}\n\t" +
39+
$"Total Number of Queries: {WorkflowQueries}";
40+
41+
public override string ToString() =>
42+
ClientInfo() + WorkflowInfo();
43+
}

‎src/CounterInterceptor/MyActivities.cs‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
3+
using System.Diagnostics;
4+
using Temporalio.Activities;
5+
6+
public class MyActivities
7+
{
8+
[Activity]
9+
public string SayHello(string name, string title) =>
10+
$"Hello {title} {name}";
11+
12+
[Activity]
13+
public string SayGoodBye(string name, string title) =>
14+
$"Goodby {title} {name}";
15+
}

‎src/CounterInterceptor/MyChildWorkflow.workflow.cs‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
3+
using Temporalio.Workflows;
4+
5+
[Workflow]
6+
public class MyChildWorkflow
7+
{
8+
private readonly ActivityOptions activityOptions = new()
9+
{
10+
StartToCloseTimeout = TimeSpan.FromSeconds(10),
11+
};
12+
13+
[WorkflowRun]
14+
public async Task<string> RunAsync(string name, string title) =>
15+
await Workflow.ExecuteActivityAsync((MyActivities act) => act.SayHello(name, title), activityOptions) +
16+
await Workflow.ExecuteActivityAsync((MyActivities act) => act.SayGoodBye(name, title), activityOptions);
17+
}

‎src/CounterInterceptor/MyCounterInterceptor.cs‎

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
3+
using System.Collections.Concurrent;
4+
using Temporalio.Activities;
5+
using Temporalio.Client;
6+
using Temporalio.Client.Interceptors;
7+
using Temporalio.Worker.Interceptors;
8+
using Temporalio.Workflows;
9+
10+
public class MyCounterInterceptor : IClientInterceptor, IWorkerInterceptor
11+
{
12+
public ConcurrentDictionary<string, Counts> Counts { get; } = new();
13+
14+
public string WorkerInfo() =>
15+
string.Join(
16+
"\n",
17+
Counts.Select(kvp => $"** Workflow ID: {kvp.Key} {kvp.Value.WorkflowInfo()}"));
18+
19+
public string ClientInfo() =>
20+
string.Join(
21+
"\n",
22+
Counts.Select(kvp => $"** Workflow ID: {kvp.Key} {kvp.Value.ClientInfo()}"));
23+
24+
public ClientOutboundInterceptor InterceptClient(ClientOutboundInterceptor nextInterceptor) =>
25+
new ClientOutbound(this, nextInterceptor);
26+
27+
public WorkflowInboundInterceptor InterceptWorkflow(WorkflowInboundInterceptor nextInterceptor) =>
28+
new WorkflowInbound(this, nextInterceptor);
29+
30+
public ActivityInboundInterceptor InterceptActivity(ActivityInboundInterceptor nextInterceptor) =>
31+
new ActivityInbound(this, nextInterceptor);
32+
33+
private void Increment(string id, Action<Counts> increment) =>
34+
increment(Counts.GetOrAdd(id, _ => new()));
35+
36+
private sealed class ClientOutbound : ClientOutboundInterceptor
37+
{
38+
private MyCounterInterceptor root;
39+
40+
public ClientOutbound(MyCounterInterceptor root, ClientOutboundInterceptor next)
41+
: base(next) => this.root = root;
42+
43+
public override Task<WorkflowHandle<TWorkflow, TResult>> StartWorkflowAsync<TWorkflow, TResult>(
44+
StartWorkflowInput input)
45+
{
46+
var id = input.Options.Id ?? "None";
47+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].ClientExecutions));
48+
return base.StartWorkflowAsync<TWorkflow, TResult>(input);
49+
}
50+
51+
public override Task SignalWorkflowAsync(SignalWorkflowInput input)
52+
{
53+
var id = input.Id;
54+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].ClientSignals));
55+
return base.SignalWorkflowAsync(input);
56+
}
57+
58+
public override Task<TResult> QueryWorkflowAsync<TResult>(QueryWorkflowInput input)
59+
{
60+
var id = input.Id;
61+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].ClientQueries));
62+
return base.QueryWorkflowAsync<TResult>(input);
63+
}
64+
}
65+
66+
private sealed class WorkflowInbound : WorkflowInboundInterceptor
67+
{
68+
private readonly MyCounterInterceptor root;
69+
70+
internal WorkflowInbound(MyCounterInterceptor root, WorkflowInboundInterceptor next)
71+
: base(next) => this.root = root;
72+
73+
public override void Init(WorkflowOutboundInterceptor outbound) =>
74+
base.Init(new WorkflowOutbound(root, outbound));
75+
76+
public override Task<object?> ExecuteWorkflowAsync(ExecuteWorkflowInput input)
77+
{
78+
// Count only if we're not replaying
79+
if (!Workflow.Unsafe.IsReplaying)
80+
{
81+
var id = Workflow.Info.WorkflowId;
82+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].WorkflowReplays));
83+
}
84+
return base.ExecuteWorkflowAsync(input);
85+
}
86+
87+
public override Task HandleSignalAsync(HandleSignalInput input)
88+
{
89+
// Count only if we're not replaying
90+
if (!Workflow.Unsafe.IsReplaying)
91+
{
92+
var id = Workflow.Info.WorkflowId;
93+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].WorkflowSignals));
94+
}
95+
return base.HandleSignalAsync(input);
96+
}
97+
98+
public override object? HandleQuery(HandleQueryInput input)
99+
{
100+
// Count only if we're not replaying
101+
if (!Workflow.Unsafe.IsReplaying)
102+
{
103+
var id = Workflow.Info.WorkflowId;
104+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].WorkflowQueries));
105+
}
106+
return base.HandleQuery(input);
107+
}
108+
}
109+
110+
private sealed class WorkflowOutbound : WorkflowOutboundInterceptor
111+
{
112+
private readonly MyCounterInterceptor root;
113+
114+
internal WorkflowOutbound(MyCounterInterceptor root, WorkflowOutboundInterceptor next)
115+
: base(next) => this.root = root;
116+
117+
public override Task<ChildWorkflowHandle<TWorkflow, TResult>> StartChildWorkflowAsync<TWorkflow, TResult>(
118+
StartChildWorkflowInput input)
119+
{
120+
// Count only if we're not replaying
121+
if (!Workflow.Unsafe.IsReplaying)
122+
{
123+
var id = Workflow.Info.WorkflowId;
124+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].WorkflowChildExecutions));
125+
}
126+
return base.StartChildWorkflowAsync<TWorkflow, TResult>(input);
127+
}
128+
}
129+
130+
private sealed class ActivityInbound : ActivityInboundInterceptor
131+
{
132+
private readonly MyCounterInterceptor root;
133+
134+
internal ActivityInbound(MyCounterInterceptor root, ActivityInboundInterceptor next)
135+
: base(next) => this.root = root;
136+
137+
public override Task<object?> ExecuteActivityAsync(ExecuteActivityInput input)
138+
{
139+
var id = ActivityExecutionContext.Current.Info.WorkflowId;
140+
root.Increment(id, c => Interlocked.Increment(ref root.Counts[id].WorkflowActivityExecutions));
141+
return base.ExecuteActivityAsync(input);
142+
}
143+
}
144+
}

‎src/CounterInterceptor/MyWorkflow.workflow.cs‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
3+
using Temporalio.Workflows;
4+
5+
[Workflow]
6+
public class MyWorkflow
7+
{
8+
private bool exit; // Automatically defaults to false
9+
10+
[WorkflowRun]
11+
public async Task<string> RunAsync()
12+
{
13+
// Wait for greeting info
14+
await Workflow.WaitConditionAsync(() =>
15+
!string.IsNullOrEmpty(Name) && !string.IsNullOrEmpty(Title));
16+
17+
// Execute Child Workflow
18+
var result = await Workflow.ExecuteChildWorkflowAsync(
19+
(MyChildWorkflow wf) => wf.RunAsync(Name, Title),
20+
new() { Id = "counter-interceptor-child" });
21+
22+
// Wait for exit signal
23+
await Workflow.WaitConditionAsync(() => exit);
24+
25+
return result;
26+
}
27+
28+
[WorkflowSignal]
29+
public async Task SignalNameAndTitleAsync(string name, string title)
30+
{
31+
Name = name;
32+
Title = title;
33+
}
34+
35+
[WorkflowQuery]
36+
public string Name { get; private set; } = string.Empty;
37+
38+
[WorkflowQuery]
39+
public string Title { get; private set; } = string.Empty;
40+
41+
[WorkflowSignal]
42+
public async Task ExitAsync() => exit = true;
43+
}

‎src/CounterInterceptor/Program.cs‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
namespace TemporalioSamples.CounterInterceptor;
2+
3+
using Temporalio.Client;
4+
using Temporalio.Worker;
5+
6+
internal class Program
7+
{
8+
private static async Task Main(string[] args)
9+
{
10+
var counterInterceptor = new MyCounterInterceptor();
11+
var client = await TemporalClient.ConnectAsync(
12+
options: new("localhost:7233")
13+
{
14+
Interceptors = new[]
15+
{
16+
counterInterceptor,
17+
},
18+
});
19+
20+
var activities = new MyActivities();
21+
22+
var taskQueue = "CounterInterceptorTaskQueue";
23+
24+
var workerOptions = new TemporalWorkerOptions(taskQueue).
25+
AddAllActivities(activities).
26+
AddWorkflow<MyWorkflow>().
27+
AddWorkflow<MyChildWorkflow>();
28+
29+
// workerOptions.Interceptors = new[] { counterInterceptor };
30+
using var worker = new TemporalWorker(
31+
client,
32+
workerOptions);
33+
34+
// Run worker until cancelled
35+
Console.WriteLine("Running worker...");
36+
37+
// Start the workers
38+
await worker.ExecuteAsync(async () =>
39+
{
40+
// Start the workflow
41+
var handle = await client.StartWorkflowAsync(
42+
(MyWorkflow wf) => wf.RunAsync(),
43+
new(id: Guid.NewGuid().ToString(), taskQueue: taskQueue));
44+
45+
Console.WriteLine("Sending name and title to workflow");
46+
await handle.SignalAsync(wf => wf.SignalNameAndTitleAsync("John", "Customer"));
47+
48+
var name = await handle.QueryAsync(wf => wf.Name);
49+
var title = await handle.QueryAsync(wf => wf.Title);
50+
51+
// Send exit signal to workflow
52+
await handle.SignalAsync(wf => wf.ExitAsync());
53+
54+
var result = await handle.GetResultAsync();
55+
56+
Console.WriteLine($"Workflow result is {result}");
57+
58+
Console.WriteLine("Query results: ");
59+
Console.WriteLine($"\tName: {name}");
60+
Console.WriteLine($"\tTitle: {title}");
61+
62+
// Print worker counter info
63+
Console.WriteLine("\nCollected Worker Counter Info:\n");
64+
Console.WriteLine(counterInterceptor.WorkerInfo());
65+
Console.WriteLine($"Number of unique workflows: {counterInterceptor.Counts.Count}");
66+
67+
// Print client counter info
68+
Console.WriteLine();
69+
Console.WriteLine("Collected Client Counter Info:\n");
70+
Console.WriteLine(counterInterceptor.ClientInfo());
71+
});
72+
}
73+
}

‎src/CounterInterceptor/README.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# dotnet-counter-interceptor
2+
The sample demonstrates:
3+
- the use of a Worker Workflow Interceptor that counts the number of Workflow Executions, Child Workflow Executions, and Activity Executions and the number of Signals and Queries. It is based
4+
off of the [Java sample](https://github.com/temporalio/samples-java/tree/main) located [here](https://github.com/temporalio/samples-java/tree/main/core/src/main/java/io/temporal/samples/countinterceptor)
5+
- the use of a Client Workflow Interceptor that counts the number of Workflow Executions and the number of Signals and Queries.
6+
7+
To run, first see [README.md](https://github.com/temporalio/samples-dotnet/blob/main/README.md) for prerequisites
8+
9+
## Run Worker and Client
10+
```bash
11+
# make sure you have temporal server running
12+
dotnet run
13+
```

0 commit comments

Comments
 (0)