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 化 + 多设备并发调试 + 后台生命周期管理
---
tests/Aging.MvpTests/BrokerE2ETests.cs | 334 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 334 insertions(+), 0 deletions(-)
diff --git a/tests/Aging.MvpTests/BrokerE2ETests.cs b/tests/Aging.MvpTests/BrokerE2ETests.cs
new file mode 100644
index 0000000..dcd0d15
--- /dev/null
+++ b/tests/Aging.MvpTests/BrokerE2ETests.cs
@@ -0,0 +1,334 @@
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+using Aging.Infrastructure.Broker;
+using Aging.Infrastructure.Core;
+using Aging.Infrastructure.Drivers;
+using Aging.ResourceContracts;
+using CanSimulatorServer;
+using DeviceBroker;
+using Xunit;
+
+namespace Aging.MvpTests
+{
+ /// <summary>
+ /// Broker 鍏ㄩ摼璺� e2e锛氬鎴风锛坅rb.db 绉熺害 + CanDeviceDriver BrokerRpc 浠ｇ悊锛夆啍
+ /// DeviceBroker 瀹夸富锛堝懡鍚嶇閬� + 鏍呮爮鍐嶆牎楠� + configHash 瀵硅处 + SessionGate 鍘熷瓙鐗囨锛夈��
+ /// 瀹夸富鐨勭墿鐞� CAN 椹卞姩缁� TCP 杩炶繘绋嬪唴鎵樼鐨勬ā鎷熷櫒鏈嶅姟锛堝浐瀹氱鍙� 19555锛屼笌 App.config 涓�鑷达級銆�
+ /// 姣忎釜娴嬭瘯浣跨敤鍞竴 brokerId 鈫� 鐙珛绠￠亾鍚嶏紝閬垮厤骞惰涓叉壈銆�
+ /// </summary>
+ public class BrokerE2ETests : IDisposable
+ {
+ private const string DevId = "CAN-B1";
+ private const int SimPort = 19555;
+
+ private readonly string _dir;
+ private readonly string _dataDb;
+ private readonly string _arbDb;
+ private readonly string _brokerId;
+ private readonly string _pipeName;
+ private readonly CancellationTokenSource _cts = new CancellationTokenSource();
+ private readonly Task _serverTask;
+ private readonly DbLeaseStore _clientLease;
+ private readonly TcpSimulatorHost _sim;
+
+ public BrokerE2ETests()
+ {
+ _dir = Path.Combine(Path.GetTempPath(), "brk_" + Guid.NewGuid().ToString("N"));
+ Directory.CreateDirectory(_dir);
+ _dataDb = Path.Combine(_dir, "data.db");
+ _arbDb = Path.Combine(_dir, "arb.db");
+ _brokerId = "broker-" + Guid.NewGuid().ToString("N").Substring(0, 8);
+ _pipeName = BrokerPipeName.For(_brokerId);
+
+ CreateDataDb();
+
+ // 妯℃嫙鍣ㄦ湇鍔″厛浜庡涓诲氨浣嶏紙瀹夸富棣栨 CONNECT 浼氭媺璧风墿鐞� TCP 浼氳瘽锛�
+ _sim = new TcpSimulatorHost();
+ _sim.StartAsync(SimPort).GetAwaiter().GetResult();
+
+ var server = new BrokerServer(_dataDb, _arbDb, _brokerId);
+ server.LoadDevices();
+ _serverTask = Task.Run(() => server.RunAsync(_cts.Token));
+ _clientLease = new DbLeaseStore(_arbDb, "local-main");
+
+ // 绛夌洃鍚氨浣嶏紙瀹㈡埛绔湰韬湁 3s 杩炴帴瓒呮椂锛屽彟鏈� BROKER_UNREACHABLE 閲嶈瘯鍏滃簳锛�
+ Thread.Sleep(250);
+ }
+
+ public void Dispose()
+ {
+ try { _cts.Cancel(); } catch { }
+ try { _serverTask.Wait(1500); } catch { }
+ try { _sim?.Dispose(); } catch { }
+ _cts.Dispose();
+ foreach (var f in Directory.GetFiles(_dir))
+ {
+ try { File.Delete(f); } catch { }
+ }
+ try { Directory.Delete(_dir); } catch { }
+ }
+
+ // ==================== 姝ｅ悜閾捐矾 ====================
+
+ [Fact]
+ public async Task WithLease_Connect_Send_Query_Succeeds()
+ {
+ var grant = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ var tx = await driver.SendCommandAsync("SEND:0x123:11 22", CancellationToken.None);
+ Assert.Contains("OK:TX", tx);
+
+ var status = await driver.QueryAsync("STATUS", CancellationToken.None);
+ Assert.Contains("STATUS", status);
+
+ // Broker 渚ф墽琛屾垚鍔熷悗绉熺害浠嶆湁鏁堬紙鏈璇噴鏀撅級
+ Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
+ _clientLease.Release(grant, "test-end");
+ }
+
+ [Fact]
+ public async Task TwoSequencedFragments_SameSession_ExecuteSerially()
+ {
+ var grant = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ // SessionGate 涓嬪悓浼氳瘽鐗囨涓ユ牸涓茶锛堝悓杩涚▼鍗曞鎴风鏈氨涓茶锛岄獙璇佷笉浜х敓浜ら敊閿欒锛�
+ var a = await driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None);
+ var b = await driver.SendCommandAsync("SEND:0x200:02", CancellationToken.None);
+ Assert.Contains("OK:TX", a);
+ Assert.Contains("OK:TX", b);
+ _clientLease.Release(grant, "test-end");
+ }
+
+ // ==================== 璐熷悜閾捐矾 ====================
+
+ [Fact]
+ public async Task NoLease_ConnectAllowed_Send_RejectedStaleLease()
+ {
+ // CONNECT/DISCONNECT 浠呯櫥璁�/閲婃斁瀹夸富鐗╃悊浼氳瘽锛岃眮鍏嶇绾︼紱
+ // SEND 鏈夋�荤嚎杈撳嚭锛屾棤绉熺害蹇呴』鎷掔粷锛圛NV-01 绂佹鏃犵绾︾洿鎺� I/O锛�
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ var ex = await Record.ExceptionAsync(
+ () => driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None));
+ Assert.NotNull(ex);
+ Assert.Contains("LEASE_STALE", ex.Message);
+ }
+
+ [Fact]
+ public async Task ConfigHashMismatch_Rejected()
+ {
+ GrabLease(); // 鏈夌绾︼紝浣嗗搱甯屽涓嶄笂 鈫� INV-04 鎷︽埅
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, "deadbeefdeadbeef");
+
+ var ex = await Record.ExceptionAsync(
+ () => RetryConnectAsync(driver, expectThrow: true));
+ Assert.NotNull(ex);
+ Assert.Contains("CONFIG_MISMATCH", ex.Message);
+ }
+
+ [Fact]
+ public async Task ReleasedLease_Send_Rejected()
+ {
+ var grant = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ // 绉熺害褰掕繕鍚庡啀 I/O 鈫� 鏍呮爮蹇呴』鎷掔粷锛堢姝㈡棤绉熺害鐩存帴 I/O锛�
+ _clientLease.Release(grant, "released-by-test");
+ var ex = await Record.ExceptionAsync(
+ () => driver.SendCommandAsync("SEND:0x100:01", CancellationToken.None));
+ Assert.NotNull(ex);
+ Assert.Contains("LEASE_STALE", ex.Message);
+ }
+
+ [Fact]
+ public async Task MaintenanceLease_BrokeredWrite_Succeeds()
+ {
+ // 璋冭瘯椤电淮鎶ゅ啓鏉冮摼璺細缁存姢绉熺害鍐欏叆 arb.db锛圤wnerType=Maintenance锛夊悗锛�
+ // Broker 鏍呮爮鍚屾牱鏀捐锛堝崰鐢ㄥ彲瑙�/鎷掔粷璇箟宸茬敱鐢熶骇鍦烘櫙鐢ㄤ緥瑕嗙洊锛�
+ var outcome = _clientLease.Acquire(new AcquireRequest
+ {
+ OwnerId = "maint-session-1",
+ OwnerType = OwnerTypeV3.Maintenance,
+ ResourceKeys = new[] { DevId },
+ TtlSeconds = 30
+ });
+ Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
+
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+ var tx = await driver.SendCommandAsync("SEND:0x300:AB", CancellationToken.None);
+ Assert.Contains("OK:TX", tx);
+
+ _clientLease.Release(outcome.Grants[0], "maint-end");
+ }
+
+ // ==================== 杈呭姪 ====================
+
+ [Fact]
+ public async Task BrokerReceive_PiggybackPush_DeliveredToProxyDriver()
+ {
+ // 楠屾敹鐐癸細鍏变韩璁惧(Broker)鑳芥敹鍒板悓鎬荤嚎瀵圭鍙戞潵鐨勫抚锛堢粡瀹夸富鎹庡甫鎺ㄩ�侊級銆�
+ // 淇鍘嗗彶涓� BrokeredCanDriver 鏃犳帴鏀朵簨浠躲�侀潰鏉挎敹涓嶅埌 RX 鐨勯〗鐥囥��
+ var grant = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ CanFrameish received = null;
+ driver.DriverFrameReceived += (s, f) => received = new CanFrameish { Id = f.FrameIdHex, Data = f.DataHex };
+
+ // 瀵圭锛氬悓妯℃嫙鍣ㄥ悓鎬荤嚎 CAN_E2E 鐨勮妭鐐� 1锛屾湰鍦扮洿杩炲彂甯�
+ using var peer = new CanDeviceDriver("CAN-PEER", "CAN_E2E:1", "127.0.0.1", SimPort);
+ peer.ConnectAsync(CancellationToken.None).GetAwaiter().GetResult();
+ peer.SendCommandAsync("SEND:0x7E7:EE FF", CancellationToken.None).GetAwaiter().GetResult();
+
+ // 瑙﹀彂瀹夸富 drain 鎹庡甫锛氫换涓�闈炴帴鏀剁被鏌ヨ锛圫TATUS锛夊嵆鍙甫鍥炴帹閫�
+ for (int i = 0; i < 20 && received == null; i++)
+ {
+ await driver.QueryAsync("STATUS", CancellationToken.None);
+ if (received == null) await Task.Delay(50);
+ }
+
+ Assert.NotNull(received);
+ Assert.Equal("0x7E7", received.Id);
+ Assert.Contains("EE FF", received.Data);
+
+ _clientLease.Release(grant, "test-end");
+ }
+
+ private sealed class CanFrameish { public string Id; public string Data; public string Src; }
+
+ [Fact]
+ public async Task BrokerResponderEcho_ViaBusEcho_DeliveredToProxyDriver()
+ {
+ // 澶嶇幇鐢ㄦ埛鍦烘櫙锛氫粎 Broker 璁惧鑷繁鍙戝抚 + 鎬荤嚎搴旂瓟鍣ㄥ紑鍚� 鈫� 搴旀敹鍒� RESPONDER 鍥炴樉銆�
+ var grant = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driver = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driver.ConnectAsync(CancellationToken.None));
+
+ CanFrameish echo = null;
+ driver.DriverFrameReceived += (s, f) => echo = new CanFrameish { Id = f.FrameIdHex, Data = f.DataHex, Src = f.SourceNodeId };
+
+ // 缁� RPC 寮�鍚�荤嚎搴旂瓟鍣�
+ var on = await driver.QueryAsync("BUS_ECHO:ON", CancellationToken.None);
+ Assert.StartsWith("OK:", on);
+
+ // Broker 璁惧鑷繁鍙戜竴甯�
+ await driver.SendCommandAsync("SEND:0x111:AA BB CC", CancellationToken.None);
+
+ // 杞 STATUS 瑙﹀彂瀹夸富 drain 鎹庡甫锛岀瓑鍥炴樉閫佽揪
+ for (int i = 0; i < 30 && echo == null; i++)
+ {
+ await driver.QueryAsync("STATUS", CancellationToken.None);
+ if (echo == null) await Task.Delay(50);
+ }
+
+ Assert.NotNull(echo);
+ Assert.Equal("0x111", echo.Id);
+ Assert.Contains("RESPONDER", echo.Src);
+
+ _clientLease.Release(grant, "test-end");
+ }
+
+ // ==================== 杈呭姪锛堝師鏈夛級 ====================
+
+ private LeaseGrant GrabLease()
+ {
+ var outcome = _clientLease.Acquire(new AcquireRequest
+ {
+ OwnerId = "e2e-" + Guid.NewGuid().ToString("N"),
+ OwnerType = OwnerTypeV3.Production,
+ ResourceKeys = new[] { DevId },
+ TtlSeconds = 30
+ });
+ Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
+ return outcome.Grants[0];
+ }
+
+ private static async Task RetryAsync(Func<Task> action, int attempts = 4)
+ {
+ Exception last = null;
+ for (int i = 0; i < attempts; i++)
+ {
+ try { await action().ConfigureAwait(false); return; }
+ catch (InvalidOperationException ex) when (ex.Message.Contains("BROKER_UNREACHABLE"))
+ {
+ last = ex; // 瀹夸富灏氭湭寮�濮嬬洃鍚紝鐭殏閲嶈瘯
+ await Task.Delay(400).ConfigureAwait(false);
+ }
+ }
+ throw last ?? new InvalidOperationException("閲嶈瘯浠嶅け璐�");
+ }
+
+ /// <summary>璐熷悜鐢ㄤ緥锛氭彙鎵嬪彲閲嶈瘯锛堢閬撴湭灏变綅锛夛紝浣嗘爡鏍�/鍝堝笇鎷掔粷灞炵粓鎬佸簲鐩存帴涓婃姏銆�</summary>
+ private async Task RetryConnectAsync(CanDeviceDriver driver, bool expectThrow)
+ {
+ Exception last = null;
+ for (int i = 0; i < 4; i++)
+ {
+ try
+ {
+ await driver.ConnectAsync(CancellationToken.None).ConfigureAwait(false);
+ return; // 鎰忓鎴愬姛鈥斺�旂敱澶栧眰鏂█鎹曟崏
+ }
+ catch (InvalidOperationException ex) when (ex.Message.Contains("BROKER_UNREACHABLE"))
+ {
+ last = ex;
+ await Task.Delay(400).ConfigureAwait(false);
+ }
+ catch (InvalidOperationException)
+ {
+ throw; // LEASE_STALE / CONFIG_MISMATCH 缁堟��
+ }
+ }
+ throw last ?? new InvalidOperationException("鏈幏寰楅鏈熸嫆缁�");
+ }
+
+ private void CreateDataDb()
+ {
+ using var conn = new System.Data.SQLite.SQLiteConnection(
+ $"Data Source={_dataDb};Version=3;");
+ conn.Open();
+ using (var ddl = conn.CreateCommand())
+ {
+ ddl.CommandText = @"
+ CREATE TABLE DeviceDefinition (
+ DeviceId TEXT PRIMARY KEY, Kind TEXT NOT NULL, DriverType TEXT NOT NULL,
+ Transport TEXT NOT NULL, Endpoint TEXT NOT NULL,
+ ExecutionMode TEXT NOT NULL DEFAULT 'Local',
+ AuthorityId TEXT NOT NULL DEFAULT '', BrokerId TEXT NOT NULL DEFAULT '',
+ OwnerProcessId TEXT NOT NULL DEFAULT '', PhysicalSessionKey TEXT NOT NULL DEFAULT '',
+ EffectiveVersion TEXT NOT NULL DEFAULT '', CreatedAt TEXT NOT NULL, UpdatedAt TEXT NOT NULL);";
+ ddl.ExecuteNonQuery();
+ }
+ using (var ins = conn.CreateCommand())
+ {
+ ins.CommandText = @"
+ INSERT INTO DeviceDefinition VALUES (
+ 'CAN-B1','CanDevice','CAN','Can','CAN_E2E:0',
+ 'Broker',@auth,@broker,'','can/CAN_E2E/main',
+ 'V1','2026-11-08 00:00:00','2026-11-08 00:00:00');";
+ ins.Parameters.AddWithValue("@auth", _brokerId);
+ ins.Parameters.AddWithValue("@broker", _brokerId);
+ ins.ExecuteNonQuery();
+ }
+ }
+ }
+}
--
Gitblit v1.9.1