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

将FlowTrigger触发器整合成接口的形式方便替换

Browse Source
This commit is contained in:
fengjiayi
2024-12-23 23:19:10 +08:00
parent 0f9c9b9988
commit 5941f75313
29 changed files with 403 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
Library/Utils/ChannelFlowTrigger.cs
View File

@@ -1,5 +1,6 @@
using Serein.Library.Api;
using System;
using System.Collections.Concurrent;
using System.Threading;
@@ -12,18 +13,22 @@ namespace Serein.Library.Utils
public class ChannelFlowTrigger<TSignal>
public class ChannelFlowTrigger<TSignal> : IFlowTrigger<TSignal>
{
// 使用并发字典管理每个枚举信号对应的 Channel
private readonly ConcurrentDictionary<TSignal, Channel<(TriggerType,object)>> _channels = new ConcurrentDictionary<TSignal, Channel<(TriggerType, object)>>();
private readonly ConcurrentDictionary<TSignal, Channel<TriggerResult<object>>> _channels = new ConcurrentDictionary<TSignal, Channel<TriggerResult<object>>>();
/// <summary>
/// 创建信号并指定超时时间，到期后自动触发（异步方法）
/// 获取或创建指定信号的 Channel
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务</returns>
public async Task<(TriggerType, TResult)> WaitDataWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
/// <returns>对应的 Channel</returns>
private Channel<TriggerResult<object>> GetOrCreateChannel(TSignal signal)
{
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<TriggerResult<object>>());
}
public async Task<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
{
var channel = GetOrCreateChannel(signal);
var cts = new CancellationTokenSource();
@@ -34,7 +39,11 @@ namespace Serein.Library.Utils
try
{
await Task.Delay(outTime, cts.Token);
await channel.Writer.WriteAsync((TriggerType.Overtime, null));
var outResult = new TriggerResult<object>()
{
Type = TriggerDescription.Overtime
};
await channel.Writer.WriteAsync(outResult);
}
catch (OperationCanceledException)
{
@@ -43,45 +52,51 @@ namespace Serein.Library.Utils
}, cts.Token);
// 等待信号传入（超时或手动触发）
(var type, var result) = await channel.Reader.ReadAsync();
var result = await WaitTriggerAsync<TResult>(signal); // 返回一个可以超时触发的等待任务
return result;
return (type, result.ToConvert<TResult>());
}
/// <summary>
/// 创建信号，直到触发
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>等待任务</returns>
public async Task<TResult> WaitData<TResult>(TSignal signal)
public async Task<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSignal signal)
{
var channel = GetOrCreateChannel(signal);
// 等待信号传入（超时或手动触发）
(var type, var result) = await channel.Reader.ReadAsync();
return result.ToConvert<TResult>();
var result = await channel.Reader.ReadAsync();
if (result.Value is TResult data)
{
return new TriggerResult<TResult>()
{
Value = data,
Type = TriggerDescription.External,
};
}
else
{
return new TriggerResult<TResult>()
{
Type = TriggerDescription.TypeInconsistency,
};
}
}
/// <summary>
/// 触发信号
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>是否成功触发</returns>
public bool TriggerSignal(TSignal signal, object value)
public async Task<bool> InvokeTriggerAsync<TResult>(TSignal signal, TResult value)
{
if (_channels.TryGetValue(signal, out var channel))
{
// 手动触发信号
channel.Writer.TryWrite((TriggerType.External,value));
var result = new TriggerResult<object>()
{
Type = TriggerDescription.External,
Value = value
};
await channel.Writer.WriteAsync(result);
return true;
}
return false;
}
/// <summary>
/// 取消所有任务
/// </summary>
public void CancelAllTasks()
public void CancelAllTrigger()
{
foreach (var channel in _channels.Values)
{
@@ -89,16 +104,6 @@ namespace Serein.Library.Utils
}
_channels.Clear();
}
/// <summary>
/// 获取或创建指定信号的 Channel
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>对应的 Channel</returns>
private Channel<(TriggerType, object)> GetOrCreateChannel(TSignal signal)
{
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<(TriggerType, object)>());
}
}

