fix(dcl): remove unused poll-based WriteBatchAndWaitAsync from IDataConnection and both adapters (review 2026-10-21 02/U3)
This commit is contained in:
@@ -31,11 +31,12 @@ IDataConnection : IAsyncDisposable
|
|||||||
├── ReadBatch(tagPaths) → values
|
├── ReadBatch(tagPaths) → values
|
||||||
├── Write(tagPath, value) → void
|
├── Write(tagPath, value) → void
|
||||||
├── WriteBatch(values) → void
|
├── WriteBatch(values) → void
|
||||||
├── WriteBatchAndWait(values, flagPath, flagValue, responsePath, responseValue, timeout) → bool
|
|
||||||
├── Status → ConnectionHealth
|
├── Status → ConnectionHealth
|
||||||
└── Disconnected → event Action?
|
└── Disconnected → event Action?
|
||||||
```
|
```
|
||||||
|
|
||||||
|
There is deliberately no write-and-wait handshake on the adapter interface: the script-facing `Attributes.WriteBatchAndWaitAsync` composes a `WriteTagBatchRequest` (batch + trigger write) with the Instance Actor's event-driven attribute waiter, so the handshake can never fall back to a polling read loop that bypasses the waiter (the adapters' former poll-based `WriteBatchAndWaitAsync` was unused and was removed — review 2026-10-21).
|
||||||
|
|
||||||
The `Disconnected` event is raised by an adapter when it detects an unexpected connection loss (server offline, network failure, keep-alive timeout). The `DataConnectionActor` subscribes to this event to trigger the reconnection state machine. Additional protocols can be added by implementing this interface.
|
The `Disconnected` event is raised by an adapter when it detects an unexpected connection loss (server offline, network failure, keep-alive timeout). The `DataConnectionActor` subscribes to this event to trigger the reconnection state machine. Additional protocols can be added by implementing this interface.
|
||||||
|
|
||||||
### Batch Capability Seam
|
### Batch Capability Seam
|
||||||
|
|||||||
@@ -56,16 +56,6 @@ public interface IDataConnection : IAsyncDisposable
|
|||||||
/// <param name="cancellationToken">Cancellation token.</param>
|
/// <param name="cancellationToken">Cancellation token.</param>
|
||||||
/// <returns>A dictionary of tag paths to their write results.</returns>
|
/// <returns>A dictionary of tag paths to their write results.</returns>
|
||||||
Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default);
|
Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default);
|
||||||
/// <summary>Writes a batch of values, then writes a flag and waits for a specific response value within the timeout.</summary>
|
|
||||||
/// <param name="values">Tag values to write before the flag.</param>
|
|
||||||
/// <param name="flagPath">Tag path of the trigger flag.</param>
|
|
||||||
/// <param name="flagValue">Value to write to the flag tag.</param>
|
|
||||||
/// <param name="responsePath">Tag path to monitor for the expected response value.</param>
|
|
||||||
/// <param name="responseValue">The response value that indicates completion.</param>
|
|
||||||
/// <param name="timeout">Maximum time to wait for the response.</param>
|
|
||||||
/// <param name="cancellationToken">Cancellation token.</param>
|
|
||||||
/// <returns><c>true</c> if the response value was observed within the timeout; otherwise <c>false</c>.</returns>
|
|
||||||
Task<bool> WriteBatchAndWaitAsync(IDictionary<string, object?> values, string flagPath, object? flagValue, string responsePath, object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default);
|
|
||||||
/// <summary>Current connection health status.</summary>
|
/// <summary>Current connection health status.</summary>
|
||||||
ConnectionHealth Status { get; }
|
ConnectionHealth Status { get; }
|
||||||
|
|
||||||
|
|||||||
@@ -1972,8 +1972,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
|||||||
/// the timeout, or an adapter exception is translated into a failed
|
/// the timeout, or an adapter exception is translated into a failed
|
||||||
/// <see cref="WriteTagBatchResponse"/> returned synchronously to the caller — never a
|
/// <see cref="WriteTagBatchResponse"/> returned synchronously to the caller — never a
|
||||||
/// dropped reply. NOTE: this deliberately composes the batch + trigger primitives and
|
/// dropped reply. NOTE: this deliberately composes the batch + trigger primitives and
|
||||||
/// uses the EXISTING event-driven WaitForAttribute waiter for the wait half; it does
|
/// uses the EXISTING event-driven WaitForAttribute waiter for the wait half (the
|
||||||
/// NOT call the adapter's poll-based <c>WriteBatchAndWaitAsync</c>.
|
/// adapters' former poll-based <c>WriteBatchAndWaitAsync</c> was removed — review
|
||||||
|
/// 2026-10-21 02/U3 — so no polling handshake can bypass that waiter).
|
||||||
/// </summary>
|
/// </summary>
|
||||||
private void HandleWriteBatch(WriteTagBatchRequest request)
|
private void HandleWriteBatch(WriteTagBatchRequest request)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -420,35 +420,6 @@ public class MxGatewayDataConnection
|
|||||||
return results.ToDictionary(w => w.TagPath, w => new WriteResult(w.Success, w.Error));
|
return results.ToDictionary(w => w.TagPath, w => new WriteResult(w.Success, w.Error));
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
|
||||||
public async Task<bool> WriteBatchAndWaitAsync(
|
|
||||||
IDictionary<string, object?> values, string flagPath, object? flagValue,
|
|
||||||
string responsePath, object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default)
|
|
||||||
{
|
|
||||||
await WriteBatchAsync(values, cancellationToken);
|
|
||||||
await WriteAsync(flagPath, flagValue, cancellationToken);
|
|
||||||
|
|
||||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
|
||||||
timeoutCts.CancelAfter(timeout);
|
|
||||||
try
|
|
||||||
{
|
|
||||||
while (!timeoutCts.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
var r = await ReadAsync(responsePath, timeoutCts.Token);
|
|
||||||
// r.Value is a TagValue wrapper; compare its underlying scalar. String
|
|
||||||
// projection tolerates numeric type differences across the gRPC boundary.
|
|
||||||
if (r.Success && string.Equals(r.Value?.Value?.ToString(), responseValue?.ToString(), StringComparison.Ordinal))
|
|
||||||
return true;
|
|
||||||
await Task.Delay(TimeSpan.FromMilliseconds(200), timeoutCts.Token);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
// Timeout elapsed (the linked CTS, not the caller's token) — fall through to false.
|
|
||||||
}
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public async Task<BrowseChildrenResult> BrowseChildrenAsync(
|
public async Task<BrowseChildrenResult> BrowseChildrenAsync(
|
||||||
string? parentNodeId,
|
string? parentNodeId,
|
||||||
|
|||||||
@@ -416,34 +416,6 @@ public class OpcUaDataConnection
|
|||||||
return _client.SearchAddressSpaceAsync(query, maxDepth, maxResults, cancellationToken);
|
return _client.SearchAddressSpaceAsync(query, maxDepth, maxResults, cancellationToken);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
|
||||||
public async Task<bool> WriteBatchAndWaitAsync(
|
|
||||||
IDictionary<string, object?> values, string flagPath, object? flagValue,
|
|
||||||
string responsePath, object? responseValue, TimeSpan timeout,
|
|
||||||
CancellationToken cancellationToken = default)
|
|
||||||
{
|
|
||||||
// Write all values including the flag
|
|
||||||
var allValues = new Dictionary<string, object?>(values) { [flagPath] = flagValue };
|
|
||||||
var writeResults = await WriteBatchAsync(allValues, cancellationToken);
|
|
||||||
|
|
||||||
if (writeResults.Values.Any(r => !r.Success))
|
|
||||||
return false;
|
|
||||||
|
|
||||||
// Poll for response value within timeout
|
|
||||||
var deadline = DateTimeOffset.UtcNow + timeout;
|
|
||||||
while (DateTimeOffset.UtcNow < deadline)
|
|
||||||
{
|
|
||||||
cancellationToken.ThrowIfCancellationRequested();
|
|
||||||
var readResult = await ReadAsync(responsePath, cancellationToken);
|
|
||||||
if (readResult.Success && readResult.Value != null && Equals(readResult.Value.Value, responseValue))
|
|
||||||
return true;
|
|
||||||
|
|
||||||
await Task.Delay(100, cancellationToken);
|
|
||||||
}
|
|
||||||
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
private void StopHeartbeatMonitor()
|
private void StopHeartbeatMonitor()
|
||||||
{
|
{
|
||||||
_staleMonitor?.Dispose();
|
_staleMonitor?.Dispose();
|
||||||
|
|||||||
-7
@@ -163,13 +163,6 @@ public sealed class FakeBatchDataConnection
|
|||||||
=> Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(
|
=> Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(
|
||||||
values.ToDictionary(kv => kv.Key, _ => new WriteResult(true, null)));
|
values.ToDictionary(kv => kv.Key, _ => new WriteResult(true, null)));
|
||||||
|
|
||||||
// Fake behaviour: Reports that the handshake matched without writing anything.
|
|
||||||
/// <inheritdoc />
|
|
||||||
public Task<bool> WriteBatchAndWaitAsync(
|
|
||||||
IDictionary<string, object?> values, string flagPath, object? flagValue, string responsePath,
|
|
||||||
object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default)
|
|
||||||
=> Task.FromResult(true);
|
|
||||||
|
|
||||||
// Fake behaviour: Captures the alarm callback and returns a generated subscription id.
|
// Fake behaviour: Captures the alarm callback and returns a generated subscription id.
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
public Task<string> SubscribeAlarmsAsync(
|
public Task<string> SubscribeAlarmsAsync(
|
||||||
|
|||||||
-53
@@ -186,59 +186,6 @@ public class MxGatewayDataConnectionTests
|
|||||||
Assert.Equal("rejected", r.ErrorMessage);
|
Assert.Equal("rejected", r.ErrorMessage);
|
||||||
}
|
}
|
||||||
|
|
||||||
// ── WriteBatchAndWait ──
|
|
||||||
|
|
||||||
/// <summary>Verifies that a batch write and wait returns true when the response value matches.</summary>
|
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
||||||
[Fact]
|
|
||||||
public async Task WriteBatchAndWait_returns_true_when_response_matches()
|
|
||||||
{
|
|
||||||
var writeCalls = new List<string>();
|
|
||||||
var fake = new FakeMxGatewayClient
|
|
||||||
{
|
|
||||||
WriteHandler = writes =>
|
|
||||||
{
|
|
||||||
writeCalls.AddRange(writes.Select(w => w.TagPath));
|
|
||||||
return writes.Select(w => new MxWriteOutcome(w.TagPath, true, null)).ToList();
|
|
||||||
},
|
|
||||||
// Response path already reads the expected value.
|
|
||||||
ReadHandler = tags => tags.Select(t =>
|
|
||||||
new MxReadOutcome(t, true, "DONE", QualityCode.Good, DateTimeOffset.UtcNow, null)).ToList()
|
|
||||||
};
|
|
||||||
var adapter = NewAdapter(fake);
|
|
||||||
await adapter.ConnectAsync(Details());
|
|
||||||
|
|
||||||
var ok = await adapter.WriteBatchAndWaitAsync(
|
|
||||||
new Dictionary<string, object?> { ["V"] = 1 }, "Flag", 1, "Resp", "DONE",
|
|
||||||
TimeSpan.FromSeconds(5));
|
|
||||||
|
|
||||||
Assert.True(ok);
|
|
||||||
// Values written before the flag, and the flag itself.
|
|
||||||
Assert.Contains("V", writeCalls);
|
|
||||||
Assert.Contains("Flag", writeCalls);
|
|
||||||
}
|
|
||||||
|
|
||||||
/// <summary>Verifies that a batch write and wait returns false on timeout.</summary>
|
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
||||||
[Fact]
|
|
||||||
public async Task WriteBatchAndWait_returns_false_on_timeout()
|
|
||||||
{
|
|
||||||
var fake = new FakeMxGatewayClient
|
|
||||||
{
|
|
||||||
WriteHandler = writes => writes.Select(w => new MxWriteOutcome(w.TagPath, true, null)).ToList(),
|
|
||||||
ReadHandler = tags => tags.Select(t =>
|
|
||||||
new MxReadOutcome(t, true, "NEVER", QualityCode.Good, DateTimeOffset.UtcNow, null)).ToList()
|
|
||||||
};
|
|
||||||
var adapter = NewAdapter(fake);
|
|
||||||
await adapter.ConnectAsync(Details());
|
|
||||||
|
|
||||||
var ok = await adapter.WriteBatchAndWaitAsync(
|
|
||||||
new Dictionary<string, object?> { ["V"] = 1 }, "Flag", 1, "Resp", "DONE",
|
|
||||||
TimeSpan.FromMilliseconds(300));
|
|
||||||
|
|
||||||
Assert.False(ok);
|
|
||||||
}
|
|
||||||
|
|
||||||
// ── browse ──
|
// ── browse ──
|
||||||
|
|
||||||
/// <summary>Verifies that browsing children maps the children and the truncated flag.</summary>
|
/// <summary>Verifies that browsing children maps the children and the truncated flag.</summary>
|
||||||
|
|||||||
@@ -152,14 +152,6 @@ public sealed class SimulatedDataConnection : IDataConnection, IBatchSubscribabl
|
|||||||
return Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(results);
|
return Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(results);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Accepts the batch and reports that the awaited response matched immediately.
|
|
||||||
/// <inheritdoc />
|
|
||||||
public Task<bool> WriteBatchAndWaitAsync(
|
|
||||||
IDictionary<string, object?> values, string flagPath, object? flagValue,
|
|
||||||
string responsePath, object? responseValue, TimeSpan timeout,
|
|
||||||
CancellationToken cancellationToken = default)
|
|
||||||
=> Task.FromResult(true);
|
|
||||||
|
|
||||||
/// <summary>Releases the connection; the simulated connection holds no resources.</summary>
|
/// <summary>Releases the connection; the simulated connection holds no resources.</summary>
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||||
|
|||||||
Reference in New Issue
Block a user