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

在ChannelFlowTrigger的基础上添加了观察者模式，使其具备通知多个消费者的能力。

Browse Source
This commit is contained in:
fengjiayi
2024-10-08 12:01:10 +08:00
parent 21a6f30c3a
commit 0bab770f0a
5 changed files with 92 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Library/Serein.Library.csproj
View File

@@ -16,6 +16,7 @@
<ItemGroup>
<PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
<PackageReference Include="System.Reactive" Version="6.0.1" />
<PackageReference Include="System.Threading.Channels" Version="8.0.0" />
<PackageReference Include="System.ValueTuple" Version="4.5.0" />
</ItemGroup>

98
Library/Utils/ChannelFlowTrigger.cs
View File

@@ -1,11 +1,10 @@
using System;
using System;
using System.Collections.Concurrent;
using System.Reactive.Subjects;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
namespace Serein.Library.NodeFlow.Tool
{
/// <summary>
@@ -22,50 +21,69 @@ namespace Serein.Library.NodeFlow.Tool
/// </summary>
Overtime
}
public class TriggerData
{
public TriggerType Type { get; set; }
public object Value { get; set; }
}
/// <summary>
/// 信号触发器类，带有消息广播功能
/// </summary>
public class ChannelFlowTrigger<TSignal> where TSignal : struct, Enum
{
// 使用并发字典管理每个枚举信号对应的 Channel
private readonly ConcurrentDictionary<TSignal, Channel<TriggerData>> _channels = new ConcurrentDictionary<TSignal, Channel<TriggerData>>();
// 使用并发字典管理每个信号对应的广播列表
private readonly ConcurrentDictionary<TSignal, Subject<TriggerData>> _subscribers = new ConcurrentDictionary<TSignal, Subject<TriggerData>>();
/// <summary>
/// 获取或创建指定信号的 Channel
/// 获取或创建指定信号的 Subject（消息广播者）
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>对应的 Channel</returns>
private Channel<TriggerData> GetOrCreateChannel(TSignal signal)
/// <returns>对应的 Subject</returns>
private Subject<TriggerData> GetOrCreateSubject(TSignal signal)
{
return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded<TriggerData>());
return _subscribers.GetOrAdd(signal, _ => new Subject<TriggerData>());
}
/// <summary>
/// 创建信号并指定超时时间的Channel.
/// 订阅指定信号的消息
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <param name="observer">订阅者</param>
/// <returns>取消订阅的句柄</returns>
public IDisposable Subscribe(TSignal signal, IObserver<TriggerData> observer)
{
var subject = GetOrCreateSubject(signal);
return subject.Subscribe(observer); // 返回取消订阅的句柄
}
/// <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 subject = GetOrCreateSubject(signal);
var cts = new CancellationTokenSource();
// 异步任务：超时后自动触发信号
_ = Task.Run(async () =>
{
try
{
await Task.Delay(outTime, cts.Token);
if(!cts.IsCancellationRequested) // 如果还没有被取消
if (!cts.IsCancellationRequested) // 如果还没有被取消
{
TriggerData triggerData = new TriggerData()
{
Value = outValue,
Type = TriggerType.Overtime,
};
await channel.Writer.WriteAsync(triggerData);
subject.OnNext(triggerData); // 广播给所有订阅者
subject.OnCompleted(); // 通知订阅结束
}
}
catch (OperationCanceledException)
@@ -74,35 +92,44 @@ namespace Serein.Library.NodeFlow.Tool
}
finally
{
cts?.Cancel();
cts?.Dispose(); // 确保 cts 被释放
}
}, cts.Token);
// 返回一个等待的任务
return await WaitForSignalAsync(signal);
}
// 等待信号传入（超时或手动触发）
var result = await channel.Reader.ReadAsync();
cts?.Cancel();
/// <summary>
/// 等待指定信号的触发
/// </summary>
/// <param name="signal">枚举信号标识符</param>
/// <returns>等待任务</returns>
public async Task<TriggerData> WaitForSignalAsync(TSignal signal)
{
var taskCompletionSource = new TaskCompletionSource<TriggerData>();
var subscription = Subscribe(signal, new Observer<TriggerData>(taskCompletionSource.SetResult));
var result = await taskCompletionSource.Task;
subscription.Dispose(); // 取消订阅
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))
if (_subscribers.TryGetValue(signal, out var subject))
{
TriggerData triggerData = new TriggerData()
{
Value = value,
Type = TriggerType.External,
};
// 手动触发信号
channel.Writer.TryWrite(triggerData);
subject.OnNext(triggerData); // 广播给所有订阅者
//subject.OnCompleted(); // 通知订阅结束
return true;
}
return false;
@@ -113,13 +140,32 @@ namespace Serein.Library.NodeFlow.Tool
/// </summary>
public void CancelAllTasks()
{
foreach (var channel in _channels.Values)
foreach (var subject in _subscribers.Values)
{
channel.Writer.Complete();
subject.OnCompleted(); // 通知所有订阅者结束
}
_channels.Clear();
_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);
}
}
}

