Files
UA-.NETStandard/Libraries/Opc.Ua.Server/Fluent/NodeManagerBuilder.cs
Copilot b3518b53df [Server]: Queue Events from the Server Node using multiple channel consumers (#3823)
# Description

This pull request introduces a scalable consumer task model for the
`MonitoredNode2` class, optimizing the handling of event notifications
for notifier nodes. Any notifier node can opt into (or out of) dynamic
scaling of consumer tasks based on the number of event monitored items,
improving throughput and responsiveness under high event loads. The
Server node (`ObjectIds.Server`) is always auto-opted-in. The
implementation includes safe scaling up and down of consumers, robust
disposal logic, Fluent API integration, and tests to verify correct
delivery of events.

**Scalable consumer task management:**

* Added an `enableMultipleEventConsumers` constructor parameter to
`MonitoredNode2` that enables dynamic scaling of consumer tasks to match
the number of event monitored items. The Server node
(`ObjectIds.Server`) always auto-opts-in regardless of this parameter.
* Added `AllowMultipleEventConsumers(NodeId, bool enable = true)`
protected method on `AsyncCustomNodeManager` with an internal
`MultiConsumerNodeIds` registry, so any node manager can
programmatically opt in or out specific nodes.
* Added `IsMultipleEventConsumerNode(NodeId)` to the `IAsyncNodeManager`
interface, allowing consumers to query multi-consumer opt-in status
through the interface without casting to `AsyncCustomNodeManager`.
* Both `MonitoredNodeMonitoredItemManager` and
`SamplingGroupMonitoredItemManager` use the `IAsyncNodeManager`
interface method when creating `MonitoredNode2` instances.

**Fluent API integration:**

* Added `AllowMultipleEventConsumers(bool enable = true)` method to
`INodeBuilder` and its `NodeBuilder` implementation, allowing any
notifier node to be opted in or out via the fluent builder pipeline
(e.g. `builder.Node("myNotifier").AllowMultipleEventConsumers()`).
* The `NodeManagerBuilder` routes the registration to the owning
`AsyncCustomNodeManager`'s multi-consumer registry, supporting both
opt-in and opt-out.

**Testing and verification:**

* Added unit tests to verify multi-consumer behavior for both Server
node (auto-opt-in) and non-Server nodes (explicit opt-in), as well as
default single-consumer behavior for nodes without opt-in.
* Fixed race condition in permission cache tests
(`OnMonitoredNodeChanged_DefaultPermissionsChanged_CacheInvalidated` and
`InvalidatePermissionCacheForSession` tests) that caused flaky CI
failures. The tests now signal on `QueueValue` (which fires after the
cache is populated) instead of `ValidateRolePermissionsAsync` (which
fires before), ensuring cache invalidation operates on a populated
cache.

## Related Issues

## Checklist

- [ ] I have signed the
[CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf)
and read the
[CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md)
doc.
- [x] I have added tests that prove my fix is effective or that my
feature works and increased code coverage.
- [ ] I have added all necessary documentation.
- [x] I have verified that my changes do not introduce (new) build or
analyzer warnings.
- [ ] I ran **all** tests locally using the **UA.slnx** solution against
at least .net **framework** and .net **10**, and all passed.
- [x] I fixed **all** failing and flaky tests in the CI pipelines and
**all** CodeQL warnings.
- [x] I have addressed **all** PR feedback received.

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
2026-06-02 07:14:36 +02:00

597 lines
23 KiB
C#

/* ========================================================================
* 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;
using System.Collections.Generic;
namespace Opc.Ua.Server.Fluent
{
/// <summary>
/// Default implementation of <see cref="INodeManagerBuilder"/> and
/// <see cref="IFluentDispatcher"/>. Built and owned by the source-generated
/// <c>NodeManagerBase</c> (or by a hand-written manager that wants to opt
/// in to the fluent surface).
/// </summary>
/// <remarks>
/// <para>
/// All wiring happens during the user's <c>Configure</c> delegate, which
/// runs once per manager activation immediately after
/// <c>LoadPredefinedNodes</c> populates the address space. After
/// <see cref="Seal"/> is called the builder rejects further <c>Node(...)</c>
/// calls; the dispatcher remains live and fields per-node lookups during
/// runtime.
/// </para>
/// <para>
/// Threading: <c>Configure</c> runs synchronously on the thread that
/// activates the manager; the dispatcher's dictionaries are populated
/// once and read-only thereafter, so no synchronization is needed at
/// dispatch time.
/// </para>
/// </remarks>
public sealed class NodeManagerBuilder : INodeManagerBuilder, IFluentDispatcher
{
/// <summary>
/// Creates a new builder for the supplied <paramref name="nodeManager"/>.
/// </summary>
/// <param name="context">
/// System context that flows through <c>Configure</c>; typically the
/// manager's <c>SystemContext</c>.
/// </param>
/// <param name="nodeManager">The node manager being wired.</param>
/// <param name="defaultNamespaceIndex">
/// Namespace index used when a browse-path segment omits an explicit
/// <c>ns=N;</c> prefix. Typically the manager's first registered
/// namespace.
/// </param>
/// <param name="rootResolver">
/// Delegate that locates a root <see cref="NodeState"/> for a given
/// <see cref="QualifiedName"/>. Typically backed by the manager's
/// <c>PredefinedNodes</c> dictionary.
/// </param>
/// <param name="nodeIdResolver">
/// Delegate that locates a <see cref="NodeState"/> by absolute
/// <see cref="NodeId"/>. Typically backed by the manager's
/// <c>PredefinedNodes</c> dictionary.
/// </param>
/// <param name="typeIdResolver">
/// Delegate that returns every <see cref="NodeState"/> whose
/// <c>TypeDefinitionId</c> matches the supplied <see cref="NodeId"/>.
/// Typically a generated walk over the manager's predefined nodes.
/// </param>
/// <exception cref="ArgumentNullException">
/// <paramref name="context"/>, <paramref name="nodeManager"/>,
/// <paramref name="rootResolver"/>, <paramref name="nodeIdResolver"/>,
/// or <paramref name="typeIdResolver"/> is null.
/// </exception>
public NodeManagerBuilder(
ISystemContext context,
IAsyncNodeManager nodeManager,
ushort defaultNamespaceIndex,
Func<QualifiedName, NodeState> rootResolver,
Func<NodeId, NodeState> nodeIdResolver,
Func<NodeId, IReadOnlyList<NodeState>> typeIdResolver)
{
Context = context ?? throw new ArgumentNullException(nameof(context));
NodeManager = nodeManager ?? throw new ArgumentNullException(nameof(nodeManager));
m_defaultNamespaceIndex = defaultNamespaceIndex;
m_rootResolver = rootResolver ?? throw new ArgumentNullException(nameof(rootResolver));
m_nodeIdResolver = nodeIdResolver ?? throw new ArgumentNullException(nameof(nodeIdResolver));
m_typeIdResolver = typeIdResolver ?? throw new ArgumentNullException(nameof(typeIdResolver));
}
/// <inheritdoc/>
public ISystemContext Context { get; }
/// <inheritdoc/>
public IAsyncNodeManager NodeManager { get; }
/// <inheritdoc/>
public IFluentDispatcher Dispatcher => this;
/// <summary>
/// Marks the builder as no longer accepting new <c>Node(...)</c>
/// lookups. Existing per-node builders remain functional but the
/// generator-emitted manager calls this once <c>Configure</c>
/// returns to fail-fast on stray late wiring attempts.
/// </summary>
public void Seal()
{
m_sealed = true;
}
/// <inheritdoc/>
public INodeBuilder Node(string browsePath)
{
ThrowIfSealed();
NodeState node = BrowsePathResolver.Resolve(
Context,
browsePath,
m_defaultNamespaceIndex,
m_rootResolver);
return new NodeBuilder(this, node);
}
/// <inheritdoc/>
public INodeBuilder<TState> Node<TState>(string browsePath)
where TState : NodeState
{
ThrowIfSealed();
NodeState node = BrowsePathResolver.Resolve(
Context,
browsePath,
m_defaultNamespaceIndex,
m_rootResolver);
if (node is not TState typed)
{
throw ServiceResultException.Create(
StatusCodes.BadTypeMismatch,
"Browse path '{0}' resolved to {1}, which is not assignable to {2}.",
browsePath,
node.GetType().Name,
typeof(TState).Name);
}
return new NodeBuilder<TState>(this, typed);
}
/// <inheritdoc/>
public INodeBuilder Node(NodeId nodeId)
{
ThrowIfSealed();
NodeState node = ResolveNodeId(nodeId);
return new NodeBuilder(this, node);
}
/// <inheritdoc/>
public INodeBuilder<TState> Node<TState>(NodeId nodeId)
where TState : NodeState
{
ThrowIfSealed();
NodeState node = ResolveNodeId(nodeId);
if (node is not TState typed)
{
throw ServiceResultException.Create(
StatusCodes.BadTypeMismatch,
"NodeId '{0}' resolved to {1}, which is not assignable to {2}.",
nodeId,
node.GetType().Name,
typeof(TState).Name);
}
return new NodeBuilder<TState>(this, typed);
}
/// <inheritdoc/>
public INodeBuilder NodeFromTypeId(NodeId typeDefinitionId)
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, (QualifiedName)null!);
return new NodeBuilder(this, node);
}
/// <inheritdoc/>
public INodeBuilder NodeFromTypeId(NodeId typeDefinitionId, QualifiedName browseName)
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, browseName);
return new NodeBuilder(this, node);
}
/// <inheritdoc/>
public INodeBuilder<TState> NodeFromTypeId<TState>(NodeId typeDefinitionId)
where TState : NodeState
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, (QualifiedName)null!);
if (node is not TState typed)
{
throw ServiceResultException.Create(
StatusCodes.BadTypeMismatch,
"TypeDefinitionId '{0}' resolved to {1}, which is not assignable to {2}.",
typeDefinitionId,
node.GetType().Name,
typeof(TState).Name);
}
return new NodeBuilder<TState>(this, typed);
}
/// <inheritdoc/>
public INodeBuilder<TState> NodeFromTypeId<TState>(NodeId typeDefinitionId, QualifiedName browseName)
where TState : NodeState
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, browseName);
if (node is not TState typed)
{
throw ServiceResultException.Create(
StatusCodes.BadTypeMismatch,
"TypeDefinitionId '{0}' (browse name '{1}') resolved to {2}, which is not assignable to {3}.",
typeDefinitionId,
browseName,
node.GetType().Name,
typeof(TState).Name);
}
return new NodeBuilder<TState>(this, typed);
}
/// <inheritdoc/>
public IVariableBuilder<TValue> Variable<TValue>(string browsePath)
{
ThrowIfSealed();
NodeState node = BrowsePathResolver.Resolve(
Context,
browsePath,
m_defaultNamespaceIndex,
m_rootResolver);
return ToVariableBuilder<TValue>(node, browsePath);
}
/// <inheritdoc/>
public IVariableBuilder<TValue> Variable<TValue>(NodeId nodeId)
{
ThrowIfSealed();
NodeState node = ResolveNodeId(nodeId);
return ToVariableBuilder<TValue>(node, FormatNodeId(nodeId));
}
/// <inheritdoc/>
public IVariableBuilder<TValue> VariableFromTypeId<TValue>(NodeId typeDefinitionId)
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, (QualifiedName)null!);
return ToVariableBuilder<TValue>(node, FormatNodeId(typeDefinitionId));
}
/// <inheritdoc/>
public IVariableBuilder<TValue> VariableFromTypeId<TValue>(NodeId typeDefinitionId, QualifiedName browseName)
{
ThrowIfSealed();
NodeState node = ResolveByTypeDefinition(typeDefinitionId, browseName);
return ToVariableBuilder<TValue>(
node,
CoreUtils.Format(
"{0} (browse name '{1}')",
FormatNodeId(typeDefinitionId),
browseName));
}
internal VariableBuilder<TValue> ToVariableBuilder<TValue>(NodeState node, string lookupHint)
{
if (node is not BaseVariableState variable)
{
throw ServiceResultException.Create(
StatusCodes.BadTypeMismatch,
"Lookup '{0}' resolved to {1}, which is not a BaseVariableState.",
lookupHint,
node.GetType().Name);
}
return new VariableBuilder<TValue>(this, variable);
}
/// <summary>
/// Event-source registry owned by the
/// <see cref="FluentNodeManagerBase"/>; populated via
/// <see cref="AttachEventSources"/> immediately after the
/// builder is constructed and before <c>Configure</c> runs.
/// </summary>
/// <remarks>
/// Hand-written managers that derive from
/// <see cref="CustomNodeManager2"/> rather than
/// <see cref="FluentNodeManagerBase"/> leave this property
/// <c>null</c>; the <c>Publish</c> extensions surface a
/// targeted error in that case.
/// </remarks>
internal EventSourceRegistry? EventSources { get; private set; }
/// <summary>
/// Wires the supplied registry into this builder so the
/// <c>Publish</c> extensions can route source registrations to
/// the owning manager. Called once by
/// <see cref="FluentNodeManagerBase"/>; subsequent calls throw.
/// </summary>
/// <exception cref="ServiceResultException">
/// An <see cref="EventSourceRegistry"/> is already attached to
/// this builder.
/// </exception>
/// <exception cref="ArgumentNullException">
/// <paramref name="registry"/> is null.
/// </exception>
internal void AttachEventSources(EventSourceRegistry registry)
{
if (EventSources != null)
{
throw ServiceResultException.Create(
StatusCodes.BadInvalidState,
"An EventSourceRegistry is already attached to this builder.");
}
EventSources = registry ?? throw new ArgumentNullException(nameof(registry));
}
private static string FormatNodeId(NodeId nodeId)
{
// NodeId is a readonly struct so the caller may pass `default`;
// .IsNull guards both the default-struct case and a constructed
// NodeId with no identifier.
return nodeId.IsNull ? "(null)" : nodeId.ToString();
}
/// <inheritdoc/>
public bool TryHandleHistoryRead(
ISystemContext context,
NodeState node,
HistoryReadDetails details,
TimestampsToReturn timestampsToReturn,
bool releaseContinuationPoints,
HistoryReadValueId nodeToRead,
HistoryReadResult result,
out ServiceResult status)
{
if (node != null &&
m_historyRead.TryGetValue(node.NodeId, out HistoryReadHandler? handler))
{
status = handler(
context,
node,
details,
timestampsToReturn,
releaseContinuationPoints,
nodeToRead,
result);
return true;
}
status = ServiceResult.Good;
return false;
}
/// <inheritdoc/>
public bool TryHandleHistoryUpdate(
ISystemContext context,
NodeState node,
HistoryUpdateDetails nodeToUpdate,
HistoryUpdateResult result,
out ServiceResult status)
{
if (node != null &&
m_historyUpdate.TryGetValue(node.NodeId, out HistoryUpdateHandler? handler))
{
status = handler(context, node, nodeToUpdate, result);
return true;
}
status = ServiceResult.Good;
return false;
}
/// <inheritdoc/>
public void NotifyMonitoredItemCreated(
ISystemContext context,
NodeState source,
ISampledDataChangeMonitoredItem monitoredItem)
{
if (source != null &&
m_monitoredItemCreated.TryGetValue(source.NodeId, out MonitoredItemCreatedHandler? handler))
{
handler(context, source, monitoredItem);
}
}
/// <inheritdoc/>
public void NotifyNodeAdded(ISystemContext context, NodeState node)
{
if (node != null &&
m_nodeAdded.TryGetValue(node.NodeId, out NodeLifecycleHandler? handler))
{
handler(context, node);
}
}
/// <inheritdoc/>
public void NotifyNodeRemoved(ISystemContext context, NodeState node)
{
if (node != null &&
m_nodeRemoved.TryGetValue(node.NodeId, out NodeLifecycleHandler? handler))
{
handler(context, node);
}
}
internal void RegisterHistoryRead(NodeState node, HistoryReadHandler handler)
{
ThrowIfDuplicate(m_historyRead, node, "OnHistoryRead");
m_historyRead[node.NodeId] = handler;
}
internal void RegisterHistoryUpdate(NodeState node, HistoryUpdateHandler handler)
{
ThrowIfDuplicate(m_historyUpdate, node, "OnHistoryUpdate");
m_historyUpdate[node.NodeId] = handler;
}
internal void RegisterMonitoredItemCreated(NodeState node, MonitoredItemCreatedHandler handler)
{
ThrowIfDuplicate(m_monitoredItemCreated, node, "OnMonitoredItemCreated");
m_monitoredItemCreated[node.NodeId] = handler;
}
internal void RegisterNodeAdded(NodeState node, NodeLifecycleHandler handler)
{
ThrowIfDuplicate(m_nodeAdded, node, "OnNodeAdded");
m_nodeAdded[node.NodeId] = handler;
}
internal void RegisterNodeRemoved(NodeState node, NodeLifecycleHandler handler)
{
ThrowIfDuplicate(m_nodeRemoved, node, "OnNodeRemoved");
m_nodeRemoved[node.NodeId] = handler;
}
internal void RegisterMultiConsumerNode(NodeState node, bool enable)
{
if (NodeManager is not AsyncCustomNodeManager acnm)
{
throw ServiceResultException.Create(
StatusCodes.BadConfigurationError,
"AllowMultipleEventConsumers requires the node manager to derive from AsyncCustomNodeManager. " +
"Manager type '{0}' does not qualify.",
NodeManager?.GetType().FullName ?? "(unknown)");
}
if (enable)
{
acnm.MultiConsumerNodeIds[node.NodeId] = true;
}
else
{
acnm.MultiConsumerNodeIds.Remove(node.NodeId);
}
}
private NodeState ResolveNodeId(NodeId nodeId)
{
if (nodeId.IsNull)
{
throw ServiceResultException.Create(
StatusCodes.BadNodeIdInvalid,
"NodeId is null or empty.");
}
return m_nodeIdResolver(nodeId) ??
throw ServiceResultException.Create(
StatusCodes.BadNodeIdUnknown,
"NodeId '{0}' did not resolve to a predefined node.",
nodeId);
}
private NodeState ResolveByTypeDefinition(NodeId typeDefinitionId, QualifiedName browseName)
{
if (typeDefinitionId.IsNull)
{
throw ServiceResultException.Create(
StatusCodes.BadNodeIdInvalid,
"TypeDefinitionId is null or empty.");
}
IReadOnlyList<NodeState> candidates = m_typeIdResolver(typeDefinitionId)
?? [];
if (candidates.Count == 0)
{
throw ServiceResultException.Create(
StatusCodes.BadNodeIdUnknown,
"No predefined node has TypeDefinitionId '{0}'.",
typeDefinitionId);
}
if (browseName.IsNull)
{
if (candidates.Count > 1)
{
throw ServiceResultException.Create(
StatusCodes.BadBrowseNameDuplicated,
"TypeDefinitionId '{0}' is ambiguous: {1} matching instances found. " +
"Pass a QualifiedName disambiguator to NodeFromTypeId.",
typeDefinitionId,
candidates.Count);
}
return candidates[0];
}
NodeState? match = null;
for (int i = 0; i < candidates.Count; i++)
{
if (candidates[i].BrowseName == browseName)
{
if (match != null)
{
throw ServiceResultException.Create(
StatusCodes.BadBrowseNameDuplicated,
"TypeDefinitionId '{0}' has multiple instances with browse name '{1}'.",
typeDefinitionId,
browseName);
}
match = candidates[i];
}
}
if (match == null)
{
throw ServiceResultException.Create(
StatusCodes.BadNodeIdUnknown,
"TypeDefinitionId '{0}' has no instance with browse name '{1}'.",
typeDefinitionId,
browseName);
}
return match;
}
private void ThrowIfSealed()
{
if (m_sealed)
{
throw ServiceResultException.Create(
StatusCodes.BadInvalidState,
"Cannot wire additional nodes after the builder has been sealed. " +
"All Node(...) calls must occur inside the Configure delegate.");
}
}
private static void ThrowIfDuplicate<T>(
Dictionary<NodeId, T> map,
NodeState node,
string what)
{
if (map.ContainsKey(node.NodeId))
{
throw ServiceResultException.Create(
StatusCodes.BadConfigurationError,
"Node '{0}' (id '{1}') already has a {2} handler registered.",
node.BrowseName,
node.NodeId,
what);
}
}
private readonly ushort m_defaultNamespaceIndex;
private readonly Func<QualifiedName, NodeState> m_rootResolver;
private readonly Func<NodeId, NodeState> m_nodeIdResolver;
private readonly Func<NodeId, IReadOnlyList<NodeState>> m_typeIdResolver;
private bool m_sealed;
private readonly Dictionary<NodeId, HistoryReadHandler> m_historyRead = [];
private readonly Dictionary<NodeId, HistoryUpdateHandler> m_historyUpdate = [];
private readonly Dictionary<NodeId, MonitoredItemCreatedHandler> m_monitoredItemCreated = [];
private readonly Dictionary<NodeId, NodeLifecycleHandler> m_nodeAdded = [];
private readonly Dictionary<NodeId, NodeLifecycleHandler> m_nodeRemoved = [];
}
}