From b544d5cccd76735e034c8055918a50927b437474 Mon Sep 17 00:00:00 2001
From: wenxiangchao <wenxiangchao@ihengxin.com.cn>
Date: 星期二, 10 十一月 2026 08:29:04 +0800
Subject: [PATCH] fix: 维护模式重复进入失败与按钮灰死 + Broker 并发验证用例
---
Aging.Infrastructure/Drivers/BrokerRpcTransport.cs | 152 ++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 152 insertions(+), 0 deletions(-)
diff --git a/Aging.Infrastructure/Drivers/BrokerRpcTransport.cs b/Aging.Infrastructure/Drivers/BrokerRpcTransport.cs
new file mode 100644
index 0000000..26796d6
--- /dev/null
+++ b/Aging.Infrastructure/Drivers/BrokerRpcTransport.cs
@@ -0,0 +1,152 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using Aging.Infrastructure.Broker;
+using Aging.ResourceContracts;
+using log4net;
+
+namespace Aging.Infrastructure.Drivers
+{
+ /// <summary>
+ /// Broker 浼犺緭鍚庣锛氬懡浠ょ粡鍛藉悕绠￠亾 RPC 濮旀墭 DeviceBroker 瀹夸富鎵ц锛堢墿鐞嗕細璇濈敱瀹夸富鍞竴鎸佹湁锛夈��
+ /// 涓夐噸鍏冲崱锛堢櫧鍚嶅崟/configHash/绉熺害鏍呮爮锛変笌 SessionGate 鍏ㄥ湪瀹夸富渚э紝鏈被鍙仛鎻愪氦涓庢�佹槧灏勩��
+ /// 鎺ユ敹甯х敱瀹夸富闅忓搷搴斾互 "|PUSHED:n|EV:RECV鈥�" 鍚庣紑鎹庡甫杩斿洖锛岃В鏋愬悗缁� FramePushed 涓婃姏锛埪�5.4 鎹庡甫妯″瀷锛夈��
+ /// </summary>
+ internal sealed class BrokerRpcTransport : ICanTransport
+ {
+ private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerRpcTransport));
+
+ private readonly BrokerClientChannel _channel;
+ private readonly string _deviceId;
+ private readonly string _configHash;
+ private bool _disposed;
+
+ public bool IsConnected => _channel.IsConnected;
+ public event Action<string> FramePushed;
+ public event Action TransportBroken;
+
+ public BrokerRpcTransport(string deviceId, string pipeName, string configHash)
+ {
+ _deviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
+ _configHash = configHash ?? "";
+ string instanceId = $"{Environment.MachineName}/{Process.GetCurrentProcess().Id}";
+ _channel = new BrokerClientChannel(pipeName, instanceId);
+ }
+
+ // busName/nodeIndex 瀵� Broker 鏃犳剰涔夛紙鐗╃悊浼氳瘽鍦ㄥ涓绘寜璁惧绔偣寤虹珛锛夛紝CONNECT 鍗虫敞鍐�
+ public Task ConnectAsync(string busName, int nodeIndex, CancellationToken ct)
+ => SubmitExpectOkAsync("CONNECT", "", ct);
+
+ public Task DisconnectAsync()
+ => SubmitExpectOkAsync("DISCONNECT", "", CancellationToken.None);
+
+ public async Task<string> TransceiveAsync(string line, CancellationToken ct)
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(BrokerRpcTransport));
+
+ // line 褰㈠ "SEND:鈥�" 鎴� "QUERY:鈥�"锛屾槧灏勫洖瀹夸富鐧藉悕鍗� opCode + payload锛�
+ // SEND 鈫� 瀹夸富鐨� device.Driver.SendCommandAsync 闇�瀹屾暣 "SEND:frameId:data"锛宲ayload 淇濈暀鍓嶇紑
+ // QUERY 鈫� 瀹夸富鐨� device.Driver.QueryAsync 闇�鍘� QUERY: 鍓嶇紑鐨勮８鏌ヨ
+ string opCode;
+ string payload;
+ if (line.StartsWith("SEND:", StringComparison.Ordinal))
+ {
+ opCode = "SEND";
+ payload = line;
+ }
+ else if (line.StartsWith("QUERY:", StringComparison.Ordinal))
+ {
+ opCode = "QUERY";
+ payload = line.Substring("QUERY:".Length);
+ }
+ else
+ {
+ opCode = "QUERY";
+ payload = line;
+ }
+
+ var result = await SubmitAsync(opCode, payload, ct).ConfigureAwait(false);
+ DrainPiggyback(result.Readback, out string clean);
+ return clean;
+ }
+
+ private async Task SubmitExpectOkAsync(string opCode, string payload, CancellationToken ct)
+ {
+ var r = await SubmitAsync(opCode, payload, ct).ConfigureAwait(false);
+ // CONNECT/DISCONNECT 鏃犳湁鏁堣鍥炲抚锛屽拷鐣ユ崕甯�
+ _ = r.Readback;
+ }
+
+ private async Task<BrokerIpc.ResultBody> SubmitAsync(string opCode, string payload, CancellationToken ct)
+ {
+ try
+ {
+ await _channel.EnsureConnectedAsync(ct).ConfigureAwait(false);
+ var body = new BrokerIpc.SubmitBody
+ {
+ DeviceId = _deviceId,
+ OperationCode = opCode,
+ Payload = payload ?? "",
+ ResourceKey = _deviceId,
+ ClientInstanceId = $"{Environment.MachineName}/{Process.GetCurrentProcess().Id}"
+ };
+ var r = await _channel.SubmitAsync(body, _configHash, ct).ConfigureAwait(false);
+ ThrowForState(opCode, r);
+ return r;
+ }
+ catch (Exception ex) when (!(ex is ObjectDisposedException))
+ {
+ Log.Error($"[BrokerRpcCAN] {_deviceId} {opCode} 澶辫触: {ex.Message}", ex);
+ throw;
+ }
+ }
+
+ /// <summary>浠� readback 灏鹃儴鍓ョ |PUSHED:n|EV:RECV鈥� 娈靛苟瑙﹀彂 FramePushed銆�</summary>
+ private void DrainPiggyback(string readback, out string clean)
+ {
+ clean = readback ?? "";
+ if (string.IsNullOrEmpty(clean)) return;
+
+ int idx = clean.IndexOf("|PUSHED:", StringComparison.Ordinal);
+ if (idx < 0) return;
+
+ string pushSection = clean.Substring(idx + "|PUSHED:".Length);
+ clean = clean.Substring(0, idx);
+
+ // pushSection 褰㈠ "n|EV:RECV:鈥EV:RECV:鈥�"
+ var segs = pushSection.Split('|');
+ for (int i = 1; i < segs.Length; i++)
+ {
+ if (segs[i].StartsWith("EV:", StringComparison.Ordinal))
+ FramePushed?.Invoke(segs[i]);
+ }
+ }
+
+ private void ThrowForState(string opCode, BrokerIpc.ResultBody r)
+ {
+ switch (r.State)
+ {
+ case "Succeeded":
+ return;
+ case "RejectedStaleLease":
+ throw new InvalidOperationException($"LEASE_STALE: {_deviceId} 绉熺害鏍呮爮鏍￠獙澶辫触: {r.ErrorCode} {r.Readback}");
+ case "RejectedConfigMismatch":
+ throw new InvalidOperationException($"CONFIG_MISMATCH: Broker 鐢熸晥閰嶇疆涓庢湰鍦板揩鐓т笉涓�鑷�(INV-04): {r.Readback}");
+ case "RejectedQueueFull":
+ throw new InvalidOperationException($"QUEUE_FULL: {r.Readback}");
+ case "Unknown":
+ throw new TimeoutException($"UNKNOWN: {_deviceId} 鍔ㄤ綔 {opCode} 缁撴灉鏈煡锛岀姝㈤噸璇曞亣瀹氭湭鎵ц(INV-06): {r.Readback}");
+ default:
+ throw new InvalidOperationException($"Broker 鎵ц澶辫触[{r.ErrorCode}]: {r.Readback}");
+ }
+ }
+
+ public void Dispose()
+ {
+ if (_disposed) return;
+ _disposed = true;
+ _channel.Dispose();
+ }
+ }
+}
--
Gitblit v1.9.1