Skip to content

Commit db42f91

Browse files
danielgerlagdanielgerlag
authored
Merge pull request #775 from glucaci/postMiddleware
Add execute middleware phase
2 parents 674929f + 045cb49 commit db42f91

File tree

9 files changed

+303
-48
lines changed

9 files changed

+303
-48
lines changed

‎ReleaseNotes/3.4.0.md‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
# Workflow Core 3.4.0
2+
3+
## Execute Workflow Middleware
4+
5+
These middleware get run after each workflow execution and can be used to perform additional actions or build metrics/statistics for all workflows in your app.
6+
7+
The following example illustrates how you can use a execute workflow middleware to build [prometheus](https://prometheus.io/) metrics.
8+
9+
Note that you use `WorkflowMiddlewarePhase.ExecuteWorkflow` to specify that it runs after each workflow execution.
10+
11+
**Important:** You should call `next` as part of the workflow middleware to ensure that the next workflow in the chain runs.
12+
13+
```cs
14+
public class MetricsMiddleware : IWorkflowMiddleware
15+
{
16+
private readonly ConcurrentHashSet<string>() _suspendedWorkflows =
17+
new ConcurrentHashSet<string>();
18+
19+
private readonly Counter _completed;
20+
private readonly Counter _suspended;
21+
22+
public MetricsMiddleware()
23+
{
24+
_completed = Prometheus.Metrics.CreateCounter(
25+
"workflow_completed", "Workflow completed");
26+
27+
_suspended = Prometheus.Metrics.CreateCounter(
28+
"workflow_suspended", "Workflow suspended");
29+
}
30+
31+
public WorkflowMiddlewarePhase Phase =>
32+
WorkflowMiddlewarePhase.ExecuteWorkflow;
33+
34+
public Task HandleAsync(
35+
WorkflowInstance workflow,
36+
WorkflowDelegate next)
37+
{
38+
switch (workflow.Status)
39+
{
40+
case WorkflowStatus.Complete:
41+
if (_suspendedWorkflows.TryRemove(workflow.Id))
42+
{
43+
_suspended.Dec();
44+
}
45+
_completed.Inc();
46+
break;
47+
case WorkflowStatus.Suspended:
48+
_suspended.Inc();
49+
break;
50+
}
51+
52+
return next();
53+
}
54+
}
55+
```

‎src/WorkflowCore/Interface/IWorkflowMiddleware.cs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,12 @@ public enum WorkflowMiddlewarePhase
1616
/// <summary>
1717
/// The middleware should run after a workflow completes.
1818
/// </summary>
19-
PostWorkflow
19+
PostWorkflow,
20+
21+
/// <summary>
22+
/// The middleware should run after each workflow execution.
23+
/// </summary>
24+
ExecuteWorkflow
2025
}
2126

2227
/// <summary>

‎src/WorkflowCore/Interface/IWorkflowMiddlewareRunner.cs‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
namespace WorkflowCore.Interface
55
{
66
/// <summary>
7-
/// Runs workflow pre/post middleware.
7+
/// Runs workflow pre/post and execute middleware.
88
/// </summary>
99
public interface IWorkflowMiddlewareRunner
1010
{
@@ -29,5 +29,16 @@ public interface IWorkflowMiddlewareRunner
2929
/// <param name="def">The <see cref="WorkflowDefinition"/> definition.</param>
3030
/// <returns>A task that will complete when all middleware has run.</returns>
3131
Task RunPostMiddleware(WorkflowInstance workflow, WorkflowDefinition def);
32+
33+
/// <summary>
34+
/// Runs workflow-level middleware that is set to run at the
35+
/// <see cref="WorkflowMiddlewarePhase.ExecuteWorkflow"/> phase. Middleware will be run in the
36+
/// order in which they were registered with DI with middleware declared earlier starting earlier and
37+
/// completing later.
38+
/// </summary>
39+
/// <param name="workflow">The <see cref="WorkflowInstance"/> to run for.</param>
40+
/// <param name="def">The <see cref="WorkflowDefinition"/> definition.</param>
41+
/// <returns>A task that will complete when all middleware has run.</returns>
42+
Task RunExecuteMiddleware(WorkflowInstance workflow, WorkflowDefinition def);
3243
}
3344
}

‎src/WorkflowCore/Models/WorkflowDefinition.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ public class WorkflowDefinition
1818
public WorkflowErrorHandling DefaultErrorBehavior { get; set; }
1919

2020
public Type OnPostMiddlewareError { get; set; }
21+
public Type OnExecuteMiddlewareError { get; set; }
2122

