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 并发验证用例
---
tests/Aging.MvpTests/BrokerE2ETests.cs | 521 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 521 insertions(+), 0 deletions(-)
diff --git a/tests/Aging.MvpTests/BrokerE2ETests.cs b/tests/Aging.MvpTests/BrokerE2ETests.cs
new file mode 100644
index 0000000..b9315ea
--- /dev/null
+++ b/tests/Aging.MvpTests/BrokerE2ETests.cs
@@ -0,0 +1,521 @@
+using System;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+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");
+ }
+
+ // ==================== 骞跺彂涓庣敓鍛藉懆鏈� ====================
+
+ /// <summary>鍗曞鎴风杩炵画 5 娆″揩閫熷彂閫侊細楠岃瘉 SessionGate 涓茶鍖栦笉涓㈠懡浠ゃ�佺绾︿笉琚閲婃斁銆�</summary>
+ [Fact]
+ public async Task SingleClient_RapidSend5Times_AllSucceed_LeaseStaysActive()
+ {
+ 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));
+
+ for (int i = 0; i < 5; i++)
+ {
+ var tx = await driver.SendCommandAsync($"SEND:0x{i:X3}:AA", CancellationToken.None);
+ Assert.Contains("OK:TX", tx);
+ }
+
+ // 5 娆℃搷浣滃悗绉熺害浠� Active锛堟湭琚剰澶栭噴鏀炬垨杩囨湡锛�
+ Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
+ _clientLease.Release(grant, "test-end");
+ }
+
+ /// <summary>
+ /// 涓や釜 Broker 浠ｇ悊椹卞姩鍚屾椂鎸佹湁鍚屼竴璁惧鐨勬椿璺冪绾︼紝骞跺彂鍙戦�佸悇 5 娆°��
+ /// 鐢变簬 arb.db 浠� ResourceKey 涓轰富閿彧鍏佽涓�琛� Active锛岀浜屼釜绉熺害閫氳繃鐩存帴 SQL 鎻掑叆妯℃嫙
+ /// 锛圡VP 鏍呮爮浠呮牎楠� Phase=Active + 鏈繃鏈燂紝涓嶆瘮瀵� OwnerId锛夈��
+ /// 楠岃瘉 SessionGate(SemaphoreSlim=1) 灏嗕袱涓苟鍙戣姹備覆琛屾墽琛岋紝涓よ�呭潎鎴愬姛銆�
+ /// </summary>
+ [Fact]
+ public async Task TwoClients_ConcurrentSend_BothSucceed_SessionGateSerializes()
+ {
+ // 瀹㈡埛绔� A锛氱粡姝ｅ父 Acquire 鑾峰彇 Production 绉熺害
+ var grantA = GrabLease();
+ var hash = DeviceConfigHash.Compute(_dataDb);
+ using var driverA = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driverA.ConnectAsync(CancellationToken.None));
+
+ // 瀹㈡埛绔� B锛氱洿鎺ュ線 arb.db 鎻掑叆涓�鏉� Maintenance Active 绉熺害琛岋紝妯℃嫙鍙︿竴涓寔鏈夎��
+ InsertDirectLeaseRow(DevId, "concurrent-client-B", OwnerTypeV3.Maintenance, ttlSeconds: 30);
+ using var driverB = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash);
+ await RetryAsync(() => driverB.ConnectAsync(CancellationToken.None));
+
+ // 骞跺彂鍙戦�佸悇 5 娆�
+ var tasksA = Enumerable.Range(0, 5).Select(i =>
+ driverA.SendCommandAsync($"SEND:0x{i:X3}:AA", CancellationToken.None)).ToArray();
+ var tasksB = Enumerable.Range(0, 5).Select(i =>
+ driverB.SendCommandAsync($"SEND:0x{i + 0x10:X3}:BB", CancellationToken.None)).ToArray();
+
+ var results = await Task.WhenAll(tasksA.Concat(tasksB));
+ Assert.All(results, r => Assert.Contains("OK:TX", r));
+
+ // 涓や釜绉熺害琛屽潎淇濇寔 Active
+ Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
+ _clientLease.Release(grantA, "test-end-A");
+ }
+
+ /// <summary>
+ /// 涓や釜 Production 瀹㈡埛绔珵浜夊悓涓�璁惧绉熺害锛氱涓�涓垚鍔熻幏鍙栵紝绗簩涓簲寰楀埌 ConflictHeld銆�
+ /// </summary>
+ [Fact]
+ public void TwoProductionClients_SecondGetsConflictHeld()
+ {
+ var grantA = GrabLease();
+ Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
+
+ // 绗簩涓鎴风锛堜笉鍚� OwnerId锛屽悓 OwnerType=Production锛�
+ var secondLease = new DbLeaseStore(_arbDb, "second-client");
+ var outcome = secondLease.Acquire(new AcquireRequest
+ {
+ OwnerId = "competing-client",
+ OwnerType = OwnerTypeV3.Production,
+ ResourceKeys = new[] { DevId },
+ TtlSeconds = 30,
+ WaitTimeoutMs = 0
+ });
+
+ Assert.Equal(AcquireResultCode.ConflictHeld, outcome.Code);
+ Assert.NotNull(outcome.Holder);
+
+ _clientLease.Release(grantA, "test-end");
+ }
+
+ /// <summary>
+ /// 瀹屾暣鐢熷懡鍛ㄦ湡锛欰cquire 鈫� Connect 鈫� Send 鈫� Release 鈫� 鍐� Acquire 鈫� 鍐� Send銆�
+ /// 楠岃瘉绉熺害閲婃斁鍚庤祫婧愬洖鍒� Free锛屽彲琚噸鏂拌幏鍙栧苟缁х画浣跨敤銆�
+ /// </summary>
+ [Fact]
+ public async Task LeaseLifecycle_AcquireSendReleaseAcquireSendAgain_AllSucceed()
+ {
+ var hash = DeviceConfigHash.Compute(_dataDb);
+
+ // 绗竴杞細鑾峰彇绉熺害 鈫� 杩炴帴 鈫� 鍙戦�� 鈫� 閲婃斁
+ LeaseGrant grant1;
+ {
+ var outcome = _clientLease.Acquire(new AcquireRequest
+ {
+ OwnerId = "lifecycle-1",
+ OwnerType = OwnerTypeV3.Production,
+ ResourceKeys = new[] { DevId },
+ TtlSeconds = 30
+ });
+ Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
+ grant1 = outcome.Grants[0];
+ }
+
+ using (var driver1 = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash))
+ {
+ await RetryAsync(() => driver1.ConnectAsync(CancellationToken.None));
+ var tx = await driver1.SendCommandAsync("SEND:0x100:AA", CancellationToken.None);
+ Assert.Contains("OK:TX", tx);
+ }
+
+ Assert.True(_clientLease.Release(grant1, "lifecycle-1-end"));
+ Assert.Equal(LeasePhase.Free, _clientLease.PhaseOf(DevId));
+
+ // 绗簩杞細閲嶆柊鑾峰彇 鈫� 杩炴帴 鈫� 鍙戦��
+ LeaseGrant grant2;
+ {
+ var outcome = _clientLease.Acquire(new AcquireRequest
+ {
+ OwnerId = "lifecycle-2",
+ OwnerType = OwnerTypeV3.Production,
+ ResourceKeys = new[] { DevId },
+ TtlSeconds = 30
+ });
+ Assert.Equal(AcquireResultCode.Acquired, outcome.Code);
+ grant2 = outcome.Grants[0];
+ }
+
+ using (var driver2 = CanDeviceDriver.CreateBrokerProxy(DevId, "CAN_E2E:0", _pipeName, hash))
+ {
+ await RetryAsync(() => driver2.ConnectAsync(CancellationToken.None));
+ var tx = await driver2.SendCommandAsync("SEND:0x200:BB", CancellationToken.None);
+ Assert.Contains("OK:TX", tx);
+ }
+
+ _clientLease.Release(grant2, "lifecycle-2-end");
+ }
+
+ /// <summary>
+ /// 骞跺彂鍘嬪姏锛氬崟瀹㈡埛绔� 20 涓� Task 鍚屾椂鍙戦�侊紝楠岃瘉 SessionGate + 绠￠亾鍐欏叆閿佷笉宕╂簝銆�
+ /// </summary>
+ [Fact]
+ public async Task SingleClient_ConcurrentSend20Tasks_AllSucceed()
+ {
+ 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 tasks = Enumerable.Range(0, 20).Select(i =>
+ driver.SendCommandAsync($"SEND:0x{i:X3}:DD", CancellationToken.None)).ToArray();
+
+ var results = await Task.WhenAll(tasks);
+ Assert.All(results, r => Assert.Contains("OK:TX", r));
+
+ Assert.Equal(LeasePhase.Active, _clientLease.PhaseOf(DevId));
+ _clientLease.Release(grant, "test-end");
+ }
+
+ // ==================== 杈呭姪 ====================
+
+ /// <summary>缁曡繃 Acquire 浜掓枼锛岀洿鎺ュ悜 arb.db 鎻掑叆涓�鏉� Active 绉熺害琛岋紙娴嬭瘯骞跺彂鍦烘櫙鐢級銆�</summary>
+ private void InsertDirectLeaseRow(string deviceId, string ownerId, OwnerTypeV3 ownerType, int ttlSeconds)
+ {
+ long now = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+ using var conn = new System.Data.SQLite.SQLiteConnection(
+ $"Data Source={_arbDb};Version=3;");
+ conn.Open();
+ using var cmd = conn.CreateCommand();
+ cmd.CommandText = @"
+ INSERT OR REPLACE INTO resource_state
+ (ResourceKey, Generation, Phase, LeaseId, OwnerId, OwnerType,
+ ProcessInstanceId, MachineName, HostProcessId, Epoch, Reason,
+ AcquiredUtcMs, ExpiresUtcMs, UpdatedUtcMs)
+ VALUES (@key, 1, 1, @lid, @oid, @otype, '', 'local', 0, 1, 'direct-insert',
+ @now, @exp, @now)";
+ cmd.Parameters.AddWithValue("@key", DbLeaseStore.NormKey(deviceId));
+ cmd.Parameters.AddWithValue("@lid", "lease_" + Guid.NewGuid().ToString("N").Substring(0, 12));
+ cmd.Parameters.AddWithValue("@oid", ownerId);
+ cmd.Parameters.AddWithValue("@otype", (int)ownerType);
+ cmd.Parameters.AddWithValue("@now", now);
+ cmd.Parameters.AddWithValue("@exp", now + ttlSeconds * 1000L);
+ cmd.ExecuteNonQuery();
+ }
+
+ 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