diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs index 45d358f5..543a199c 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs @@ -19,7 +19,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus; /// since Modbus has no native push model. /// public sealed class ModbusDriver - : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable + : IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable { /// Authored tags this driver could not build into its poll table (Gitea #541). private readonly SkippedTagLedger _skippedTags = new(); @@ -80,6 +80,13 @@ public sealed class ModbusDriver /// Occurs when a subscribed tag value changes. public event EventHandler? OnDataChange; + /// + /// Occurs once per poll cycle that produced at least one change, carrying that cycle's + /// changed tags. Both data-change events fire by design — this one once per cycle, + /// once per changed tag — so a consumer subscribes to EXACTLY + /// ONE of the two; subscribing both double-delivers. + /// + public event EventHandler? OnDataChangeBatch; /// Occurs when host connectivity status changes. public event EventHandler? OnHostStatusChanged; @@ -125,7 +132,8 @@ public sealed class ModbusDriver OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)); }, onError: HandlePollError, - backoffCap: PollBackoffCap); + backoffCap: PollBackoffCap, + onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items))); } /// Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide (05/STAB-8). diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs index 04ebf8ae..14378ca3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs @@ -30,7 +30,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7; /// /// public sealed class S7Driver - : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDisposable, IAsyncDisposable + : IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDisposable, IAsyncDisposable { /// Authored tags this driver could not build into its poll table (Gitea #541). private readonly SkippedTagLedger _skippedTags = new(); @@ -68,7 +68,8 @@ public sealed class S7Driver OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, tagRef, snapshot)), minInterval: TimeSpan.FromMilliseconds(100), onError: HandlePollError, - backoffCap: TimeSpan.FromSeconds(30)); + backoffCap: TimeSpan.FromSeconds(30), + onBatch: (handle, items) => OnDataChangeBatch?.Invoke(this, new DataChangeBatchEventArgs(handle, items))); } // ---- ISubscribable + IHostConnectivityProbe state ---- @@ -93,6 +94,13 @@ public sealed class S7Driver /// Occurs when a subscribed tag value or status code changes. public event EventHandler? OnDataChange; + /// + /// Occurs once per poll cycle that produced at least one change, carrying that cycle's + /// changed tags. Both data-change events fire by design — this one once per cycle, + /// once per changed tag — so a consumer subscribes to EXACTLY + /// ONE of the two; subscribing both double-delivers. + /// + public event EventHandler? OnDataChangeBatch; /// Occurs when host connectivity status changes. public event EventHandler? OnHostStatusChanged; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs new file mode 100644 index 00000000..ee2535ec --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs @@ -0,0 +1,138 @@ +using System.Collections.Concurrent; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Modbus; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests; + +/// +/// Pins the driver-side half of the batched value pipeline: the Modbus driver is an +/// and forwards the PollGroupEngine's per-cycle batch +/// callback to . Both events fire by design — +/// the batch once per poll cycle, the per-tag one per changed tag — so a consumer subscribes to +/// exactly one of the two. +/// +[Trait("Category", "Unit")] +public sealed class ModbusBatchSubscriptionTests +{ + /// + /// Lightweight fake transport the subscription tests drive through — only the FC03 + /// (Read Holding Registers) path is used. Mutating + /// between polls is how each test simulates a PLC value change. + /// + private sealed class FakeTransport : IModbusTransport + { + public readonly ushort[] HoldingRegisters = new ushort[256]; + /// Simulates connecting to the Modbus device. + /// The cancellation token. + public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask; + + /// Simulates sending a Modbus PDU. + /// The Modbus unit ID. + /// The protocol data unit. + /// The cancellation token. + public Task SendAsync(byte unitId, byte[] pdu, CancellationToken ct) + { + if (pdu[0] != 0x03) return Task.FromException(new NotSupportedException("FC not supported")); + var addr = (ushort)((pdu[1] << 8) | pdu[2]); + var qty = (ushort)((pdu[3] << 8) | pdu[4]); + var resp = new byte[2 + qty * 2]; + resp[0] = 0x03; + resp[1] = (byte)(qty * 2); + for (var i = 0; i < qty; i++) + { + resp[2 + i * 2] = (byte)(HoldingRegisters[addr + i] >> 8); + resp[3 + i * 2] = (byte)(HoldingRegisters[addr + i] & 0xFF); + } + return Task.FromResult(resp); + } + /// Disposes the transport asynchronously. + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } + + private static (ModbusDriver drv, FakeTransport fake) NewDriver(params ModbusTagDefinition[] tags) + { + var fake = new FakeTransport(); + var opts = new ModbusDriverOptions { Host = "fake", RawTags = ModbusRawTags.Entries(tags)}; + return (new ModbusDriver(opts, "modbus-1", _ => fake), fake); + } + + /// The driver advertises the batched-subscription capability. + [Fact] + public void Driver_is_batch_subscribable() + { + var (drv, _) = NewDriver(new ModbusTagDefinition("Level", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16)); + (drv is IBatchSubscribable).ShouldBeTrue(); + } + + /// + /// One poll cycle that changed N tags raises the batch event ONCE carrying N items, and + /// raises the per-tag event N times with identical content. The invariant asserted across + /// the whole run — the concatenation of every batch's items equals the per-tag event + /// sequence — is what makes the two events interchangeable for a downstream consumer. + /// + [Fact] + public async Task Batch_event_fires_once_per_cycle_with_per_tag_events() + { + var (drv, fake) = NewDriver( + new ModbusTagDefinition("Level", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16), + new ModbusTagDefinition("Temp", ModbusRegion.HoldingRegisters, 1, ModbusDataType.Int16), + new ModbusTagDefinition("Flow", ModbusRegion.HoldingRegisters, 2, ModbusDataType.Int16)); + await drv.InitializeAsync("{}", CancellationToken.None); + fake.HoldingRegisters[0] = 100; + fake.HoldingRegisters[1] = 200; + fake.HoldingRegisters[2] = 300; + + var perTag = new ConcurrentQueue(); + var batches = new ConcurrentQueue(); + drv.OnDataChange += (_, e) => perTag.Enqueue(e); + ((IBatchSubscribable)drv).OnDataChangeBatch += (_, e) => batches.Enqueue(e); + + var handle = await drv.SubscribeAsync(["Level", "Temp", "Flow"], TimeSpan.FromMilliseconds(100), CancellationToken.None); + + // Cycle 1 — the initial-data push: all three tags raise. + await WaitForCountAsync(batches, 1, TimeSpan.FromSeconds(2)); + batches.Count.ShouldBe(1); + batches.Single().Items.Count.ShouldBe(3); + perTag.Count.ShouldBe(3); + + // Cycle 2 — every tag changes, so the next cycle again batches all three. + fake.HoldingRegisters[0] = 101; + fake.HoldingRegisters[1] = 201; + fake.HoldingRegisters[2] = 301; + await WaitForCountAsync(batches, 2, TimeSpan.FromSeconds(3)); + + // Unsubscribe drains the in-flight poll loop, so the queues are stable from here. + await drv.UnsubscribeAsync(handle, CancellationToken.None); + + var allBatches = batches.ToArray(); + allBatches.Length.ShouldBeGreaterThanOrEqualTo(2); + allBatches[1].Items.Count.ShouldBe(3); + + // Every batch belongs to the subscription it was raised for, and is never empty. + foreach (var b in allBatches) + { + b.SubscriptionHandle.ShouldBeSameAs(handle); + b.Items.ShouldNotBeEmpty(); + } + + // Identical content: flattening the batches reproduces the per-tag event stream exactly + // (same references, same snapshots, same order). + var flattened = allBatches.SelectMany(b => b.Items).ToArray(); + var perTagArray = perTag.ToArray(); + flattened.Length.ShouldBe(perTagArray.Length); + for (var i = 0; i < flattened.Length; i++) + { + flattened[i].FullReference.ShouldBe(perTagArray[i].FullReference); + flattened[i].Snapshot.ShouldBe(perTagArray[i].Snapshot); + } + } + + private static async Task WaitForCountAsync(ConcurrentQueue q, int target, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (q.Count < target && DateTime.UtcNow < deadline) + await Task.Delay(25); + } +}