2223
public TimeSpan? DefaultErrorRetryInterval { get; set; }
2324

‎src/WorkflowCore/Services/WorkflowExecutor.cs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,12 @@ public async Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow, Can
9999
ProcessAfterExecutionIteration(workflow, def, wfResult);
100100
await DetermineNextExecutionTime(workflow, def);
101101

102+
using (var scope = _serviceProvider.CreateScope())
103+
{
104+
var middlewareRunner = scope.ServiceProvider.GetRequiredService<IWorkflowMiddlewareRunner>();
105+
await middlewareRunner.RunExecuteMiddleware(workflow, def);
106+
}
107+
102108
return wfResult;
103109
}
104110

@@ -213,7 +219,9 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
213219
workflow.NextExecution = null;
214220

215221
if (workflow.Status == WorkflowStatus.Complete)
222+
{
216223
return;
224+
}
217225

218226
foreach (var pointer in workflow.ExecutionPointers.Where(x => x.Active && (x.Children ?? new List<string>()).Count == 0))
219227
{
@@ -243,7 +251,9 @@ private async Task DetermineNextExecutionTime(WorkflowInstance workflow, Workflo
243251
}
244252

245253
if ((workflow.NextExecution != null) || (workflow.ExecutionPointers.Any(x => x.EndTime == null)))
254+
{
246255
return;
256+
}
247257

248258
workflow.Status = WorkflowStatus.Complete;
249259
workflow.CompleteTime = _datetimeProvider.UtcNow;

‎src/WorkflowCore/Services/WorkflowMiddlewareRunner.cs‎

Lines changed: 35 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
namespace WorkflowCore.Services
1010
{
11-
/// <inheritdoc />
11+
/// <inheritdoc cref="IWorkflowMiddlewareRunner" />
1212
public class WorkflowMiddlewareRunner : IWorkflowMiddlewareRunner
1313
{
1414
private static readonly WorkflowDelegate NoopWorkflowDelegate = () => Task.CompletedTask;
@@ -17,80 +17,74 @@ public class WorkflowMiddlewareRunner : IWorkflowMiddlewareRunner
1717

1818
public WorkflowMiddlewareRunner(
1919
IEnumerable<IWorkflowMiddleware> middleware,
20-
IServiceProvider serviceProvider
21-
)
20+
IServiceProvider serviceProvider)
2221
{
2322
_middleware = middleware;
2423
_serviceProvider = serviceProvider;
2524
}
2625

27-
28-
/// <summary>
29-
/// Runs workflow-level middleware that is set to run at the
30-
/// <see cref="WorkflowMiddlewarePhase.PreWorkflow"/> phase. Middleware will be run in the
31-
/// order in which they were registered with DI with middleware declared earlier starting earlier and
32-
/// completing later.
33-
/// </summary>
34-
/// <param name="workflow">The <see cref="WorkflowInstance"/> to run for.</param>
35-
/// <param name="def">The <see cref="WorkflowDefinition"/> definition.</param>
36-
/// <returns>A task that will complete when all middleware has run.</returns>
26+
/// <inheritdoc cref="IWorkflowMiddlewareRunner.RunPreMiddleware"/>
3727
public async Task RunPreMiddleware(WorkflowInstance workflow, WorkflowDefinition def)
3828
{
3929
var preMiddleware = _middleware
40-
.Where(m => m.Phase == WorkflowMiddlewarePhase.PreWorkflow)
41-
.ToArray();
30+
.Where(m => m.Phase == WorkflowMiddlewarePhase.PreWorkflow);
4231

4332
await RunWorkflowMiddleware(workflow, preMiddleware);
4433
}
4534

46-
/// <summary>
47-
/// Runs workflow-level middleware that is set to run at the
48-
/// <see cref="WorkflowMiddlewarePhase.PostWorkflow"/> phase. Middleware will be run in the
49-
/// order in which they were registered with DI with middleware declared earlier starting earlier and
50-
/// completing later.
51-
/// </summary>
52-
/// <param name="workflow">The <see cref="WorkflowInstance"/> to run for.</param>
53-
/// <param name="def">The <see cref="WorkflowDefinition"/> definition.</param>
54-
/// <returns>A task that will complete when all middleware has run.</returns>
55-
public async Task RunPostMiddleware(WorkflowInstance workflow, WorkflowDefinition def)
35+
/// <inheritdoc cref="IWorkflowMiddlewareRunner.RunPostMiddleware"/>
36+
public Task RunPostMiddleware(WorkflowInstance workflow, WorkflowDefinition def)
5637
{
57-
var postMiddleware = _middleware
58-
.Where(m => m.Phase == WorkflowMiddlewarePhase.PostWorkflow)
59-
.ToArray();
38+
return RunWorkflowMiddlewareWithErrorHandling(
39+
workflow,
40+
WorkflowMiddlewarePhase.PostWorkflow,
41+
def.OnPostMiddlewareError);
42+
}
43+
44+
/// <inheritdoc cref="IWorkflowMiddlewareRunner.RunExecuteMiddleware"/>
45+
public Task RunExecuteMiddleware(WorkflowInstance workflow, WorkflowDefinition def)
46+
{
47+
return RunWorkflowMiddlewareWithErrorHandling(
48+
workflow,
49+
WorkflowMiddlewarePhase.ExecuteWorkflow,
50+
def.OnExecuteMiddlewareError);
51+
}
52+
53+
public async Task RunWorkflowMiddlewareWithErrorHandling(
54+
WorkflowInstance workflow,
55+
WorkflowMiddlewarePhase phase,
56+
Type middlewareErrorType)
57+
{
58+
var middleware = _middleware.Where(m => m.Phase == phase);
6059

6160
try
6261
{
63-
await RunWorkflowMiddleware(workflow, postMiddleware);
62+
await RunWorkflowMiddleware(workflow, middleware);
6463
}
6564
catch (Exception exception)
6665
{
67-
// On error, determine which error handler to run and then run it
68-
var errorHandlerType = def.OnPostMiddlewareError ?? typeof(IWorkflowMiddlewareErrorHandler);
66+
var errorHandlerType = middlewareErrorType ?? typeof(IWorkflowMiddlewareErrorHandler);
67+
6968
using (var scope = _serviceProvider.CreateScope())
7069
{
7170
var typeInstance = scope.ServiceProvider.GetService(errorHandlerType);
72-
if (typeInstance != null && typeInstance is IWorkflowMiddlewareErrorHandler handler)
71+
if (typeInstance is IWorkflowMiddlewareErrorHandler handler)
7372
{
7473
await handler.HandleAsync(exception);
7574
}
7675
}
7776
}
7877
}
7978

80-
private static async Task RunWorkflowMiddleware(
79+
private static Task RunWorkflowMiddleware(
8180
WorkflowInstance workflow,
82-
IEnumerable<IWorkflowMiddleware> middlewareCollection
83-
)
81+
IEnumerable<IWorkflowMiddleware> middlewareCollection)
8482
{
85-
// Build the middleware chain
86-
var middlewareChain = middlewareCollection
83+
return middlewareCollection
8784
.Reverse()
8885
.Aggregate(
8986
NoopWorkflowDelegate,
90-
(previous, middleware) => () => middleware.HandleAsync(workflow, previous)
91-
);
92-
93-
await middlewareChain();
87+
(previous, middleware) => () => middleware.HandleAsync(workflow, previous))();
9488
}
9589
}
9690
}

