| | | 1 | | using System.Text.Json; |
| | | 2 | | using Bpmn.Model; |
| | | 3 | | using Bpmn.Semantics; |
| | | 4 | | using Elsa.Bpmn.Activities; |
| | | 5 | | using Elsa.Bpmn.Signals; |
| | | 6 | | using Elsa.Extensions; |
| | | 7 | | using Elsa.Workflows; |
| | | 8 | | using Elsa.Workflows.Memory; |
| | | 9 | | using Elsa.Workflows.Options; |
| | | 10 | | |
| | | 11 | | namespace Elsa.Bpmn.Hosting; |
| | | 12 | | |
| | | 13 | | /// <summary> |
| | | 14 | | /// Translates the interpreter's three host commands onto <see cref="ActivityExecutionContext"/>. |
| | | 15 | | /// </summary> |
| | | 16 | | /// <remarks> |
| | | 17 | | /// <list type="table"> |
| | | 18 | | /// <item><term><c>StartWork</c></term><description><see cref="ActivityExecutionContext.ScheduleActivityAsync(IActivit |
| | | 19 | | /// <item><term><c>CancelWorkSubtree</c></term><description><c>CancelActivityAsync</c>, which already walks the child |
| | | 20 | | /// <item><term><c>SignalEnclosingScope</c></term><description><c>SendSignalAsync</c>, which bubbles to ancestors.</de |
| | | 21 | | /// </list> |
| | | 22 | | /// </remarks> |
| | 182 | 23 | | internal sealed class BpmnCommandApplier(ActivityExecutionContext scopeContext, BpmnProcess process, BpmnScopeMemory mem |
| | | 24 | | { |
| | | 25 | | /// <summary> |
| | | 26 | | /// Applies a command list <b>in the order returned</b>. |
| | | 27 | | /// </summary> |
| | | 28 | | /// <remarks> |
| | | 29 | | /// The ordering carries meaning and is not an implementation detail. An interrupting boundary event emits the |
| | | 30 | | /// boundary path's <c>StartWork</c> <i>before</i> the teardown that retires the host it interrupted, and a host |
| | | 31 | | /// that tidied up first would be applying a different process. |
| | | 32 | | /// </remarks> |
| | | 33 | | public async ValueTask ApplyAsync(IReadOnlyList<BpmnHostCommand> commands) |
| | | 34 | | { |
| | | 35 | | // Refused before anything in the batch is applied. Applying commands one at a time and refusing only once |
| | | 36 | | // the offending StartWork is reached would leave earlier commands in the same batch already applied and |
| | | 37 | | // saved via memory.SaveWork() below — and under ContinueWithIncidentsStrategy that throw is absorbed into |
| | | 38 | | // an incident rather than surfaced, so the workflow would carry on with a half-applied batch and half-saved |
| | | 39 | | // memory instead of the refusal stopping it clean. |
| | 638 | 40 | | foreach (var start in commands.OfType<BpmnHostCommand.StartWork>()) |
| | | 41 | | { |
| | 138 | 42 | | if (process.FindWorkActivity(start.BindingRef) is BpmnProcess { IsRootScope: true } nested) |
| | | 43 | | { |
| | 2 | 44 | | throw new InvalidOperationException( |
| | 2 | 45 | | $"BPMN element '{start.ElementId}' binds process activity '{nested.Id}' as the work of scope '{proce |
| | 2 | 46 | | + "A nested scope's start events are internal to the process around it, not workflow entry points, s |
| | | 47 | | } |
| | | 48 | | } |
| | | 49 | | |
| | 658 | 50 | | foreach (var command in commands) |
| | | 51 | | { |
| | | 52 | | switch (command) |
| | | 53 | | { |
| | | 54 | | case BpmnHostCommand.StartWork start: |
| | 135 | 55 | | await StartWorkAsync(start); |
| | 135 | 56 | | break; |
| | | 57 | | case BpmnHostCommand.CancelWorkSubtree cancel: |
| | 10 | 58 | | await CancelWorkSubtreeAsync(cancel); |
| | 10 | 59 | | break; |
| | | 60 | | case BpmnHostCommand.SignalEnclosingScope signal: |
| | 4 | 61 | | await SignalEnclosingScopeAsync(signal); |
| | 4 | 62 | | break; |
| | | 63 | | default: |
| | | 64 | | // The command hierarchy is closed, so this can only be reached by a library version that added a |
| | | 65 | | // command this host has never heard of. Refusing is the only honest answer: silently skipping it |
| | | 66 | | // would run a different process than the one the interpreter decided on. |
| | 0 | 67 | | throw new NotSupportedException($"The BPMN host command '{command.GetType().Name}' is not supported |
| | | 68 | | } |
| | | 69 | | |
| | 149 | 70 | | memory.SaveWork(); |
| | | 71 | | } |
| | 180 | 72 | | } |
| | | 73 | | |
| | | 74 | | private async ValueTask StartWorkAsync(BpmnHostCommand.StartWork start) |
| | | 75 | | { |
| | 135 | 76 | | var activity = process.FindWorkActivity(start.BindingRef) |
| | 135 | 77 | | ?? throw new InvalidOperationException( |
| | 135 | 78 | | $"BPMN element '{start.ElementId}' binds work '{start.BindingRef}', which activity '{process. |
| | | 79 | | |
| | | 80 | | // The rule that a nested scope registers no start triggers is enforced by ApplyAsync's pre-scan, before any |
| | | 81 | | // command in the batch is applied — not here, where earlier commands in the same batch could already have |
| | | 82 | | // been applied and persisted. |
| | 135 | 83 | | var workflowExecutionContext = scopeContext.WorkflowExecutionContext; |
| | | 84 | | |
| | | 85 | | // The child's context is created up front so that this scope has its id before the child ever runs, and can |
| | | 86 | | // key the unit of work on it. The alternative — recognising the child by ActivityExecutionContext.Tag — is |
| | | 87 | | // unsound across nested scopes, because the completion-callback dispatch rewrites the receiving context's Tag. |
| | 135 | 88 | | var childContext = await workflowExecutionContext.CreateActivityExecutionContextAsync(activity, new ActivityInvo |
| | 135 | 89 | | { |
| | 135 | 90 | | Owner = scopeContext, |
| | 135 | 91 | | Variables = BuildIterationVariables(start.IterationScope), |
| | 135 | 92 | | SchedulingActivityExecutionId = scopeContext.Id |
| | 135 | 93 | | }); |
| | | 94 | | |
| | | 95 | | // The correlation is opaque interpreter state that must travel with the work and, when the work is a nested |
| | | 96 | | // BPMN process, arrive there as its InvocationCorrelation. It goes on the child's own context rather than on |
| | | 97 | | // the activity instance: an activity object is shared by every concurrent execution of one definition, so |
| | | 98 | | // writing per-invocation state onto it corrupts as soon as two instances run at once. |
| | 135 | 99 | | BpmnScopeMemory.Write(childContext, BpmnScopeHost.InvocationCorrelationPropertyKey, start.Correlation); |
| | | 100 | | |
| | 135 | 101 | | childContext.Taint(); |
| | 135 | 102 | | workflowExecutionContext.AddActivityExecutionContext(childContext); |
| | | 103 | | |
| | | 104 | | // Recorded before scheduling, so the work is live from the moment anything could report against it. This |
| | | 105 | | // never checks for an existing record on the same (BindingRef, IterationId): the port guarantees the |
| | | 106 | | // interpreter never issues a second StartWork for a slot it already holds live, so a duplicate here would be |
| | | 107 | | // an interpreter contract breach, not a host-side race. Were it to happen anyway, Records is append-only and |
| | | 108 | | // keyed by handle rather than by slot, so nothing gets overwritten or stranded — the older record, and the |
| | | 109 | | // context behind it, stay exactly as reachable as before. What would go wrong is the snapshot then reporting |
| | | 110 | | // two live entries for one slot, which is the interpreter's invariant to keep, not this host's to enforce. |
| | 135 | 111 | | memory.Work.Records.Add(new BpmnWorkRecord |
| | 135 | 112 | | { |
| | 135 | 113 | | Handle = memory.Work.NextHandle(), |
| | 135 | 114 | | BindingRef = start.BindingRef, |
| | 135 | 115 | | IterationId = start.IterationScope?.IterationId, |
| | 135 | 116 | | ElementId = start.ElementId, |
| | 135 | 117 | | ChildContextId = childContext.Id |
| | 135 | 118 | | }); |
| | | 119 | | |
| | | 120 | | // A named instance method, not a lambda: completion callbacks are rehydrated by method name. |
| | 135 | 121 | | await scopeContext.ScheduleActivityAsync(activity, new ScheduleWorkOptions |
| | 135 | 122 | | { |
| | 135 | 123 | | CompletionCallback = process.OnWorkCompletedAsync, |
| | 135 | 124 | | ExistingActivityExecutionContext = childContext, |
| | 135 | 125 | | SchedulingActivityExecutionId = scopeContext.Id |
| | 135 | 126 | | }); |
| | 135 | 127 | | } |
| | | 128 | | |
| | | 129 | | private async ValueTask CancelWorkSubtreeAsync(BpmnHostCommand.CancelWorkSubtree cancel) |
| | | 130 | | { |
| | | 131 | | // The interpreter only ever names a handle this scope reported as live, so a miss means the work has already |
| | | 132 | | // gone — a race BPMN produces routinely, and one the interpreter absorbs on the way back in. |
| | 10 | 133 | | if (memory.Work.FindByHandle(cancel.Handle) is not { } record) |
| | 0 | 134 | | return; |
| | | 135 | | |
| | 10 | 136 | | memory.Work.Remove(record); |
| | | 137 | | |
| | | 138 | | // Saved here rather than left to the end-of-command save in ApplyAsync: cancelling the subtree runs arbitrary |
| | | 139 | | // activity code (CancelSignal handlers, cancellation notifications), any of which can throw, and under the |
| | | 140 | | // continue-with-incidents strategy such a throw is absorbed into an incident rather than left to crash the |
| | | 141 | | // process — so the end-of-command save would never run. Saving the removal now, before the teardown, keeps |
| | | 142 | | // the persisted ledger from claiming work this scope just tore down, and is what lets a completion callback |
| | | 143 | | // that arrives for that work anyway find no live record and be discarded (see |
| | | 144 | | // BpmnScopeHost.OnWorkCompletedAsync) instead of being handed to the interpreter as real work. |
| | 10 | 145 | | memory.SaveWork(); |
| | | 146 | | |
| | 10 | 147 | | if (BpmnWorkTeardown.FindContext(scopeContext.WorkflowExecutionContext, record.ChildContextId) is not { } childC |
| | 0 | 148 | | return; |
| | | 149 | | |
| | | 150 | | // On the scope-completion path (an armed listener retired when its scope completes), this call is measured |
| | | 151 | | // redundant with Elsa's own CompleteActivityAsync, which already cancels a completed container's |
| | | 152 | | // non-completed children: see BpmnEventSubprocessTests.MessageEventSubprocess_RetiresTheStillArmedListenerWhenT |
| | | 153 | | // where every assertion but the ledger removal above still holds with this call skipped. The ledger removal |
| | | 154 | | // is this host's own contribution and is not redundant. The fault-teardown path (BpmnScopeHost, tearing down |
| | | 155 | | // a claimed fault's sibling work) is a different call site and was not part of that measurement. |
| | 10 | 156 | | await BpmnWorkTeardown.CancelSubtreeAsync(childContext, $"element '{cancel.ElementId}', {cancel.Reason}"); |
| | 10 | 157 | | } |
| | | 158 | | |
| | | 159 | | private ValueTask SignalEnclosingScopeAsync(BpmnHostCommand.SignalEnclosingScope signal) => |
| | 4 | 160 | | scopeContext.SendSignalAsync(new BpmnScopeSignal(signal.Code, signal.Payload)); |
| | | 161 | | |
| | | 162 | | private static ICollection<Variable>? BuildIterationVariables(BpmnIterationScope? iterationScope) => |
| | 155 | 163 | | iterationScope?.Values.Select(value => new Variable(value.Key, ToClrValue(value.Value))).ToList(); |
| | | 164 | | |
| | 20 | 165 | | private static object? ToClrValue(BpmnValue value) => value.Json is not { } json |
| | 20 | 166 | | ? null |
| | 20 | 167 | | : json.ValueKind switch |
| | 20 | 168 | | { |
| | 3 | 169 | | JsonValueKind.String => json.GetString(), |
| | 17 | 170 | | JsonValueKind.Number => json.TryGetInt64(out var integer) ? integer : json.GetDouble(), |
| | 0 | 171 | | JsonValueKind.True => true, |
| | 0 | 172 | | JsonValueKind.False => false, |
| | 0 | 173 | | JsonValueKind.Null or JsonValueKind.Undefined => null, |
| | 0 | 174 | | _ => json.GetRawText() |
| | 20 | 175 | | }; |
| | | 176 | | } |