zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-04-02 14:36:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

移除了中断相关的后台代码与UI交互（待重写）；重写运行时节点获取参数的方法；重写了节点容器的互动；完善了WebSocket远程交互；完善了项目文件的加载；

Browse Source
This commit is contained in:
fengjiayi
2024-12-26 16:42:05 +08:00
parent 56b22be8c0
commit 7a6f8c407b
37 changed files with 880 additions and 1127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

353
Library/Utils/ChannelFlowInterrupt.cs
View File

@@ -1,353 +0,0 @@
#region plan 2
using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
namespace Serein.Library.Utils
{
/// <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 () =>
{
try
{
await Task.Delay(outTime, cts.Token);
if (!cts.Token.IsCancellationRequested)
{
await channel.Writer.WriteAsync(CancelType.Overtime);
}
}
catch (OperationCanceledException)
{
// 超时任务被取消
}
finally
{
cts?.Dispose();
}
}, cts.Token);
// 等待信号传入（超时或手动触发）
try
{
var result = await channel.Reader.ReadAsync();
return result;
}
catch
{
return CancelType.Error;
}
}
/// <summary>
/// 创建信号，直到手动触发（异步方法）
/// </summary>
/// <param name="signal">信号标识符</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 async Task<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 = await channel.Reader.ReadAsync();
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

57
Library/Utils/TaskFlowTrigger.cs → Library/Utils/FlowTrigger/ChannelFlowInterrupt.cs
View File