‎src/WorkflowCore/WorkflowCore.csproj‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,12 @@
1515
<GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
1616
<GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
1717
<Description>Workflow Core is a light weight workflow engine targeting .NET Standard.</Description>
18-
<Version>3.3.6</Version>
19-
<AssemblyVersion>3.3.6.0</AssemblyVersion>
20-
<FileVersion>3.3.6.0</FileVersion>
18+
<Version>3.4.0</Version>
19+
<AssemblyVersion>3.4.0.0</AssemblyVersion>
20+
<FileVersion>3.4.0.0</FileVersion>
2121
<PackageReleaseNotes></PackageReleaseNotes>
2222
<PackageIconUrl>https://github.com/danielgerlag/workflow-core/raw/master/src/logo.png</PackageIconUrl>
23-
<PackageVersion>3.3.6</PackageVersion>
23+
<PackageVersion>3.4.0</PackageVersion>
2424
</PropertyGroup>
2525

2626
<ItemGroup>

‎test/WorkflowCore.UnitTests/Services/WorkflowExecutorFixture.cs‎

Lines changed: 69 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,14 @@ public WorkflowExecutorFixture()
4444

4545
Options = new WorkflowOptions(A.Fake<IServiceCollection>());
4646

47+
var stepExecutionScope = A.Fake<IServiceScope>();
48+
A.CallTo(() => ScopeProvider.CreateScope(A<IStepExecutionContext>._)).Returns(stepExecutionScope);
49+
A.CallTo(() => stepExecutionScope.ServiceProvider).Returns(ServiceProvider);
50+
4751
var scope = A.Fake<IServiceScope>();
48-
A.CallTo(() => ScopeProvider.CreateScope(A<IStepExecutionContext>._)).Returns(scope);
52+
var scopeFactory = A.Fake<IServiceScopeFactory>();
53+
A.CallTo(() => ServiceProvider.GetService(typeof(IServiceScopeFactory))).Returns(scopeFactory);
54+
A.CallTo(() => scopeFactory.CreateScope()).Returns(scope);
4955
A.CallTo(() => scope.ServiceProvider).Returns(ServiceProvider);
5056