1
Net462DllTest/LogicControl/PlcLogicControl.cs
View File

@@ -87,6 +87,7 @@ namespace Net462DllTest.LogicControl
[NodeAction(NodeType.Flipflop, "等待变量更新", ReturnType = typeof(int))]
public async Task<IFlipflopContext> WaitTask(PlcVarName varName = PlcVarName.ErrorCode)
{
try
{
TriggerData triggerData = await MyPlc.CreateChannelWithTimeoutAsync(varName, TimeSpan.FromMinutes(120), 0);

34
Net462DllTest/LogicControl/ViewLogicControl.cs
View File

@@ -59,15 +59,10 @@ namespace Net462DllTest.LogicControl
this.ViewManagement = ViewManagement;
}
[NodeAction(NodeType.Init)]
public void Init(IDynamicContext context)
{
context.Env.IOC.Register<ViewManagement>();
}
#region 触发器节点
[NodeAction(NodeType.Flipflop, "等待信号触发", ReturnType = typeof(int))]
[NodeAction(NodeType.Flipflop, "等待视图命令", ReturnType = typeof(int))]
public async Task<IFlipflopContext> WaitTask(CommandSignal command = CommandSignal.Command_1)
{
try
@@ -91,26 +86,29 @@ namespace Net462DllTest.LogicControl
}
#endregion
[NodeAction(NodeType.Action, "打开窗体（指定枚举值）")]
public void OpenForm(IDynamicContext context, FromValue fromId = FromValue.FromWorkBenchView, bool isTop = true)
{
var fromType = EnumHelper.GetBoundValue<FromValue, Type>(fromId, attr => attr.Value);
if (fromType is null) return;
if (context.Env.IOC.Instantiate(fromType) is Form form)
{
ViewManagement.OpenView(form, isTop);
}
}
//[NodeAction(NodeType.Action, "打开窗体（指定枚举值）")]
//public void OpenForm(IDynamicContext context,
// FromValue fromId = FromValue.FromWorkBenchView,
// bool isTop = true)
//{
// var fromType = EnumHelper.GetBoundValue<FromValue, Type>(fromId, attr => attr.Value);
// if (fromType is null) return;
// if (context.Env.IOC.Instantiate(fromType) is Form form)
// {
// ViewManagement.OpenView(form, isTop);
// }
//}
[NodeAction(NodeType.Action, "打开窗体（使用转换器）")]
[NodeAction(NodeType.Action, "打开窗体（转换器）")]
public void OpenForm2([EnumTypeConvertor(typeof(FromValue))] Form form, bool isTop = true)
{
// 枚举转换为对应的Type并自动实例化
ViewManagement.OpenView(form, isTop);
}
[NodeAction(NodeType.Action, "关闭窗体")]
[NodeAction(NodeType.Action, "关闭指定类型的所有窗体")]
public void CloseForm(IDynamicContext context, FromValue fromId = FromValue.FromWorkBenchView)
{
var fromType = EnumHelper.GetBoundValue<FromValue, Type>(fromId, attr => attr.Value);

3
NodeFlow/FlowStarter.cs
View File

@@ -371,8 +371,7 @@ namespace Serein.NodeFlow
{
try
{
var waitTask = singleFlipFlopNode.ExecutingAsync(context); // 获取触发器等待Task
var newFlowData = await waitTask;
var newFlowData = await singleFlipFlopNode.ExecutingAsync(context); // 获取触发器等待Task
await NodeModelBase.RefreshFlowDataAndExpInterrupt(context, singleFlipFlopNode, newFlowData); // 全局触发器触发后刷新该触发器的节点数据
if (singleFlipFlopNode.NextOrientation != ConnectionType.None)
{