/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System.Collections.Generic; using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Threading; using System.Threading.Tasks; using Opc.Ua.Client; namespace Opc.Ua { /// /// The client side interface with support for batching according to operation limits. /// public class SessionClientBatched : SessionClient { /// /// The operation limits are used to batch the service requests. /// public OperationLimits OperationLimits { get; } = new(); /// /// The server capabilities /// public ServerCapabilities ServerCapabilities { get; } = new(); /// /// Intializes the object with a channel and logger factory. /// /// /// protected SessionClientBatched( ITransportChannel channel, ITelemetryContext telemetry) : base(channel, telemetry) { m_telemetry = telemetry; } /// public override ValueTask BrowseAsync( RequestHeader? requestHeader, ViewDescription? view, uint requestedMaxReferencesPerNode, ArrayOf nodesToBrowse, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerBrowse; if (operationLimit == 0 || operationLimit >= nodesToBrowse.Count) { return base.BrowseAsync( requestHeader, view, requestedMaxReferencesPerNode, nodesToBrowse, ct); } return BrowseBatchedAsync( requestHeader, view, requestedMaxReferencesPerNode, nodesToBrowse, operationLimit, ct); async ValueTask BrowseBatchedAsync( RequestHeader? requestHeader, ViewDescription? view, uint requestedMaxReferencesPerNode, ArrayOf nodesToBrowse, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); BrowseResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToBrowse.Count, operationLimit); foreach (ArrayOf nodesToBrowseBatch in nodesToBrowse.Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.BrowseAsync( requestHeader, view, requestedMaxReferencesPerNode, nodesToBrowseBatch, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, nodesToBrowseBatch); ValidateDiagnosticInfos(batchDiagnosticInfos, nodesToBrowseBatch); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask BrowseNextAsync( RequestHeader? requestHeader, bool releaseContinuationPoints, ArrayOf continuationPoints, CancellationToken ct) { ushort operationLimit = ServerCapabilities.MaxBrowseContinuationPoints; if (operationLimit == 0 || operationLimit >= continuationPoints.Count) { return base.BrowseNextAsync( requestHeader, releaseContinuationPoints, continuationPoints, ct); } return BrowseNextBatchedAsync( requestHeader, releaseContinuationPoints, continuationPoints, operationLimit, ct); async ValueTask BrowseNextBatchedAsync( RequestHeader? requestHeader, bool releaseContinuationPoints, ArrayOf continuationPoints, ushort operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); BrowseNextResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, continuationPoints.Count, operationLimit); foreach (ArrayOf continuationPointsBatch in continuationPoints.Batch(operationLimit)) { requestHeader.RequestHandle = 0; response = await base.BrowseNextAsync( requestHeader, releaseContinuationPoints, continuationPointsBatch, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, continuationPointsBatch); ValidateDiagnosticInfos(batchDiagnosticInfos, continuationPointsBatch); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask TranslateBrowsePathsToNodeIdsAsync( RequestHeader? requestHeader, ArrayOf browsePaths, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerTranslateBrowsePathsToNodeIds; if (operationLimit == 0 || operationLimit >= browsePaths.Count) { return base.TranslateBrowsePathsToNodeIdsAsync( requestHeader, browsePaths, ct); } return TranslateBrowsePathsToNodeIdsBatchedAsync(requestHeader, browsePaths, operationLimit, ct); async ValueTask TranslateBrowsePathsToNodeIdsBatchedAsync( RequestHeader? requestHeader, ArrayOf browsePaths, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); TranslateBrowsePathsToNodeIdsResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, browsePaths.Count, operationLimit); foreach (ArrayOf batchBrowsePaths in browsePaths.Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.TranslateBrowsePathsToNodeIdsAsync( requestHeader, batchBrowsePaths, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchBrowsePaths); ValidateDiagnosticInfos(batchDiagnosticInfos, batchBrowsePaths); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask RegisterNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToRegister, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerRegisterNodes; if (operationLimit == 0 || operationLimit >= nodesToRegister.Count) { return base.RegisterNodesAsync( requestHeader, nodesToRegister, ct); } return RegisterNodesBatchedAsync(requestHeader, nodesToRegister, operationLimit, ct); async ValueTask RegisterNodesBatchedAsync( RequestHeader? requestHeader, ArrayOf nodesToRegister, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); RegisterNodesResponse? response = null; var registeredNodeIds = new List(); foreach (ArrayOf batchNodesToRegister in nodesToRegister.Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.RegisterNodesAsync( requestHeader, batchNodesToRegister, ct).ConfigureAwait(false); ArrayOf batchRegisteredNodeIds = response.RegisteredNodeIds; ValidateResponse(batchRegisteredNodeIds, batchNodesToRegister); registeredNodeIds.AddRange(batchRegisteredNodeIds); } ValidateResponse(response); response.RegisteredNodeIds = registeredNodeIds; return response; } } /// public override ValueTask UnregisterNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToUnregister, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerRegisterNodes; if (operationLimit == 0 || operationLimit >= nodesToUnregister.Count) { return base.UnregisterNodesAsync( requestHeader, nodesToUnregister, ct); } return UnregisterNodesBatchedAsync(requestHeader, nodesToUnregister, operationLimit, ct); async ValueTask UnregisterNodesBatchedAsync( RequestHeader? requestHeader, ArrayOf nodesToUnregister, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); UnregisterNodesResponse? response = null; foreach (ArrayOf batchNodesToUnregister in nodesToUnregister .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.UnregisterNodesAsync( requestHeader, batchNodesToUnregister, ct).ConfigureAwait(false); } ValidateResponse(response); return response; } } /// public override ValueTask ReadAsync( RequestHeader? requestHeader, double maxAge, TimestampsToReturn timestampsToReturn, ArrayOf nodesToRead, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerRead; if (operationLimit == 0 || operationLimit >= nodesToRead.Count) { return base.ReadAsync( requestHeader, maxAge, timestampsToReturn, nodesToRead, ct); } return ReadBatchedAsync( requestHeader, maxAge, timestampsToReturn, nodesToRead, operationLimit, ct); async ValueTask ReadBatchedAsync( RequestHeader? requestHeader, double maxAge, TimestampsToReturn timestampsToReturn, ArrayOf nodesToRead, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); ReadResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToRead.Count, operationLimit); foreach (ArrayOf batchAttributesToRead in nodesToRead .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.ReadAsync( requestHeader, maxAge, timestampsToReturn, batchAttributesToRead, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchAttributesToRead); ValidateDiagnosticInfos(batchDiagnosticInfos, batchAttributesToRead); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask HistoryReadAsync( RequestHeader? requestHeader, ExtensionObject historyReadDetails, TimestampsToReturn timestampsToReturn, bool releaseContinuationPoints, ArrayOf nodesToRead, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerHistoryReadData; if (historyReadDetails.TypeId == DataTypeIds.ReadEventDetails || historyReadDetails.TryGetValue(out ReadEventDetails? _)) { operationLimit = OperationLimits.MaxNodesPerHistoryReadEvents; } if (operationLimit == 0 || operationLimit >= nodesToRead.Count) { return base.HistoryReadAsync( requestHeader, historyReadDetails, timestampsToReturn, releaseContinuationPoints, nodesToRead, ct); } return HistoryReadBatchedAsync( requestHeader, historyReadDetails, timestampsToReturn, releaseContinuationPoints, nodesToRead, operationLimit, ct); async ValueTask HistoryReadBatchedAsync( RequestHeader? requestHeader, ExtensionObject historyReadDetails, TimestampsToReturn timestampsToReturn, bool releaseContinuationPoints, ArrayOf nodesToRead, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); HistoryReadResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToRead.Count, operationLimit); foreach (ArrayOf batchNodesToRead in nodesToRead .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.HistoryReadAsync( requestHeader, historyReadDetails, timestampsToReturn, releaseContinuationPoints, batchNodesToRead, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchNodesToRead); ValidateDiagnosticInfos(batchDiagnosticInfos, batchNodesToRead); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask WriteAsync( RequestHeader? requestHeader, ArrayOf nodesToWrite, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerWrite; if (operationLimit == 0 || operationLimit >= nodesToWrite.Count) { return base.WriteAsync( requestHeader, nodesToWrite, ct); } return WriteBatchedAsync(requestHeader, nodesToWrite, operationLimit, ct); async ValueTask WriteBatchedAsync( RequestHeader? requestHeader, ArrayOf nodesToWrite, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); WriteResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToWrite.Count, operationLimit); foreach (ArrayOf batchNodesToWrite in nodesToWrite .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.WriteAsync( requestHeader, batchNodesToWrite, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchNodesToWrite); ValidateDiagnosticInfos(batchDiagnosticInfos, batchNodesToWrite); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask HistoryUpdateAsync( RequestHeader? requestHeader, ArrayOf historyUpdateDetails, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerHistoryUpdateData; if (historyUpdateDetails.Count > 0 && (historyUpdateDetails[0].TypeId == DataTypeIds.UpdateEventDetails || historyUpdateDetails[0].TryGetValue(out UpdateEventDetails? _))) { operationLimit = OperationLimits.MaxNodesPerHistoryUpdateEvents; } if (operationLimit == 0 || operationLimit >= historyUpdateDetails.Count) { return base.HistoryUpdateAsync( requestHeader, historyUpdateDetails, ct); } return HistoryUpdateBatchedAsync(requestHeader, historyUpdateDetails, operationLimit, ct); async ValueTask HistoryUpdateBatchedAsync( RequestHeader? requestHeader, ArrayOf historyUpdateDetails, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); HistoryUpdateResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, historyUpdateDetails.Count, operationLimit); foreach (ArrayOf batchHistoryUpdateDetails in historyUpdateDetails .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.HistoryUpdateAsync( requestHeader, batchHistoryUpdateDetails, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchHistoryUpdateDetails); ValidateDiagnosticInfos(batchDiagnosticInfos, batchHistoryUpdateDetails); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask CallAsync( RequestHeader? requestHeader, ArrayOf methodsToCall, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerMethodCall; if (operationLimit == 0 || operationLimit >= methodsToCall.Count) { return base.CallAsync( requestHeader, methodsToCall, ct); } return CallBatchedAsync(requestHeader, methodsToCall, operationLimit, ct); async ValueTask CallBatchedAsync( RequestHeader? requestHeader, ArrayOf methodsToCall, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); CallResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, methodsToCall.Count, operationLimit); foreach (ArrayOf batchMethodsToCall in methodsToCall .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.CallAsync( requestHeader, batchMethodsToCall, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchMethodsToCall); ValidateDiagnosticInfos(batchDiagnosticInfos, batchMethodsToCall); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask CreateMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToCreate, CancellationToken ct) { uint operationLimit = OperationLimits.MaxMonitoredItemsPerCall; if (operationLimit == 0 || operationLimit >= itemsToCreate.Count) { return base.CreateMonitoredItemsAsync( requestHeader, subscriptionId, timestampsToReturn, itemsToCreate, ct); } return CreateMonitoredItemsBatchedAsync( requestHeader, subscriptionId, timestampsToReturn, itemsToCreate, operationLimit, ct); async ValueTask CreateMonitoredItemsBatchedAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToCreate, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); CreateMonitoredItemsResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, itemsToCreate.Count, operationLimit); foreach (ArrayOf batchItemsToCreate in itemsToCreate .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.CreateMonitoredItemsAsync( requestHeader, subscriptionId, timestampsToReturn, batchItemsToCreate, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchItemsToCreate); ValidateDiagnosticInfos(batchDiagnosticInfos, batchItemsToCreate); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask ModifyMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToModify, CancellationToken ct) { uint operationLimit = OperationLimits.MaxMonitoredItemsPerCall; if (operationLimit == 0 || operationLimit >= itemsToModify.Count) { return base.ModifyMonitoredItemsAsync( requestHeader, subscriptionId, timestampsToReturn, itemsToModify, ct); } return ModifyMonitoredItemsBatchedAsync( requestHeader, subscriptionId, timestampsToReturn, itemsToModify, operationLimit, ct); async ValueTask ModifyMonitoredItemsBatchedAsync( RequestHeader? requestHeader, uint subscriptionId, TimestampsToReturn timestampsToReturn, ArrayOf itemsToModify, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); ModifyMonitoredItemsResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, itemsToModify.Count, operationLimit); foreach (ArrayOf batchItemsToModify in itemsToModify .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.ModifyMonitoredItemsAsync( requestHeader, subscriptionId, timestampsToReturn, batchItemsToModify, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchItemsToModify); ValidateDiagnosticInfos(batchDiagnosticInfos, batchItemsToModify); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask SetMonitoringModeAsync( RequestHeader? requestHeader, uint subscriptionId, MonitoringMode monitoringMode, ArrayOf monitoredItemIds, CancellationToken ct) { uint operationLimit = OperationLimits.MaxMonitoredItemsPerCall; if (operationLimit == 0 || operationLimit >= monitoredItemIds.Count) { return base.SetMonitoringModeAsync( requestHeader, subscriptionId, monitoringMode, monitoredItemIds, ct); } return SetMonitoringModeBatchedAsync( requestHeader, subscriptionId, monitoringMode, monitoredItemIds, operationLimit, ct); async ValueTask SetMonitoringModeBatchedAsync( RequestHeader? requestHeader, uint subscriptionId, MonitoringMode monitoringMode, ArrayOf monitoredItemIds, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); SetMonitoringModeResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, monitoredItemIds.Count, operationLimit); foreach (ArrayOf batchMonitoredItemIds in monitoredItemIds .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.SetMonitoringModeAsync( requestHeader, subscriptionId, monitoringMode, batchMonitoredItemIds, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchMonitoredItemIds); ValidateDiagnosticInfos(batchDiagnosticInfos, batchMonitoredItemIds); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask SetTriggeringAsync( RequestHeader? requestHeader, uint subscriptionId, uint triggeringItemId, ArrayOf linksToAdd, ArrayOf linksToRemove, CancellationToken ct) { uint operationLimit = OperationLimits.MaxMonitoredItemsPerCall; if (operationLimit == 0 || operationLimit >= linksToAdd.Count + linksToRemove.Count) { return base.SetTriggeringAsync( requestHeader, subscriptionId, triggeringItemId, linksToAdd, linksToRemove, ct); } return SetTriggeringBatchedAsync( requestHeader, subscriptionId, triggeringItemId, linksToAdd, linksToRemove, operationLimit, ct); async ValueTask SetTriggeringBatchedAsync( RequestHeader? requestHeader, uint subscriptionId, uint triggeringItemId, ArrayOf linksToAdd, ArrayOf linksToRemove, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); SetTriggeringResponse? response = null; InitResponseCollections( out List? addResults, out List? addDiagnosticInfos, out List stringTable, linksToAdd.Count, operationLimit); InitResponseCollections( out List? removeResults, out List? removeDiagnosticInfos, out _, linksToRemove.Count, operationLimit); foreach (ArrayOf batchLinksToAdd in linksToAdd .Batch((int)operationLimit)) { ArrayOf batchLinksToRemove; if (operationLimit == 0) { batchLinksToRemove = linksToRemove; linksToRemove = []; } else if (batchLinksToAdd.Count < operationLimit) { batchLinksToRemove = linksToRemove[..((int)operationLimit - batchLinksToAdd.Count)]; linksToRemove = linksToRemove[batchLinksToRemove.Count..]; } else { batchLinksToRemove = []; } requestHeader.RequestHandle = 0; response = await base.SetTriggeringAsync( requestHeader, subscriptionId, triggeringItemId, batchLinksToAdd, batchLinksToRemove, ct).ConfigureAwait(false); ArrayOf batchAddResults = response.AddResults; ArrayOf batchAddDiagnosticInfos = response.AddDiagnosticInfos; ArrayOf batchRemoveResults = response.RemoveResults; ArrayOf batchRemoveDiagnosticInfos = response.RemoveDiagnosticInfos; ValidateResponse(batchAddResults, batchLinksToAdd); ValidateDiagnosticInfos(batchAddDiagnosticInfos, batchLinksToAdd); ValidateResponse(batchRemoveResults, batchLinksToRemove); ValidateDiagnosticInfos(batchRemoveDiagnosticInfos, batchLinksToRemove); AddResponses( ref addResults, ref addDiagnosticInfos, ref stringTable, batchAddResults, batchAddDiagnosticInfos, response.ResponseHeader.StringTable); AddResponses( ref removeResults, ref removeDiagnosticInfos, ref stringTable, batchRemoveResults, batchRemoveDiagnosticInfos, response.ResponseHeader.StringTable); } if (linksToRemove.Count > 0) { foreach (ArrayOf batchLinksToRemove in linksToRemove .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; var batchLinksToAdd = ArrayOf.Empty(); response = await base.SetTriggeringAsync( requestHeader, subscriptionId, triggeringItemId, batchLinksToAdd, batchLinksToRemove, ct).ConfigureAwait(false); ArrayOf batchAddResults = response.AddResults; ArrayOf batchAddDiagnosticInfos = response.AddDiagnosticInfos; ArrayOf batchRemoveResults = response.RemoveResults; ArrayOf batchRemoveDiagnosticInfos = response.RemoveDiagnosticInfos; ValidateResponse(batchAddResults, batchLinksToAdd); ValidateDiagnosticInfos(batchAddDiagnosticInfos, batchLinksToAdd); ValidateResponse(batchRemoveResults, batchLinksToRemove); ValidateDiagnosticInfos(batchRemoveDiagnosticInfos, batchLinksToRemove); AddResponses( ref addResults, ref addDiagnosticInfos, ref stringTable, batchAddResults, batchAddDiagnosticInfos, response.ResponseHeader.StringTable); AddResponses( ref removeResults, ref removeDiagnosticInfos, ref stringTable, batchRemoveResults, batchRemoveDiagnosticInfos, response.ResponseHeader.StringTable); } } ValidateResponse(response); response.AddResults = addResults; response.AddDiagnosticInfos = addDiagnosticInfos; response.RemoveResults = removeResults; response.RemoveDiagnosticInfos = removeDiagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask DeleteMonitoredItemsAsync( RequestHeader? requestHeader, uint subscriptionId, ArrayOf monitoredItemIds, CancellationToken ct) { uint operationLimit = OperationLimits.MaxMonitoredItemsPerCall; if (operationLimit == 0 || operationLimit >= monitoredItemIds.Count) { return base.DeleteMonitoredItemsAsync( requestHeader, subscriptionId, monitoredItemIds, ct); } return DeleteMonitoredItemsBatchedAsync( requestHeader, subscriptionId, monitoredItemIds, operationLimit, ct); async ValueTask DeleteMonitoredItemsBatchedAsync( RequestHeader? requestHeader, uint subscriptionId, ArrayOf monitoredItemIds, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); DeleteMonitoredItemsResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, monitoredItemIds.Count, operationLimit); foreach (ArrayOf batchMonitoredItemIds in monitoredItemIds .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.DeleteMonitoredItemsAsync( requestHeader, subscriptionId, batchMonitoredItemIds, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchMonitoredItemIds); ValidateDiagnosticInfos(batchDiagnosticInfos, batchMonitoredItemIds); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask AddNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToAdd, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerNodeManagement; if (operationLimit == 0 || operationLimit >= nodesToAdd.Count) { return base.AddNodesAsync( requestHeader, nodesToAdd, ct); } return AddNodesBatchedAsync(requestHeader, nodesToAdd, operationLimit, ct); async ValueTask AddNodesBatchedAsync( RequestHeader? requestHeader, ArrayOf nodesToAdd, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); AddNodesResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToAdd.Count, operationLimit); foreach (ArrayOf batchNodesToAdd in nodesToAdd .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.AddNodesAsync(requestHeader, batchNodesToAdd, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchNodesToAdd); ValidateDiagnosticInfos(batchDiagnosticInfos, batchNodesToAdd); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask AddReferencesAsync( RequestHeader? requestHeader, ArrayOf referencesToAdd, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerNodeManagement; if (operationLimit == 0 || operationLimit >= referencesToAdd.Count) { return base.AddReferencesAsync( requestHeader, referencesToAdd, ct); } return AddReferencesBatchedAsync(requestHeader, referencesToAdd, operationLimit, ct); async ValueTask AddReferencesBatchedAsync( RequestHeader? requestHeader, ArrayOf referencesToAdd, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); AddReferencesResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, referencesToAdd.Count, operationLimit); foreach (ArrayOf batchReferencesToAdd in referencesToAdd .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.AddReferencesAsync( requestHeader, batchReferencesToAdd, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchReferencesToAdd); ValidateDiagnosticInfos(batchDiagnosticInfos, batchReferencesToAdd); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask DeleteNodesAsync( RequestHeader? requestHeader, ArrayOf nodesToDelete, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerNodeManagement; if (operationLimit == 0 || operationLimit >= nodesToDelete.Count) { return base.DeleteNodesAsync( requestHeader, nodesToDelete, ct); } return DeleteNodesBatchedAsync(requestHeader, nodesToDelete, operationLimit, ct); async ValueTask DeleteNodesBatchedAsync( RequestHeader? requestHeader, ArrayOf nodesToDelete, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); DeleteNodesResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, nodesToDelete.Count, operationLimit); foreach (ArrayOf batchNodesToDelete in nodesToDelete .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.DeleteNodesAsync( requestHeader, batchNodesToDelete, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchNodesToDelete); ValidateDiagnosticInfos(batchDiagnosticInfos, batchNodesToDelete); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// public override ValueTask DeleteReferencesAsync( RequestHeader? requestHeader, ArrayOf referencesToDelete, CancellationToken ct) { uint operationLimit = OperationLimits.MaxNodesPerNodeManagement; if (operationLimit == 0 || operationLimit >= referencesToDelete.Count) { return base.DeleteReferencesAsync( requestHeader, referencesToDelete, ct); } return DeleteReferencesBatchedAsync(requestHeader, referencesToDelete, operationLimit, ct); async ValueTask DeleteReferencesBatchedAsync( RequestHeader? requestHeader, ArrayOf referencesToDelete, uint operationLimit, CancellationToken ct) { using Activity? activity = m_telemetry.StartActivity(); requestHeader ??= new RequestHeader(); DeleteReferencesResponse? response = null; InitResponseCollections( out List? results, out List? diagnosticInfos, out List stringTable, referencesToDelete.Count, operationLimit); foreach (ArrayOf batchReferencesToDelete in referencesToDelete .Batch((int)operationLimit)) { requestHeader.RequestHandle = 0; response = await base.DeleteReferencesAsync( requestHeader, batchReferencesToDelete, ct).ConfigureAwait(false); ArrayOf batchResults = response.Results; ArrayOf batchDiagnosticInfos = response.DiagnosticInfos; ValidateResponse(batchResults, batchReferencesToDelete); ValidateDiagnosticInfos(batchDiagnosticInfos, batchReferencesToDelete); AddResponses( ref results, ref diagnosticInfos, ref stringTable, batchResults, batchDiagnosticInfos, response.ResponseHeader.StringTable); } ValidateResponse(response); response.Results = results; response.DiagnosticInfos = diagnosticInfos; response.ResponseHeader.StringTable = stringTable; return response; } } /// /// Check response for null /// /// /// private static void ValidateResponse([NotNull] IServiceResponse? response) { if (response is null) { throw ServiceResultException.Unexpected( "Empty ops should have been covered in fast path"); } } /// /// Initialize the collections for a service call. /// /// /// /// /// /// /// /// /// Preset the result collections with null if the operation limit /// is sufficient or with the final size if batching is necessary. /// private static void InitResponseCollections( out List results, out List diagnosticInfos, out List stringTable, int count, uint operationLimit) { Debug.Assert(count > operationLimit); results = new List(count); diagnosticInfos = new List(count); stringTable = []; } /// /// Add the result of a batched service call to the results. /// /// /// /// /// /// /// /// /// /// Assigns the batched collection result to the result if the result /// collection is not initialized, adds the range to the result /// collections otherwise. /// The string table indexes are updated in the diagnostic infos if necessary. /// private static void AddResponses( ref List results, ref List diagnosticInfos, ref List stringTable, ArrayOf batchedResults, ArrayOf batchedDiagnosticInfos, ArrayOf batchedStringTable) { bool hasDiagnosticInfos = diagnosticInfos.Count > 0; bool hasEmptyDiagnosticInfos = diagnosticInfos.Count == 0 && results.Count > 0; bool hasBatchDiagnosticInfos = batchedDiagnosticInfos.Count > 0; int correctionCount = 0; if (hasBatchDiagnosticInfos && hasEmptyDiagnosticInfos) { correctionCount = results.Count; } else if (!hasBatchDiagnosticInfos && hasDiagnosticInfos) { correctionCount = batchedResults.Count; } if (correctionCount > 0) { // fill missing diagnostics infos with null entries for (int i = 0; i < correctionCount; i++) { // DiagnosticInfoCollection accepts null sentinels even though the // declared element type is non-nullable; null marks a "missing" slot. diagnosticInfos.Add(null!); } } else if (batchedStringTable.Count > 0) { // correct indexes in the string table int stringTableOffset = stringTable.Count; foreach (DiagnosticInfo? diagnosticInfo in batchedDiagnosticInfos) { UpdateDiagnosticInfoIndexes(diagnosticInfo, stringTableOffset); } } results.AddRange(batchedResults); diagnosticInfos.AddRange(batchedDiagnosticInfos); stringTable.AddRange(batchedStringTable); static void UpdateDiagnosticInfoIndexes(DiagnosticInfo? diagnosticInfo, int stringTableOffset) { int depth = 0; while (diagnosticInfo != null && depth++ < DiagnosticInfo.MaxInnerDepth) { if (diagnosticInfo.LocalizedText >= 0) { diagnosticInfo.LocalizedText += stringTableOffset; } if (diagnosticInfo.Locale >= 0) { diagnosticInfo.Locale += stringTableOffset; } if (diagnosticInfo.NamespaceUri >= 0) { diagnosticInfo.NamespaceUri += stringTableOffset; } if (diagnosticInfo.SymbolicId >= 0) { diagnosticInfo.SymbolicId += stringTableOffset; } diagnosticInfo = diagnosticInfo.InnerDiagnosticInfo; } } } private readonly ITelemetryContext m_telemetry; } }