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

更改了日志输出，更改了ChannelFlowTrigger存在的内存泄漏（取消超时机制）

Browse Source
This commit is contained in:
fengjiayi
2024-09-21 10:06:44 +08:00
parent a1ecd259dd
commit 3537a49784
15 changed files with 624 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
NodeFlow/Base/NodeModelBaseFunc.cs
View File

@@ -11,6 +11,7 @@ using System.Linq;
using System.Net.Http.Headers;
using System.Text;
using System.Threading.Tasks;
using System.Xml.Linq;
using static Serein.Library.Utils.ChannelFlowInterrupt;
namespace Serein.NodeFlow.Base
@@ -35,6 +36,7 @@ namespace Serein.NodeFlow.Base
this.DebugSetting.InterruptClass = InterruptClass.Branch;
this.DebugSetting.IsInterrupt = true;
}
/// <summary>
/// 不再中断
/// </summary>
@@ -90,22 +92,6 @@ namespace Serein.NodeFlow.Base
node.MethodDetails.ExplicitDatas[i].DataValue = pd.Value;
}
}
//if (control is ConditionNodeControl conditionNodeControl)
//{
// conditionNodeControl.ViewModel.IsCustomData = pd.state;
// conditionNodeControl.ViewModel.CustomData = pd.value;
// conditionNodeControl.ViewModel.Expression = pd.expression;
//}
//else if (control is ExpOpNodeControl expOpNodeControl)
//{
// expOpNodeControl.ViewModel.Expression = pd.expression;
//}
//else
//{
// node.MethodDetails.ExplicitDatas[i].IsExplicitData = pd.state;
// node.MethodDetails.ExplicitDatas[i].DataValue = pd.value;
//}
return this;
}
@@ -120,101 +106,85 @@ namespace Serein.NodeFlow.Base
/// <returns></returns>
public async Task StartExecute(IDynamicContext context)
{
var cts = context.SereinIoc.GetOrRegisterInstantiate<CancellationTokenSource>();
CancellationTokenSource cts = null;
Stack<NodeModelBase> stack = new Stack<NodeModelBase>();
stack.Push(this);
while (stack.Count > 0 && !cts.IsCancellationRequested) // 循环中直到栈为空才会退出循环
try
{
// 从栈中弹出一个节点作为当前节点进行处理
var currentNode = stack.Pop();
cts = context.SereinIoc.Get<CancellationTokenSource>(FlowStarter.FlipFlopCtsName);
// 设置方法执行的对象
if (currentNode.MethodDetails?.ActingInstance == null && currentNode.MethodDetails?.ActingInstanceType is not null)
Stack<NodeModelBase> stack = new Stack<NodeModelBase>();
stack.Push(this);
while (stack.Count > 0 && !cts.IsCancellationRequested) // 循环中直到栈为空才会退出循环
{
currentNode.MethodDetails.ActingInstance ??= context.SereinIoc.GetOrRegisterInstantiate(currentNode.MethodDetails.ActingInstanceType);
}
// 从栈中弹出一个节点作为当前节点进行处理
var currentNode = stack.Pop();
//if (TryCreateInterruptTask(context, currentNode, out Task<CancelType>? task))
//{
// var cancelType = await task!;
// await Console.Out.WriteLineAsync($"[{currentNode.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
//}
#region 执行相关
// 首先执行上游分支
var upstreamNodes = currentNode.SuccessorNodes[ConnectionType.Upstream];
for (int i = upstreamNodes.Count - 1; i >= 0; i--)
{
if (upstreamNodes[i].DebugSetting.IsEnable) // 排除未启用的上游节点
// 设置方法执行的对象
if (currentNode.MethodDetails?.ActingInstance == null && currentNode.MethodDetails?.ActingInstanceType is not null)
{
upstreamNodes[i].PreviousNode = currentNode;
await upstreamNodes[i].StartExecute(context); // 执行流程节点的上游分支
currentNode.MethodDetails.ActingInstance ??= context.SereinIoc.GetOrRegisterInstantiate(currentNode.MethodDetails.ActingInstanceType);
}
}
currentNode.FlowData = currentNode.ExecutingAsync(context); // 流程中正常执行
// 判断是否为触发器节点，如果是，则开始等待。
//if (currentNode.MethodDetails != null && currentNode.MethodDetails.MethodDynamicType == NodeType.Flipflop)
//{
// currentNode.FlowData = await currentNode.ExecutingFlipflopAsync(context); // 流程中遇到了触发器
//}
//else
//{
//}
#endregion
#region 执行完成
if (currentNode.NextOrientation == ConnectionType.None) break; // 不再执行
// 选择后继分支
var nextNodes = currentNode.SuccessorNodes[currentNode.NextOrientation];
// 将下一个节点集合中的所有节点逆序推入栈中
for (int i = nextNodes.Count - 1; i >= 0; i--)
{
// 排除未启用的节点
if (nextNodes[i].DebugSetting.IsEnable)
#region 执行相关
// 首先执行上游分支
#if false
var upstreamNodes = currentNode.SuccessorNodes[ConnectionType.Upstream];
for (int i = upstreamNodes.Count - 1; i >= 0; i--)
{
nextNodes[i].PreviousNode = currentNode;
stack.Push(nextNodes[i]);
if (upstreamNodes[i].DebugSetting.IsEnable) // 排除未启用的上游节点
{
upstreamNodes[i].PreviousNode = currentNode;
await upstreamNodes[i].StartExecute(context); // 执行流程节点的上游分支
}
}
#endif
currentNode.FlowData = await currentNode.ExecutingAsync(context); // 流程中正常执行
#endregion
#region 执行完成
if (currentNode.NextOrientation == ConnectionType.None) break; // 不再执行
// 选择后继分支
var nextNodes = currentNode.SuccessorNodes[currentNode.NextOrientation];
// 将下一个节点集合中的所有节点逆序推入栈中
for (int i = nextNodes.Count - 1; i >= 0; i--)
{
// 排除未启用的节点
if (nextNodes[i].DebugSetting.IsEnable)
{
nextNodes[i].PreviousNode = currentNode;
stack.Push(nextNodes[i]);
}
}
}
#endregion
#endregion
}
}
finally
{
cts?.Dispose();
}
}
public static bool TryCreateInterruptTask(IDynamicContext context, NodeModelBase currentNode, out Task<CancelType>? task)
{
if (!currentNode.DebugSetting.IsInterrupt)
{
task = null;
return false;
}
Task<CancelType>? result = null;
bool haveTask = false;
bool haveTask;
Console.WriteLine($"[{currentNode.MethodDetails.MethodName}]在当前分支中断");
if (currentNode.DebugSetting.InterruptClass == InterruptClass.None)
{
haveTask = false;
task = null;
currentNode.DebugSetting.IsInterrupt = false; // 纠正设置
}
if (currentNode.DebugSetting.InterruptClass == InterruptClass.Branch) // 中断当前分支
else if (currentNode.DebugSetting.InterruptClass == InterruptClass.Branch) // 中断当前分支
{
currentNode.DebugSetting.IsInterrupt = true;
haveTask = true;
task = context.FlowEnvironment.ChannelFlowInterrupt.CreateChannelWithTimeoutAsync(currentNode.Guid, TimeSpan.FromSeconds(1));
currentNode.CancelInterruptCallback ??= () => context.FlowEnvironment.ChannelFlowInterrupt.TriggerSignal(currentNode.Guid);
task = context.FlowEnvironment.ChannelFlowInterrupt.CreateChannelWithTimeoutAsync(currentNode.Guid, TimeSpan.FromSeconds(60 * 30)); // 中断30分钟
}
else
{
@@ -233,16 +203,19 @@ namespace Serein.NodeFlow.Base
public virtual async Task<object?> ExecutingAsync(IDynamicContext context)
{
#region 调试中断
if (TryCreateInterruptTask(context, this, out Task<CancelType>? task))
if (DebugSetting.IsInterrupt && TryCreateInterruptTask(context, this, out Task<CancelType>? task)) // 执行节点前检查中断
{
string guid = this.Guid.ToString();
this.CancelInterruptCallback ??= () => context.FlowEnvironment.ChannelFlowInterrupt.TriggerSignal(guid);
var cancelType = await task!;
task?.ToString();
await Console.Out.WriteLineAsync($"[{this.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
}
#endregion
MethodDetails md = MethodDetails;
var del = md.MethodDelegate;
var del = md.MethodDelegate.Clone();
object instance = md.ActingInstance;
var haveParameter = md.ExplicitDatas.Length > 0;
@@ -250,13 +223,42 @@ namespace Serein.NodeFlow.Base
try
{
// Action/Func([方法作用的实例],[可能的参数值],[可能的返回值])
object?[]? parameters = GetParameters(context, md);
object? result = (haveParameter, haveResult) switch
{
(false, false) => Execution((Action<object>)del, instance), // 调用节点方法，返回null
(true, false) => Execution((Action<object, object?[]?>)del, instance, GetParameters(context, md)), // 调用节点方法，返回null
(true, false) => Execution((Action<object, object?[]?>)del, instance, parameters), // 调用节点方法，返回null
(false, true) => Execution((Func<object, object?>)del, instance), // 调用节点方法，返回方法传回类型
(true, true) => Execution((Func<object, object?[]?, object?>)del, instance, GetParameters(context, md)), // 调用节点方法，获取入参参数，返回方法忏悔类型
(true, true) => Execution((Func<object, object?[]?, object?>)del, instance, parameters), // 调用节点方法，获取入参参数，返回方法忏悔类型
};
//object?[]? parameters;
//object? result = null;
//if ( haveParameter )
//{
// var data = GetParameters(context, md);
// if (data[0] is Int32 count && count > 1)
// {
// }
// parameters = [instance, data];
//}
//else
//{
// parameters = [instance];
//}
//if (haveResult)
//{
// result = del.DynamicInvoke(parameters);
//}
//else
//{
// del.DynamicInvoke(parameters);
//}
NextOrientation = ConnectionType.IsSucceed;
return result;
}
@@ -307,7 +309,7 @@ namespace Serein.NodeFlow.Base
// 用正确的大小初始化参数数组
if (md.ExplicitDatas.Length == 0)
{
return [];// md.ActingInstance
return null;// md.ActingInstance
}
object?[]? parameters = new object[md.ExplicitDatas.Length];
@@ -344,7 +346,7 @@ namespace Serein.NodeFlow.Base
{
parameters[i] = ed.DataType switch
{
Type t when t == previousDataType => context, // 上下文
//Type t when t == previousDataType => inputParameter, // 上下文
Type t when t == typeof(IDynamicContext) => context, // 上下文
Type t when t == typeof(MethodDetails) => md, // 节点方法描述
Type t when t == typeof(NodeModelBase) => this, // 节点实体类
@@ -387,45 +389,5 @@ namespace Serein.NodeFlow.Base
#endregion
/// <summary>
/// json文本反序列化为对象
/// </summary>
/// <param name="value"></param>
/// <param name="targetType"></param>
/// <returns></returns>
private dynamic? ConvertValue(string value, Type targetType)
{
try
{
if (!string.IsNullOrEmpty(value))
{
return JsonConvert.DeserializeObject(value, targetType);
}
else
{
return null;
}
}
catch (JsonReaderException ex)
{
Console.WriteLine(ex);
return value;
}
catch (JsonSerializationException ex)
{
// 如果无法转为对应的JSON对象
int startIndex = ex.Message.IndexOf("to type '") + "to type '".Length; // 查找类型信息开始的索引
int endIndex = ex.Message.IndexOf('\''); // 查找类型信息结束的索引
var typeInfo = ex.Message[startIndex..endIndex]; // 提取出错类型信息，该怎么传出去？
Console.WriteLine("无法转为对应的JSON对象:" + typeInfo);
return null;
}
catch // (Exception ex)
{
return value;
}
}
}
}

16
NodeFlow/FlowEnvironment.cs
View File

@@ -9,6 +9,7 @@ using Serein.NodeFlow.Model;
using Serein.NodeFlow.Tool;
using System.Collections.Concurrent;
using System.Reflection;
using System.Xml.Linq;
using static Serein.NodeFlow.FlowStarter;
namespace Serein.NodeFlow
@@ -196,9 +197,20 @@ namespace Serein.NodeFlow
}
public void Exit()
{
foreach (var node in Nodes.Values)
{
if (typeof(IDisposable).IsAssignableFrom(node?.FlowData?.GetType()) && node.FlowData is IDisposable disposable)
{
disposable?.Dispose();
}
node!.FlowData = null;
}
ChannelFlowInterrupt?.CancelAllTasks();
flowStarter?.Exit();
OnFlowRunComplete?.Invoke(new FlowEventArgs());
GC.Collect();
}
/// <summary>
@@ -452,7 +464,9 @@ namespace Serein.NodeFlow
for (int i = 0; i < pnc.Value.Count; i++)
{
NodeModelBase? pNode = pnc.Value[i];
pNode.SuccessorNodes[pCType].RemoveAt(i);
//pNode.SuccessorNodes[pCType].RemoveAt(i);
pNode.SuccessorNodes[pCType].Remove(pNode);
OnNodeConnectChange?.Invoke(new NodeConnectChangeEventArgs(pNode.Guid,
remoteNode.Guid,
pCType,

205
NodeFlow/FlowStarter.cs
View File

@@ -7,6 +7,7 @@ using Serein.Library.Web;
using Serein.NodeFlow.Base;
using Serein.NodeFlow.Model;
using System.ComponentModel.Design;
using System.Runtime.CompilerServices;
using static Serein.Library.Utils.ChannelFlowInterrupt;
namespace Serein.NodeFlow
@@ -43,10 +44,12 @@ namespace Serein.NodeFlow
Completion,
}
/// <summary>
/// 控制触发器的结束
/// 控制触发器
/// </summary>
private NodeRunCts FlipFlopCts { get; set; } = null;
public const string FlipFlopCtsName = "<>.FlowFlipFlopCts";
public bool IsStopStart = false;
/// <summary>
/// 运行状态
/// </summary>
@@ -68,6 +71,16 @@ namespace Serein.NodeFlow
/// </summary>
private IDynamicContext Context { get; set; } = null;
private void CheckStartState()
{
if (IsStopStart)
{
throw new Exception("停止启动");
}
}
/// <summary>
/// 开始运行
/// </summary>
@@ -95,14 +108,14 @@ namespace Serein.NodeFlow
#region 选择运行环境的上下文
// 判断使用哪一种流程上下文
var isNetFramework = true;
var isNetFramework = false;
if (isNetFramework)
{
Context = new Serein.Library.Framework.NodeFlow.DynamicContext(SereinIOC, env);
}
else
{
Context = new Serein.Library.Core.NodeFlow.DynamicContext(SereinIOC, env);
Context = new Serein.Library.Core.NodeFlow.DynamicContext(SereinIOC, env); // 从起始节点启动流程时创建上下文
}
#endregion
@@ -119,24 +132,36 @@ namespace Serein.NodeFlow
{
nodeMd.ActingInstance = null;
}
SereinIOC.Reset(); // 开始运行时清空ioc中注册的实例
// 初始化ioc容器中的类型对象
foreach (var md in thisRuningMds)
{
if(md.ActingInstanceType != null)
if (md.ActingInstanceType != null)
{
SereinIOC.Register(md.ActingInstanceType);
}
}
SereinIOC.Build(); // 流程启动前的初始化
foreach (var md in thisRuningMds)
{
if (md.ActingInstanceType != null)
else
{
md.ActingInstance = SereinIOC.GetOrRegisterInstantiate(md.ActingInstanceType);
await Console.Out.WriteLineAsync($"{md.MethodName} - 没有类型声明");
IsStopStart = true;
}
}
CheckStartState();
SereinIOC.Build(); // 流程启动前的初始化
foreach (var md in thisRuningMds)
{
md.ActingInstance = SereinIOC.GetOrRegisterInstantiate(md.ActingInstanceType);
if(md.ActingInstance is null)
{
await Console.Out.WriteLineAsync($"{md.MethodName} - 无法获取类型[{md.ActingInstanceType}]的实例");
IsStopStart = true;
}
}
CheckStartState();
//foreach (var md in flipflopNodes.Select(it => it.MethodDetails).ToArray())
//{
@@ -144,34 +169,34 @@ namespace Serein.NodeFlow
//}
#endregion
#region 检查并修正初始化、加载时、退出时方法作用的对象，保证后续不会报错
foreach (var md in initMethods) // 初始化
{
md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
}
foreach (var md in loadingMethods) // 加载
{
md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
}
foreach (var md in exitMethods) // 初始化
{
md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
}
#region 检查并修正初始化、加载时、退出时方法作用的对象，保证后续不会报错（已注释）
//foreach (var md in initMethods) // 初始化
//{
// md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
//}
//foreach (var md in loadingMethods) // 加载
//{
// md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
//}
//foreach (var md in exitMethods) // 初始化
//{
// md.ActingInstance ??= Context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
//}
#endregion
#region 执行初始化，绑定IOC容器，再执行加载时
object?[]? args = [Context];
//object?[]? args = [Context];
foreach (var md in initMethods) // 初始化
{
object?[]? data = [md.ActingInstance, args];
md.MethodDelegate.DynamicInvoke(data);
((Action<object, object?[]?>)md.MethodDelegate).Invoke(md.ActingInstance, [Context]);
}
Context.SereinIoc.Build(); // 绑定初始化时注册的类型
foreach (var md in loadingMethods) // 加载
{
object?[]? data = [md.ActingInstance, args];
md.MethodDelegate.DynamicInvoke(data);
//object?[]? data = [md.ActingInstance, args];
//md.MethodDelegate.DynamicInvoke(data);
((Action<object, object?[]?>)md.MethodDelegate).Invoke(md.ActingInstance, [Context]);
}
Context.SereinIoc.Build(); // 预防有人在加载时才注册类型，再绑定一次
#endregion
@@ -179,38 +204,43 @@ namespace Serein.NodeFlow
#region 设置流程退出时的回调函数
ExitAction = () =>
{
SereinIOC.Run<WebServer>(web => {
web?.Stop();
});
foreach (MethodDetails? md in exitMethods)
{
object?[]? data = [md.ActingInstance, args];
md.MethodDelegate.DynamicInvoke(data);
}
if (Context != null && Context.NodeRunCts != null && !Context.NodeRunCts.IsCancellationRequested)
{
Context.NodeRunCts.Cancel();
}
if (FlipFlopCts != null && !FlipFlopCts.IsCancellationRequested)
{
FlipFlopCts.Cancel();
((Action<object, object?[]?>)md.MethodDelegate).Invoke(md.ActingInstance, [Context]);
}
//if (Context != null && Context.NodeRunCts != null && !Context.NodeRunCts.IsCancellationRequested)
//{
// Context.NodeRunCts.Cancel();
//}
//if (FlipFlopCts != null && !FlipFlopCts.IsCancellationRequested)
//{
// FlipFlopCts?.Cancel();
// FlipFlopCts?.Dispose();
//}
FlowState = RunState.Completion;
FlipFlopState = RunState.Completion;
};
#endregion
#region 开始启动流程
CancellationTokenSource FlipFlopCts = null;
try
{
if (flipflopNodes.Count > 0)
{
FlipFlopState = RunState.Running;
// 如果存在需要启动的触发器，则开始启动
FlipFlopCts = SereinIOC.GetOrRegisterInstantiate<NodeRunCts>();
FlipFlopCts = new CancellationTokenSource();
SereinIOC.CustomRegisterInstance(FlipFlopCtsName, FlipFlopCts,false);
// 使用 TaskCompletionSource 创建未启动的触发器任务
var tasks = flipflopNodes.Select(async node =>
{
@@ -220,18 +250,25 @@ namespace Serein.NodeFlow
}
await startNode.StartExecute(Context); // 开始运行时从起始节点开始运行
// 等待结束
if (FlipFlopCts != null)
if(FlipFlopState == RunState.Running && FlipFlopCts is not null)
{
while (!FlipFlopCts.IsCancellationRequested)
{
await Task.Delay(100);
}
}
//FlipFlopCts?.Dispose();
}
catch (Exception ex)
{
await Console.Out.WriteLineAsync(ex.ToString());
}
}
finally
{
FlipFlopCts?.Dispose();
FlowState = RunState.Completion;
}
#endregion
}
@@ -253,36 +290,70 @@ namespace Serein.NodeFlow
/// <returns></returns>
private async Task FlipflopExecute(IFlowEnvironment flowEnvironment,SingleFlipflopNode singleFlipFlopNode)
{
var context = new DynamicContext(SereinIOC, flowEnvironment);
CancellationTokenSource cts = null;
var context = new DynamicContext(SereinIOC, flowEnvironment); // 启动全局触发器时新建上下文
MethodDetails md = singleFlipFlopNode.MethodDetails;
var del = md.MethodDelegate;
// 设置方法执行的对象
if (md?.ActingInstance == null && md?.ActingInstanceType is not null)
{
md.ActingInstance ??= context.SereinIoc.GetOrRegisterInstantiate(md.ActingInstanceType);
}
object?[]? parameters = singleFlipFlopNode.GetParameters(context, singleFlipFlopNode.MethodDetails); // 启动全局触发器时获取入参参数
// 设置委托对象
var func = md.ExplicitDatas.Length == 0 ?
(Func<object, object, Task<IFlipflopContext>>)del :
(Func<object, object[], Task<IFlipflopContext>>)del;
bool t = md.ExplicitDatas.Length == 0;
try
{
while (!FlipFlopCts.IsCancellationRequested)
cts = Context.SereinIoc.Get<CancellationTokenSource>(FlipFlopCtsName);
while (!cts.IsCancellationRequested)
{
IFlipflopContext flipflopContext = await func.Invoke(md.ActingInstance, parameters);// 开始等待全局触发器的触发
var connectionType = flipflopContext.State.ToContentType();
if (connectionType != ConnectionType.None)
singleFlipFlopNode.FlowData = await singleFlipFlopNode.ExecutingAsync(context);
if (singleFlipFlopNode.NextOrientation != ConnectionType.None)
{
await GlobalFlipflopExecute(context, singleFlipFlopNode, connectionType);
var nextNodes = singleFlipFlopNode.SuccessorNodes[singleFlipFlopNode.NextOrientation];
for (int i = nextNodes.Count - 1; i >= 0; i--)
{
if (nextNodes[i].DebugSetting.IsEnable) // 排除未启用的后继节点
{
nextNodes[i].PreviousNode = singleFlipFlopNode;
await nextNodes[i].StartExecute(context); // 执行流程节点的后继分支
}
}
}
//if(t)
//{
// IFlipflopContext flipflopContext = await ((Func<object, Task<IFlipflopContext>>)del.Clone()).Invoke(md.ActingInstance);// 开始等待全局触发器的触发
// var connectionType = flipflopContext.State.ToContentType();
// if (connectionType != ConnectionType.None)
// {
// await GlobalFlipflopExecute(context, singleFlipFlopNode, connectionType, cts);
// }
//}
//else
//{
// IFlipflopContext flipflopContext = await ((Func<object, object[], Task<IFlipflopContext>>)del.Clone()).Invoke(md.ActingInstance, parameters);// 开始等待全局触发器的触发
// var connectionType = flipflopContext.State.ToContentType();
// if (connectionType != ConnectionType.None)
// {
// await GlobalFlipflopExecute(context, singleFlipFlopNode, connectionType, cts);
// }
//}
}
}
catch (Exception ex)
{
await Console.Out.WriteLineAsync(ex.ToString());
}
finally
{
cts?.Cancel();
}
}
/// <summary>
@@ -292,15 +363,12 @@ namespace Serein.NodeFlow
/// <param name="singleFlipFlopNode">被触发的全局触发器</param>
/// <param name="connectionType">分支类型</param>
/// <returns></returns>
public async Task GlobalFlipflopExecute(IDynamicContext context, SingleFlipflopNode singleFlipFlopNode, ConnectionType connectionType)
public async Task GlobalFlipflopExecute(IDynamicContext context, SingleFlipflopNode singleFlipFlopNode,
ConnectionType connectionType, CancellationTokenSource cts)
{
if (FlipFlopCts.IsCancellationRequested )
{
return;
}
bool skip = true;
var cts = context.SereinIoc.GetOrRegisterInstantiate<CancellationTokenSource>();
Stack<NodeModelBase> stack = new Stack<NodeModelBase>();
stack.Push(singleFlipFlopNode);
@@ -311,10 +379,10 @@ namespace Serein.NodeFlow
var currentNode = stack.Pop();
// 设置方法执行的对象
if (currentNode.MethodDetails?.ActingInstance == null && currentNode.MethodDetails?.ActingInstanceType is not null)
{
currentNode.MethodDetails.ActingInstance ??= context.SereinIoc.GetOrRegisterInstantiate(currentNode.MethodDetails.ActingInstanceType);
}
//if (currentNode.MethodDetails?.ActingInstance == null && currentNode.MethodDetails?.ActingInstanceType is not null)
//{
// currentNode.MethodDetails.ActingInstance ??= context.SereinIoc.GetOrRegisterInstantiate(currentNode.MethodDetails.ActingInstanceType);
//}
// 首先执行上游分支
var upstreamNodes = currentNode.SuccessorNodes[ConnectionType.Upstream];
@@ -331,7 +399,8 @@ namespace Serein.NodeFlow
}
else
{
currentNode.FlowData = await currentNode.ExecutingAsync(context);
currentNode.FlowData = await currentNode.ExecutingAsync(context);
if (currentNode.NextOrientation == ConnectionType.None)
{

12
NodeFlow/Model/SingleActionNode.cs
View File

@@ -15,12 +15,12 @@ namespace Serein.NodeFlow.Model
if (base.MethodDetails.ExplicitDatas.Length > 0)
{
return MethodDetails.ExplicitDatas
.Select(it => new Parameterdata
{
State = it.IsExplicitData,
Value = it.DataValue,
})
.ToArray();
.Select(it => new Parameterdata
{
State = it.IsExplicitData,
Value = it.DataValue,
})
.ToArray();
}
else
{

40
NodeFlow/Model/SingleFlipflopNode.cs
View File

@@ -22,31 +22,49 @@ namespace Serein.NodeFlow.Model
public override async Task<object?> ExecutingAsync(IDynamicContext context)
{
#region 执行前中断
if (TryCreateInterruptTask(context, this, out Task<CancelType>? task))
if (DebugSetting.IsInterrupt && TryCreateInterruptTask(context, this, out Task<CancelType>? task)) // 执行触发前
{
string guid = this.Guid.ToString();
this.CancelInterruptCallback ??= () => context.FlowEnvironment.ChannelFlowInterrupt.TriggerSignal(guid);
var cancelType = await task!;
task?.ToString();
await Console.Out.WriteLineAsync($"[{this.MethodDetails.MethodName}]中断已{(cancelType == CancelType.Manual ? "手动取消" : "自动取消")}，开始执行后继分支");
}
#endregion
MethodDetails md = MethodDetails;
Delegate del = md.MethodDelegate;
var del = md.MethodDelegate.Clone();
object instance = md.ActingInstance;
var haveParameter = md.ExplicitDatas.Length >= 0;
// Task<IFlipflopContext>? flipflopTask = null;
try
{
// 调用委托并获取结果
Task<IFlipflopContext> flipflopTask = haveParameter switch
Task<IFlipflopContext> flipflopTask = md.ExplicitDatas.Length switch
{
true => ((Func<object, object?[]?, Task<IFlipflopContext>>)del).Invoke(instance, GetParameters(context, md)), // 执行流程中的触发器方法时获取入参参数
false => ((Func<object, Task<IFlipflopContext>>)del).Invoke(instance),
0 => ((Func<object, Task<IFlipflopContext>>)del).Invoke(md.ActingInstance),
_ => ((Func<object, object?[]?, Task<IFlipflopContext>>)del).Invoke(md.ActingInstance, GetParameters(context, md)), // 执行流程中的触发器方法时获取入参参数
};
//object?[]? parameters;
//object? result = null;
//if (haveParameter)
//{
// var data = GetParameters(context, md);
// parameters = [instance, data];
//}
//else
//{
// parameters = [instance];
//}
//flipflopTask = del.DynamicInvoke(parameters) as Task<IFlipflopContext>;
//if (flipflopTask == null)
//{
// throw new FlipflopException(base.MethodDetails.MethodName + "触发器返回值非 Task<IFlipflopContext> 类型");
//}
IFlipflopContext flipflopContext = (await flipflopTask) ?? throw new FlipflopException("没有返回上下文");
NextOrientation = flipflopContext.State.ToContentType();
if(flipflopContext.TriggerData.Type == Library.NodeFlow.Tool.TriggerType.Overtime)
if(flipflopContext.TriggerData is null || flipflopContext.TriggerData.Type == Library.NodeFlow.Tool.TriggerType.Overtime)
{
throw new FlipflopException("");
throw new FlipflopException(base.MethodDetails.MethodName + "触发器超时触发。Guid"+base.Guid);
}
return flipflopContext.TriggerData.Value;
}
@@ -62,6 +80,10 @@ namespace Serein.NodeFlow.Model
RuningException = ex;
return null;
}
finally
{
// flipflopTask?.Dispose();
}
}
internal override Parameterdata[] GetParameterdatas()