From 559890c482f5973f06aafdfec9550fee5ba3c7f2 Mon Sep 17 00:00:00 2001
From: wenxiangchao <wenxiangchao@ihengxin.com.cn>
Date: 星期一, 09 十一月 2026 17:02:15 +0800
Subject: [PATCH] feat: CAN 模拟器 TCP 化 + 多设备并发调试 + 后台生命周期管理
---
Aging.Infrastructure/Broker/BrokerClientChannel.cs | 283 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 283 insertions(+), 0 deletions(-)
diff --git a/Aging.Infrastructure/Broker/BrokerClientChannel.cs b/Aging.Infrastructure/Broker/BrokerClientChannel.cs
new file mode 100644
index 0000000..e57f703
--- /dev/null
+++ b/Aging.Infrastructure/Broker/BrokerClientChannel.cs
@@ -0,0 +1,283 @@
+using System;
+using System.Buffers.Binary;
+using System.Collections.Concurrent;
+using System.IO;
+using System.IO.Pipes;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Aging.ResourceContracts;
+using log4net;
+
+namespace Aging.Infrastructure.Broker
+{
+ /// <summary>
+ /// DeviceBroker 鍛藉悕绠￠亾瀹㈡埛绔�氶亾锛圴3.0 搂13 鍚屾満鍩虹嚎锛夈��
+ /// 淇鐜版湁 PipeClient 鐨勪笁澶ч棶棰橈細鍝嶅簲鍏宠仈鍙揪銆佽秴鏃剁敓鏁堛�佸彇娑堝彲浼犳挱銆�
+ /// 甯ф牸寮忥細4 瀛楄妭澶х闀垮害鍓嶇紑 + UTF8 JSON 灏佸锛涜姹�-鍝嶅簲缁� RequestId/CorrelationId 閰嶅銆�
+ /// 绾跨▼妯″瀷锛氬崟鍚庡彴璇诲惊鐜� + pending 瀛楀吀锛涘啓鍏ョ敤淇″彿閲忎覆琛屽寲銆傜嚎绋嬪畨鍏紝鍙澶氱绾﹀叡浜��
+ /// </summary>
+ public sealed class BrokerClientChannel : IDisposable
+ {
+ private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerClientChannel));
+
+ private readonly string _pipeName;
+ private readonly string _clientInstanceId;
+ private readonly SemaphoreSlim _writeLock = new SemaphoreSlim(1, 1);
+ private readonly ConcurrentDictionary<string, TaskCompletionSource<IpcEnvelope>> _pending = new();
+
+ private NamedPipeClientStream? _pipe;
+ private Task? _readLoop;
+ private volatile bool _disposed;
+ private volatile bool _connected;
+ private int _inFlight;
+
+ /// <summary>鎻℃墜鎴愬姛鍚庣殑 Broker 绾厓锛堥噸鍚悗閫掑锛岀敤浜庢湰鍦伴檲鏃у垽瀹氾級</summary>
+ public long ServerEpoch { get; private set; }
+
+ /// <summary>Broker 鐢熸晥閰嶇疆鍝堝笇锛圛NV-04 瀵硅处鍙傜収锛�</summary>
+ public string BrokerConfigHash { get; private set; } = "";
+
+ public string BrokerId { get; private set; } = "";
+
+ public bool IsConnected => _connected && !_disposed;
+
+ /// <summary>杩炴帴/鎻℃墜榛樿瓒呮椂锛堟绉掞級</summary>
+ public int ConnectTimeoutMs { get; set; } = 3000;
+
+ /// <summary>鍗曟 RPC 榛樿绛夊緟涓婇檺锛堟绉掞級</summary>
+ public int DefaultRequestTimeoutMs { get; set; } = 10_000;
+
+ public BrokerClientChannel(string pipeName, string clientInstanceId)
+ {
+ _pipeName = pipeName ?? throw new ArgumentNullException(nameof(pipeName));
+ _clientInstanceId = clientInstanceId ?? throw new ArgumentNullException(nameof(clientInstanceId));
+ }
+
+ /// <summary>
+ /// 纭繚杩炴帴骞跺畬鎴� Hello 鎻℃墜锛堝箓绛夛紱鏂嚎鍚庡彲閲嶅懠锛夈��
+ /// 鍗忚涓荤増鏈笉鍖归厤鏃舵姏鍑� InvalidOperationException 骞舵嫆缁濆悗缁姹傘��
+ /// </summary>
+ public async Task EnsureConnectedAsync(CancellationToken ct)
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
+ if (_connected) return;
+
+ var pipe = new NamedPipeClientStream("localhost", _pipeName,
+ PipeDirection.InOut, PipeOptions.Asynchronous);
+ try
+ {
+ using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
+ {
+ timeoutCts.CancelAfter(ConnectTimeoutMs);
+ await pipe.ConnectAsync(timeoutCts.Token).ConfigureAwait(false);
+ }
+ }
+ catch (Exception ex)
+ {
+ pipe.Dispose();
+ throw new InvalidOperationException($"BROKER_UNREACHABLE: 鏃犳硶杩炴帴 DeviceBroker 绠￠亾 [{_pipeName}]: {ex.Message}", ex);
+ }
+
+ _pipe = pipe;
+ _connected = true;
+ _readLoop = Task.Run(() => ReadLoopAsync(pipe, ct));
+
+ // Hello 鎻℃墜
+ var hello = NewEnvelope(IpcMessageTypes.Hello);
+ var resp = await SendAsync(hello, ct).ConfigureAwait(false);
+ var result = BrokerIpc.DeserializeBody<HelloResult>(resp.BodyJson);
+ if (result == null)
+ throw new InvalidOperationException("BROKER_HANDSHAKE_FAILED: Hello 鍝嶅簲鏃犳晥");
+ if (result.ProtocolVersion != BrokerIpc.ProtocolVersion)
+ {
+ _connected = false;
+ throw new InvalidOperationException(
+ $"BROKER_PROTOCOL_MISMATCH: 瀹㈡埛绔� v{BrokerIpc.ProtocolVersion}, Broker v{result.ProtocolVersion}");
+ }
+ ServerEpoch = result.ServerEpoch;
+ BrokerConfigHash = result.ConfigHash;
+ BrokerId = result.BrokerId;
+ Log.Info($"[BrokerChannel] 宸茶繛鎺� {_pipeName}: BrokerId={BrokerId}, epoch={ServerEpoch}, configHash={ShortHash(BrokerConfigHash)}");
+ }
+
+ /// <summary>
+ /// 鍙戦�佷竴甯ц姹傚苟绛夊緟鍏跺叧鑱斿搷搴斻�傝秴鏃�/鏂嚎鍧囪繑鍥炲彲鍖哄垎鐨勯敊璇皝濂楋紙涓嶆姏瑁稿紓甯革級銆�
+ /// </summary>
+ public async Task<IpcEnvelope> SendAsync(IpcEnvelope request, CancellationToken ct)
+ {
+ if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
+ var pipe = _pipe;
+ if (!_connected || pipe == null)
+ return ErrorEnvelope(request, "BROKER_UNREACHABLE", "鏈繛鎺� DeviceBroker");
+
+ var tcs = new TaskCompletionSource<IpcEnvelope>(TaskCreationOptions.RunContinuationsAsynchronously);
+ _pending[request.RequestId] = tcs;
+ Interlocked.Increment(ref _inFlight);
+ try
+ {
+ byte[] frame;
+ try
+ {
+ frame = BrokerIpc.EncodeFrame(BrokerIpc.EncodeEnvelope(request));
+ }
+ catch (Exception ex)
+ {
+ return ErrorEnvelope(request, "PROTOCOL_ERROR", ex.Message);
+ }
+
+ // 鍐欏叆涓茶鍖栵紱pipe 鍗婂弻宸ヤ氦閿欎笅 async 鍐欏彲鑳芥寕姝伙紝鏁呭啓鍏ヤ笉鍔犲簲鐢ㄧ骇瓒呮椂锛堜緷璧栬渚� deadline + 鏂嚎鎰熺煡锛�
+ await _writeLock.WaitAsync(ct).ConfigureAwait(false);
+ try
+ {
+ await pipe.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
+ await pipe.FlushAsync(ct).ConfigureAwait(false);
+ }
+ catch (IOException ex)
+ {
+ Fault($"绠￠亾鍐欏け璐�: {ex.Message}");
+ return ErrorEnvelope(request, "BROKER_IO_ERROR", ex.Message);
+ }
+ finally
+ {
+ _writeLock.Release();
+ }
+
+ using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
+ {
+ int waitMs = request.DeadlineUtcMs > 0
+ ? (int)Math.Max(1, request.DeadlineUtcMs - BrokerIpc.NowUtcMs())
+ : DefaultRequestTimeoutMs;
+ var delay = Task.Delay(waitMs, timeoutCts.Token);
+ var winner = await Task.WhenAny(tcs.Task, delay).ConfigureAwait(false);
+ if (winner == tcs.Task)
+ {
+ timeoutCts.Cancel(); // 閲婃斁 delay 瀹氭椂鍣�
+ return await tcs.Task.ConfigureAwait(false);
+ }
+ // INV-06锛氳姹傚凡鍐欏嚭涓斿彲鑳藉湪閫� 鈫� 缁撴灉鏈煡(Unknown)锛岀姝㈠綋鏈墽琛�
+ return UnknownEnvelope(request, "BROKER_TIMEOUT",
+ "绛夊緟鍝嶅簲瓒呮椂锛岀粨鏋滄湭鐭ワ紝鎸夋湭瀹屾垚澶勭悊");
+ }
+ }
+ finally
+ {
+ _pending.TryRemove(request.RequestId, out _);
+ Interlocked.Decrement(ref _inFlight);
+ }
+ }
+
+ /// <summary>渚挎嵎鏂规硶锛氭彁浜や竴娆� OneShot 璁惧鍔ㄤ綔锛岃繑鍥炵粨鏋滀綋銆�</summary>
+ public async Task<BrokerIpc.ResultBody> SubmitAsync(BrokerIpc.SubmitBody body, string configHash, CancellationToken ct)
+ {
+ await EnsureConnectedAsync(ct).ConfigureAwait(false);
+ var env = NewEnvelope(IpcMessageTypes.SubmitTask);
+ env.ConfigHash = configHash ?? "";
+ env.BodyJson = BrokerIpc.SerializeBody(body);
+ var resp = await SendAsync(env, ct).ConfigureAwait(false);
+ var result = BrokerIpc.DeserializeBody<BrokerIpc.ResultBody>(resp.BodyJson)
+ ?? new BrokerIpc.ResultBody { State = "Unknown", ErrorCode = "BAD_RESPONSE" };
+ return result;
+ }
+
+ // ==================== 鍐呴儴 ====================
+
+ private async Task ReadLoopAsync(NamedPipeClientStream pipe, CancellationToken ct)
+ {
+ var lenBuf = new byte[4];
+ try
+ {
+ while (!ct.IsCancellationRequested && !_disposed)
+ {
+ await ReadExactAsync(pipe, lenBuf, 4, ct).ConfigureAwait(false);
+ int len = BinaryPrimitives.ReadInt32BigEndian(lenBuf);
+ if (len <= 0 || len > BrokerIpc.MaxFrameBytes)
+ throw new IOException($"闈炴硶甯ч暱搴� {len}");
+ var payload = new byte[len];
+ await ReadExactAsync(pipe, payload, len, ct).ConfigureAwait(false);
+
+ IpcEnvelope env;
+ try
+ {
+ env = BrokerIpc.DecodeEnvelope(Encoding.UTF8.GetString(payload));
+ }
+ catch (Exception)
+ {
+ continue; // 鏃犳硶瑙ｆ瀽鐨勫崟甯т涪寮冿紝涓嶅奖鍝嶅悗缁叧鑱�
+ }
+
+ if (!string.IsNullOrEmpty(env.CorrelationId)
+ && _pending.TryGetValue(env.CorrelationId, out var tcs))
+ {
+ tcs.TrySetResult(env);
+ }
+ }
+ }
+ catch (OperationCanceledException) { /* 姝ｅ父閫�鍑� */ }
+ catch (Exception ex)
+ {
+ if (!_disposed) Fault($"璇诲惊鐜粓姝�: {ex.Message}");
+ }
+ }
+
+ private static async Task ReadExactAsync(Stream s, byte[] buf, int count, CancellationToken ct)
+ {
+ int off = 0;
+ while (off < count)
+ {
+ int n = await s.ReadAsync(buf, off, count - off, ct).ConfigureAwait(false);
+ if (n <= 0) throw new EndOfStreamException("绠￠亾宸插叧闂�");
+ off += n;
+ }
+ }
+
+ /// <summary>杩炴帴澶辨晥锛氱疆闄嶇骇鎬侊紝绛夊緟涓嬫 EnsureConnectedAsync 閲嶈繛锛涘湪閫旇姹傜敱鍏惰嚜韬秴鏃舵敹鍙ｃ��</summary>
+ private void Fault(string reason)
+ {
+ if (!_connected) return;
+ _connected = false;
+ Log.Warn($"[BrokerChannel] 杩炴帴澶辨晥({reason})锛屽皢鍦ㄤ笅娆¤姹傛椂閲嶈繛");
+ try { _pipe?.Dispose(); } catch { }
+ _pipe = null;
+ }
+
+ private IpcEnvelope NewEnvelope(string type) => new IpcEnvelope
+ {
+ ProtocolVersion = BrokerIpc.ProtocolVersion,
+ MessageType = type,
+ RequestId = Guid.NewGuid().ToString("N"),
+ CorrelationId = "",
+ ClientInstanceId = _clientInstanceId,
+ AuthorityId = BrokerId
+ };
+
+ private static IpcEnvelope ErrorEnvelope(IpcEnvelope? request, string code, string message)
+ => BuildEnvelope(request, "Failed", code, message);
+
+ private static IpcEnvelope UnknownEnvelope(IpcEnvelope? request, string code, string message)
+ => BuildEnvelope(request, "Unknown", code, message);
+
+ private static IpcEnvelope BuildEnvelope(IpcEnvelope? request, string state, string code, string message)
+ => new IpcEnvelope
+ {
+ ProtocolVersion = BrokerIpc.ProtocolVersion,
+ MessageType = "Result",
+ CorrelationId = request?.RequestId ?? "",
+ BodyJson = "{\"State\":\"" + state + "\",\"ErrorCode\":\"" + code + "\",\"Readback\":\"" +
+ message.Replace("\"", "'") + "\"}"
+ };
+
+ private static string ShortHash(string h)
+ => string.IsNullOrEmpty(h) ? "" : (h.Length > 8 ? h.Substring(0, 8) : h);
+
+ public void Dispose()
+ {
+ if (_disposed) return;
+ _disposed = true;
+ _connected = false;
+ try { _pipe?.Dispose(); } catch { }
+ _pipe = null;
+ _writeLock.Dispose();
+ }
+ }
+}
--
Gitblit v1.9.1