zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-03-11 20:19:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化了中断功能。

Browse Source
This commit is contained in:
fengjiayi
2024-09-22 17:37:32 +08:00
parent c930c870a6
commit b63a5f4c62
45 changed files with 4821 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
NodeFlow/Base/NodeModelBaseData.cs
View File

@@ -78,10 +78,42 @@ namespace Serein.NodeFlow.Base
/// </summary>
public Exception RuningException { get; set; } = null;
/// <summary>
/// 当前传递数据（执行了节点对应的方法，才会存在值）
/// 控制FlowData在同一时间只会被同一个线程更改。
/// </summary>
protected object? FlowData { get; set; } = null;
private readonly ReaderWriterLockSlim _flowDataLock = new ReaderWriterLockSlim();
private object? _flowData;
/// <summary>
/// 当前传递数据（执行了节点对应的方法，才会存在值）。
/// </summary>
protected object? FlowData
{
get
{
_flowDataLock.EnterReadLock();
try
{
return _flowData;
}
finally
{
_flowDataLock.ExitReadLock();
}
}
set
{
_flowDataLock.EnterWriteLock();
try
{
_flowData = value;
}
finally
{
_flowDataLock.ExitWriteLock();
}
}
}
}

143
NodeFlow/Base/NodeModelBaseFunc.cs
View File

