mirror of
https://gitee.com/langsisi_admin/serein-flow
synced 2026-03-03 00:00:49 +08:00
使用异步重构了节点执行方法，将触发器节点与其他节点统一。使用Channel代替Tcs更改了信号触发，使其符合异步编程的习惯。增加了节点是否启用勾选框、参数遮罩勾选框，节点右键面板增加中断功能（试验）。增加了选择后被选择的节点的视觉效果。更改平移缩放逻辑，使其更加符合一般的使用习惯。
This commit is contained in:
fengjiayi
122
Library/Utils/ChannelFlowInterrupt.cs
Normal file
122
Library/Utils/ChannelFlowInterrupt.cs
Normal file
@@ -0,0 +1,122 @@
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace Serein.Library.Utils
|
||||
{
|
||||
public class ChannelFlowInterrupt
|
||||
{
|
||||
/// <summary>
|
||||
/// 中断取消类型
|
||||
/// </summary>
|
||||
public enum CancelType
|
||||
{
|
||||
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 () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(outTime, cts.Token);
|
||||
await channel.Writer.WriteAsync(CancelType.Overtime);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// 超时任务被取消
|
||||
}
|
||||
}, 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;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 取消所有任务
|
||||
/// </summary>
|
||||
public void CancelAllTasks()
|
||||
{
|
||||
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)
|
||||
{
|
||||
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<CancelType>());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
115
Library/Utils/ChannelFlowTrigger.cs
Normal file
115
Library/Utils/ChannelFlowTrigger.cs
Normal file
@@ -0,0 +1,115 @@
|
||||
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
|
||||
namespace Serein.Library.NodeFlow.Tool
|
||||
{
|
||||
/// <summary>
|
||||
/// 触发类型
|
||||
/// </summary>
|
||||
public enum TriggerType
|
||||
{
|
||||
/// <summary>
|
||||
/// 外部触发
|
||||
/// </summary>
|
||||
External,
|
||||
/// <summary>
|
||||
/// 超时触发
|
||||
/// </summary>
|
||||
Overtime
|
||||
}
|
||||
public class TriggerData
|
||||
{
|
||||
public TriggerType Type { get; set; }
|
||||
public object Value { get; set; }
|
||||
}
|
||||
|
||||
|
||||
public class ChannelFlowTrigger<TSignal> where TSignal : struct, Enum
|
||||
{
|
||||
// 使用并发字典管理每个枚举信号对应的 Channel
|
||||
private readonly ConcurrentDictionary<TSignal, Channel<TriggerData>> _channels = new ConcurrentDictionary<TSignal, Channel<TriggerData>>();
|
||||
|
||||
/// <summary>
|
||||
/// 创建信号并指定超时时间，到期后自动触发（异步方法）
|
||||
/// </summary>
|
||||
/// <param name="signal">枚举信号标识符</param>
|
||||
/// <param name="outTime">超时时间</param>
|
||||
/// <returns>等待任务</returns>
|
||||
public async Task<TriggerData> CreateChannelWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime, TResult outValue)
|
||||
{
|
||||
var channel = GetOrCreateChannel(signal);
|
||||
var cts = new CancellationTokenSource();
|
||||
|
||||
// 异步任务：超时后自动触发信号
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(outTime, cts.Token);
|
||||
TriggerData triggerData = new TriggerData()
|
||||
{
|
||||
Value = outValue,
|
||||
Type = TriggerType.Overtime,
|
||||
};
|
||||
await channel.Writer.WriteAsync(triggerData);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// 超时任务被取消
|
||||
}
|
||||
}, cts.Token);
|
||||
|
||||
// 等待信号传入（超时或手动触发）
|
||||
var result = await channel.Reader.ReadAsync();
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 触发信号
|
||||
/// </summary>
|
||||
/// <param name="signal">枚举信号标识符</param>
|
||||
/// <returns>是否成功触发</returns>
|
||||
public bool TriggerSignal<TResult>(TSignal signal, TResult value)
|
||||
{
|
||||
if (_channels.TryGetValue(signal, out var channel))
|
||||
{
|
||||
TriggerData triggerData = new TriggerData()
|
||||
{
|
||||
Value = value,
|
||||
Type = TriggerType.External,
|
||||
};
|
||||
// 手动触发信号
|
||||
channel.Writer.TryWrite(triggerData);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 取消所有任务
|
||||
/// </summary>
|
||||
public void CancelAllTasks()
|
||||
{
|
||||
foreach (var channel in _channels.Values)
|
||||
{
|
||||
channel.Writer.Complete();
|
||||
}
|
||||
_channels.Clear();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 获取或创建指定信号的 Channel
|
||||
/// </summary>
|
||||
/// <param name="signal">枚举信号标识符</param>
|
||||
/// <returns>对应的 Channel</returns>
|
||||
private Channel<TriggerData> GetOrCreateChannel(TSignal signal)
|
||||
{
|
||||
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<TriggerData>());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,64 +0,0 @@
|
||||
using Serein.Library.Ex;
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
namespace Serein.Library.Core.NodeFlow.Tool
|
||||
{
|
||||
//public class TcsSignalException : Exception
|
||||
//{
|
||||
// public FlowStateType FsState { get; set; }
|
||||
// public TcsSignalException(string? message) : base(message)
|
||||
// {
|
||||
// FsState = FlowStateType.Error;
|
||||
// }
|
||||
//}
|
||||
|
||||
public class TcsSignalFlipflop<TSignal> where TSignal : struct, Enum
|
||||
{
|
||||
public ConcurrentDictionary<TSignal, TaskCompletionSource<object>> TcsEvent { get; } = new ConcurrentDictionary<TSignal, TaskCompletionSource<object>>();
|
||||
|
||||
public ConcurrentDictionary<TSignal, object> TcsLock { get; } = new ConcurrentDictionary<TSignal, object>();
|
||||
|
||||
/// <summary>
|
||||
/// 触发信号
|
||||
/// </summary>
|
||||
/// <typeparam name="T"></typeparam>
|
||||
/// <param name="signal">信号</param>
|
||||
/// <param name="value">传递的参数</param>
|
||||
/// <returns>是否成功触发</returns>
|
||||
public bool TriggerSignal<T>(TSignal signal, T value)
|
||||
{
|
||||
var tcsLock = TcsLock.GetOrAdd(signal, new object());
|
||||
lock (tcsLock)
|
||||
{
|
||||
if (TcsEvent.TryRemove(signal, out var waitTcs))
|
||||
{
|
||||
waitTcs.SetResult(value);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public TaskCompletionSource<object> CreateTcs(TSignal signal)
|
||||
{
|
||||
var tcsLock = TcsLock.GetOrAdd(signal, new object());
|
||||
lock (tcsLock)
|
||||
{
|
||||
var tcs = TcsEvent.GetOrAdd(signal, new TaskCompletionSource<object>());
|
||||
return tcs;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public void CancelTask()
|
||||
{
|
||||
foreach (var tcs in TcsEvent.Values)
|
||||
{
|
||||
tcs.SetException(new FlipflopException("任务取消"));
|
||||
}
|
||||
TcsEvent.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user