18
Library/Utils/EnumHelper.cs
View File

@@ -77,7 +77,7 @@ namespace Serein.Library.Utils
/// <param name="enumValue">枚举值</param>
/// <param name="valueSelector">特性成员选择</param>
/// <returns></returns>
public static TResult GetBoundValue<TEnum, TAttribute, TResult>(TEnum enumValue,
public static TResult GetAttributeValue<TEnum, TAttribute, TResult>(TEnum enumValue,
Func<TAttribute, TResult> valueSelector)
where TEnum : Enum
where TAttribute : Attribute
@@ -88,6 +88,22 @@ namespace Serein.Library.Utils
return attribute != null ? valueSelector(attribute) : default;
}
/// <summary>
/// 从枚举值从获取自定义特性的成员，并自动转换类型
/// </summary>
/// <typeparam name="TEnum">枚举类型</typeparam>
/// <typeparam name="TAttribute">自定义特性类型</typeparam>
/// <param name="enumValue">枚举值</param>
/// <returns></returns>
public static TAttribute GetAttribute<TEnum, TAttribute>(TEnum enumValue)
where TEnum : Enum
where TAttribute : Attribute
{
var fieldInfo = typeof(TEnum).GetField(enumValue.ToString());
var attribute = fieldInfo.GetCustomAttribute<TAttribute>();
return attribute;
}
}

136
Library/Utils/SingleSyncFlowTrigger.cs Normal file
View File

@@ -0,0 +1,136 @@
using Newtonsoft.Json.Linq;
using Serein.Library.Api;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Subjects;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
namespace Serein.Library.Utils
{
/// <summary>
/// 同步的单体消息触发器
/// </summary>
/// <typeparam name="TSingle"></typeparam>
public class SingleSyncFlowTrigger<TSingle> : IFlowTrigger<TSingle>
{
private readonly ConcurrentDictionary<TSingle, Queue<TaskCompletionSource<TriggerResult<object>>>> _syncChannel
= new ConcurrentDictionary<TSingle, Queue<TaskCompletionSource<TriggerResult<object>>>>();
public void CancelAllTrigger()
{
foreach (var triggers in _syncChannel.Values)
{
foreach (var trigger in triggers)
{
trigger.SetCanceled();
}
}
}
public Task<bool> InvokeTriggerAsync<TResult>(TSingle signal, TResult value)
{
if(_syncChannel.TryGetValue(signal, out var tcss))
{
var tcs = tcss.Dequeue();
var result = new TriggerResult<object>
{
Type = TriggerDescription.External,
Value = value,
};
tcs.SetResult(result);
return Task.FromResult(true);
}
return Task.FromResult(false);
}
public async Task<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSingle signal)
{
if (!_syncChannel.TryGetValue(signal,out var tcss))
{
tcss = new Queue<TaskCompletionSource<TriggerResult<object>>>();
_syncChannel.TryAdd(signal, tcss);
}
var taskCompletionSource = new TaskCompletionSource<TriggerResult<object>>();
tcss.Enqueue(taskCompletionSource);
var result = await taskCompletionSource.Task;
if (result.Value is TResult result2)
{
return new TriggerResult<TResult>
{
Type = TriggerDescription.External,
Value = result2,
};
}
else
{
return new TriggerResult<TResult>
{
Type = TriggerDescription.TypeInconsistency,
};
}
}
public async Task<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSingle signal, TimeSpan outTime)
{
if (!_syncChannel.TryGetValue(signal, out var tcss))
{
tcss = new Queue<TaskCompletionSource<TriggerResult<object>>>();
_syncChannel.TryAdd(signal, tcss);
}
var taskCompletionSource = new TaskCompletionSource<TriggerResult<object>>();
tcss.Enqueue(taskCompletionSource);
var cts = new CancellationTokenSource();
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
{
try
{
await Task.Delay(outTime, cts.Token);
if (!cts.IsCancellationRequested) // 如果还没有被取消
{
var outResult = new TriggerResult<object>()
{
Type = TriggerDescription.Overtime
};
taskCompletionSource.SetResult(outResult); // 超时触发
}
}
catch (OperationCanceledException)
{
// 超时任务被取消
}
finally
{
cts?.Dispose(); // 确保 cts 被释放
}
}, cts.Token);
var result = await taskCompletionSource.Task;
cts?.Cancel();
if (result.Value is TResult result2)
{
return new TriggerResult<TResult>
{
Type = result.Type,
Value = result2,
};
}
else
{
return new TriggerResult<TResult>
{
Type = result.Type,
};
}
}
}
}

