zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-04-06 08:06:33 +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 eff0de410b
51 changed files with 5258 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
Library/Api/IFlowEnvironment.cs
View File

@@ -5,6 +5,7 @@ using System;
using System.Collections.Generic;
using System.Reflection;
using System.Threading.Tasks;
using static Serein.Library.Utils.ChannelFlowInterrupt;
namespace Serein.Library.Api
{
@@ -226,7 +227,7 @@ namespace Serein.Library.Api
/// 节点触发中断事件
/// </summary>
/// <param name="eventArgs"></param>
public delegate void NodeInterruptTriggerHandler(NodeInterruptTriggerEventArgs eventArgs);
public delegate void ExpInterruptTriggerHandler(InterruptTriggerEventArgs eventArgs);
/// <summary>
/// 监视的节点数据发生变化
@@ -270,23 +271,54 @@ namespace Serein.Library.Api
/// <summary>
/// 节点触发了中断事件参数
/// </summary>
public class NodeInterruptTriggerEventArgs : FlowEventArgs
public class InterruptTriggerEventArgs : FlowEventArgs
{
public NodeInterruptTriggerEventArgs(string nodeGuid)
public enum InterruptTriggerType
{
NodeGuid = nodeGuid;
/// <summary>
/// 主动监视中断
/// </summary>
Monitor,
/// <summary>
/// 表达式中断
/// </summary>
Exp,
}
public InterruptTriggerEventArgs(string nodeGuid, string expression, InterruptTriggerType type)
{
this.NodeGuid = nodeGuid;
this.Expression = expression;
this.Type = type;
}
/// <summary>
/// 中断的节点Guid
/// </summary>
public string NodeGuid { get; protected set; }
public string Expression { get; protected set; }
public InterruptTriggerType Type { get; protected set; }
}
public interface IFlowEnvironment
{
ChannelFlowInterrupt ChannelFlowInterrupt { get; set; }
/// <summary>
/// 环境名称
/// </summary>
string EnvName {get;}
/// <summary>
/// 是否全局中断
/// </summary>
bool IsGlobalInterrupt { get; }
/// <summary>
/// 设置中断时的中断级别
/// </summary>
//InterruptClass EnvInterruptClass { get; set; }
/// <summary>
/// 调试管理
/// </summary>
//ChannelFlowInterrupt ChannelFlowInterrupt { get; set; }
/// <summary>
/// 加载Dll
@@ -334,9 +366,9 @@ namespace Serein.Library.Api
event NodeInterruptStateChangeHandler OnNodeInterruptStateChange;
/// <summary>
/// 节点触发中断
/// 触发中断
/// </summary>
event NodeInterruptTriggerHandler OnNodeInterruptTrigger;
event ExpInterruptTriggerHandler OnInterruptTrigger;
/// <summary>
@@ -417,21 +449,51 @@ namespace Serein.Library.Api
/// <param name="nodeGuid">被中断的节点Guid</param>
/// <param name="interruptClass">新的中断级别</param>
/// <returns></returns>
bool NodeInterruptChange(string nodeGuid,InterruptClass interruptClass);
bool SetNodeInterrupt(string nodeGuid, InterruptClass interruptClass);
/// <summary>
/// 添加中断表达式
/// </summary>
/// <param name="nodeGuid"></param>
/// <param name="expression"></param>
/// <returns></returns>
bool AddInterruptExpression(string nodeGuid,string expression);
/// <summary>
/// /// <summary>
/// 设置节点数据监视状态
/// </summary>
/// <param name="nodeGuid">需要监视的节点Guid</param>
/// <param name="isMonitor">是否监视</param>
void SetNodeFLowDataMonitorState(string nodeGuid, bool isMonitor);
/// <summary>
/// 节点数据更新通知
/// 流程启动器调用，节点数据更新通知
/// </summary>
/// <param name="nodeGuid"></param>
void FlowDataUpdateNotification(string nodeGuid, object flowData);
/// <param name="nodeGuid">更新了数据的节点Guid</param>
/// <param name="flowData">更新的数据</param>
void FlowDataNotification(string nodeGuid, object flowData);
/// <summary>
/// 流程启动器调用，节点触发了中断
/// </summary>
/// <param name="nodeGuid">被中断的节点Guid</param>
/// <param name="expression">被触发的表达式</param>
/// <param name="type">中断类型。0主动监视，1表达式</param>
void TriggerInterrupt(string nodeGuid,string expression, InterruptTriggerEventArgs.InterruptTriggerType type);
/// <summary>
/// 全局中断
/// </summary>
/// <param name="signal"></param>
/// <param name="interruptClass"></param>
/// <returns></returns>
Task<CancelType> GetOrCreateGlobalInterruptAsync();
}
}

