serein-flow/NodeFlow/FlowWorkManagement.cs

using Microsoft.CodeAnalysis;
using Serein.Library;
using Serein.Library.Api;
using Serein.Library.Utils;
using Serein.NodeFlow.Model;
using Serein.NodeFlow.Tool;
using System;
using System.Collections.Concurrent;
using System.Xml.Linq;

namespace Serein.NodeFlow
{
    /// <summary>
    /// 流程任务管理
    /// </summary>
    public class FlowWorkManagement
    {
        /// <summary>
        /// 触发器对应的Cts
        /// </summary>
        private ConcurrentDictionary<SingleFlipflopNode, CancellationTokenSource> dictGlobalFlipflop = [];

        


        /// <summary>
        /// 结束运行时需要执行的方法
        /// </summary>
        private  Func<Task>? ExitAction { get; set; }
        /// <summary>
        /// 初始化选项
        /// </summary>
        public FlowWorkLibrary WorkLibrary { get; }

        /// <summary>
        /// 流程任务管理
        /// </summary>
        /// <param name="library"></param>
        public FlowWorkManagement(FlowWorkLibrary library)
        {
            WorkLibrary = library;
          
        }

        /// <summary>
        /// 初始化啊
        /// </summary>
        /// <returns></returns>
        public async Task<bool> RunAsync(CancellationToken token)
        {
            NodeModelBase? startNode = WorkLibrary.Nodes.FirstOrDefault(node => node.IsStart);
            if (startNode is null)
            {
                return false;
            }

            if (!RegisterAllType())
            {
                return false;
            };
            var initState = await TryInit();
            if (!initState)
            {
                return false;
            };
            var loadState = await TryLoadAsync();
            if (!loadState)
            {
                return false;
            };
            var task = CallFlipflopNode();
            await CallStartNode(startNode);
            await task;
            await CallExit();
            return true;
        }

        #region 初始化
        private bool RegisterAllType()
        {
            var env = WorkLibrary.Environment;
            var nodeMds = WorkLibrary.Nodes.Select(item => item.MethodDetails).ToList(); // 获取环境中所有节点的方法信息 
            var allMds = new List<MethodDetails>();
            allMds.AddRange(nodeMds.Where(md => md?.ActingInstanceType is not null));
            allMds.AddRange(WorkLibrary.InitMds.Where(md => md?.ActingInstanceType is not null));
            allMds.AddRange(WorkLibrary.LoadMds.Where(md => md?.ActingInstanceType is not null));
            allMds.AddRange(WorkLibrary.ExitMds.Where(md => md?.ActingInstanceType is not null));
            var isSuccessful = true;
            foreach (var md in allMds)
            {
                if (md.ActingInstanceType != null)
                {
                    env.IOC.Register(md.ActingInstanceType);
                }
                else
                {
                    SereinEnv.WriteLine(InfoType.ERROR, "{md.MethodName} - 没有类型声明");
                    isSuccessful = false ;
                }
            }
            env.IOC.Build(); // 绑定初始化时注册的类型
            foreach (var md in allMds)
            {
                var instance = env.IOC.Get(md.ActingInstanceType);
                if (instance is null)
                {
                    SereinEnv.WriteLine(InfoType.ERROR, $"{md.MethodName} - 无法获取类型[{md.ActingInstanceType}]的实例");
                    isSuccessful = false;
                }
            }

            return isSuccessful;
        }

        private async Task<bool> TryInit()
        {
            var env = WorkLibrary.Environment;
            var initMds = WorkLibrary.InitMds;
            var pool = WorkLibrary.FlowContextPool;
            var ioc = WorkLibrary.Environment.IOC;
            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);
            }
            env.IOC.Build(); // 绑定初始化时注册的类型
            var isSuccessful = true;
            return isSuccessful;
        }
        private async Task<bool> TryLoadAsync()
        {
            var env = WorkLibrary.Environment;
            var loadMds = WorkLibrary.LoadMds;
            var pool = WorkLibrary.FlowContextPool;
            var ioc = WorkLibrary.Environment.IOC;
            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);
            }
            env.IOC.Build(); // 绑定初始化时注册的类型
            var isSuccessful = true;
            return isSuccessful;

        }
        private async Task<bool> CallExit()
        {
            var env = WorkLibrary.Environment;
            var mds = WorkLibrary.ExitMds;
            var pool = WorkLibrary.FlowContextPool;
            var ioc = WorkLibrary.Environment.IOC;

            ioc.Run<FlowInterruptTool>(fit => fit.CancelAllTrigger());// 取消所有中断
            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 Task CallFlipflopNode()
        {
            var env = WorkLibrary.Environment;
            var flipflopNodes = WorkLibrary.Nodes.Where(item => item is SingleFlipflopNode node
                                                  && !node.IsStart
                                                  && node.DebugSetting.IsEnable
                                                  && node.NotExitPreviousNode())
                                        .Select(item => (SingleFlipflopNode)item);
                                        //.ToList();// 获取需要再运行开始之前启动的触发器节点

            if (flipflopNodes.Count() > 0)
            {
                var tasks = flipflopNodes.Select(async node =>
                {
                    await RunGlobalFlipflopAsync(env, node); // 启动流程时启动全局触发器
                });
                Task.WhenAll(tasks);
            }
            return Task.CompletedTask;
        }
        private async Task CallStartNode(NodeModelBase startNode)
        {
            var pool = WorkLibrary.FlowContextPool;
            var token = WorkLibrary.CancellationTokenSource.Token;
            var context = pool.Allocate();
            await startNode.StartFlowAsync(context, token);
            context.Exit();
            pool.Free(context);
            return;
        }

        #endregion

        /// <summary>
        /// 从选定的节点开始运行
        /// </summary>
        /// <param name="env"></param>
        /// <param name="startNode"></param>
        /// <returns></returns>
        public async Task StartFlowInSelectNodeAsync(IFlowEnvironment env, NodeModelBase startNode)
        {
            var pool = WorkLibrary.FlowContextPool;
            var context = pool.Allocate();
            var token = WorkLibrary.CancellationTokenSource.Token;
            await startNode.StartFlowAsync(context, token); // 开始运行时从选定节点开始运行
            context.Reset();
            pool.Free(context);
        }


     

        /// <summary>
        /// 尝试添加全局触发器
        /// </summary>
        /// <param name="singleFlipFlopNode"></param>
        /// <param name="env"></param>
        public async Task RunGlobalFlipflopAsync(IFlowEnvironment env, SingleFlipflopNode singleFlipFlopNode)
        {
            if (dictGlobalFlipflop.TryAdd(singleFlipFlopNode, new CancellationTokenSource()))
            {
                var cts = dictGlobalFlipflop[singleFlipFlopNode];
                await FlipflopExecuteAsync(singleFlipFlopNode, cts.Token);
            }
        }

        /// <summary>
        /// 尝试移除全局触发器
        /// </summary>
        /// <param name="singleFlipFlopNode"></param>
        public void TerminateGlobalFlipflopRuning(SingleFlipflopNode singleFlipFlopNode)
        {
            if (dictGlobalFlipflop.TryRemove(singleFlipFlopNode, out var cts))
            {
                if (!cts.IsCancellationRequested)
                {
                    cts.Cancel();
                }
                cts.Dispose();
            }
        }
        
        /// <summary>
        /// 终结所有全局触发器
        /// </summary>
        private void TerminateAllGlobalFlipflop()
        {
            foreach ((var node, var cts) in dictGlobalFlipflop)
            {
                if (!cts.IsCancellationRequested)
                {
                    cts.Cancel();
                }
                cts.Dispose();
            }
            dictGlobalFlipflop.Clear();
        }

        /// <summary>
        /// 启动全局触发器
        /// </summary>
        /// <param name="singleFlipFlopNode">需要全局监听信号的触发器</param>
        /// <param name="singleToken">单个触发器持有的</param>
        /// <returns></returns>
        private async Task FlipflopExecuteAsync(SingleFlipflopNode singleFlipFlopNode,
                                                CancellationToken singleToken)
        {

            var pool = WorkLibrary.FlowContextPool;
            while (!singleToken.IsCancellationRequested && !singleToken.IsCancellationRequested)
            {
                try
                {
                    var context = pool.Allocate(); // 启动全局触发器时新建上下文
                    var newFlowData = await singleFlipFlopNode.ExecutingAsync(context, singleToken); // 获取触发器等待Task
                    context.AddOrUpdate(singleFlipFlopNode, newFlowData);
                    if (context.NextOrientation == ConnectionInvokeType.None)
                    {
                        continue;
                    }
                    _ = Task.Run(() => CallSubsequentNode(singleFlipFlopNode, singleToken, pool, context)); // 重新启动触发器

                }
                catch (FlipflopException ex) 
                {
                    SereinEnv.WriteLine(InfoType.ERROR, $"触发器[{singleFlipFlopNode.MethodDetails.MethodName}]因非预期异常终止。"+ex.Message);
                    if (ex.Type == FlipflopException.CancelClass.CancelFlow)
                    {
                        break;
                    }
                }
                catch (Exception ex)
                {
                    SereinEnv.WriteLine(InfoType.ERROR, $"触发器[{singleFlipFlopNode.Guid}]异常。"+ ex.Message);
                    await Task.Delay(100);
                }
            }

        }

        /// <summary>
        /// 全局触发器触发后的调用
        /// </summary>
        /// <param name="singleFlipFlopNode"></param>
        /// <param name="singleToken"></param>
        /// <param name="pool"></param>
        /// <param name="context"></param>
        /// <returns></returns>
        private static async Task? CallSubsequentNode(SingleFlipflopNode singleFlipFlopNode, CancellationToken singleToken, ObjectPool<IDynamicContext> pool, IDynamicContext 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], singleFlipFlopNode); // 设置调用关系

                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], singleFlipFlopNode);
                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); // 启动执行触发器后继分支的节点
            }

            context.Reset();
            pool.Free(context);
        }

        /// <summary>
        /// 结束流程
        /// </summary>
        public void Exit()
        {
            ExitAction?.Invoke();

        }

    }
}