129
Library/Utils/FlowTrigger.cs → Library/Utils/TaskFlowTrigger.cs
View File

@@ -1,4 +1,6 @@
using Serein.Library.Utils;
using Newtonsoft.Json.Linq;
using Serein.Library.Api;
using Serein.Library.Utils;
using System;
using System.Collections.Concurrent;
using System.Reactive.Linq;
@@ -6,13 +8,16 @@ using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Transactions;
namespace Serein.Library
{
/// <summary>
/// 触发类型
/// </summary>
public enum TriggerType
public enum TriggerDescription
{
/// <summary>
/// 外部触发
@@ -21,32 +26,37 @@ namespace Serein.Library
/// <summary>
/// 超时触发
/// </summary>
Overtime
Overtime,
/// <summary>
/// 触发了，但类型不一致
/// </summary>
TypeInconsistency
}
public class TriggerResult<T>
public class TriggerResult<TResult>
{
public TriggerType Type { get; set; }
public T Value { get; set; }
public TriggerDescription Type { get; set; }
public TResult Value { get; set; }
}
/// <summary>
/// 信号触发器类，带有消息广播功能。
/// 使用枚举作为标记，创建
/// </summary>
public class FlowTrigger<TSignal> where TSignal : struct, Enum
public class TaskFlowTrigger<TSignal> : IFlowTrigger<TSignal> where TSignal : struct, Enum
{
// 使用并发字典管理每个信号对应的广播列表
private readonly ConcurrentDictionary<TSignal, Subject<(TriggerType,object)>> _subscribers = new ConcurrentDictionary<TSignal, Subject<(TriggerType, object)>>();
private readonly ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>> _subscribers = new ConcurrentDictionary<TSignal, Subject<TriggerResult<object>>>();
/// <summary>
/// 获取或创建指定信号的 Subject（消息广播者）
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>对应的 Subject</returns>
private Subject<(TriggerType, object)> GetOrCreateSubject(TSignal signal)
private Subject<TriggerResult<object>> GetOrCreateSubject(TSignal signal)
{
return _subscribers.GetOrAdd(signal, _ => new Subject<(TriggerType, object)>());
return _subscribers.GetOrAdd(signal, _ => new Subject<TriggerResult<object>>());
}
/// <summary>
@@ -55,20 +65,23 @@ namespace Serein.Library
/// <param name="signal">枚举信号标识符</param>
/// <param name="observer">订阅者</param>
/// <returns>取消订阅的句柄</returns>
public IDisposable Subscribe(TSignal signal, IObserver<(TriggerType, object)> observer)
private IDisposable Subscribe<TResult>(TSignal signal, Action<TriggerResult<object>> action)
{
IObserver<TriggerResult<object>> observer = new Observer<TriggerResult<object>>(action);
var subject = GetOrCreateSubject(signal);
// (IObserver<(TriggerType, object)>)
return subject.Subscribe(observer); // 返回取消订阅的句柄
}
/// <summary>
/// 创建信号并指定超时时间，触发时通知所有订阅者
/// 等待触发器并指定超时的时间
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <typeparam name="TResult">返回值类型</typeparam>
/// <param name="signal">等待信号</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务，返回值为：状态（超时触发，手动触发），数据（超时触发时会使用设置好的数据）</returns>
public async Task<(TriggerType, TResult)> CreateTaskWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime, TResult outValue)
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerWithTimeoutAsync<TResult>(TSignal signal, TimeSpan outTime)
{
var subject = GetOrCreateSubject(signal);
var cts = new CancellationTokenSource();
@@ -81,7 +94,11 @@ namespace Serein.Library
await Task.Delay(outTime, cts.Token);
if (!cts.IsCancellationRequested) // 如果还没有被取消
{
subject.OnNext((TriggerType.Overtime, outValue)); // 广播给所有订阅者
var outResult = new TriggerResult<object>()
{
Type = TriggerDescription.Overtime
};
subject.OnNext(outResult); // 广播给所有订阅者
subject.OnCompleted(); // 通知订阅结束
}
}
@@ -94,65 +111,67 @@ namespace Serein.Library
cts?.Dispose(); // 确保 cts 被释放
}
}, cts.Token);
var result = await WaitSignalAsync<TResult>(signal);// 返回一个可以超时触发的等待任务
var result = await WaitTriggerAsync<TResult>(signal); // 返回一个可以超时触发的等待任务
return result;
}
/// <summary>
/// 创建等待任务，触发时通知所有订阅者
/// 等待触发
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <param name="outTime">超时时间</param>
/// <returns>等待任务</returns>
public async Task<TResult> CreateTaskAsync<TResult>(TSignal signal)
/// <typeparam name="TResult"></typeparam>
/// <param name="signal"></param>
/// <returns></returns>
public async Task<TriggerResult<TResult>> WaitTriggerAsync<TResult>(TSignal signal)
{
var subject = GetOrCreateSubject(signal);
(_,var result) = await WaitSignalAsync<TResult>(signal);
return result;// 返回一个等待的任务
}
/// <summary>
/// 等待指定信号的触发
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>等待任务</returns>
public async Task<(TriggerType, TResult)> WaitSignalAsync<TResult>(TSignal signal)
{
var taskCompletionSource = new TaskCompletionSource<(TriggerType, object)>();
var subscription = Subscribe(signal, new Observer<(TriggerType, object)>(taskCompletionSource.SetResult));
(var type,var result) = await taskCompletionSource.Task;
var taskCompletionSource = new TaskCompletionSource<TriggerResult<object>>();
var subscription = Subscribe<TResult>(signal, taskCompletionSource.SetResult);
var result = await taskCompletionSource.Task;
subscription.Dispose(); // 取消订阅
return (type, result.ToConvert<TResult>());
if(result.Value is TResult data)
{
return new TriggerResult<TResult>()
{
Value = data,
Type = TriggerDescription.External,
};
}
else
{
return new TriggerResult<TResult>()
{
Type = TriggerDescription.TypeInconsistency,
};
}
}
/// <summary>
/// 手动触发信号，并广播给所有订阅者
/// </summary>
/// <typeparam name="TResult">触发类型</typeparam>
/// <param name="signal">枚举信号标识符</param>
/// <param name="value">传递的数据</param>
/// <returns>是否成功触发</returns>
public bool Trigger<TResult>(TSignal signal, TResult value)
public Task<bool> InvokeTriggerAsync<TResult>(TSignal signal, TResult value)
{
if (_subscribers.TryGetValue(signal, out var subject))
{
subject.OnNext((TriggerType.External, value)); // 广播给所有订阅者
//subject.OnCompleted(); // 通知订阅结束
return true;
var result = new TriggerResult<object>()
{
Type = TriggerDescription.External,
Value = value
};
subject.OnNext(result); // 广播给所有订阅者
subject.OnCompleted(); // 通知订阅结束
return Task.FromResult(true);
}
return false;
return Task.FromResult(false);
}
/// <summary>
/// 取消所有任务
/// </summary>
public void CancelAllTasks()
public void CancelAllTrigger()
{
foreach (var subject in _subscribers.Values)
{
@@ -160,9 +179,11 @@ namespace Serein.Library
}
_subscribers.Clear();
}
}
/// <summary>
/// 观察者类，用于包装 Action
/// </summary>