dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(drivers): Modbus + S7 implement IBatchSubscribable via PollGroupEngine batch callback

Browse Source
This commit is contained in:
Joseph Doherty
2026-08-15 12:14:43 -04:00
parent 536fa29224
commit ba55a5b48e
3 changed files with 158 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus/ModbusDriver.cs
+10 -2
View File
@@ -19,7 +19,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
/// <see cref="PollGroupEngine"/> since Modbus has no native push model.
/// </remarks>
public sealed class ModbusDriver
: IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable
: IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, IPerCallHostResolver, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table (Gitea #541).</summary>
private readonly SkippedTagLedger _skippedTags = new();
@@ -80,6 +80,13 @@ public sealed class ModbusDriver
/// <summary>Occurs when a subscribed tag value changes.</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>
/// 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,
/// <see cref="OnDataChange"/> once per changed tag — so a consumer subscribes to EXACTLY
/// ONE of the two; subscribing both double-delivers.
/// </summary>
public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
/// <summary>Occurs when host connectivity status changes.</summary>
public event EventHandler<HostStatusChangedEventArgs>? 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)));
}
/// <summary>Upper bound on the poll-loop failure backoff — adopts the S7-proven 30 s cap fleet-wide (05/STAB-8).</summary>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.S7/S7Driver.cs
+10 -2
View File
@@ -30,7 +30,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.S7;
/// </para>
/// </remarks>
public sealed class S7Driver
: IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDisposable, IAsyncDisposable
: IDriver, ITagDiscovery, IReadable, IWritable, IBatchSubscribable, IHostConnectivityProbe, ISkippedTagReporter, IDisposable, IAsyncDisposable
{
/// <summary>Authored tags this driver could not build into its poll table (Gitea #541).</summary>
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
/// <summary>Occurs when a subscribed tag value or status code changes.</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>
/// 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,
/// <see cref="OnDataChange"/> once per changed tag — so a consumer subscribes to EXACTLY
/// ONE of the two; subscribing both double-delivers.
/// </summary>
public event EventHandler<DataChangeBatchEventArgs>? OnDataChangeBatch;
/// <summary>Occurs when host connectivity status changes.</summary>
public event EventHandler<HostStatusChangedEventArgs>? OnHostStatusChanged;
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Modbus.Tests/ModbusBatchSubscriptionTests.cs
+138
View File
@@ -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;
/// <summary>
/// Pins the driver-side half of the batched value pipeline: the Modbus driver is an
/// <see cref="IBatchSubscribable"/> and forwards the <c>PollGroupEngine</c>'s per-cycle batch
/// callback to <see cref="IBatchSubscribable.OnDataChangeBatch"/>. 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.
/// </summary>
[Trait("Category", "Unit")]
public sealed class ModbusBatchSubscriptionTests
{
/// <summary>
/// Lightweight fake transport the subscription tests drive through — only the FC03
/// (Read Holding Registers) path is used. Mutating <see cref="HoldingRegisters"/>
/// between polls is how each test simulates a PLC value change.
/// </summary>
private sealed class FakeTransport : IModbusTransport
{
public readonly ushort[] HoldingRegisters = new ushort[256];
/// <summary>Simulates connecting to the Modbus device.</summary>
/// <param name="ct">The cancellation token.</param>
public Task ConnectAsync(CancellationToken ct) => Task.CompletedTask;
/// <summary>Simulates sending a Modbus PDU.</summary>
/// <param name="unitId">The Modbus unit ID.</param>
/// <param name="pdu">The protocol data unit.</param>
/// <param name="ct">The cancellation token.</param>
public Task<byte[]> SendAsync(byte unitId, byte[] pdu, CancellationToken ct)
{
if (pdu[0] != 0x03) return Task.FromException<byte[]>(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);
}
/// <summary>Disposes the transport asynchronously.</summary>
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);
}
/// <summary>The driver advertises the batched-subscription capability.</summary>
[Fact]
public void Driver_is_batch_subscribable()
{
var (drv, _) = NewDriver(new ModbusTagDefinition("Level", ModbusRegion.HoldingRegisters, 0, ModbusDataType.Int16));
(drv is IBatchSubscribable).ShouldBeTrue();
}
/// <summary>
/// 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.
/// </summary>
[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<DataChangeEventArgs>();
var batches = new ConcurrentQueue<DataChangeBatchEventArgs>();
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<T>(ConcurrentQueue<T> q, int target, TimeSpan timeout)
{
var deadline = DateTime.UtcNow + timeout;
while (q.Count < target && DateTime.UtcNow < deadline)
await Task.Delay(25);
}
}