5157
A.CallTo(() => DateTimeProvider.Now).Returns(DateTime.Now);
@@ -63,6 +69,10 @@ public WorkflowExecutorFixture()
6369
.RunPostMiddleware(A<WorkflowInstance>._, A<WorkflowDefinition>._))
6470
.Returns(Task.CompletedTask);
6571

72+
A.CallTo(() => MiddlewareRunner
73+
.RunExecuteMiddleware(A<WorkflowInstance>._, A<WorkflowDefinition>._))
74+
.Returns(Task.CompletedTask);
75+
6676
A.CallTo(() => StepExecutor.ExecuteStep(A<IStepExecutionContext>._, A<IStepBody>._))
6777
.ReturnsLazily(call =>
6878
call.Arguments[1].As<IStepBody>().RunAsync(
@@ -105,6 +115,64 @@ public void should_execute_active_step()
105115
A.CallTo(() => ResultProcesser.ProcessExecutionResult(instance, A<WorkflowDefinition>.Ignored, A<ExecutionPointer>.Ignored, step1, A<ExecutionResult>.Ignored, A<WorkflowExecutorResult>.Ignored)).MustHaveHappened();
106116
}
107117

118+
[Fact(DisplayName = "Should call execute middleware when not completed")]
119+
public void should_call_execute_middleware_when_not_completed()
120+
{
121+
//arrange
122+
var step1Body = A.Fake<IStepBody>();
123+
A.CallTo(() => step1Body.RunAsync(A<IStepExecutionContext>.Ignored)).Returns(ExecutionResult.Next());
124+
WorkflowStep step1 = BuildFakeStep(step1Body);
125+
Given1StepWorkflow(step1, "Workflow", 1);
126+
127+
var instance = new WorkflowInstance
128+
{
129+
WorkflowDefinitionId = "Workflow",
130+
Version = 1,
131+
Status = WorkflowStatus.Runnable,
132+
NextExecution = 0,
133+
Id = "001",
134+
ExecutionPointers = new ExecutionPointerCollection(new List<ExecutionPointer>
135+
{
136+
new ExecutionPointer { Id = "1", Active = true, StepId = 0 }
137+
})
138+
};
139+
140+
//act
141+
Subject.Execute(instance);
142+
143+
//assert
144+
A.CallTo(() => MiddlewareRunner.RunExecuteMiddleware(instance, A<WorkflowDefinition>.Ignored)).MustHaveHappened();
145+
}
146+
147+
[Fact(DisplayName = "Should not call post middleware when not completed")]
148+
public void should_not_call_post_middleware_when_not_completed()
149+
{
150+
//arrange
151+
var step1Body = A.Fake<IStepBody>();
152+
A.CallTo(() => step1Body.RunAsync(A<IStepExecutionContext>.Ignored)).Returns(ExecutionResult.Next());
153+
WorkflowStep step1 = BuildFakeStep(step1Body);
154+
Given1StepWorkflow(step1, "Workflow", 1);
155+
156+
var instance = new WorkflowInstance
157+
{
158+
WorkflowDefinitionId = "Workflow",
159+
Version = 1,
160+
Status = WorkflowStatus.Runnable,
161+
NextExecution = 0,
162+
Id = "001",
163+
ExecutionPointers = new ExecutionPointerCollection(new List<ExecutionPointer>
164+
{
165+
new ExecutionPointer { Id = "1", Active = true, StepId = 0 }
166+
})
167+
};
168+
169+
//act
170+
Subject.Execute(instance);
171+
172+
//assert
173+
A.CallTo(() => MiddlewareRunner.RunPostMiddleware(instance, A<WorkflowDefinition>.Ignored)).MustNotHaveHappened();
174+
}
175+
108176
[Fact(DisplayName = "Should trigger step hooks")]
109177
public void should_trigger_step_hooks()
110178
{

0 commit comments

Comments
 (0)