@@ -27,15 +27,6 @@ namespace Serein.NodeFlow.Base
#region 调试中断
public Action? CancelInterruptCallback;
/// <summary>
/// 中断节点
/// </summary>
public void Interrupt()
{
this.DebugSetting.InterruptClass = InterruptClass.Branch;
}
/// <summary>
/// 不再中断
@@ -43,9 +34,9 @@ namespace Serein.NodeFlow.Base
public void CancelInterrupt()
{
this.DebugSetting.InterruptClass = InterruptClass.None;
CancelInterruptCallback?.Invoke();
CancelInterruptCallback = null;
DebugSetting.CancelInterruptCallback?.Invoke();
}
#endregion
#region 导出/导入项目文件节点信息
@@ -105,6 +96,17 @@ namespace Serein.NodeFlow.Base
/// <returns></returns>
public async Task StartExecute(IDynamicContext context)
{
if (DebugSetting.InterruptClass != InterruptClass.None) // 执行触发前
{
var cancelType = await this.DebugSetting.GetInterruptTask();
//if (cancelType == CancelType.Discard)
//{
// this.NextOrientation = ConnectionType.None;
// return;
//}
await Console.Out.WriteLineAsync($"[{this.MethodDetails.MethodName}]中断已{cancelType}，开始执行后继分支");
return;
}
Stack<NodeModelBase> stack = new Stack<NodeModelBase>();
stack.Push(this);
@@ -126,27 +128,30 @@ namespace Serein.NodeFlow.Base
var upstreamNodes = currentNode.SuccessorNodes[ConnectionType.Upstream];
for (int i = upstreamNodes.Count - 1; i >= 0; i--)
{
if (upstreamNodes[i].DebugSetting.IsEnable) // 排除未启用的上游节点
// 筛选出启用的节点、未被中断的节点
if (upstreamNodes[i].DebugSetting.IsEnable && upstreamNodes[i].DebugSetting.InterruptClass == InterruptClass.None)
{
upstreamNodes[i].PreviousNode = currentNode;
var upNewFlowData = await upstreamNodes[i].ExecutingAsync(context); // 执行流程节点的上游分支
await FlowRefreshDataOrInterrupt(context, upstreamNodes[i], upNewFlowData); // 执行上游分支后刷新上游节点数据
await upstreamNodes[i].StartExecute(context); // 执行流程节点的上游分支
}
}
// 执行当前节点
var newFlowData = await currentNode.ExecutingAsync(context);
await FlowRefreshDataOrInterrupt(context, currentNode, newFlowData); // 执行当前节点后刷新数据
#endregion
#region 执行完成
if (cts == null || cts.IsCancellationRequested || currentNode.NextOrientation == ConnectionType.None)
{
// 不再执行
break;
}
await RefreshFlowDataAndExpInterrupt(context, currentNode, newFlowData); // 执行当前节点后刷新数据
#endregion
#region 执行完成
// 选择后继分支
var nextNodes = currentNode.SuccessorNodes[currentNode.NextOrientation];
@@ -154,8 +159,8 @@ namespace Serein.NodeFlow.Base
// 将下一个节点集合中的所有节点逆序推入栈中
for (int i = nextNodes.Count - 1; i >= 0; i--)
{
// 排除未启用的节点
if (nextNodes[i].DebugSetting.IsEnable)
// 筛选出启用的节点、未被中断的节点
if (nextNodes[i].DebugSetting.IsEnable && nextNodes[i].DebugSetting.InterruptClass == InterruptClass.None)
{
nextNodes[i].PreviousNode = currentNode;
stack.Push(nextNodes[i]);
@@ -165,6 +170,15 @@ namespace Serein.NodeFlow.Base
}
}
public void ThorwExitExecuting(CancellationTokenSource cts,NodeModelBase currentNode)
{
if (cts == null || cts.IsCancellationRequested || currentNode.NextOrientation == ConnectionType.None)
{
// 不再执行
throw new Exception("退出方法节点的执行");
}
}
/// <summary>
/// 执行节点对应的方法
/// </summary>
@@ -173,12 +187,16 @@ namespace Serein.NodeFlow.Base
public virtual async Task<object?> ExecutingAsync(IDynamicContext context)
{
#region 调试中断
if (DebugSetting.InterruptClass != InterruptClass.None && TryCreateInterruptTask(context, this, out Task<CancelType>? task)) // 执行节点前检查中断
if (DebugSetting.InterruptClass != InterruptClass.None) // 执行触发前
{
string guid = this.Guid.ToString();
this.CancelInterruptCallback ??= () => context.FlowEnvironment.ChannelFlowInterrupt.TriggerSignal(guid);
var cancelType = await task!;
await Console.Out.WriteLineAsync($"[{this.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
var cancelType = await this.DebugSetting.GetInterruptTask();
//if(cancelType == CancelType.Discard)
//{
// this.NextOrientation = ConnectionType.None;
// return null;
//}
await Console.Out.WriteLineAsync($"[{this.MethodDetails.MethodName}]中断已{cancelType}，开始执行后继分支");
}
#endregion
@@ -213,7 +231,6 @@ namespace Serein.NodeFlow.Base
}
#region 节点转换的委托类型
public static object? Execution(Action<object> del, object instance)
{
@@ -326,64 +343,52 @@ namespace Serein.NodeFlow.Base
/// 更新节点数据，并检查监视表达式
/// </summary>
/// <param name="newData"></param>
public static async Task FlowRefreshDataOrInterrupt(IDynamicContext context , NodeModelBase nodeModel, object? newData = null)
public static async Task RefreshFlowDataAndExpInterrupt(IDynamicContext context , NodeModelBase nodeModel, object? newData = null)
{
string guid = nodeModel.Guid;
if (newData is not null)
// 检查是否存在监视表达式
if (newData is not null && nodeModel.DebugSetting.InterruptExpressions.Count > 0)
{
// 判断是否存在表达式
bool isInterrupt = false;
// 判断监视表达式
for (int i = 0; i < nodeModel.DebugSetting.InterruptExpression.Count && !isInterrupt; i++)
// 表达式环境下判断是否需要执行中断
bool isExpInterrupt = false;
// 判断执行监视表达式，直到为 true 时退出
for (int i = 0; i < nodeModel.DebugSetting.InterruptExpressions.Count && !isExpInterrupt; i++)
{
string? exp = nodeModel.DebugSetting.InterruptExpression[i];
isInterrupt = SereinConditionParser.To(newData, exp);
string? exp = nodeModel.DebugSetting.InterruptExpressions[i];
isExpInterrupt = SereinConditionParser.To(newData, exp);
}
if (isInterrupt) // 触发中断
if (isExpInterrupt) // 触发中断
{
nodeModel.Interrupt();
if(TryCreateInterruptTask(context, nodeModel, out Task<CancelType>? task))
InterruptClass interruptClass = InterruptClass.Branch; // 分支中断
if (context.FlowEnvironment.SetNodeInterrupt(nodeModel.Guid, interruptClass))
{
nodeModel.CancelInterruptCallback ??= () => context.FlowEnvironment.ChannelFlowInterrupt.TriggerSignal(guid);
var cancelType = await task!;
await Console.Out.WriteLineAsync($"[{nodeModel.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
var cancelType = await nodeModel.DebugSetting.GetInterruptTask();
//if (cancelType == CancelType.Discard)
//{
// nodeModel.NextOrientation = ConnectionType.None;
// return;
//}
await Console.Out.WriteLineAsync($"[{nodeModel.MethodDetails.MethodName}]中断已{cancelType}，开始执行后继分支");
}
}
}
nodeModel.FlowData = newData;
//else if (nodeModel.DebugSetting.InterruptClass != InterruptClass.None)
//{
// var cancelType = await nodeModel.DebugSetting.InterruptTask;
// await Console.Out.WriteLineAsync($"[{nodeModel.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
//}
nodeModel.FlowData = newData; // 替换数据
// 节点是否监视了数据，如果是，调用环境接口触发其相关事件。
if (nodeModel.DebugSetting.IsMonitorFlowData)
{
context.FlowEnvironment.FlowDataUpdateNotification(guid, newData);
context.FlowEnvironment.FlowDataNotification(guid, newData);
}
}
public static bool TryCreateInterruptTask(IDynamicContext context, NodeModelBase currentNode, out Task<CancelType>? task)
{
bool haveTask;
Console.WriteLine($"[{currentNode.MethodDetails.MethodName}]在当前分支中断");
if (currentNode.DebugSetting.InterruptClass == InterruptClass.None)
{
haveTask = false;
task = null;
}
else if (currentNode.DebugSetting.InterruptClass == InterruptClass.Branch) // 中断当前分支
{
haveTask = true;
task = context.FlowEnvironment.ChannelFlowInterrupt.CreateChannelWithTimeoutAsync(currentNode.Guid, TimeSpan.FromSeconds(60 * 30)); // 中断30分钟
}
else
{
haveTask = false;
task = null;
}
return haveTask;
}
/// <summary>
/// 释放对象
/// </summary>