using Microsoft.CodeAnalysis;
using Serein.Library;
using Serein.Library.Api;
using Serein.Library.Utils;
using Serein.NodeFlow.Model;
using Serein.NodeFlow.Model.Nodes;
using Serein.NodeFlow.Tool;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks.Dataflow;
using System.Xml.Linq;
namespace Serein.NodeFlow.Services
{
///
/// 流程任务管理
///
public class FlowWorkManagement
{
///
/// 触发器对应的Cts
///
private ConcurrentDictionary _globalFlipflops = [];
///
/// 结束运行时需要执行的方法
///
private Func? _exitAction { get; set; }
///
/// 初始化选项
///
public FlowWorkOptions WorkOptions { get; }
///
/// 流程任务管理
///
///
public FlowWorkManagement(FlowWorkOptions options)
{
WorkOptions = options;
}
///
/// 初始化
///
///
public async Task RunAsync(CancellationToken token)
{
var sw = Stopwatch.StartNew();
var checkpoints = new Dictionary();
#region 注册所有节点所属的类的类型,如果注册失败则退出
List nodes = new List();
var flowTask = WorkOptions.Flows.Values.ToArray();
foreach (var item in flowTask)
{
var temp = item?.GetNodes?.Invoke() ;
if (temp is null)
continue;
nodes.AddRange(temp);
}
if (!RegisterAllType(nodes))
{
return false;
}
checkpoints["注册所有节点类型"] = sw.Elapsed; // 记录注册所有节点类型的时间
#endregion
#region 调用所有流程类的Init、Load事件
var initState = await TryInit();
if (!initState)
return false;
checkpoints["调用Init事件"] = sw.Elapsed; // 记录调用Init事件的时间
var loadState = await TryLoadAsync();
if (!loadState)
return false;
checkpoints["调用Load事件"] = sw.Elapsed; // 记录调用Load事件的时间
#endregion
var last = TimeSpan.Zero;
foreach (var kv in checkpoints)
{
SereinEnv.WriteLine(InfoType.INFO, $"{kv.Key} 耗时: {(kv.Value - last).TotalMilliseconds} ms");
last = kv.Value;
}
// 开始调用流程
foreach (var kvp in WorkOptions.Flows)
{
var guid = kvp.Key;
var flow = kvp.Value;
var flowNodes = flow.GetNodes?.Invoke();
if (flowNodes is null)
continue;
IFlowNode? startNode = flow.GetStartNode?.Invoke();
// 找到流程的起始节点,开始运行
if (startNode is null)
continue;
// 是否后台运行当前画布流程
if (flow.IsWaitStartFlow)
{
_ = Task.Run(async () => await CallNode(startNode), token); // 后台调用流程中的触发器
}
else
{
await CallNode(startNode);
}
await CallFlipflopNode(flow); // 后台调用流程中的触发器
}
// 等待流程运行完成
await CallExit();
return true;
}
#region 初始化
///
/// 初始化节点所需的所有类型
///
///
private bool RegisterAllType(List nodes)
{
var env = WorkOptions.Environment;
var ioc = WorkOptions.FlowIOC;
var nodeMds = nodes.Select(item => item.MethodDetails).ToList(); // 获取环境中所有节点的方法信息
var allMds = new List();
allMds.AddRange(nodeMds.Where(md => md?.ActingInstanceType is not null));
allMds.AddRange(WorkOptions.InitMds.Where(md => md?.ActingInstanceType is not null));
allMds.AddRange(WorkOptions.LoadMds.Where(md => md?.ActingInstanceType is not null));
allMds.AddRange(WorkOptions.ExitMds.Where(md => md?.ActingInstanceType is not null));
var isSuccessful = true;
foreach (var md in allMds)
{
if (md.ActingInstanceType != null)
{
ioc.Register(md.ActingInstanceType);
}
else
{
SereinEnv.WriteLine(InfoType.ERROR, "{md.MethodName} - 没有类型声明");
isSuccessful = false ;
}
}
ioc.Build(); // 绑定初始化时注册的类型
foreach (var md in allMds)
{
var instance = ioc.Get(md.ActingInstanceType);
if (instance is null)
{
SereinEnv.WriteLine(InfoType.ERROR, $"{md.MethodName} - 无法获取类型[{md.ActingInstanceType}]的实例");
isSuccessful = false;
}
}
return isSuccessful;
}
///
/// 尝试初始化
///
///
///
private async Task TryInit()
{
var env = WorkOptions.Environment;
var initMds = WorkOptions.InitMds;
var pool = WorkOptions.FlowContextPool;
var ioc = WorkOptions.FlowIOC;
foreach (var md in initMds) // 初始化
{
if (!env.TryGetDelegateDetails(md.AssemblyName, md.MethodName, out var dd)) // 流程运行初始化
{
throw new Exception("不存在对应委托");
}
var context = pool.Allocate();
var instance = ioc.Get(md.ActingInstanceType);
await dd.InvokeAsync(instance, [context]);
context.Reset();
pool.Free(context);
}
ioc.Build(); // 绑定初始化时注册的类型
var isSuccessful = true;
return isSuccessful;
}
///
/// 尝试加载流程
///
///
///
private async Task TryLoadAsync()
{
var env = WorkOptions.Environment;
var loadMds = WorkOptions.LoadMds;
var pool = WorkOptions.FlowContextPool;
var ioc = WorkOptions.FlowIOC;
foreach (var md in loadMds) // 加载时
{
if (!env.TryGetDelegateDetails(md.AssemblyName, md.MethodName, out var dd)) // 流程运行初始化
{
throw new Exception("不存在对应委托");
}
var context = pool.Allocate();
var instance = ioc.Get(md.ActingInstanceType);
await dd.InvokeAsync(instance, [context]);
context.Reset();
pool.Free(context);
}
ioc.Build(); // 绑定初始化时注册的类型
var isSuccessful = true;
return isSuccessful;
}
///
/// 结束流程时调用的方法
///
///
///
private async Task CallExit()
{
var env = WorkOptions.Environment;
var mds = WorkOptions.ExitMds;
var pool = WorkOptions.FlowContextPool;
var ioc = WorkOptions.FlowIOC;
foreach (var md in mds) // 结束时
{
if (!env.TryGetDelegateDetails(md.AssemblyName, md.MethodName, out var dd)) // 流程运行初始化
{
throw new Exception("不存在对应委托");
}
var context = pool.Allocate();
var instance = ioc.Get(md.ActingInstanceType);
await dd.InvokeAsync(instance, [context]);
context.Reset();
pool.Free(context);
}
TerminateAllGlobalFlipflop(); // 确保所有触发器不再运行
SereinEnv.ClearFlowGlobalData(); // 清空全局数据缓存
NativeDllHelper.FreeLibrarys(); // 卸载所有已加载的 Native Dll
var isSuccessful = true;
return isSuccessful;
}
///
/// 调用流程中的触发器节点
///
///
///
private async Task CallFlipflopNode(FlowTask flow)
{
var env = WorkOptions.Environment;
var nodes = flow.GetNodes?.Invoke();
if (nodes is null)
{
SereinEnv.WriteLine(InfoType.WARN, "流程中没有触发器节点可供执行");
return;
}
var flipflopNodes = nodes.Where(item => item is SingleFlipflopNode node
&& node.DebugSetting.IsEnable
&& node.NotExitPreviousNode())
.OfType()
.Select(async node =>
{
await RunGlobalFlipflopAsync(env, node); // 启动流程时启动全局触发器
});
var tasks = flipflopNodes.ToArray();
await Task.WhenAll(tasks);
}
///
/// 从某个节点开始执行
///
///
///
private async Task CallNode(IFlowNode startNode)
{
var pool = WorkOptions.FlowContextPool;
var token = WorkOptions.CancellationTokenSource.Token;
var context = pool.Allocate();
await startNode.StartFlowAsync(context, token);
context.Reset();
pool.Free(context);
return;
}
#endregion
///
/// 从选定的节点开始运行
///
///
///
public async Task StartFlowInSelectNodeAsync(IFlowNode startNode)
{
var pool = WorkOptions.FlowContextPool;
var sw = Stopwatch.StartNew();
var checkpoints = new Dictionary();
var context = pool.Allocate();
checkpoints["准备Context"] = sw.Elapsed;
var result = await startNode.StartFlowAsync(context, WorkOptions.CancellationTokenSource.Token); // 开始运行时从选定节点开始运行
checkpoints["执行流程"] = sw.Elapsed;
context.Reset();
pool.Free(context);
checkpoints["释放Context"] = sw.Elapsed;
_ = Task.Run(() =>
{
//var checkpoints = new Dictionary();
var last = TimeSpan.Zero;
foreach (var kv in checkpoints)
{
SereinEnv.WriteLine(InfoType.INFO, $"{kv.Key} 耗时: {(kv.Value - last).TotalMilliseconds} ms");
last = kv.Value;
}
});
return result;
}
///
/// 运行全局触发器
///
///
///
public async Task RunGlobalFlipflopAsync(IFlowEnvironment env, SingleFlipflopNode singleFlipFlopNode)
{
using var cts = new CancellationTokenSource();
if (_globalFlipflops.TryAdd(singleFlipFlopNode, cts))
{
await FlipflopExecuteAsync(singleFlipFlopNode, cts.Token);
}
}
///
/// 尝试移除全局触发器
///
///
public void TerminateGlobalFlipflopRuning(SingleFlipflopNode singleFlipFlopNode)
{
if (_globalFlipflops.TryRemove(singleFlipFlopNode, out var cts))
{
if (!cts.IsCancellationRequested)
{
cts.Cancel();
}
cts.Dispose();
}
}
///
/// 终结所有全局触发器
///
private void TerminateAllGlobalFlipflop()
{
foreach ((var node, var cts) in _globalFlipflops)
{
if (!cts.IsCancellationRequested)
{
cts.Cancel();
}
cts.Dispose();
}
_globalFlipflops.Clear();
}
///
/// 启动全局触发器
///
/// 需要全局监听信号的触发器
/// 单个触发器持有的
///
private async Task FlipflopExecuteAsync(SingleFlipflopNode flipflopNode,
CancellationToken token)
{
var pool = WorkOptions.FlowContextPool;
while (true)
{
if(token.IsCancellationRequested)
{
break;
}
var context = pool.Allocate(); // 从上下文池取出新的实例
try
{
var result = await flipflopNode.ExecutingAsync(context, token); // 等待触发获取结果
context.AddOrUpdateFlowData(flipflopNode.Guid, result);
if (context.NextOrientation == ConnectionInvokeType.None)
{
continue;
}
await CallSuccessorNodesAsync(flipflopNode, token, pool, context);
}
catch (FlipflopException ex)
{
SereinEnv.WriteLine(InfoType.ERROR, $"触发器[{flipflopNode.MethodDetails.MethodName}]因非预期异常终止。"+ex.Message);
if (ex.Type == FlipflopException.CancelClass.CancelFlow)
{
break;
}
}
catch (Exception ex)
{
SereinEnv.WriteLine(InfoType.ERROR, $"触发器[{flipflopNode.Guid}]异常。"+ ex.Message);
await Task.Delay(100);
}
finally
{
context.Reset();
pool.Free(context);
}
}
}
///
/// 全局触发器触发后的调用
///
///
///
///
///
///
private static async Task CallSuccessorNodesAsync(SingleFlipflopNode singleFlipFlopNode, CancellationToken singleToken, ObjectPool pool, IFlowContext context)
{
var flowState = context.NextOrientation; // 记录一下流程状态
var nextNodes = singleFlipFlopNode.SuccessorNodes[ConnectionInvokeType.Upstream]; // 优先调用上游分支
for (int i = nextNodes.Count - 1; i >= 0 && !singleToken.IsCancellationRequested; i--)
{
// 筛选出启用的节点
if (!nextNodes[i].DebugSetting.IsEnable)
{
continue;
}
context.SetPreviousNode(nextNodes[i].Guid, singleFlipFlopNode.Guid); // 设置调用关系
if (nextNodes[i].DebugSetting.IsInterrupt) // 执行触发前检查终端
{
await nextNodes[i].DebugSetting.GetInterruptTask.Invoke();
await Console.Out.WriteLineAsync($"[{nextNodes[i].MethodDetails.MethodName}]中断已取消,开始执行后继分支");
}
await nextNodes[i].StartFlowAsync(context, singleToken); // 启动执行触发器后继分支的节点
}
nextNodes = singleFlipFlopNode.SuccessorNodes[flowState]; // 调用对应分支
for (int i = nextNodes.Count - 1; i >= 0 && !singleToken.IsCancellationRequested; i--)
{
// 筛选出启用的节点
if (!nextNodes[i].DebugSetting.IsEnable)
{
continue;
}
context.SetPreviousNode(nextNodes[i].Guid, singleFlipFlopNode.Guid);
if (nextNodes[i].DebugSetting.IsInterrupt) // 执行触发前
{
await nextNodes[i].DebugSetting.GetInterruptTask.Invoke();
await Console.Out.WriteLineAsync($"[{nextNodes[i].MethodDetails.MethodName}]中断已取消,开始执行后继分支");
}
await nextNodes[i].StartFlowAsync(context, singleToken); // 启动执行触发器后继分支的节点
}
}
///
/// 结束流程
///
public void Exit()
{
_exitAction?.Invoke();
}
}
}