Fix unexpected FlowJoin(WaitAny) behavior in cyclic flow. (#6592)
* Fix unexpected FlowJoin(WaitAny) behavior in cyclic flow. Visited counting mechanism now applies to canceled branches. * change FlowJoin to be in pending (instead of running) state if it's not ready to complete --------- Co-authored-by: n.rulyev <n.rulyev@bars.group>
This commit is contained in:
parent
3629a2ee69
commit
0dac514973
|
|
@ -35,26 +35,15 @@ public class FlowJoin : Activity, IJoinNode
|
|||
/// <inheritdoc />
|
||||
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
|
||||
{
|
||||
var mode = context.Get(Mode);
|
||||
|
||||
switch (mode)
|
||||
{
|
||||
case FlowJoinMode.WaitAll:
|
||||
{
|
||||
if (Flowchart.CanWaitAllProceed(context))
|
||||
{
|
||||
Flowchart.CancelAncestorActivatesAsync(context);
|
||||
await context.CompleteActivityAsync();
|
||||
}
|
||||
|
||||
break;
|
||||
}
|
||||
case FlowJoinMode.WaitAny:
|
||||
{
|
||||
Flowchart.CancelAncestorActivatesAsync(context);
|
||||
await context.CompleteActivityAsync();
|
||||
break;
|
||||
}
|
||||
}
|
||||
Flowchart.CancelAncestorActivatesAsync(context);
|
||||
await context.CompleteActivityAsync();
|
||||
}
|
||||
|
||||
protected override bool CanExecute(ActivityExecutionContext context)
|
||||
=> context.Get(Mode) switch
|
||||
{
|
||||
FlowJoinMode.WaitAny => true,
|
||||
FlowJoinMode.WaitAll => Flowchart.CanWaitAllProceed(context),
|
||||
_ => true
|
||||
};
|
||||
}
|
||||
|
|
@ -321,13 +321,20 @@ public class Flowchart : Container
|
|||
}
|
||||
}
|
||||
|
||||
var scheduleWorkOptions = new ScheduleWorkOptions
|
||||
if (joinContext is not { Status: ActivityStatus.Running })
|
||||
{
|
||||
CompletionCallback = OnChildCompletedAsync,
|
||||
ExistingActivityExecutionContext = joinContext
|
||||
};
|
||||
await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions);
|
||||
return true;
|
||||
var scheduleWorkOptions = new ScheduleWorkOptions
|
||||
{
|
||||
CompletionCallback = OnChildCompletedAsync,
|
||||
ExistingActivityExecutionContext = joinContext
|
||||
};
|
||||
await flowchartContext.ScheduleActivityAsync(outboundActivity, scheduleWorkOptions);
|
||||
return true;
|
||||
}
|
||||
else
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public static bool CanWaitAllProceed(ActivityExecutionContext context)
|
||||
|
|
@ -415,5 +422,13 @@ public class Flowchart : Container
|
|||
private async ValueTask OnActivityCanceledAsync(CancelSignal signal, SignalContext context)
|
||||
{
|
||||
await CompleteIfNoPendingWorkAsync(context.ReceiverActivityExecutionContext);
|
||||
|
||||
var flowchartContext = context.ReceiverActivityExecutionContext!;
|
||||
var flowchart = (Flowchart)flowchartContext.Activity;
|
||||
var flowGraph = flowchart.GetFlowGraph(flowchartContext);
|
||||
var flowScope = flowchart.GetFlowScope(flowchartContext);
|
||||
|
||||
// Propagate canceled connections visited count by scheduling with Outcomes.Empty
|
||||
await flowchart.ScheduleOutboundActivitiesAsync(flowGraph, flowScope, flowchartContext, context.SenderActivityExecutionContext.Activity, Outcomes.Empty);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue