Skip to content

Commit 34e5e3f

Browse files
stidsborgstidsborg
committed
Renamed FlowState -> FlowExecutionState
1 parent d43aa95 commit 34e5e3f

8 files changed

Lines changed: 38 additions & 38 deletions

File tree

‎Core/Cleipnir.ResilientFunctions.Tests/TestTemplates/FunctionTests/EffectTests.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -331,7 +331,7 @@ await store.EffectsStore.GetEffectResults(storedId),
331331
session,
332332
clearChildren: true
333333
);
334-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
334+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
335335

336336
effect.TryGet<int>("alias", out _).ShouldBeFalse();
337337

@@ -379,7 +379,7 @@ public async Task ExistingEffectsFuncIsOnlyInvokedAfterGettingValue(Task<IFuncti
379379
storageSession: null,
380380
clearChildren: true
381381
);
382-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
382+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
383383

384384
// Verify the effect is immediately available (eager loading)
385385
effect.TryGet<int>("test_alias", out var result).ShouldBeTrue();
@@ -711,7 +711,7 @@ public async Task CaptureUsingAtLeastOnceWithoutFlushResiliencyDelaysFlush(Task<
711711
session,
712712
clearChildren: true
713713
);
714-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
714+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
715715

716716
var result = await effect.Capture(() => "hello world", ResiliencyLevel.AtLeastOnceDelayFlush);
717717
result.ShouldBe("hello world");

‎Core/Cleipnir.ResilientFunctions.Tests/UtilsTests/PrintEffectsTests.cs‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ public void PrintSingleCompletedEffect()
4444
clearChildren: true
4545
);
4646

47-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
47+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
4848
var output = effect.ExecutionTree();
4949

5050
var expected = "└─ ✓ [1]\n";
@@ -76,7 +76,7 @@ public void PrintEffectWithAlias()
7676
clearChildren: true
7777
);
7878

79-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
79+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
8080
var output = effect.ExecutionTree();
8181

8282
var expected = "└─ ✓ [1] my-effect\n";
@@ -112,7 +112,7 @@ public void PrintFailedEffect()
112112
clearChildren: true
113113
);
114114

115-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
115+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
116116
var output = effect.ExecutionTree();
117117

118118
var expected = "└─ ✗ [1] failed-operation (System.InvalidOperationException)\n";
@@ -144,7 +144,7 @@ public void PrintStartedEffect()
144144
clearChildren: true
145145
);
146146

147-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
147+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
148148
var output = effect.ExecutionTree();
149149

150150
var expected = "└─ ⋯ [1] in-progress\n";
@@ -190,7 +190,7 @@ public void PrintEffectHierarchy()
190190
clearChildren: true
191191
);
192192

193-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
193+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
194194
var output = effect.ExecutionTree();
195195

196196
var expected =
@@ -246,7 +246,7 @@ public void PrintDeepEffectHierarchy()
246246
clearChildren: true
247247
);
248248

249-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
249+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
250250
var output = effect.ExecutionTree();
251251

252252
var expected =
@@ -296,7 +296,7 @@ public void PrintMultipleRootEffects()
296296
clearChildren: true
297297
);
298298

299-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
299+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
300300
var output = effect.ExecutionTree();
301301

302302
var expected =
@@ -331,7 +331,7 @@ public void PrintComplexEffectTree()
331331
clearChildren: true
332332
);
333333

334-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
334+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
335335
var output = effect.ExecutionTree();
336336

337337
var expected =
@@ -366,7 +366,7 @@ public void PrintEffectTreeWithMissingIntermediateEffect()
366366
clearChildren: false
367367
);
368368

369-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
369+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
370370
var output = effect.ExecutionTree();
371371

372372
var expected =
@@ -399,7 +399,7 @@ public void PrintEffectTreeWithMultipleMissingAncestors()
399399
clearChildren: false
400400
);
401401

402-
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
402+
var effect = new Effect(effectResults, utcNow: () => DateTime.UtcNow, new FlowTimeouts(), new FlowExecutionState(storedId, subflows: 1, waitingSubflows: 0, new FlowTimeouts(), completed: ForeverTask.Instance));
403403
var output = effect.ExecutionTree();
404404

405405
var expected =

Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowState.cs renamed to Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowExecutionState.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ public enum FlowStatus
1515
Completed = 2
1616
}
1717

18-
public class FlowState
18+
public class FlowExecutionState
1919
{
2020
private readonly Lock _lock = new();
2121
private readonly TaskCompletionSource _suspendedTcs = new();
@@ -44,7 +44,7 @@ public FlowStatus Status
4444
}
4545
}
4646