@@ -10,41 +10,15 @@ using System.Threading.Channels;
using System.Threading.Tasks;
using System.Transactions;
namespace Serein.Library
namespace Serein.Library.Utils
{
/// <summary>
/// 触发类型
/// </summary>
public enum TriggerDescription
{
/// <summary>
/// 外部触发
/// </summary>
External,
/// <summary>
/// 超时触发
/// </summary>
Overtime,
/// <summary>
/// 触发了，但类型不一致
/// </summary>
TypeInconsistency
}
public class TriggerResult<TResult>
{
public TriggerDescription Type { get; set; }
public TResult Value { get; set; }
}
/// <summary>
/// 信号触发器类，带有消息广播功能。
/// 使用枚举作为标记，创建
/// </summary>
public class TaskFlowTrigger<TSignal> : IFlowTrigger<TSignal> where TSignal : struct, Enum
public class ValueTaskFlowTrigger<TSignal>
{
// 使用并发字典管理每个信号对应的广播列表
private readonly ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>> _subscribers = new ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>>();
@@ -73,7 +47,7 @@ namespace Serein.Library
}
/// <summary>
/// 等待触发器并指定超时的时间
/// </summary>
@@ -81,7 +55,7 @@ namespace Serein.Library
/// <param name="signal">等待信号</param>
/// <param name="outTime">超时时间</param>
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
public async ValueTask<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
{
var subject = GetOrCreateSubject(signal);
var cts = new CancellationTokenSource();
@@ -121,13 +95,14 @@ namespace Serein.Library
/// <typeparam name="TResult"></typeparam>
/// <param name="signal"></param>
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSignal signal)
public async ValueTask<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSignal signal)
{
var taskCompletionSource = new TaskCompletionSource<TriggerResult<object>>();
var subscription = Subscribe<TResult>(signal, taskCompletionSource.SetResult);
var result = await taskCompletionSource.Task;
subscription.Dispose(); // 取消订阅
if(result.Value is TResult data)
if (result.Value is TResult data)
{
return new TriggerResult<TResult>()
{
@@ -184,23 +159,5 @@ namespace Serein.Library
/// <summary>
/// 观察者类，用于包装 Action
/// </summary>
public class Observer<T> : IObserver<T>
{
private readonly Action<T> _onNext;
public Observer(Action<T> onNext)
{
_onNext = onNext;
}
public void OnCompleted() { }
public void OnError(Exception error) { }
public void OnNext(T value)
{
_onNext?.Invoke(value);
}
}
}

6
Library/Utils/ChannelFlowTrigger.cs → Library/Utils/FlowTrigger/ChannelFlowTrigger.cs
View File

@@ -1,6 +1,4 @@

using Serein.Library.Api;
using Serein.Library.Api;
using System;
using System.Collections.Concurrent;
using System.Threading;
@@ -13,7 +11,7 @@ namespace Serein.Library.Utils
public class ChannelFlowTrigger<TSignal> : IFlowTrigger<TSignal>
public class ChannelFlowTrigger<TSignal> : IFlowTrigger<TSignal>
{
// 使用并发字典管理每个枚举信号对应的 Channel
private readonly ConcurrentDictionary<TSignal, Channel<TriggerResult<object>>> _channels = new ConcurrentDictionary<TSignal, Channel<TriggerResult<object>>>();

160
Library/Utils/FlowTrigger/TaskFlowTrigger.cs Normal file
View File

@@ -0,0 +1,160 @@
using Microsoft.Extensions.ObjectPool;
using Newtonsoft.Json.Linq;
using Serein.Library.Api;
using Serein.Library.Utils;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Transactions;
namespace Serein.Library.Utils
{
/// <summary>
/// 信号触发器类，带有消息广播功能。
/// 使用枚举作为标记，创建
/// </summary>
public class TaskFlowTrigger<TSignal> : IFlowTrigger<TSignal>
{
// 使用并发字典管理每个信号对应的广播列表
private readonly ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>> _subscribers = new ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>>();
private readonly TriggerResultPool _triggerResultPool = new TriggerResultPool();
/// <summary>
/// 获取或创建指定信号的 Subject（消息广播者）
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>对应的 Subject</returns>
private Subject<TriggerResult<object>> GetOrCreateSubject(TSignal signal)
{
return _subscribers.GetOrAdd(signal, _ => new Subject<TriggerResult<object>>());
}
/// <summary>
/// 订阅指定信号的消息
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <param name="action">订阅者</param>
/// <returns>取消订阅的句柄</returns>
private IDisposable Subscribe<TResult>(TSignal signal, Action<TriggerResult<object>> action)
{
IObserver<TriggerResult<object>> observer = new Observer<TriggerResult<object>>(action);
var subject = GetOrCreateSubject(signal);
return subject.Subscribe(observer); // 返回取消订阅的句柄
}
/// <summary>
/// 等待触发器并指定超时的时间
/// </summary>
/// <typeparam name="TResult">返回值类型</typeparam>
/// <param name="signal">等待信号</param>
/// <param name="outTime">超时时间</param>
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
{
var subject = GetOrCreateSubject(signal);
var cts = new CancellationTokenSource();
// 超时任务：延迟后触发超时信号
var timeoutTask = Task.Delay(outTime, cts.Token).ContinueWith(t =>
{
if (!cts.Token.IsCancellationRequested)
{
var outResult = _triggerResultPool.Get();
outResult.Type = TriggerDescription.Overtime;
subject.OnNext(outResult);
subject.OnCompleted();
}
}, cts.Token, TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default);
var result = await WaitTriggerAsync<TResult>(signal); // 获取触发的结果
cts.Cancel(); // 取消超时任务
await timeoutTask; // 确保超时任务完成
cts.Dispose();
return result;
}
/// <summary>
/// 等待触发
/// </summary>
/// <typeparam name="TResult"></typeparam>
/// <param name="signal"></param>
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSignal signal)
{
var taskCompletionSource = new TaskCompletionSource<TriggerResult<object>>();
var subscription = Subscribe<TResult>(signal, taskCompletionSource.SetResult);
var result = await taskCompletionSource.Task;
subscription.Dispose(); // 取消订阅
var result2 = result.Value is TResult data
? new TriggerResult<TResult> { Value = data, Type = TriggerDescription.External }
: new TriggerResult<TResult> { Type = TriggerDescription.TypeInconsistency };
_triggerResultPool.Return(result); // 将结果归还池中
return result2;
}
/// <summary>
/// 手动触发信号，并广播给所有订阅者
/// </summary>
/// <typeparam name="TResult">触发类型</typeparam>
/// <param name="signal">枚举信号标识符</param>
/// <param name="value">传递的数据</param>
/// <returns>是否成功触发</returns>
public Task<bool> InvokeTriggerAsync<TResult>(TSignal signal, TResult value)
{
if (_subscribers.TryGetValue(signal, out var subject))
{
var result = _triggerResultPool.Get();
result.Type = TriggerDescription.External;
result.Value = value;
subject.OnNext(result); // 广播给所有订阅者
subject.OnCompleted(); // 通知订阅结束
return Task.FromResult(true);
}
return Task.FromResult(false);
}
/// <summary>
/// 取消所有任务
/// </summary>
public void CancelAllTrigger()
{
foreach (var subject in _subscribers.Values)
{
subject.OnCompleted(); // 通知所有订阅者结束
}
_subscribers.Clear();
}
}
/// <summary>
/// 观察者类，用于包装 Action
/// </summary>
public class Observer<T> : IObserver<T>
{
private readonly Action<T> _onNext;
public Observer(Action<T> onNext)
{
_onNext = onNext;
}
public void OnCompleted() { }
public void OnError(Exception error) { }
public void OnNext(T value)
{
_onNext?.Invoke(value);
}
}
}

112
Library/Utils/FlowTrigger/TriggerResult.cs Normal file
View File

@@ -0,0 +1,112 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Serein.Library.Utils
{
public class TriggerResult<TResult>
{
public TriggerDescription Type { get; set; }
public TResult Value { get; set; }
}
/// <summary>
/// 对象池队列
/// </summary>
public class ConcurrentExpandingObjectPool<T> where T : class, new()
{
private readonly ConcurrentQueue<T> _pool; // 存储池中对象的队列
public ConcurrentExpandingObjectPool(int initialCapacity)
{
// 初始化对象池，初始容量为 initialCapacity
_pool = new ConcurrentQueue<T>();
// 填充初始对象
for (int i = 0; i < initialCapacity; i++)
{
_pool.Enqueue(new T());
}
}
/// <summary>
/// 获取一个对象，如果池中没有对象，则动态创建新的对象
/// </summary>
/// <returns>池中的一个对象</returns>
public T Get()
{
// 尝试从池中获取一个对象
if (!_pool.TryDequeue(out var item))
{
// 如果池为空，则创建一个新的对象
item = new T();
}
return item;
}
/// <summary>
/// 将一个对象归还到池中
/// </summary>
/// <param name="item">需要归还的对象</param>
public void Return(T item)
{
// 将对象归还到池中
_pool.Enqueue(item);
}
/// <summary>
/// 获取当前池中的对象数
/// </summary>
public int CurrentSize => _pool.Count;
/// <summary>
/// 清空池中的所有对象
/// </summary>
public void Clear()
{
while (_pool.TryDequeue(out _)) { } // 清空队列
}
}
/// <summary>
/// 使用 ObjectPool 来复用 TriggerResult 对象
/// </summary>
// 示例 TriggerResult 对象池
public class TriggerResultPool
{
private readonly ConcurrentExpandingObjectPool<TriggerResult<object>> _objectPool;
public TriggerResultPool(int defaultCapacity = 30)
{
_objectPool = new ConcurrentExpandingObjectPool<TriggerResult<object>>(defaultCapacity);
}
public TriggerResult<object> Get() => _objectPool.Get();
public void Return(TriggerResult<object> result) => _objectPool.Return(result);
}
/// <summary>
/// 触发类型
/// </summary>
public enum TriggerDescription
{
/// <summary>
/// 外部触发
/// </summary>
External,
/// <summary>
/// 超时触发
/// </summary>
Overtime,
/// <summary>
/// 触发了，但类型不一致
/// </summary>
TypeInconsistency
}
}