feat(siteruntime): event-driven Attributes.WriteBatchAndWaitAsync (batched DCL write + trigger + existing WaitForAttribute waiter) + compile mirror
This commit is contained in:
@@ -245,6 +245,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
break;
|
||||
case SubscribeTagsRequest:
|
||||
case WriteTagRequest:
|
||||
case WriteTagBatchRequest:
|
||||
case UnsubscribeTagsRequest:
|
||||
case SubscribeAlarmsRequest:
|
||||
case UnsubscribeAlarmsRequest:
|
||||
@@ -331,6 +332,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
case WriteTagRequest req:
|
||||
HandleWrite(req);
|
||||
break;
|
||||
case WriteTagBatchRequest req:
|
||||
HandleWriteBatch(req);
|
||||
break;
|
||||
case TagValueReceived tvr:
|
||||
HandleTagValueReceived(tvr);
|
||||
break;
|
||||
@@ -457,6 +461,7 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
break;
|
||||
case SubscribeTagsRequest:
|
||||
case WriteTagRequest:
|
||||
case WriteTagBatchRequest:
|
||||
case SubscribeAlarmsRequest:
|
||||
Stash.Stash();
|
||||
break;
|
||||
@@ -1059,6 +1064,67 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers
|
||||
}).PipeTo(sender);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Batch write counterpart of <see cref="HandleWrite"/>. Writes every value in
|
||||
/// <see cref="WriteTagBatchRequest.Values"/> to the device in ONE adapter
|
||||
/// <c>WriteBatchAsync</c> round-trip, then (if present) writes the trigger flag with
|
||||
/// a single <c>WriteAsync</c>. Both legs share one <see cref="DataConnectionOptions.WriteTimeout"/>
|
||||
/// budget (DataConnectionLayer-005). Any failed value in the batch, a failed trigger,
|
||||
/// the timeout, or an adapter exception is translated into a failed
|
||||
/// <see cref="WriteTagBatchResponse"/> returned synchronously to the caller — never a
|
||||
/// dropped reply. NOTE: this deliberately composes the batch + trigger primitives and
|
||||
/// uses the EXISTING event-driven WaitForAttribute waiter for the wait half; it does
|
||||
/// NOT call the adapter's poll-based <c>WriteBatchAndWaitAsync</c>.
|
||||
/// </summary>
|
||||
private void HandleWriteBatch(WriteTagBatchRequest request)
|
||||
{
|
||||
_log.Debug("[{0}] Batch-writing {1} tag(s)", _connectionName, request.Values.Count);
|
||||
var sender = Sender;
|
||||
var cts = new CancellationTokenSource(_options.WriteTimeout);
|
||||
|
||||
async Task<WriteTagBatchResponse> RunAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
var results = await _adapter.WriteBatchAsync(
|
||||
new Dictionary<string, object?>(request.Values), cts.Token);
|
||||
var failed = results.Values.Where(r => !r.Success).Select(r => r.ErrorMessage).ToList();
|
||||
if (failed.Count > 0)
|
||||
return new WriteTagBatchResponse(
|
||||
request.CorrelationId, false,
|
||||
"Batch write failed: " + string.Join("; ", failed), DateTimeOffset.UtcNow);
|
||||
|
||||
if (!string.IsNullOrEmpty(request.TriggerTagPath))
|
||||
{
|
||||
var tr = await _adapter.WriteAsync(request.TriggerTagPath, request.TriggerValue, cts.Token);
|
||||
if (!tr.Success)
|
||||
return new WriteTagBatchResponse(
|
||||
request.CorrelationId, false,
|
||||
"Trigger write failed: " + tr.ErrorMessage, DateTimeOffset.UtcNow);
|
||||
}
|
||||
|
||||
return new WriteTagBatchResponse(request.CorrelationId, true, null, DateTimeOffset.UtcNow);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return new WriteTagBatchResponse(
|
||||
request.CorrelationId, false,
|
||||
$"Write timeout after {_options.WriteTimeout.TotalSeconds:F0}s", DateTimeOffset.UtcNow);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return new WriteTagBatchResponse(
|
||||
request.CorrelationId, false, ex.GetBaseException().Message, DateTimeOffset.UtcNow);
|
||||
}
|
||||
finally
|
||||
{
|
||||
cts.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
RunAsync().PipeTo(sender);
|
||||
}
|
||||
|
||||
// ── OPC UA Tag Browser (interactive design-time query) ──
|
||||
|
||||
/// <summary>
|
||||
|
||||
@@ -46,6 +46,7 @@ public class DataConnectionManagerActor : ReceiveActor
|
||||
Receive<SubscribeAlarmsRequest>(HandleRouteAlarms);
|
||||
Receive<UnsubscribeAlarmsRequest>(HandleRouteAlarms);
|
||||
Receive<WriteTagRequest>(HandleRouteWrite);
|
||||
Receive<WriteTagBatchRequest>(HandleRouteWriteBatch);
|
||||
Receive<RemoveConnectionCommand>(HandleRemoveConnection);
|
||||
Receive<GetAllHealthReports>(HandleGetAllHealthReports);
|
||||
Receive<BrowseNodeCommand>(HandleBrowse);
|
||||
@@ -141,6 +142,26 @@ public class DataConnectionManagerActor : ReceiveActor
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Routes a <see cref="WriteTagBatchRequest"/> to the child
|
||||
/// <see cref="DataConnectionActor"/> that owns the named connection — the batch
|
||||
/// counterpart of <see cref="HandleRouteWrite"/>. The manager owns only the
|
||||
/// unknown-connection failure (the same split as every other routed message);
|
||||
/// the child resolves connected/not-connected and the per-write outcomes.
|
||||
/// </summary>
|
||||
private void HandleRouteWriteBatch(WriteTagBatchRequest request)
|
||||
{
|
||||
if (_connectionActors.TryGetValue(request.ConnectionName, out var actor))
|
||||
actor.Forward(request);
|
||||
else
|
||||
{
|
||||
_log.Warning("No connection actor for {0}", request.ConnectionName);
|
||||
Sender.Tell(new WriteTagBatchResponse(
|
||||
request.CorrelationId, false,
|
||||
$"Unknown connection: {request.ConnectionName}", DateTimeOffset.UtcNow));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Routes a <see cref="BrowseNodeCommand"/> from the central UI's OPC UA
|
||||
/// Tag Browser to the child <see cref="DataConnectionActor"/> that owns the
|
||||
|
||||
Reference in New Issue
Block a user