/* ========================================================================
* 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;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
namespace Opc.Ua.Server.Fluent
{
///
/// Coordinates the lifecycle of Publish-registered event sources
/// for a single node manager. Activation is driven off
/// : each call to the manager's
/// SubscribeToEvents override triggers a reconcile pass that
/// starts iterators for newly-monitored sources and cancels iterators
/// for sources whose subscriber count dropped to zero. Disposing the
/// registry cancels every running iterator and waits for them to drain
/// (bounded by ).
///
///
///
/// The registry runs a single background reconcile loop that consumes
/// signals via a . Signals coalesce — multiple
/// rapid sub/unsub calls produce at most one extra reconcile pass.
/// Each registered source runs in its own background
/// so backpressure on one source does not block another.
///
///
/// Threading: takes a short private lock to
/// mutate the source dictionary; everything else runs lock-free off
/// the reconcile worker.
///
///
internal sealed class EventSourceRegistry : IDisposable
{
public EventSourceRegistry(
FluentNodeManagerBase owner,
ILogger logger)
{
m_owner = owner ?? throw new ArgumentNullException(nameof(owner));
m_logger = logger;
m_reconcileSignal = new SemaphoreSlim(0, 1);
m_managerCts = new CancellationTokenSource();
m_reconcileTask = Task.Run(() => RunReconcileLoopAsync(m_managerCts.Token));
}
///
/// Registers as the event source for
/// . Auto-promotes the notifier's
/// flag and, when
/// requested in , also registers the
/// notifier as a root notifier with the owning manager. Throws if a
/// source is already registered for the same notifier.
///
///
/// Designed to be called from the manager's Configure
/// delegate, which runs single-threaded before clients connect.
/// In that context the synchronous wait on
///
/// (when
/// is set) cannot deadlock because no other thread is contending
/// on the manager's monitored-item semaphore.
///
///
/// is null.
///
///
/// is null.
///
///
/// A source is already registered for .
///
///
/// is
/// set and adding as a root notifier
/// failed.
///
public void Register(
BaseObjectState notifier,
Func> factory,
EventPublishOptions? options)
{
if (notifier == null)
{
throw new ArgumentNullException(nameof(notifier));
}
if (factory == null)
{
throw new ArgumentNullException(nameof(factory));
}
options ??= new EventPublishOptions();
ValidateOptions(options);
lock (m_sourcesLock)
{
ThrowIfDisposed();
if (m_sources.ContainsKey(notifier.NodeId))
{
throw ServiceResultException.Create(
StatusCodes.BadConfigurationError,
"Node '{0}' (id '{1}') already has a Publish event source registered.",
notifier.BrowseName,
notifier.NodeId);
}
// Auto-promote EventNotifier so clients can subscribe to events on
// the node — Boiler-style models do not set this flag by default.
if ((notifier.EventNotifier & EventNotifiers.SubscribeToEvents) == 0)
{
notifier.EventNotifier |= EventNotifiers.SubscribeToEvents;
m_logger?.LogDebug(
"Publish: promoted EventNotifier of '{Browse}' (id '{NodeId}') to include SubscribeToEvents.",
notifier.BrowseName,
notifier.NodeId);
}
m_sources[notifier.NodeId] = new SourceEntry(notifier, factory, options);
}
// Root-notifier registration runs eagerly OUTSIDE m_sourcesLock so
// existing Server-level event monitored items get attached
// immediately. Lazy activation gated on AreEventsMonitored cannot
// bootstrap a root notifier (the gate is on the wrong node).
if (options.RegisterAsRootNotifier)
{
try
{
m_owner.AddRootNotifierFromFluentAsync(notifier, CancellationToken.None)
.GetAwaiter()
.GetResult();
m_logger?.LogDebug(
"Publish: registered '{Browse}' (id '{NodeId}') as a root notifier (RegisterAsRootNotifier=true).",
notifier.BrowseName,
notifier.NodeId);
}
catch (Exception ex)
{
lock (m_sourcesLock)
{
m_sources.Remove(notifier.NodeId);
}
throw ServiceResultException.Create(
StatusCodes.BadConfigurationError,
ex,
"Publish: failed to add '{0}' (id '{1}') as a root notifier.",
notifier.BrowseName,
notifier.NodeId);
}
}
SignalReconcile();
}
private static void ValidateOptions(EventPublishOptions options)
{
TimeSpan timeout = options.CancellationTimeout;
if (timeout != Timeout.InfiniteTimeSpan && timeout < TimeSpan.Zero)
{
throw new ArgumentOutOfRangeException(
nameof(options),
timeout,
"EventPublishOptions.CancellationTimeout must be non-negative or Timeout.InfiniteTimeSpan.");
}
}
///
/// Signals the registry to walk every source and reconcile its
/// activation state against .
/// Called by after the base
/// SubscribeToEvents implementation has updated the ref-count
/// recursively.
///
public void SignalReconcile()
{
if (Volatile.Read(ref m_disposed) != 0)
{
return;
}
// Coalesce: only one reconcile pass is pending at a time.
try
{
m_reconcileSignal.Release();
}
catch (SemaphoreFullException)
{
// Another signal is already pending — that's the whole point.
}
catch (ObjectDisposedException)
{
// Lost the race with Dispose — that's fine, the registry is
// shutting down and no further reconciliation is needed.
}
}
///
/// Cancels every running iterator, waits for them to drain (bounded
/// by each source's ),
/// and stops the reconcile loop. Idempotent.
///
public void Dispose()
{
if (Interlocked.Exchange(ref m_disposed, 1) != 0)
{
return;
}
// Stop the reconcile loop first so it cannot race with us.
try
{
m_managerCts.Cancel();
}
catch (ObjectDisposedException)
{
}
// Reconcile loop awaits the signal — releasing here unblocks
// its WaitAsync so it can observe the cancellation token.
try
{
m_reconcileSignal.Release();
}
catch (SemaphoreFullException)
{
}
catch (ObjectDisposedException)
{
}
// Wait for the reconcile task to actually exit. We must wait long
// enough to cover the worst case where the loop is mid-pass when
// the cancel fires: the in-flight ReconcileAll may DeactivateSource
// each registered source, and each DeactivateSource blocks on its
// worker task for up to entry.Options.CancellationTimeout. Use the
// largest per-source timeout, plus a safety margin, instead of a
// hardcoded 5s that could be exceeded by a single slow source.
TimeSpan waitFor = ComputeReconcileWaitTimeout();
try
{
m_reconcileTask.Wait(waitFor);
}
catch (AggregateException)
{
}
// Deactivate every source so their iterators get cancelled.
List snapshot;
lock (m_sourcesLock)
{
snapshot = [.. m_sources.Values];
m_sources.Clear();
}
foreach (SourceEntry entry in snapshot)
{
DeactivateSource(entry, force: true);
}
m_reconcileSignal.Dispose();
m_managerCts.Dispose();
}
private TimeSpan ComputeReconcileWaitTimeout()
{
TimeSpan maxPerSource = TimeSpan.Zero;
lock (m_sourcesLock)
{
foreach (SourceEntry entry in m_sources.Values)
{
TimeSpan t = entry.Options.CancellationTimeout;
if (t == Timeout.InfiniteTimeSpan)
{
return Timeout.InfiniteTimeSpan;
}
if (t > maxPerSource)
{
maxPerSource = t;
}
}
}
// The reconcile loop only needs the larger of its in-flight pass and
// a small bookkeeping margin; per-source deactivation runs again on
// the disposer thread below.
return maxPerSource + TimeSpan.FromSeconds(5);
}
private async Task RunReconcileLoopAsync(CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
try
{
await m_reconcileSignal.WaitAsync(ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return;
}
catch (ObjectDisposedException)
{
return;
}
if (ct.IsCancellationRequested)
{
return;
}
ReconcileAll();
}
}
private void ReconcileAll()
{
List snapshot;
lock (m_sourcesLock)
{
snapshot = [.. m_sources.Values];
}
foreach (SourceEntry entry in snapshot)
{
try
{
bool wantActive = entry.Options.AlwaysOn || entry.Notifier.AreEventsMonitored;
if (wantActive && entry.WorkerCts == null)
{
ActivateSource(entry);
}
else if (!wantActive && entry.WorkerCts != null)
{
DeactivateSource(entry, force: false);
}
}
catch (Exception ex)
{
m_logger?.LogError(
ex,
"Publish: reconcile pass failed for '{Browse}' (id '{NodeId}').",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
}
}
}
private void ActivateSource(SourceEntry entry)
{
var cts = CancellationTokenSource.CreateLinkedTokenSource(m_managerCts.Token);
entry.WorkerCts = cts;
Volatile.Write(ref entry.LeakedFaulted, 0);
entry.WorkerTask = Task.Run(() => RunSourceAsync(entry, cts.Token));
m_logger?.LogDebug(
"Publish: activated source for '{Browse}' (id '{NodeId}').",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
}
private void DeactivateSource(SourceEntry entry, bool force)
{
CancellationTokenSource? cts = entry.WorkerCts;
Task? worker = entry.WorkerTask;
entry.WorkerCts = null;
entry.WorkerTask = null;
if (cts == null)
{
return;
}
try
{
cts.Cancel();
}
catch (ObjectDisposedException)
{
}
try
{
bool completed = worker?.Wait(entry.Options.CancellationTimeout) ?? true;
if (!completed)
{
Volatile.Write(ref entry.LeakedFaulted, 1);
m_logger?.LogWarning(
"Publish: source for '{Browse}' (id '{NodeId}') did not honor cancellation within {Timeout}; further yielded events will be discarded.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId,
entry.Options.CancellationTimeout);
}
}
catch (AggregateException ex)
{
ex.Handle(e => e is OperationCanceledException);
}
finally
{
cts.Dispose();
}
if (force)
{
m_logger?.LogDebug(
"Publish: tore down source for '{Browse}' (id '{NodeId}') on dispose.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
}
else
{
m_logger?.LogDebug(
"Publish: deactivated source for '{Browse}' (id '{NodeId}').",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
}
}
private async Task RunSourceAsync(SourceEntry entry, CancellationToken ct)
{
ISystemContext systemContext = m_owner.SystemContext;
IAsyncEnumerable stream;
try
{
stream = entry.Factory(entry.Notifier, systemContext, ct);
if (stream == null)
{
m_logger?.LogError(
"Publish: factory for '{Browse}' (id '{NodeId}') returned a null stream.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
return;
}
}
catch (OperationCanceledException)
{
return;
}
catch (Exception ex)
{
m_logger?.LogError(
ex,
"Publish: factory invocation for '{Browse}' (id '{NodeId}') threw.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
try
{
entry.Options.OnError?.Invoke(ex);
}
catch
{
// Swallow secondary errors from the user hook.
}
return;
}
try
{
await foreach (BaseEventState e in stream.WithCancellation(ct).ConfigureAwait(false))
{
if (ct.IsCancellationRequested ||
Volatile.Read(ref entry.LeakedFaulted) != 0)
{
return;
}
DispatchEvent(entry, systemContext, e);
}
}
catch (OperationCanceledException)
{
// Normal shutdown.
}
catch (Exception ex)
{
m_logger?.LogError(
ex,
"Publish: iterator for '{Browse}' (id '{NodeId}') threw — stopping that source only.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
try
{
entry.Options.OnError?.Invoke(ex);
}
catch
{
}
}
}
private void DispatchEvent(SourceEntry entry, ISystemContext context, BaseEventState e)
{
if (e == null)
{
return;
}
try
{
if (!entry.Options.SkipDefaultPopulation)
{
PopulateDefaults(entry.Notifier, context, e);
}
entry.Notifier.ReportEvent(context, e);
}
catch (Exception ex)
{
m_logger?.LogError(
ex,
"Publish: ReportEvent for '{Browse}' (id '{NodeId}') threw — dropping this event and continuing iterator.",
entry.Notifier.BrowseName,
entry.Notifier.NodeId);
try
{
entry.Options.OnError?.Invoke(ex);
}
catch
{
}
}
}
///
/// Mirrors BaseEventState.Initialize for fields the user
/// did not populate themselves: EventId (random uuid),
/// EventType / TypeDefinitionId (default for the
/// CLR type), SourceNode, SourceName, Time,
/// ReceiveTime, Severity (Medium when 0),
/// Message.
///
private static void PopulateDefaults(BaseObjectState notifier, ISystemContext context, BaseEventState e)
{
if (e.EventId == null || e.EventId.Value.IsNull)
{
e.EventId = PropertyState.With(
e,
Uuid.NewUuid().ToByteString());
}
if (e.EventType == null || e.EventType.Value.IsNull)
{
NodeId defaultType = e.GetDefaultTypeDefinitionId(context);
e.EventType = PropertyState.With(e, defaultType);
if (e.TypeDefinitionId.IsNull)
{
e.TypeDefinitionId = defaultType;
}
}
else if (e.TypeDefinitionId.IsNull)
{
e.TypeDefinitionId = e.EventType.Value;
}
if ((e.SourceNode == null || e.SourceNode.Value.IsNull) &&
!notifier.NodeId.IsNull)
{
e.SourceNode = PropertyState.With(e, notifier.NodeId);
}
if ((e.SourceName == null || e.SourceName.Value == null) &&
!notifier.BrowseName.IsNull)
{
e.SourceName = PropertyState.With(
e,
notifier.BrowseName.Name ?? string.Empty);
}
if (e.Time == null || e.Time.Value.IsNull)
{
e.Time = PropertyState.With(e, DateTimeUtc.Now);
}
if (e.ReceiveTime == null || e.ReceiveTime.Value.IsNull)
{
e.ReceiveTime = PropertyState.With(
e,
DateTimeUtc.Now);
}
if (e.Severity == null || e.Severity.Value == 0)
{
e.Severity = PropertyState.With(
e,
(ushort)EventSeverity.Medium);
}
e.Message ??= PropertyState.With(
e,
new LocalizedText(string.Empty));
}
private void ThrowIfDisposed()
{
if (Volatile.Read(ref m_disposed) != 0)
{
throw new ObjectDisposedException(nameof(EventSourceRegistry));
}
}
private sealed class SourceEntry
{
public SourceEntry(
BaseObjectState notifier,
Func> factory,
EventPublishOptions options)
{
Notifier = notifier;
Factory = factory;
Options = options;
}
public BaseObjectState Notifier { get; }
public Func> Factory { get; }
public EventPublishOptions Options { get; }
public CancellationTokenSource? WorkerCts;
public Task? WorkerTask;
public int LeakedFaulted;
}
private readonly FluentNodeManagerBase m_owner;
private readonly ILogger m_logger;
private readonly SemaphoreSlim m_reconcileSignal;
private readonly CancellationTokenSource m_managerCts;
private readonly Task m_reconcileTask;
private readonly Lock m_sourcesLock = new();
private readonly Dictionary m_sources = [];
private int m_disposed;
}
}