From 0bab770f0a913dd5f2ead7d6cb7f98a26b1f84be Mon Sep 17 00:00:00 2001
From: fengjiayi <12821976+ning_xi@user.noreply.gitee.com>
Date: Tue, 8 Oct 2024 12:01:10 +0800
Subject: [PATCH] =?UTF-8?q?=E5=9C=A8ChannelFlowTrigger=E7=9A=84=E5=9F=BA?=
=?UTF-8?q?=E7=A1=80=E4=B8=8A=E6=B7=BB=E5=8A=A0=E4=BA=86=E8=A7=82=E5=AF=9F?=
=?UTF-8?q?=E8=80=85=E6=A8=A1=E5=BC=8F=EF=BC=8C=E4=BD=BF=E5=85=B6=E5=85=B7?=
=?UTF-8?q?=E5=A4=87=E9=80=9A=E7=9F=A5=E5=A4=9A=E4=B8=AA=E6=B6=88=E8=B4=B9?=
=?UTF-8?q?=E8=80=85=E7=9A=84=E8=83=BD=E5=8A=9B=E3=80=82?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
Library/Serein.Library.csproj | 1 +
Library/Utils/ChannelFlowTrigger.cs | 100 +++++++++++++-----
Net462DllTest/LogicControl/PlcLogicControl.cs | 1 +
.../LogicControl/ViewLogicControl.cs | 34 +++---
NodeFlow/FlowStarter.cs | 3 +-
5 files changed, 92 insertions(+), 47 deletions(-)
diff --git a/Library/Serein.Library.csproj b/Library/Serein.Library.csproj
index cfe4b49..b1a2919 100644
--- a/Library/Serein.Library.csproj
+++ b/Library/Serein.Library.csproj
@@ -16,6 +16,7 @@
+
diff --git a/Library/Utils/ChannelFlowTrigger.cs b/Library/Utils/ChannelFlowTrigger.cs
index 917abce..9363c0b 100644
--- a/Library/Utils/ChannelFlowTrigger.cs
+++ b/Library/Utils/ChannelFlowTrigger.cs
@@ -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
{
///
@@ -22,50 +21,69 @@ namespace Serein.Library.NodeFlow.Tool
///
Overtime
}
+
public class TriggerData
{
public TriggerType Type { get; set; }
public object Value { get; set; }
}
-
+ ///
+ /// 信号触发器类，带有消息广播功能
+ ///
public class ChannelFlowTrigger where TSignal : struct, Enum
{
- // 使用并发字典管理每个枚举信号对应的 Channel
- private readonly ConcurrentDictionary> _channels = new ConcurrentDictionary>();
+ // 使用并发字典管理每个信号对应的广播列表
+ private readonly ConcurrentDictionary> _subscribers = new ConcurrentDictionary>();
+
///
- /// 获取或创建指定信号的 Channel
+ /// 获取或创建指定信号的 Subject（消息广播者）
///
/// 枚举信号标识符
- /// 对应的 Channel
- private Channel GetOrCreateChannel(TSignal signal)
+ /// 对应的 Subject
+ private Subject GetOrCreateSubject(TSignal signal)
{
- return _channels.GetOrAdd(signal, _ => Channel.CreateUnbounded());
+ return _subscribers.GetOrAdd(signal, _ => new Subject());
}
+
///
- /// 创建信号并指定超时时间的Channel.
+ /// 订阅指定信号的消息
+ ///
+ /// 枚举信号标识符
+ /// 订阅者
+ /// 取消订阅的句柄
+ public IDisposable Subscribe(TSignal signal, IObserver observer)
+ {
+ var subject = GetOrCreateSubject(signal);
+ return subject.Subscribe(observer); // 返回取消订阅的句柄
+ }
+
+ ///
+ /// 创建信号并指定超时时间，触发时通知所有订阅者
///
/// 枚举信号标识符
/// 超时时间
/// 等待任务
public async Task CreateChannelWithTimeoutAsync(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);
-
- // 等待信号传入（超时或手动触发）
- var result = await channel.Reader.ReadAsync();
- cts?.Cancel();
+ // 返回一个等待的任务
+ return await WaitForSignalAsync(signal);
+ }
+
+ ///
+ /// 等待指定信号的触发
+ ///
+ /// 枚举信号标识符
+ /// 等待任务
+ public async Task WaitForSignalAsync(TSignal signal)
+ {
+ var taskCompletionSource = new TaskCompletionSource();
+ var subscription = Subscribe(signal, new Observer(taskCompletionSource.SetResult));
+ var result = await taskCompletionSource.Task;
+ subscription.Dispose(); // 取消订阅
return result;
}
///
- /// 触发信号
+ /// 手动触发信号，并广播给所有订阅者
///
/// 枚举信号标识符
/// 是否成功触发
public bool TriggerSignal(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
///
public void CancelAllTasks()
{
- foreach (var channel in _channels.Values)
+ foreach (var subject in _subscribers.Values)
{
- channel.Writer.Complete();
+ subject.OnCompleted(); // 通知所有订阅者结束
}
- _channels.Clear();
+ _subscribers.Clear();
}
+ }
+ ///
+ /// 观察者类，用于包装 Action
+ ///
+ public class Observer : IObserver
+ {
+ private readonly Action _onNext;
+
+ public Observer(Action onNext)
+ {
+ _onNext = onNext;
+ }
+
+ public void OnCompleted() { }
+ public void OnError(Exception error) { }
+ public void OnNext(T value)
+ {
+ _onNext?.Invoke(value);
+ }
}
}
diff --git a/Net462DllTest/LogicControl/PlcLogicControl.cs b/Net462DllTest/LogicControl/PlcLogicControl.cs
index a795463..a0f7940 100644
--- a/Net462DllTest/LogicControl/PlcLogicControl.cs
+++ b/Net462DllTest/LogicControl/PlcLogicControl.cs
@@ -87,6 +87,7 @@ namespace Net462DllTest.LogicControl
[NodeAction(NodeType.Flipflop, "等待变量更新", ReturnType = typeof(int))]
public async Task WaitTask(PlcVarName varName = PlcVarName.ErrorCode)
{
+
try
{
TriggerData triggerData = await MyPlc.CreateChannelWithTimeoutAsync(varName, TimeSpan.FromMinutes(120), 0);
diff --git a/Net462DllTest/LogicControl/ViewLogicControl.cs b/Net462DllTest/LogicControl/ViewLogicControl.cs
index bb09dc8..424dfda 100644
--- a/Net462DllTest/LogicControl/ViewLogicControl.cs
+++ b/Net462DllTest/LogicControl/ViewLogicControl.cs
@@ -59,15 +59,10 @@ namespace Net462DllTest.LogicControl
this.ViewManagement = ViewManagement;
}
- [NodeAction(NodeType.Init)]
- public void Init(IDynamicContext context)
- {
- context.Env.IOC.Register();
- }
#region 触发器节点
- [NodeAction(NodeType.Flipflop, "等待信号触发", ReturnType = typeof(int))]
+ [NodeAction(NodeType.Flipflop, "等待视图命令", ReturnType = typeof(int))]
public async Task 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(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(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(fromId, attr => attr.Value);
diff --git a/NodeFlow/FlowStarter.cs b/NodeFlow/FlowStarter.cs
index dc10b97..61eff64 100644
--- a/NodeFlow/FlowStarter.cs
+++ b/NodeFlow/FlowStarter.cs
@@ -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)
{