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/CanDeviceDriver.cs | 334 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 334 insertions(+), 0 deletions(-)
diff --git a/Aging.Infrastructure/Drivers/CanDeviceDriver.cs b/Aging.Infrastructure/Drivers/CanDeviceDriver.cs
new file mode 100644
index 0000000..98f4c37
--- /dev/null
+++ b/Aging.Infrastructure/Drivers/CanDeviceDriver.cs
@@ -0,0 +1,334 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Text.RegularExpressions;
+using System.Threading;
+using System.Threading.Tasks;
+using Aging.Infrastructure.Contracts;
+using CanSimulator;
+using log4net;
+
+namespace Aging.Infrastructure.Drivers
+{
+ /// <summary>
+ /// 鍏ㄧ郴缁熷敮涓�鐨� CAN 璁惧椹卞姩銆傚疄鐜� IDeviceDriver锛屽懡浠よ涔夛紙SEND/QUERY锛変笌鍘熼�傞厤鍣ㄤ竴鑷淬��
+ /// Local 涓� Broker 鐨勫樊寮備笅娌夊埌鏋勯�犳椂閫夋嫨鐨� <see cref="ICanTransport"/> 鍚庣锛�
+ /// Local 鈫� <see cref="NodeChannelTransport"/>锛圱CP 鐩磋繛 CanSimulatorServer锛�
+ /// Broker 鈫� <see cref="BrokerRpcTransport"/>锛堝懡鍚嶇閬撳鎵� DeviceBroker 瀹夸富锛�
+ /// 鎺ユ敹甯х敱鍚庣 FramePushed 浜嬩欢椹卞姩锛岃繘鍏ユ湰鍦扮紦鍐插苟瑙﹀彂 <see cref="DriverFrameReceived"/>锛�
+ /// 浣胯皟璇曢潰鏉夸笌鑷姩鍖栫▼搴忓涓ょ鎵ц妯″紡瀹屽叏鏃犳劅鐭ャ��
+ /// </summary>
+ public sealed class CanDeviceDriver : IDeviceDriver
+ {
+ private static readonly ILog Log = LogManager.GetLogger(typeof(CanDeviceDriver));
+
+ // 淇℃伅鍨嬫煡璇細鍝嶅簲鍘绘帀浼犺緭灞傜殑 OK: 鍓嶇紑锛岃繕鍘熶负閬楃暀 "VERB:payload" 褰㈡�侊紙闈㈡澘鎸夋瑙ｆ瀽锛�
+ private static readonly Regex InfoVerb =
+ new(@"^OK:([A-Z][A-Z_]*):", RegexOptions.Compiled);
+
+ private readonly ICanTransport _transport;
+ private readonly string _busName;
+ private readonly int _nodeIndex;
+ private readonly Queue<CanFrame> _recvBuffer = new();
+ private readonly object _bufferLock = new();
+
+ /// <summary>瀹夸富渚у緟鎹庡甫鎺ㄩ�佽锛圗V:RECV 鍘熸枃锛夛紝渚� BrokerServer 闅忓搷搴旇拷鍔犵粰瀹㈡埛绔��</summary>
+ private readonly Queue<string> _pendingPushLines = new();
+ private const int MaxPendingPushLines = 1000;
+ private readonly Stopwatch _latency = new();
+
+ private bool _disposed;
+ private string _lastError = string.Empty;
+ private DateTime? _lastErrorTime;
+ private long _txCount;
+ private int _reconnectCount;
+
+ public string DeviceId { get; }
+ public bool IsConnected => !_disposed && _transport.IsConnected;
+
+ public event EventHandler<DeviceConnectionEventArgs>? ConnectionChanged;
+
+ /// <summary>鎺ユ敹甯у嵆鏃朵簨浠讹紙Local/Broker 缁熶竴鐢卞悗绔帹閫侀┍鍔級</summary>
+ public event EventHandler<CanFrame>? DriverFrameReceived;
+
+ /// <summary>Local 璁惧锛氫粠绔偣瑙ｆ瀽 bus:node锛孴CP 鐩磋繛妯℃嫙鍣紙host/port 鍙栭厤缃垨瑕嗙洊鍊硷級銆�</summary>
+ public CanDeviceDriver(string deviceId, string endpoint,
+ string hostOverride = null, int? portOverride = null)
+ {
+ DeviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
+ (_busName, _nodeIndex) = ParseEndpoint(endpoint);
+ _transport = new NodeChannelTransport(hostOverride, portOverride, deviceId);
+ WireTransport();
+ }
+
+ /// <summary>Broker 璁惧锛氭敞鍏ヤ唬鐞嗕紶杈撳悗绔紙鐢� DriverFactory 渚� ExecutionMode 鏋勯�狅級銆�</summary>
+ internal CanDeviceDriver(string deviceId, string endpoint, ICanTransport transport)
+ {
+ DeviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
+ (_busName, _nodeIndex) = ParseEndpoint(endpoint);
+ _transport = transport ?? throw new ArgumentNullException(nameof(transport));
+ WireTransport();
+ }
+
+ /// <summary>Broker 璁惧浠ｇ悊宸ュ巶锛氬鎴风/Bootstrapper 鎹鏋勯�犲鎵樺涓荤殑缁熶竴椹卞姩銆�</summary>
+ public static CanDeviceDriver CreateBrokerProxy(string deviceId, string endpoint, string pipeName, string configHash)
+ => new CanDeviceDriver(deviceId, endpoint, new BrokerRpcTransport(deviceId, pipeName, configHash));
+
+ private void WireTransport()
+ {
+ _transport.FramePushed += OnFramePushed;
+ _transport.TransportBroken += () => OnConnection(false, "閫氶亾鏂紑");
+ }
+
+ private static (string, int) ParseEndpoint(string endpoint)
+ {
+ if (string.IsNullOrWhiteSpace(endpoint))
+ throw new ArgumentException("CAN 绔偣涓嶈兘涓虹┖", nameof(endpoint));
+ var parts = endpoint.Split(':');
+ if (parts.Length < 2 || !int.TryParse(parts[1], out int node) || node < 0)
+ throw new ArgumentException($"CAN 绔偣鏍煎紡搴斾负 鎬荤嚎鍚嶇О:鑺傜偣缂栧彿锛屽疄闄�: {endpoint}");
+ return (parts[0], node);
+ }
+
+ public async Task ConnectAsync(CancellationToken ct)
+ {
+ ThrowIfDisposed();
+ _latency.Restart();
+ try
+ {
+ await _transport.ConnectAsync(_busName, _nodeIndex, ct).ConfigureAwait(false);
+ _lastError = string.Empty;
+ OnConnection(true, "杩炴帴鎴愬姛");
+ }
+ catch (Exception ex)
+ {
+ _lastError = ex.Message;
+ _lastErrorTime = DateTime.Now;
+ Log.Error($"[CAN] 璁惧 {DeviceId} 杩炴帴澶辫触: {ex.Message}", ex);
+ throw;
+ }
+ finally
+ {
+ if (_latency.IsRunning) _latency.Stop();
+ }
+ }
+
+ public async Task DisconnectAsync()
+ {
+ ThrowIfDisposed();
+ try { await _transport.DisconnectAsync().ConfigureAwait(false); }
+ finally { OnConnection(false, "涓诲姩鏂紑"); }
+ }
+
+ public async Task<string> SendCommandAsync(string command, CancellationToken ct)
+ {
+ ThrowIfDisposed();
+ await EnsureConnectedAsync(ct).ConfigureAwait(false);
+ _latency.Restart();
+ try
+ {
+ // 鍛戒护涓插師鏍烽�忎紶锛圫END:frameId:data锛夛紱鍝嶅簲鍘� OK: 浼犺緭鍓嶇紑浜ょ敱 QUERY 璇箟锛�
+ // 鍙戦�佸搷搴斾繚鐣欓仐鐣� "OK:TX:鈥�" 褰㈡�侊紙涓婂眰涓庢祴璇曟寜 OK:TX 鏂█锛�
+ string resp = await _transport.TransceiveAsync(command, ct).ConfigureAwait(false);
+ Interlocked.Increment(ref _txCount);
+ return resp;
+ }
+ catch (Exception ex)
+ {
+ RecordError(ex);
+ throw;
+ }
+ finally
+ {
+ if (_latency.IsRunning) _latency.Stop();
+ }
+ }
+
+ public async Task<string> QueryAsync(string query, CancellationToken ct)
+ {
+ ThrowIfDisposed();
+ await EnsureConnectedAsync(ct).ConfigureAwait(false);
+ _latency.Restart();
+ try
+ {
+ string op = (query ?? "").Split(':')[0].ToUpperInvariant();
+
+ // 鎺ユ敹绫绘煡璇㈢敱鏈湴缂撳啿搴旂瓟锛堟帹閫佸凡瀹炴椂鍏ラ槦锛夛紝鏃犻渶寰�杩旀ā鎷熷櫒
+ if (op == "RECV") return HandleRecv();
+ if (op == "RECV_ALL") return HandleRecvAll();
+
+ string resp = await _transport.TransceiveAsync("QUERY:" + query, ct).ConfigureAwait(false);
+ return NormalizeResponse(resp);
+ }
+ catch (Exception ex)
+ {
+ RecordError(ex);
+ throw;
+ }
+ finally
+ {
+ if (_latency.IsRunning) _latency.Stop();
+ }
+ }
+
+ public DeviceDriverStatus GetStatus() => new DeviceDriverStatus
+ {
+ IsConnected = IsConnected,
+ IsHealthy = IsConnected && string.IsNullOrEmpty(_lastError),
+ LastError = _lastError,
+ LastErrorTime = _lastErrorTime,
+ LatencyMs = _latency.IsRunning ? _latency.Elapsed.TotalMilliseconds : 0,
+ ReconnectCount = _reconnectCount
+ };
+
+ // ==================== 鎺ㄩ�佷笌缂撳啿 ====================
+
+ /// <summary>鍚庣 EV:RECV 琛� 鈫� 瑙ｆ瀽涓哄抚 鈫� 鍏ユ湰鍦扮紦鍐� + 瑙﹀彂鍗虫椂浜嬩欢銆�</summary>
+ private void OnFramePushed(string line)
+ {
+ if (_disposed) return;
+ try
+ {
+ var frame = ParsePush(line);
+ if (frame == null) return;
+
+ lock (_bufferLock)
+ {
+ while (_recvBuffer.Count >= 1000) _recvBuffer.Dequeue();
+ _recvBuffer.Enqueue(frame);
+
+ // 绱Н鍘熷鎺ㄩ�佽渚涘涓绘崕甯︾粰杩滅瀹㈡埛绔紙婊″垯涓㈡渶鑰侊級
+ while (_pendingPushLines.Count >= MaxPendingPushLines) _pendingPushLines.Dequeue();
+ _pendingPushLines.Enqueue(line);
+ }
+ DriverFrameReceived?.Invoke(this, frame);
+ }
+ catch (Exception ex)
+ {
+ Log.Warn($"[CAN] {DeviceId} 鎺ㄩ�佸抚瑙ｆ瀽澶辫触: {ex.Message} [{line}]");
+ }
+ }
+
+ private static CanFrame ParsePush(string line)
+ {
+ // EV:RECV:{frameIdHex}:{dataHex}:{ts}:{srcNodeIndex}
+ if (!line.StartsWith("EV:RECV:", StringComparison.Ordinal)) return null;
+ var body = line.Substring("EV:RECV:".Length);
+ var parts = body.Split(':');
+ if (parts.Length < 2) return null;
+
+ uint id = CanFrame.ParseFrameId(parts[0]);
+ byte[] data = parts.Length >= 2 ? CanFrame.ParseData(parts[1]) : Array.Empty<byte>();
+ return new CanFrame
+ {
+ FrameId = id,
+ Format = id > 0x7FF ? CanFrameFormat.Extended : CanFrameFormat.Standard,
+ Type = CanFrameType.Data,
+ Data = data,
+ SourceNodeId = parts.Length >= 4 ? parts[3] : "",
+ Timestamp = DateTime.Now
+ };
+ }
+
+ private string HandleRecv()
+ {
+ lock (_bufferLock)
+ {
+ if (_recvBuffer.Count == 0) return "RECV:EMPTY";
+ var f = _recvBuffer.Dequeue();
+ return $"RECV:{f.FrameIdHex}:{f.DataHex}:{f.Timestamp:HH:mm:ss.fff}";
+ }
+ }
+
+ private string HandleRecvAll()
+ {
+ List<CanFrame> batch = new();
+ lock (_bufferLock)
+ {
+ while (_recvBuffer.Count > 0) batch.Add(_recvBuffer.Dequeue());
+ }
+ if (batch.Count == 0) return "RECV_ALL:0";
+ var items = batch.Select(f => $"{f.FrameIdHex}:{f.DataHex}");
+ return $"RECV_ALL:{batch.Count}:{string.Join("|", items)}";
+ }
+
+ /// <summary>
+ /// 淇℃伅鍨嬫煡璇紙VERB:鈥︼級鍘绘帀浼犺緭灞� OK: 鍓嶇紑杩樺師閬楃暀鏍煎紡锛�
+ /// 鍔ㄤ綔纭锛堜腑鏂囨彁绀虹瓑锛変繚鐣� OK:銆�
+ /// </summary>
+ private static string NormalizeResponse(string resp)
+ {
+ if (string.IsNullOrEmpty(resp)) return resp;
+ var m = InfoVerb.Match(resp);
+ if (m.Success)
+ return resp.Substring(3); // 鍘� "OK:"
+ return resp;
+ }
+
+ // ==================== 鏉傞」 ====================
+
+ /// <summary>
+ /// 瀹夸富涓撶敤锛氬彇鍑哄苟娓呯┖鑷笂娆′互鏉ョ疮绉殑鎺ㄩ�佽锛屼緵 BrokerServer 鎷兼帴鍒板搷搴� readback 灏鹃儴銆�
+ /// </summary>
+ public List<string> DrainPendingPushLines()
+ {
+ lock (_bufferLock)
+ {
+ var list = _pendingPushLines.ToList();
+ _pendingPushLines.Clear();
+ return list;
+ }
+ }
+
+ private void EnsureConnected()
+ {
+ if (!IsConnected) throw new InvalidOperationException($"[CAN] 璁惧 {DeviceId} 鏈繛鎺�");
+ }
+
+ /// <summary>鎸夐渶杩炴帴锛氶娆� I/O 鏃惰嫢鏈繛鎺ュ垯鑷姩寤虹珛閫氶亾锛堝鍚堚�滄寜闇�杩炴帴鈥濊璁★級銆�</summary>
+ private async Task EnsureConnectedAsync(CancellationToken ct)
+ {
+ if (IsConnected) return;
+ await ConnectAsync(ct).ConfigureAwait(false);
+ }
+
+ private void RecordError(Exception ex)
+ {
+ _lastError = ex.Message;
+ _lastErrorTime = DateTime.Now;
+ Log.Error($"[CAN] 璁惧 {DeviceId} 鎿嶄綔澶辫触: {ex.Message}", ex);
+ }
+
+ private void OnConnection(bool connected, string reason)
+ {
+ if (!connected) _reconnectCount++;
+ ConnectionChanged?.Invoke(this, new DeviceConnectionEventArgs
+ {
+ DeviceId = DeviceId,
+ IsConnected = connected,
+ Reason = reason
+ });
+ }
+
+ public void Dispose()
+ {
+ if (_disposed) return;
+ _disposed = true;
+ try
+ {
+ _transport.FramePushed -= OnFramePushed;
+ _transport.Dispose();
+ }
+ catch (Exception ex)
+ {
+ Log.Error($"[CAN] 璁惧 {DeviceId} 閲婃斁寮傚父: {ex.Message}", ex);
+ }
+ }
+
+ private void ThrowIfDisposed()
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(CanDeviceDriver));
+ }
+ }
+}
--
Gitblit v1.9.1