-
Notifications
You must be signed in to change notification settings - Fork 46
/
ScenarioWithEndpointBehaviorExtensions.cs
156 lines (134 loc) · 5.67 KB
/
ScenarioWithEndpointBehaviorExtensions.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
namespace ServiceControl.AcceptanceTesting
{
using System;
using System.Runtime.ExceptionServices;
using System.Threading;
using System.Threading.Tasks;
using NServiceBus;
using NServiceBus.AcceptanceTesting;
using NServiceBus.AcceptanceTesting.Support;
public static class ScenarioWithEndpointBehaviorExtensions
{
public static IScenarioWithEndpointBehavior<TContext> Done<TContext>(this IScenarioWithEndpointBehavior<TContext> endpointBehavior, Func<TContext, Task<bool>> func) where TContext : ScenarioContext
{
var behavior = new ServiceControlClient<TContext>(func);
return endpointBehavior.WithComponent(behavior).Done(ctx => behavior.Done);
}
public static EndpointBehaviorBuilder<TContext> When<TContext>(this EndpointBehaviorBuilder<TContext> endpointBehavior, Func<TContext, Task<bool>> predicate, Func<IMessageSession, TContext, Task> action) where TContext : ScenarioContext => endpointBehavior.When(ctx => predicate(ctx).GetAwaiter().GetResult(), action);
public static SequenceBuilder<TContext> Do<TContext>(
this IScenarioWithEndpointBehavior<TContext> endpointBehavior, string step,
Func<TContext, Task<bool>> handler) where TContext : ScenarioContext, ISequenceContext =>
new(endpointBehavior, step, handler);
}
public class SequenceBuilder<TContext>
where TContext : ScenarioContext, ISequenceContext
{
public SequenceBuilder(IScenarioWithEndpointBehavior<TContext> endpointBehavior, string step, Func<TContext, Task<bool>> handler)
{
this.endpointBehavior = endpointBehavior;
sequence.Do(step, handler);
}
public SequenceBuilder<TContext> Do(string step, Func<TContext, Task<bool>> handler)
{
sequence.Do(step, handler);
return this;
}
public SequenceBuilder<TContext> Do(string step, Func<TContext, Task> handler)
{
sequence.Do(step, handler);
return this;
}
public IScenarioWithEndpointBehavior<TContext> Done(Func<TContext, bool> doneCriteria = null)
{
var behavior = new ServiceControlClient<TContext>(context => sequence.Continue(context));
return endpointBehavior.WithComponent(behavior).Done(async ctx =>
{
if (!behavior.Done && !sequence.IsFinished(ctx))
{
return false;
}
if (doneCriteria == null || doneCriteria(ctx))
{
return true;
}
// If sequence is done but test is not finished, small delay to avoid tight loop check
await Task.Delay(250);
// If sequence is not finished immediately return false, since each step will enforce delays
return false;
});
}
IScenarioWithEndpointBehavior<TContext> endpointBehavior;
Sequence<TContext> sequence = new Sequence<TContext>();
}
public class ServiceControlClient<TContext>(Func<TContext, Task<bool>> checkDone) : IComponentBehavior
where TContext : ScenarioContext
{
public bool Done
{
get
{
exceptionInfo?.Throw();
return isDone;
}
}
public Task<ComponentRunner> CreateRunner(RunDescriptor run) => Task.FromResult<ComponentRunner>(new Runner(checkDone, () => isDone = true, info => exceptionInfo = info, (TContext)run.ScenarioContext));
volatile ExceptionDispatchInfo exceptionInfo;
volatile bool isDone;
class Runner(
Func<TContext, Task<bool>> isDone,
Action setDone,
Action<ExceptionDispatchInfo> setException,
TContext scenarioContext)
: ComponentRunner
{
public override string Name => "ServiceControlClient";
public override Task Start(CancellationToken cancellationToken = default) => Task.CompletedTask;
public override Task ComponentsStarted(CancellationToken cancellationToken = default)
{
tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
checkTask = Task.Run(async () =>
{
try
{
while (!tokenSource.IsCancellationRequested)
{
if (await isDone(scenarioContext))
{
setDone();
return;
}
await Task.Delay(100, tokenSource.Token);
}
}
catch (Exception e)
{
setException(ExceptionDispatchInfo.Capture(e));
}
}, tokenSource.Token);
return Task.CompletedTask;
}
public override async Task Stop(CancellationToken cancellationToken = default)
{
if (checkTask == null)
{
return;
}
await tokenSource.CancelAsync();
try
{
await checkTask;
}
catch (OperationCanceledException)
{
//Swallow
}
finally
{
tokenSource.Dispose();
}
}
Task checkTask;
CancellationTokenSource tokenSource;
}
}
}