47-
public FlowState(
47+
public FlowExecutionState(
4848
StoredId id,
4949
int subflows,
5050
int waitingSubflows,

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/FlowsManager.cs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,22 +9,22 @@ namespace Cleipnir.ResilientFunctions.CoreRuntime;
99

1010
public class FlowsManager
1111
{
12-
private readonly Dictionary<StoredId, FlowState> _dict = new();
12+
private readonly Dictionary<StoredId, FlowExecutionState> _dict = new();
1313
private readonly IFunctionStore _functionStore;
1414
private readonly Lock _lock = new();
1515

1616
public FlowsManager(IFunctionStore functionStore) => _functionStore = functionStore;
1717

18-
public FlowState CreateFlowState(StoredId id, FlowTimeouts timeouts, Task completed)
18+
public FlowExecutionState CreateFlowState(StoredId id, FlowTimeouts timeouts, Task completed)
1919
{
2020
lock (_lock)
21-
return _dict[id] = new FlowState(id, subflows: 1, waitingSubflows: 0, timeouts, completed);
21+
return _dict[id] = new FlowExecutionState(id, subflows: 1, waitingSubflows: 0, timeouts, completed);
2222
}
2323

24-
public void RemoveFlow(StoredId id, FlowState flowState)
24+
public void RemoveFlow(StoredId id, FlowExecutionState flowExecutionState)
2525
{
2626
lock (_lock)
27-
if (_dict.TryGetValue(id, out var existingState) && flowState == existingState)
27+
if (_dict.TryGetValue(id, out var existingState) && flowExecutionState == existingState)
2828
_dict.Remove(id);
2929
}
3030

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/InvocationHelper.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -386,7 +386,7 @@ await _functionStore.BulkScheduleFunctions(
386386
public MessageWriter CreateMessageWriter(StoredId storedId)
387387
=> new MessageWriter(storedId, _functionStore.MessageStore, Serializer, _replicaId);
388388

389-
public Effect CreateEffect(StoredId storedId, FlowId flowId, IReadOnlyList<StoredEffect> storedEffects, FlowTimeouts flowTimeouts, IStorageSession? storageSession, FlowState flowState)
389+
public Effect CreateEffect(StoredId storedId, FlowId flowId, IReadOnlyList<StoredEffect> storedEffects, FlowTimeouts flowTimeouts, IStorageSession? storageSession, FlowExecutionState flowExecutionState)
390390
{
391391
var effectsStore = _functionStore.EffectsStore;
392392

@@ -400,7 +400,7 @@ public Effect CreateEffect(StoredId storedId, FlowId flowId, IReadOnlyList<Store
400400
_clearChildren
401401
);
402402

403-
var effect = new Effect(effectResults, UtcNow, flowTimeouts, flowState);
403+
var effect = new Effect(effectResults, UtcNow, flowTimeouts, flowExecutionState);
404404
return effect;
405405
}
406406

@@ -412,8 +412,8 @@ public async Task<ExistingEffects> CreateExistingEffects(FlowId flowId)
412412
}
413413
public ExistingMessages CreateExistingMessages(FlowId flowId) => new(MapToStoredId(flowId), _functionStore.MessageStore, Serializer);
414414

415-
public QueueManager CreateQueueManager(FlowId flowId, StoredId storedId, Effect effect, FlowState flowState, FlowTimeouts timeouts, UnhandledExceptionHandler unhandledExceptionHandler)
416-
=> new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
415+
public QueueManager CreateQueueManager(FlowId flowId, StoredId storedId, Effect effect, FlowExecutionState flowExecutionState, FlowTimeouts timeouts, UnhandledExceptionHandler unhandledExceptionHandler)
416+
=> new(flowId, storedId, _functionStore.MessageStore, Serializer, effect, flowExecutionState, unhandledExceptionHandler, timeouts, UtcNow, _settings, _messageClearer);
417417

418418
public StoredId MapToStoredId(FlowId flowId) => StoredId.Create(_storedType, flowId.Instance.Value);
419419

‎Core/Cleipnir.ResilientFunctions/CoreRuntime/Invocation/Invoker.cs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -265,7 +265,7 @@ await _invocationHelper.PersistFunctionInStore(
265265
Workflow: null!,
266266
Disposable.Combine(disposables),
267267
QueueManager: null!,
268-
FlowState: null!,
268+
FlowExecutionState: null!,
269269
FlowTimeouts: null!
270270
);
271271

@@ -305,7 +305,7 @@ await _invocationHelper.PersistFunctionInStore(
305305
if (!success) Disposable.Combine(disposables).Dispose();
306306
}
307307
}
308-
private record PreparedInvocation(bool Persisted, Workflow Workflow, IDisposable Disposables, QueueManager QueueManager, FlowState FlowState, FlowTimeouts FlowTimeouts, IStorageSession? StorageSession = null);
308+
private record PreparedInvocation(bool Persisted, Workflow Workflow, IDisposable Disposables, QueueManager QueueManager, FlowExecutionState FlowExecutionState, FlowTimeouts FlowTimeouts, IStorageSession? StorageSession = null);
309309

310310
private async Task<PreparedReInvocation> PrepareForReInvocation(StoredId storedId, Task completed)
311311
{
@@ -372,7 +372,7 @@ private record PreparedReInvocation(
372372
Workflow Workflow,
373373
IDisposable Disposables,
374374
QueueManager QueueManager,
375-
FlowState FlowState,
375+
FlowExecutionState FlowExecutionState,
376376
FlowTimeouts FlowTimeouts,
377377
StoredId? Parent,
378378
IStorageSession? StorageSession

‎Core/Cleipnir.ResilientFunctions/Domain/Effect.cs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,14 +21,14 @@ public class Effect
2121
private readonly EffectResults effectResults;
2222
private readonly UtcNow utcNow;
2323
private readonly FlowTimeouts flowTimeouts;
24-
private readonly FlowState flowState;
24+
private readonly FlowExecutionState _flowExecutionState;
2525

26-
internal Effect(EffectResults effectResults, UtcNow utcNow, FlowTimeouts flowTimeouts, FlowState flowState)
26+
internal Effect(EffectResults effectResults, UtcNow utcNow, FlowTimeouts flowTimeouts, FlowExecutionState flowExecutionState)
2727
{
2828
this.effectResults = effectResults;
2929
this.utcNow = utcNow;
3030
this.flowTimeouts = flowTimeouts;
31-
this.flowState = flowState;
31+
this._flowExecutionState = flowExecutionState;
3232
}
3333

3434

@@ -295,11 +295,11 @@ public async Task<TSeed> AggregateEach<T, TSeed>(
295295

296296
public Task<T> RunParallelle<T>(Func<Task<T>> work)
297297
{
298-
flowState.SubflowStarted();
298+
_flowExecutionState.SubflowStarted();
299299
var task = Capture(() => Task.Run(work));
300300
return task.ContinueWith(t =>
301301
{
302-
flowState.SubflowCompleted();
302+
_flowExecutionState.SubflowCompleted();
303303
return t.GetAwaiter().GetResult();
304304
});
305305
}

‎Core/Cleipnir.ResilientFunctions/Queuing/QueueManager.cs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ internal class QueueManager : IDisposable
2727
private readonly IMessageStore _messageStore;
2828
private readonly ISerializer _serializer;
2929
private readonly Effect _effect;
30-
private readonly FlowState _flowState;
30+
private readonly FlowExecutionState _flowExecutionState;
3131
private readonly UnhandledExceptionHandler _unhandledExceptionHandler;
3232
private readonly FlowTimeouts _timeouts;
3333
private readonly UtcNow _utcNow;
@@ -52,7 +52,7 @@ public QueueManager(
5252
IMessageStore messageStore,
5353
ISerializer serializer,
5454
Effect effect,
55-
FlowState flowState,
55+
FlowExecutionState flowExecutionState,
5656
UnhandledExceptionHandler unhandledExceptionHandler,
5757
FlowTimeouts timeouts,
5858
UtcNow utcNow,
@@ -66,7 +66,7 @@ public QueueManager(
6666
_messageStore = messageStore;
6767
_serializer = serializer;
6868
_effect = effect;
69-
_flowState = flowState;
69+
_flowExecutionState = flowExecutionState;
7070
_unhandledExceptionHandler = unhandledExceptionHandler;
7171
_timeouts = timeouts;
7272
_utcNow = utcNow;
@@ -96,7 +96,7 @@ private async Task Initialize()
9696

9797
_initialized = true;
9898
_effect.RegisterQueueManager(this);
99-
_flowState.QueueManager = this;
99+
_flowExecutionState.QueueManager = this;
100100
}
101101
finally
102102
{
@@ -172,15 +172,15 @@ public async Task Push(IReadOnlyList<StoredMessage> messages)
172172
_ = Task.Delay((fireAt - now).RoundUpToZero(), delayCts.Token)
173173
.ContinueWith(_ => ExpireSubscription(subscription, timeout, messageId), TaskContinuationOptions.OnlyOnRanToCompletion);
174174

175-
_flowState.SubflowWaiting();
175+
_flowExecutionState.SubflowWaiting();
176176
try
177177
{
178178
var msgData = await subscription.Tcs.Task;
179179
return msgData?.Envelope;
180180
}
181181
finally
182182
{
183-
await _flowState.ResumeSubflow();
183+
await _flowExecutionState.ResumeSubflow();
184184

185185
await delayCts.CancelAsync();
186186
delayCts.Dispose();

0 commit comments

Comments
 (0)