19
Library/Entity/NodeDebugSetting.cs
View File

@@ -1,6 +1,8 @@
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using static Serein.Library.Utils.ChannelFlowInterrupt;
namespace Serein.Library.Entity
{
@@ -21,8 +23,21 @@ namespace Serein.Library.Entity
/// </summary>
public InterruptClass InterruptClass { get; set; } = InterruptClass.None;
/// <summary>
/// 中断表达式
/// </summary>
public List<string> InterruptExpressions { get; } = new List<string>();
public List<string> InterruptExpression { get; } = new List<string>();
/// <summary>
/// 取消中断的回调函数
/// </summary>
public Action CancelInterruptCallback { get; set; }
/// <summary>
/// 中断Task
/// </summary>
public Func<Task<CancelType>> GetInterruptTask { get; set; }
}
/// <summary>
@@ -41,7 +56,7 @@ namespace Serein.Library.Entity
/// <summary>
/// 分组中断，中断进入指定节点分组的分支。（暂未实现相关）
/// </summary>
Group,
// Group,
/// <summary>
/// 全局中断，中断全局所有节点的运行。（暂未实现相关）
/// </summary>

449
Library/Utils/ChannelFlowInterrupt.cs
View File

@@ -1,4 +1,5 @@
using System;
#region plan 2
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Channels;
@@ -6,125 +7,351 @@ using System.Threading.Tasks;
namespace Serein.Library.Utils
{
public class ChannelFlowInterrupt
/// <summary>
/// 流程中断管理
/// </summary>
public class ChannelFlowInterrupt
{
/// <summary>
/// 中断取消类型
/// </summary>
public enum CancelType
{
Manual,
Error,
Overtime
}
// 使用并发字典管理每个信号对应的 Channel
private readonly ConcurrentDictionary<string, Channel<CancelType>> _channels = new ConcurrentDictionary<string, Channel<CancelType>>();
/// <summary>
/// 创建信号并指定超时时间，到期后自动触发（异步方法）
/// </summary>
/// <param name="signal">信号标识符</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务</returns>
public async Task<CancelType> GetCreateChannelWithTimeoutAsync(string signal, TimeSpan outTime)
{
var channel = GetOrCreateChannel(signal);
var cts = new CancellationTokenSource();
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
{
/// <summary>
/// 中断取消类型
/// </summary>
public enum CancelType
try
{
Manual,
Overtime
}
// 使用并发字典管理每个信号对应的 Channel
private readonly ConcurrentDictionary<string, Channel<CancelType>> _channels = new ConcurrentDictionary<string, Channel<CancelType>>();
/// <summary>
/// 创建信号并指定超时时间，到期后自动触发（异步方法）
/// </summary>
/// <param name="signal">信号标识符</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务</returns>
public async Task<CancelType> CreateChannelWithTimeoutAsync(string signal, TimeSpan outTime)
{
var channel = GetOrCreateChannel(signal);
var cts = new CancellationTokenSource();
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
await Task.Delay(outTime, cts.Token);
if (!cts.Token.IsCancellationRequested)
{
try
{
await Task.Delay(outTime, cts.Token);
if(!cts.Token.IsCancellationRequested)
{
await channel.Writer.WriteAsync(CancelType.Overtime);
}
}
catch (OperationCanceledException)
{
// 超时任务被取消
}
finally
{
cts?.Dispose();
}
}, cts.Token);
// 等待信号传入（超时或手动触发）
var result = await channel.Reader.ReadAsync();
return result;
}
/// <summary>
/// 创建信号并指定超时时间，到期后自动触发（同步阻塞方法）
/// </summary>
/// <param name="signal">信号标识符</param>
/// <param name="timeout">超时时间</param>
public CancelType CreateChannelWithTimeoutSync(string signal, TimeSpan timeout)
{
var channel = GetOrCreateChannel(signal);
var cts = new CancellationTokenSource();
CancellationToken token = cts.Token;
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
{
try
{
await Task.Delay(timeout, token);
await channel.Writer.WriteAsync(CancelType.Overtime);
}
catch (OperationCanceledException)
{
// 任务被取消
}
});
// 同步阻塞直到信号触发或超时
var result = channel.Reader.ReadAsync().AsTask().GetAwaiter().GetResult();
return result;
}
/// <summary>
/// 触发信号
/// </summary>
/// <param name="signal">信号字符串</param>
/// <returns>是否成功触发</returns>
public bool TriggerSignal(string signal)
{
if (_channels.TryGetValue(signal, out var channel))
{
// 手动触发信号
channel.Writer.TryWrite(CancelType.Manual);
return true;
await channel.Writer.WriteAsync(CancelType.Overtime);
}
return false;
}
/// <summary>
/// 取消所有任务
/// </summary>
public void CancelAllTasks()
catch (OperationCanceledException)
{
foreach (var channel in _channels.Values)
{
channel.Writer.Complete();
}
_channels.Clear();
// 超时任务被取消
}
/// <summary>
/// 获取或创建指定信号的 Channel
/// </summary>
/// <param name="signal">信号字符串</param>
/// <returns>对应的 Channel</returns>
private Channel<CancelType> GetOrCreateChannel(string signal)
finally
{
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<CancelType>());
cts?.Dispose();
}
}, cts.Token);
// 等待信号传入（超时或手动触发）
try
{
var result = await channel.Reader.ReadAsync();
return result;
}
catch
{
return CancelType.Error;
}
}
/// <summary>
/// 创建信号，直到手动触发（异步方法）
/// </summary>
/// <param name="signal">信号标识符</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务</returns>
public async Task<CancelType> GetOrCreateChannelAsync(string signal)
{
try
{
var channel = GetOrCreateChannel(signal);
// 等待信号传入（超时或手动触发）
var result = await channel.Reader.ReadAsync();
return result;
}
catch
{
return CancelType.Manual;
}
}
/// <summary>
/// 创建信号并指定超时时间，到期后自动触发（同步阻塞方法）
/// </summary>
/// <param name="signal">信号标识符</param>
/// <param name="timeout">超时时间</param>
public CancelType CreateChannelWithTimeoutSync(string signal, TimeSpan timeout)
{
var channel = GetOrCreateChannel(signal);
var cts = new CancellationTokenSource();
CancellationToken token = cts.Token;
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
{
try
{
await Task.Delay(timeout, token);
await channel.Writer.WriteAsync(CancelType.Overtime);
}
catch (OperationCanceledException ex)
{
// 任务被取消
await Console.Out.WriteLineAsync(ex.Message);
}
});
// 同步阻塞直到信号触发或超时
var result = channel.Reader.ReadAsync().AsTask().GetAwaiter().GetResult();
return result;
}
/// <summary>
/// 触发信号
/// </summary>
/// <param name="signal">信号字符串</param>
/// <returns>是否成功触发</returns>
public bool TriggerSignal(string signal)
{
//if (_channels.TryGetValue(signal, out var channel))
//{
// // 手动触发信号
// channel.Writer.TryWrite(CancelType.Manual);
// return true;
//}
//return false;
try
{
if (_channels.TryGetValue(signal, out var channel))
{
// 手动触发信号
channel.Writer.TryWrite(CancelType.Manual);
// 完成写入，标记该信号通道不再接受新写入
channel.Writer.Complete();
// 触发后移除信号
_channels.TryRemove(signal, out _);
return true;
}
return false;
}
catch
{
return false;
}
}
/// <summary>
/// 取消所有任务
/// </summary>
public void CancelAllTasks()
{
foreach (var channel in _channels.Values)
{
try
{
channel.Writer.Complete();
}
finally
{
}
}
_channels.Clear();
}
/// <summary>
/// 获取或创建指定信号的 Channel
/// </summary>
/// <param name="signal">信号字符串</param>
/// <returns>对应的 Channel</returns>
private Channel<CancelType> GetOrCreateChannel(string signal)
{
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<CancelType>());
}
}
}
#endregion
#region plan 3
//using System;
//using System.Collections.Concurrent;
//using System.Threading;
//using System.Threading.Channels;
//using System.Threading.Tasks;
//namespace Serein.Library.Utils
//{
// /// <summary>
// /// 流程中断管理类，提供了基于 Channel 的异步中断机制
// /// </summary>
// public class ChannelFlowInterrupt
// {
// /// <summary>
// /// 中断取消类型
// /// </summary>
// public enum CancelType
// {
// Manual, // 手动触发
// Overtime, // 超时触发
// Discard // 丢弃触发
// }
// // 使用并发字典管理每个信号对应的 Channel 和状态
// private readonly ConcurrentDictionary<string, (Channel<CancelType> Channel, bool IsCancelled, bool IsDiscardMode)> _channels
// = new ConcurrentDictionary<string, (Channel<CancelType>, bool, bool)>();
// // 锁对象，用于保护并发访问
// private readonly object _lock = new object();
// /// <summary>
// /// 创建带有超时功能的信号，超时后自动触发
// /// </summary>
// public async Task<CancelType> GetCreateChannelWithTimeoutAsync(string signal, TimeSpan outTime)
// {
// var (channel, isCancelled, isDiscardMode) = GetOrCreateChannel(signal);
// // 如果信号已取消或在丢弃模式下，立即返回丢弃类型
// if (isCancelled || isDiscardMode) return CancelType.Discard;
// var cts = new CancellationTokenSource();
// _ = Task.Run(async () =>
// {
// try
// {
// await Task.Delay(outTime, cts.Token);
// if (!cts.Token.IsCancellationRequested && !isCancelled)
// {
// await channel.Writer.WriteAsync(CancelType.Overtime);
// }
// }
// catch (OperationCanceledException)
// {
// // 处理任务取消的情况
// }
// finally
// {
// cts.Dispose();
// }
// }, cts.Token);
// return await channel.Reader.ReadAsync();
// }
// /// <summary>
// /// 创建或获取现有信号，等待手动触发
// /// </summary>
// public async Task<CancelType> GetOrCreateChannelAsync(string signal)
// {
// var (channel, isCancelled, isDiscardMode) = GetOrCreateChannel(signal);
// // 如果信号已取消或在丢弃模式下，立即返回丢弃类型
// if (isCancelled || isDiscardMode) return CancelType.Discard;
// return await channel.Reader.ReadAsync();
// }
// /// <summary>
// /// 触发信号并将其移除
// /// </summary>
// public bool TriggerSignal(string signal)
// {
// lock (_lock)
// {
// if (_channels.TryGetValue(signal, out var channelInfo))
// {
// var (channel, isCancelled, isDiscardMode) = channelInfo;
// // 如果信号未被取消，则触发并标记为已取消
// if (!isCancelled)
// {
// channel.Writer.TryWrite(CancelType.Manual);
// _channels[signal] = (channel, true, false); // 标记为已取消
// _channels.TryRemove(signal, out _); // 从字典中移除信号
// return true;
// }
// }
// }
// return false;
// }
// /// <summary>
// /// 启用丢弃模式，所有后续获取的信号将直接返回丢弃类型
// /// </summary>
// /// <param name="signal">信号标识符</param>
// public void EnableDiscardMode(string signal,bool state = true)
// {
// lock (_lock)
// {
// if (_channels.TryGetValue(signal, out var channelInfo))
// {
// var (channel, isCancelled, _) = channelInfo;
// _channels[signal] = (channel, isCancelled, state); // 标记为丢弃模式
// }
// }
// }
// /// <summary>
// /// 取消所有任务
// /// </summary>
// public void CancelAllTasks()
// {
// foreach (var (channel, _, _) in _channels.Values)
// {
// try
// {
// channel.Writer.Complete();
// }
// catch
// {
// // 忽略完成时的异常
// }
// }
// _channels.Clear();
// }
// /// <summary>
// /// 获取或创建指定信号的 Channel 通道
// /// </summary>
// private (Channel<CancelType>, bool, bool) GetOrCreateChannel(string signal)
// {
// lock (_lock)
// {
// return _channels.GetOrAdd(signal, _ => (Channel.CreateUnbounded<CancelType>(), false, false));
// }
// }
// }
//}
#endregion

2
Library/Web/Attribute.cs
View File

@@ -68,7 +68,7 @@ namespace Serein.Library.Web
/// </summary>
/// <param name="http"></param>
/// <param name="url"></param>
public WebApiAttribute(API http = API.POST, bool isUrl = false, string url = "")
public WebApiAttribute(API http = API.POST, bool isUrl = true, string url = "")
{
Http = http;
Url = url;