mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-22 09:37:12 +08:00
Marc Schier
d8a95b4e9e
# Description Full-repo `dotnet format` sweep across all 73 projects in `UA.slnx` — no behavioral changes.
1839 lines
78 KiB
C#
1839 lines
78 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.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Channels;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
|
|
namespace Opc.Ua.Client.Subscriptions
|
|
{
|
|
/// <summary>
|
|
/// Manages all subscriptions of a session on the server side. This
|
|
/// includes managing the publish requests and acknowledgements.
|
|
/// The publish requests are managed by a controller that adjusts
|
|
/// the number of workers based on the number of subscriptions.
|
|
/// The workers are responsible for sending the publish requests
|
|
/// to the server and dispatching the notifications to the
|
|
/// subscriptions. The subscriptions queue acknowledgements for
|
|
/// the completed notifications as soon as they are dispatched.
|
|
/// The queued acknowledgements are then sent by the workers the
|
|
/// next publish cycle.
|
|
/// </summary>
|
|
internal sealed class SubscriptionManager : ISubscriptionManager,
|
|
IMessageAckQueue, IAsyncDisposable
|
|
{
|
|
/// <summary>
|
|
/// Create subscription manager
|
|
/// </summary>
|
|
/// <param name="session"></param>
|
|
/// <param name="loggerFactory"></param>
|
|
/// <param name="returnDiagnostics"></param>
|
|
/// <param name="timeProvider">Optional <see cref="TimeProvider"/>
|
|
/// used for elapsed-time and timer calculations. Defaults to
|
|
/// <see cref="TimeProvider.System"/> when <c>null</c>.</param>
|
|
public SubscriptionManager(ISubscriptionManagerContext session,
|
|
ILoggerFactory loggerFactory, DiagnosticsMasks returnDiagnostics,
|
|
TimeProvider? timeProvider = null)
|
|
{
|
|
m_timeProvider = timeProvider ?? TimeProvider.System;
|
|
m_session = session;
|
|
m_loggerFactory = loggerFactory;
|
|
m_logger = loggerFactory.CreateLogger<SubscriptionManager>();
|
|
ReturnDiagnostics = returnDiagnostics;
|
|
m_publishController = PublishControllerAsync(m_cts.Token);
|
|
m_acks = Channel.CreateUnboundedPrioritized(
|
|
new UnboundedPrioritizedChannelOptions<SubscriptionAcknowledgement>
|
|
{
|
|
Comparer = Comparer<SubscriptionAcknowledgement>
|
|
.Create((x, y) => x.SequenceNumber.CompareTo(y.SequenceNumber))
|
|
});
|
|
}
|
|
|
|
/// <summary>
|
|
/// If the subscriptions are transferred when a session is
|
|
/// recreated.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Default <c>false</c>, set to <c>true</c> if subscriptions
|
|
/// should be transferred after recreating the session.
|
|
/// Service must be supported by server.
|
|
/// </remarks>
|
|
public bool TransferSubscriptionsOnRecreate { get; set; }
|
|
|
|
/// <inheritdoc/>
|
|
public bool PoolNotifications { get; set; }
|
|
|
|
/// <inheritdoc/>
|
|
public DiagnosticsMasks ReturnDiagnostics { get; set; }
|
|
|
|
/// <inheritdoc/>
|
|
/// <remarks>
|
|
/// Setting this property signals the publish controller so the worker
|
|
/// pool resizes promptly — without the signal, the controller stays
|
|
/// parked on <see cref="m_publishControl"/> and the change does not
|
|
/// take effect until some other event (subscription add/remove,
|
|
/// pause/resume, worker exit) wakes it.
|
|
/// </remarks>
|
|
public int MinPublishWorkerCount
|
|
{
|
|
get;
|
|
set
|
|
{
|
|
if (field == value)
|
|
{
|
|
return;
|
|
}
|
|
field = value;
|
|
m_publishControl.Set();
|
|
}
|
|
} = 2;
|
|
|
|
/// <inheritdoc/>
|
|
/// <remarks>
|
|
/// Setting this property signals the publish controller so the worker
|
|
/// pool resizes promptly. See <see cref="MinPublishWorkerCount"/>.
|
|
/// </remarks>
|
|
public int MaxPublishWorkerCount
|
|
{
|
|
get;
|
|
set
|
|
{
|
|
if (field == value)
|
|
{
|
|
return;
|
|
}
|
|
field = value;
|
|
m_publishControl.Set();
|
|
}
|
|
} = 15;
|
|
|
|
/// <inheritdoc/>
|
|
public IEnumerable<ISubscription> Items
|
|
{
|
|
get
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
// Expose logical wrappers — callers see one
|
|
// ISubscription per logical subscription regardless
|
|
// of how many partitions back it.
|
|
return [.. m_logicals];
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public int Count
|
|
{
|
|
get
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
return m_logicals.Count;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public int GoodPublishRequestCount => m_goodPublishRequestCount;
|
|
|
|
/// <inheritdoc/>
|
|
public int BadPublishRequestCount => m_badPublishRequestCount;
|
|
|
|
/// <inheritdoc/>
|
|
public long MissingMessageCount
|
|
{
|
|
get
|
|
{
|
|
long total = 0;
|
|
lock (m_subscriptionLock)
|
|
{
|
|
foreach (IManagedSubscription s in m_subscriptions)
|
|
{
|
|
total += s.MissingMessageCount;
|
|
}
|
|
}
|
|
return total;
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public long RepublishMessageCount
|
|
{
|
|
get
|
|
{
|
|
long total = 0;
|
|
lock (m_subscriptionLock)
|
|
{
|
|
foreach (IManagedSubscription s in m_subscriptions)
|
|
{
|
|
total += s.RepublishMessageCount;
|
|
}
|
|
}
|
|
return total;
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public int PublishWorkerCount { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
internal int PublishControlCycles { get; set; }
|
|
|
|
/// <summary>
|
|
/// Created items
|
|
/// </summary>
|
|
internal List<IManagedSubscription> Created
|
|
{
|
|
get
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
return [.. m_subscriptions.Where(s => s.Created)];
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Number of created items
|
|
/// </summary>
|
|
public int CreatedCount
|
|
{
|
|
get
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
return m_subscriptions.Count(s => s.Created);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
if (Interlocked.Exchange(ref m_disposed, 1) != 0)
|
|
{
|
|
return;
|
|
}
|
|
try
|
|
{
|
|
await m_cts.CancelAsync().ConfigureAwait(false);
|
|
m_publishControl.Set();
|
|
await m_publishController.ConfigureAwait(false);
|
|
|
|
List<LogicalSubscription>? logicals;
|
|
List<IManagedSubscription>? orphans;
|
|
lock (m_subscriptionLock)
|
|
{
|
|
logicals = [.. m_logicals];
|
|
m_logicals.Clear();
|
|
// Drop the dispatch registry entries owned by the
|
|
// wrappers we are about to dispose so any partition
|
|
// not currently bound to a wrapper (transient
|
|
// pre-registration window) can still be cleaned up
|
|
// below.
|
|
var ownedByLogicals = new HashSet<IManagedSubscription>();
|
|
foreach (LogicalSubscription wrapper in logicals)
|
|
{
|
|
foreach (IManagedSubscription partition in wrapper.Partitions)
|
|
{
|
|
ownedByLogicals.Add(partition);
|
|
}
|
|
}
|
|
orphans = [];
|
|
foreach (IManagedSubscription partition in m_subscriptions)
|
|
{
|
|
if (!ownedByLogicals.Contains(partition))
|
|
{
|
|
orphans.Add(partition);
|
|
}
|
|
}
|
|
m_subscriptions.Clear();
|
|
}
|
|
foreach (LogicalSubscription wrapper in logicals)
|
|
{
|
|
await wrapper.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
foreach (IManagedSubscription partition in orphans)
|
|
{
|
|
await partition.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
m_subscriptionHistory.Clear();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
m_logger.LogError(ex, "Exception during dispose");
|
|
}
|
|
finally
|
|
{
|
|
m_cts.Dispose();
|
|
(m_acks as IDisposable)?.Dispose();
|
|
GC.SuppressFinalize(this);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Trigger publish controller
|
|
/// </summary>
|
|
public void Update()
|
|
{
|
|
m_publishControl.Set();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public ValueTask QueueAsync(
|
|
SubscriptionAcknowledgement ack, CancellationToken ct)
|
|
{
|
|
return m_acks.Writer.WriteAsync(ack, ct);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public int DropPendingForSubscription(uint subscriptionId)
|
|
{
|
|
// Drain the prioritised channel into a local buffer,
|
|
// keeping only the acks that do not target the dead
|
|
// subscription id, then re-publish the kept items.
|
|
// Concurrent workers may interleave reads or writes
|
|
// during this loop; that is intentional — a worker
|
|
// racing in is acceptable because (a) any ack it reads
|
|
// would have been sent to the server anyway, (b) any
|
|
// ack it writes is queued back into the channel. The
|
|
// contract is "no targeted acks remain when this
|
|
// method returns" — callers must invoke it BEFORE
|
|
// recreate assigns a fresh subscription id so a new
|
|
// generation cannot enter the queue.
|
|
var keep = new List<SubscriptionAcknowledgement>();
|
|
int dropped = 0;
|
|
while (m_acks.Reader.TryRead(out SubscriptionAcknowledgement? ack))
|
|
{
|
|
if (ack.SubscriptionId == subscriptionId)
|
|
{
|
|
dropped++;
|
|
}
|
|
else
|
|
{
|
|
keep.Add(ack);
|
|
}
|
|
}
|
|
foreach (SubscriptionAcknowledgement ack in keep)
|
|
{
|
|
m_acks.Writer.TryWrite(ack);
|
|
}
|
|
return dropped;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public ValueTask CompleteAsync(uint subscriptionId, CancellationToken ct)
|
|
{
|
|
IManagedSubscription? subscription;
|
|
LogicalSubscription? logical = null;
|
|
lock (m_subscriptionLock)
|
|
{
|
|
// Drop the partition from the dispatch registry.
|
|
subscription = m_subscriptions
|
|
.FirstOrDefault(s => s.Id == subscriptionId);
|
|
if (subscription == null ||
|
|
!m_subscriptions.Remove(subscription))
|
|
{
|
|
return default;
|
|
}
|
|
m_subscriptionHistory.Enqueue(subscriptionId);
|
|
|
|
// If the removed partition was the primary of any
|
|
// logical wrapper, the wrapper has no usable
|
|
// partitions left — drop it from the public registry
|
|
// so ISubscriptionManager.Items / Count reflect the
|
|
// deletion. Secondary partitions (added on demand by
|
|
// the composite collection) do not remove the
|
|
// wrapper; the wrapper keeps living as long as the
|
|
// primary is registered.
|
|
foreach (LogicalSubscription wrapper in m_logicals)
|
|
{
|
|
IReadOnlyList<IManagedSubscription> parts = wrapper.Partitions;
|
|
if (parts.Count > 0 && ReferenceEquals(parts[0], subscription))
|
|
{
|
|
logical = wrapper;
|
|
break;
|
|
}
|
|
}
|
|
if (logical != null)
|
|
{
|
|
m_logicals.Remove(logical);
|
|
}
|
|
}
|
|
while (m_subscriptionHistory.Count > kMaxSubscriptionHistory)
|
|
{
|
|
m_subscriptionHistory.TryDequeue(out _);
|
|
}
|
|
m_logger.LogInformation("{Subscription} REMOVED.", subscription);
|
|
m_publishControl.Set();
|
|
return default;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public ISubscription Add(ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options)
|
|
{
|
|
if (handler == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(handler));
|
|
}
|
|
if (options == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(options));
|
|
}
|
|
|
|
SubscriptionOptions snapshot = options.CurrentValue ?? new SubscriptionOptions();
|
|
|
|
// In multi-partition mode every partition fires its
|
|
// notification callbacks on the same logical wrapper —
|
|
// wrap the user handler in a forwarding shim that (a)
|
|
// serialises calls across partitions and (b) re-targets
|
|
// the subscription parameter to the wrapper instead of
|
|
// the partition.
|
|
ISubscriptionNotificationHandler effectiveHandler = handler;
|
|
PartitionForwardingHandler? forwardingHandler = null;
|
|
if (!snapshot.DisableUnboundedItemMode)
|
|
{
|
|
#pragma warning disable CA2000 // ownership transfers to wrapper.AttachForwardingHandler
|
|
forwardingHandler = new PartitionForwardingHandler(handler);
|
|
#pragma warning restore CA2000
|
|
effectiveHandler = forwardingHandler;
|
|
}
|
|
|
|
IManagedSubscription primary = m_session.CreateSubscription(
|
|
effectiveHandler, options, this);
|
|
|
|
// Build the placement policy + on-demand partition factory
|
|
// unless the caller opted out via DisableUnboundedItemMode.
|
|
// The factory is invoked by the composite collection inside
|
|
// its lock when no existing partition has capacity; the
|
|
// factory mints a sibling partition with the same handler
|
|
// and options, registers it in the dispatch registry, and
|
|
// wakes the publish controller so a worker can serve it.
|
|
PartitionPlacementPolicy? policy = null;
|
|
Func<IManagedSubscription>? partitionFactory = null;
|
|
LogicalSubscription wrapper;
|
|
if (snapshot.DisableUnboundedItemMode)
|
|
{
|
|
wrapper = new LogicalSubscription(primary, null, null);
|
|
}
|
|
else
|
|
{
|
|
policy = new PartitionPlacementPolicy(
|
|
snapshot.MaxMonitoredItemsPerPartition ?? 0,
|
|
snapshot.MaxPartitionCount);
|
|
// Capture the wrapper into the factory closure so newly
|
|
// minted secondary partitions can route their reactive-
|
|
// fallback callbacks back to the same logical.
|
|
LogicalSubscription? created = null;
|
|
partitionFactory = () =>
|
|
{
|
|
IManagedSubscription part = MintPartition(effectiveHandler, options);
|
|
AttachReactiveFallback(part, created);
|
|
created?.TryApplyDurableToNewPartition(part);
|
|
return part;
|
|
};
|
|
wrapper = new LogicalSubscription(primary, policy, partitionFactory,
|
|
m_timeProvider,
|
|
NormalizeIdleTimeout(snapshot.SecondaryPartitionIdleTimeout),
|
|
RemoveSecondaryPartitionAsync);
|
|
created = wrapper;
|
|
AttachReactiveFallback(primary, wrapper);
|
|
forwardingHandler!.BindLogical(wrapper);
|
|
wrapper.AttachForwardingHandler(forwardingHandler);
|
|
}
|
|
|
|
lock (m_subscriptionLock)
|
|
{
|
|
if (!m_subscriptions.Add(primary))
|
|
{
|
|
throw ServiceResultException.Create(StatusCodes.BadAlreadyExists,
|
|
"Failed to register primary partition.");
|
|
}
|
|
if (!m_logicals.Add(wrapper))
|
|
{
|
|
// Roll the primary back out so we do not leak
|
|
// a partition with no logical wrapper.
|
|
m_subscriptions.Remove(primary);
|
|
throw ServiceResultException.Create(StatusCodes.BadAlreadyExists,
|
|
"Failed to register logical subscription.");
|
|
}
|
|
m_logger.LogInformation("{Subscription} ADDED.", primary);
|
|
}
|
|
m_publishControl.Set();
|
|
return wrapper;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Wire a partition's
|
|
/// <see cref="Subscription.OnPartitionCapReached"/> hook
|
|
/// back to the owning logical wrapper so a server
|
|
/// <c>Bad_TooManyMonitoredItems</c> response marks the
|
|
/// partition no-grow in the wrapper's placement policy. No-op
|
|
/// when the partition is not a <see cref="Subscription"/>
|
|
/// (e.g. a test fake) or when the wrapper has not yet been
|
|
/// captured by the factory closure.
|
|
/// </summary>
|
|
private static void AttachReactiveFallback(
|
|
IManagedSubscription partition,
|
|
LogicalSubscription? wrapper)
|
|
{
|
|
if (wrapper == null)
|
|
{
|
|
return;
|
|
}
|
|
if (partition is Subscription concrete)
|
|
{
|
|
concrete.OnPartitionCapReached = _ => wrapper.OnPartitionCapReached(partition);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Normalize a caller-supplied
|
|
/// <see cref="SubscriptionOptions.SecondaryPartitionIdleTimeout"/>
|
|
/// for the wrapper. Per the option docs,
|
|
/// <see cref="TimeSpan.Zero"/> means immediate idle-delete
|
|
/// and <see cref="Timeout.InfiniteTimeSpan"/>
|
|
/// (the only valid negative value) disables idle-delete. Any
|
|
/// other negative value is treated as "disabled" defensively
|
|
/// because <see cref="ITimer"/> rejects them.
|
|
/// </summary>
|
|
private static TimeSpan NormalizeIdleTimeout(TimeSpan idleTimeout)
|
|
{
|
|
return idleTimeout < TimeSpan.Zero
|
|
? Timeout.InfiniteTimeSpan
|
|
: idleTimeout;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Secondary-partition disposer handed to the wrapper's
|
|
/// idle-delete timer. Drops the partition from this
|
|
/// manager's dispatch registry, drains any pending
|
|
/// acknowledgements addressed at it, and disposes the
|
|
/// partition (which sends <c>DeleteSubscriptions</c> to the
|
|
/// server). Logical wrapper book-keeping is performed by the
|
|
/// composite collection under its own lock before this
|
|
/// callback runs, so we only need to clean up manager-side
|
|
/// state here.
|
|
/// </summary>
|
|
private async ValueTask RemoveSecondaryPartitionAsync(IManagedSubscription partition)
|
|
{
|
|
uint partitionId = partition.Id;
|
|
lock (m_subscriptionLock)
|
|
{
|
|
m_subscriptions.Remove(partition);
|
|
}
|
|
if (partitionId != 0)
|
|
{
|
|
DropPendingForSubscription(partitionId);
|
|
}
|
|
try
|
|
{
|
|
await partition.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
m_logger.LogInformation(ex,
|
|
"Idle-delete of secondary partition {Partition} threw on dispose.",
|
|
partition);
|
|
}
|
|
m_publishControl.Set();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Synchronous partition factory handed to the composite
|
|
/// collection. Constructs a sibling partition with the same
|
|
/// handler + options as the logical subscription's primary,
|
|
/// registers it in the dispatch registry, and wakes the
|
|
/// publish controller so a worker is scheduled for it.
|
|
/// </summary>
|
|
private IManagedSubscription MintPartition(
|
|
ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options)
|
|
{
|
|
return MintPartitionCore(handler, options, loadState: null);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Variant of <see cref="MintPartition"/> that accepts a
|
|
/// pre-existing <see cref="SubscriptionLoadState"/> so the
|
|
/// minted partition is construction-time bound to a
|
|
/// previously-saved server-side identifier and items list.
|
|
/// Used by <see cref="RestoreGroupAsync"/> to attach
|
|
/// secondary partitions during multi-partition snapshot
|
|
/// restore.
|
|
/// </summary>
|
|
private IManagedSubscription MintPreloadedPartition(
|
|
ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options,
|
|
SubscriptionLoadState loadState)
|
|
{
|
|
return MintPartitionCore(handler, options, loadState);
|
|
}
|
|
|
|
private IManagedSubscription MintPartitionCore(
|
|
ISubscriptionNotificationHandler handler,
|
|
IOptionsMonitor<SubscriptionOptions> options,
|
|
SubscriptionLoadState? loadState)
|
|
{
|
|
IManagedSubscription partition = m_session.CreateSubscription(
|
|
handler, options, this, loadState);
|
|
lock (m_subscriptionLock)
|
|
{
|
|
if (!m_subscriptions.Add(partition))
|
|
{
|
|
throw ServiceResultException.Create(StatusCodes.BadAlreadyExists,
|
|
"Failed to register partition.");
|
|
}
|
|
if (loadState == null)
|
|
{
|
|
m_logger.LogInformation(
|
|
"{Subscription} ADDED (secondary partition).", partition);
|
|
}
|
|
else
|
|
{
|
|
m_logger.LogInformation(
|
|
"{Subscription} ADDED (preloaded secondary partition).", partition);
|
|
}
|
|
}
|
|
m_publishControl.Set();
|
|
return partition;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restore a single subscription from a snapshot previously
|
|
/// produced by <see cref="Subscription.Snapshot"/>. The
|
|
/// returned subscription is registered with the manager via the
|
|
/// same path as <see cref="Add"/>.
|
|
/// </summary>
|
|
/// <param name="handler">Notification handler for the restored
|
|
/// subscription.</param>
|
|
/// <param name="state">Snapshot captured earlier on the source
|
|
/// session.</param>
|
|
/// <param name="transferSubscriptions">
|
|
/// When <c>true</c> the saved server-side subscription id and
|
|
/// per-item server ids are preserved and an OPC UA
|
|
/// TransferSubscriptions service call is issued so the new
|
|
/// session takes over the existing server-side state. If
|
|
/// transfer is unavailable (e.g. the server returns
|
|
/// <c>BadSubscriptionIdInvalid</c>), the restore falls back to
|
|
/// recreate.
|
|
/// When <c>false</c> the V2 state machine mints fresh
|
|
/// server-side ids — equivalent to a fresh
|
|
/// <see cref="Add"/> with the saved configuration.
|
|
/// </param>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
/// <remarks>
|
|
/// Not on <see cref="ISubscriptionManager"/>: callers that want
|
|
/// stream-based restore should use <see cref="LoadAsync"/>;
|
|
/// fluent helpers and the serializer cast to the concrete
|
|
/// <see cref="SubscriptionManager"/> to reach this method.
|
|
/// </remarks>
|
|
/// <exception cref="ArgumentNullException"><paramref name="handler"/> is <c>null</c>.</exception>
|
|
internal ValueTask<ISubscription> RestoreAsync(
|
|
ISubscriptionNotificationHandler handler,
|
|
SubscriptionStateSnapshot state,
|
|
bool transferSubscriptions = false,
|
|
CancellationToken ct = default)
|
|
{
|
|
if (handler == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(handler));
|
|
}
|
|
if (state == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(state));
|
|
}
|
|
if (transferSubscriptions && state.ServerId != 0)
|
|
{
|
|
return RestoreTransferAsync(handler, state, ct);
|
|
}
|
|
return RestoreRecreateAsync(handler, state, ct);
|
|
}
|
|
|
|
private ValueTask<ISubscription> RestoreRecreateAsync(
|
|
ISubscriptionNotificationHandler handler,
|
|
SubscriptionStateSnapshot state,
|
|
CancellationToken ct)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
ISubscription subscription = Add(
|
|
handler,
|
|
new OptionsMonitor<SubscriptionOptions>(state.ToOptions()));
|
|
foreach (MonitoredItemStateSnapshot item in state.MonitoredItems)
|
|
{
|
|
subscription.MonitoredItems.TryAdd(
|
|
item.Name,
|
|
new OptionsMonitor<MonitoredItems.MonitoredItemOptions>(item.ToOptions()),
|
|
out _);
|
|
}
|
|
return new ValueTask<ISubscription>(subscription);
|
|
}
|
|
|
|
private async ValueTask<ISubscription> RestoreTransferAsync(
|
|
ISubscriptionNotificationHandler handler,
|
|
SubscriptionStateSnapshot state,
|
|
CancellationToken ct)
|
|
{
|
|
// Build the internal load-state record (decoupled from the
|
|
// public DTO so the engine has no DTO coupling). The reverse
|
|
// "items I trigger" set is reconstructed on demand from the
|
|
// triggering relationships per item.
|
|
var itemLoadStates = new List<MonitoredItemLoadState>(
|
|
state.MonitoredItems.Count);
|
|
foreach (MonitoredItemStateSnapshot item in state.MonitoredItems)
|
|
{
|
|
IReadOnlyList<string> triggeredBy = item.TriggeredByNames.IsNull
|
|
? []
|
|
: item.TriggeredByNames.ToArray() ?? [];
|
|
itemLoadStates.Add(new MonitoredItemLoadState(
|
|
item.Name,
|
|
new OptionsMonitor<MonitoredItems.MonitoredItemOptions>(item.ToOptions()),
|
|
item.ClientHandle,
|
|
item.ServerId,
|
|
triggeredBy));
|
|
}
|
|
var loadState = new SubscriptionLoadState(
|
|
state.ServerId, itemLoadStates);
|
|
|
|
SubscriptionOptions options = state.ToOptions();
|
|
var optionsMonitor = new OptionsMonitor<SubscriptionOptions>(options);
|
|
|
|
// Match the Add() path: wrap the user handler in a
|
|
// forwarding shim when running in multi-partition mode so
|
|
// post-restore growth keeps the serialised-dispatch +
|
|
// logical-subscription-parameter contract.
|
|
ISubscriptionNotificationHandler effectiveHandler = handler;
|
|
PartitionForwardingHandler? forwardingHandler = null;
|
|
if (!options.DisableUnboundedItemMode)
|
|
{
|
|
#pragma warning disable CA2000 // ownership transfers to wrapper.AttachForwardingHandler
|
|
forwardingHandler = new PartitionForwardingHandler(handler);
|
|
#pragma warning restore CA2000
|
|
effectiveHandler = forwardingHandler;
|
|
}
|
|
|
|
IManagedSubscription subscription = m_session.CreateSubscription(
|
|
effectiveHandler,
|
|
optionsMonitor,
|
|
this,
|
|
loadState);
|
|
|
|
// Wrap in a logical subscription so callers receive the
|
|
// same surface as a freshly added subscription. Pick the
|
|
// same placement policy + factory the standard Add path
|
|
// would use so post-restore growth obeys the configured
|
|
// limits.
|
|
PartitionPlacementPolicy? policy = null;
|
|
Func<IManagedSubscription>? partitionFactory = null;
|
|
LogicalSubscription wrapper;
|
|
if (options.DisableUnboundedItemMode)
|
|
{
|
|
wrapper = new LogicalSubscription(subscription, null, null);
|
|
}
|
|
else
|
|
{
|
|
policy = new PartitionPlacementPolicy(
|
|
options.MaxMonitoredItemsPerPartition ?? 0,
|
|
options.MaxPartitionCount);
|
|
LogicalSubscription? created = null;
|
|
partitionFactory = () =>
|
|
{
|
|
IManagedSubscription part = MintPartition(effectiveHandler, optionsMonitor);
|
|
AttachReactiveFallback(part, created);
|
|
created?.TryApplyDurableToNewPartition(part);
|
|
return part;
|
|
};
|
|
wrapper = new LogicalSubscription(subscription, policy, partitionFactory,
|
|
m_timeProvider,
|
|
NormalizeIdleTimeout(options.SecondaryPartitionIdleTimeout),
|
|
RemoveSecondaryPartitionAsync);
|
|
created = wrapper;
|
|
AttachReactiveFallback(subscription, wrapper);
|
|
forwardingHandler!.BindLogical(wrapper);
|
|
wrapper.AttachForwardingHandler(forwardingHandler);
|
|
}
|
|
|
|
lock (m_subscriptionLock)
|
|
{
|
|
if (!m_subscriptions.Add(subscription))
|
|
{
|
|
throw ServiceResultException.Create(StatusCodes.BadAlreadyExists,
|
|
"Failed to add restored subscription.");
|
|
}
|
|
if (!m_logicals.Add(wrapper))
|
|
{
|
|
m_subscriptions.Remove(subscription);
|
|
throw ServiceResultException.Create(StatusCodes.BadAlreadyExists,
|
|
"Failed to register restored logical subscription.");
|
|
}
|
|
m_logger.LogInformation(
|
|
"{Subscription} ADDED (transfer-pending, ServerId={ServerId}).",
|
|
subscription,
|
|
state.ServerId);
|
|
}
|
|
|
|
// Issue TransferSubscriptions for the saved server id.
|
|
// sendInitialValues honors SubscriptionOptions.SendInitialValuesOnTransfer
|
|
// (default false) — the snapshot captured the last server-
|
|
// emitted values, so requesting initial values is only
|
|
// useful when the caller wants the server to re-emit them
|
|
// to a fresh notification handler.
|
|
uint[] ids = [state.ServerId];
|
|
TransferSubscriptionsResponse response = await m_session
|
|
.TransferSubscriptionsAsync(
|
|
null,
|
|
ids.ToArrayOf(),
|
|
sendInitialValues: options.SendInitialValuesOnTransfer,
|
|
ct)
|
|
.ConfigureAwait(false);
|
|
|
|
bool transferred = false;
|
|
ResponseHeader responseHeader = response.ResponseHeader;
|
|
if (StatusCode.IsGood(responseHeader.ServiceResult))
|
|
{
|
|
ArrayOf<TransferResult> results = response.Results;
|
|
ClientBase.ValidateResponse(results, ids.ToArrayOf());
|
|
if (results.Count > 0 && StatusCode.IsGood(results[0].StatusCode))
|
|
{
|
|
transferred = await subscription.TryCompleteTransferAsync(
|
|
results[0].AvailableSequenceNumbers.IsNull
|
|
? []
|
|
: [.. results[0].AvailableSequenceNumbers],
|
|
ct).ConfigureAwait(false);
|
|
}
|
|
else if (results.Count > 0)
|
|
{
|
|
m_logger.LogWarning(
|
|
"{Subscription}: TransferSubscriptions per-item " +
|
|
"result Bad ({Status}); falling back to recreate.",
|
|
subscription,
|
|
results[0].StatusCode);
|
|
}
|
|
}
|
|
else if (responseHeader.ServiceResult == StatusCodes.BadServiceUnsupported)
|
|
{
|
|
m_logger.LogWarning(
|
|
"{Subscription}: server does not support " +
|
|
"TransferSubscriptions; falling back to recreate.",
|
|
subscription);
|
|
}
|
|
else
|
|
{
|
|
m_logger.LogWarning(
|
|
"{Subscription}: TransferSubscriptions service-level " +
|
|
"result Bad ({Status}); falling back to recreate.",
|
|
subscription,
|
|
responseHeader.ServiceResult);
|
|
}
|
|
|
|
if (!transferred && subscription is Subscription loaded)
|
|
{
|
|
await loaded.ResetToRecreateAsync(ct).ConfigureAwait(false);
|
|
// Await re-creation under a bounded timeout so LoadAsync
|
|
// only returns after the server has assigned a fresh
|
|
// SubscriptionId. Items still need ApplyChangesAsync to
|
|
// round-trip but that runs in the same state-manager
|
|
// iteration as CreateAsync.
|
|
using CancellationTokenSource timeoutCts = m_timeProvider
|
|
.CreateCancellationTokenSource(TimeSpan.FromSeconds(15));
|
|
using var linkedCts = CancellationTokenSource
|
|
.CreateLinkedTokenSource(ct, timeoutCts.Token);
|
|
try
|
|
{
|
|
await loaded.WaitForCreatedAsync(linkedCts.Token)
|
|
.ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
|
{
|
|
m_logger.LogWarning(
|
|
"{Subscription}: re-creation did not complete " +
|
|
"within 15s after failed transfer; returning anyway.",
|
|
subscription);
|
|
}
|
|
}
|
|
|
|
m_publishControl.Set();
|
|
return wrapper;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Restore a logical subscription whose partitions span more
|
|
/// than one snapshot record (multi-partition snapshot
|
|
/// produced by <see cref="LogicalSubscription.SnapshotAllPartitions"/>).
|
|
/// The primary snapshot (<c>PartitionIndex == 0</c>) is
|
|
/// restored via the existing single-snapshot path; each
|
|
/// secondary snapshot is then attached to the same wrapper
|
|
/// as an additional partition with its server-side
|
|
/// identifier preserved (transfer) or recreated (recreate).
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// The caller is expected to pre-sort
|
|
/// <paramref name="snapshots"/> by <c>PartitionIndex</c> and
|
|
/// to have validated the grouping (contiguous indexes,
|
|
/// exactly one primary, shared <c>LogicalGroupId</c>) — see
|
|
/// <see cref="SubscriptionManagerSerializer.LoadAsync"/>
|
|
/// which performs the validation before dispatching here.
|
|
/// </remarks>
|
|
internal async ValueTask<ISubscription> RestoreGroupAsync(
|
|
ISubscriptionNotificationHandler handler,
|
|
IReadOnlyList<SubscriptionStateSnapshot> snapshots,
|
|
bool transferSubscriptions,
|
|
CancellationToken ct)
|
|
{
|
|
if (handler == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(handler));
|
|
}
|
|
if (snapshots == null || snapshots.Count == 0)
|
|
{
|
|
throw new ArgumentException(
|
|
"Group must contain at least the primary snapshot.",
|
|
nameof(snapshots));
|
|
}
|
|
|
|
// Primary first — single-partition restore returns the
|
|
// wrapper that owns all subsequent secondaries.
|
|
ISubscription primaryWrapper = await RestoreAsync(
|
|
handler, snapshots[0], transferSubscriptions, ct)
|
|
.ConfigureAwait(false);
|
|
if (snapshots.Count == 1)
|
|
{
|
|
return primaryWrapper;
|
|
}
|
|
|
|
if (primaryWrapper is not LogicalSubscription wrapper)
|
|
{
|
|
throw new InvalidOperationException(
|
|
"Restore returned a subscription that does not " +
|
|
"support multi-partition grouping.");
|
|
}
|
|
|
|
// Reconstruct the options the secondaries share with
|
|
// the primary. The primary's wrapper was built from
|
|
// snapshots[0] so any field accessible via ToOptions()
|
|
// is valid for every partition in the group.
|
|
SubscriptionOptions options = snapshots[0].ToOptions();
|
|
var optionsMonitor = new OptionsMonitor<SubscriptionOptions>(options);
|
|
|
|
for (int i = 1; i < snapshots.Count; i++)
|
|
{
|
|
ct.ThrowIfCancellationRequested();
|
|
await AttachSecondaryPartitionAsync(
|
|
wrapper, handler, optionsMonitor,
|
|
snapshots[i], transferSubscriptions, ct)
|
|
.ConfigureAwait(false);
|
|
}
|
|
return wrapper;
|
|
}
|
|
|
|
private async ValueTask AttachSecondaryPartitionAsync(
|
|
LogicalSubscription wrapper,
|
|
ISubscriptionNotificationHandler handler,
|
|
OptionsMonitor<SubscriptionOptions> optionsMonitor,
|
|
SubscriptionStateSnapshot snap,
|
|
bool transferSubscriptions,
|
|
CancellationToken ct)
|
|
{
|
|
// Build the per-partition load state from the snapshot.
|
|
var itemLoadStates = new List<MonitoredItemLoadState>(
|
|
snap.MonitoredItems.Count);
|
|
foreach (MonitoredItemStateSnapshot item in snap.MonitoredItems)
|
|
{
|
|
itemLoadStates.Add(new MonitoredItemLoadState(
|
|
item.Name,
|
|
new OptionsMonitor<MonitoredItems.MonitoredItemOptions>(item.ToOptions()),
|
|
item.ClientHandle,
|
|
item.ServerId,
|
|
item.TriggeredByNames.IsNull
|
|
? []
|
|
: item.TriggeredByNames.ToArray() ?? []));
|
|
}
|
|
var loadState = new SubscriptionLoadState(snap.ServerId, itemLoadStates);
|
|
|
|
// Mint the partition via the preloaded-variant factory so
|
|
// it is wired into the dispatch registry from the moment
|
|
// it is constructed. Route the partition's notification
|
|
// callbacks through the wrapper's forwarding handler so
|
|
// restored secondaries observe the same serialization and
|
|
// logical-subscription identity contract that fresh-Add
|
|
// secondaries get; without this the raw user handler is
|
|
// invoked concurrently from multiple partitions on the
|
|
// restore path, breaking the single-threaded dispatch
|
|
// guarantee callers rely on.
|
|
ISubscriptionNotificationHandler effective =
|
|
wrapper.ForwardingHandler ?? handler;
|
|
IManagedSubscription partition = MintPreloadedPartition(
|
|
effective, optionsMonitor, loadState);
|
|
AttachReactiveFallback(partition, wrapper);
|
|
|
|
// Append to the wrapper's partition list so the composite
|
|
// collection's policy + index account for it.
|
|
wrapper.AppendPreloadedPartition(partition);
|
|
|
|
// Best-effort transfer: when the saved ServerId is
|
|
// non-zero AND the caller asked for transfer, issue
|
|
// TransferSubscriptions; on failure fall back to recreate
|
|
// via the partition's own state machine.
|
|
if (transferSubscriptions && snap.ServerId != 0 &&
|
|
partition is Subscription concrete)
|
|
{
|
|
bool transferred = false;
|
|
try
|
|
{
|
|
TransferSubscriptionsResponse response =
|
|
await m_session.TransferSubscriptionsAsync(
|
|
null,
|
|
new uint[] { snap.ServerId }.ToArrayOf(),
|
|
sendInitialValues: optionsMonitor.CurrentValue.SendInitialValuesOnTransfer,
|
|
ct).ConfigureAwait(false);
|
|
if (StatusCode.IsGood(response.ResponseHeader.ServiceResult)
|
|
&& response.Results.Count > 0
|
|
&& StatusCode.IsGood(response.Results[0].StatusCode))
|
|
{
|
|
transferred = await concrete.TryCompleteTransferAsync(
|
|
response.Results[0].AvailableSequenceNumbers.IsNull
|
|
? []
|
|
: [.. response.Results[0].AvailableSequenceNumbers],
|
|
ct).ConfigureAwait(false);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
m_logger.LogWarning(ex,
|
|
"{Subscription}: TransferSubscriptions threw on " +
|
|
"secondary partition restore; falling back to recreate.",
|
|
partition);
|
|
}
|
|
if (!transferred)
|
|
{
|
|
await concrete.ResetToRecreateAsync(ct).ConfigureAwait(false);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public ValueTask SaveAsync(
|
|
System.IO.Stream stream,
|
|
IServiceMessageContext messageContext,
|
|
IEnumerable<ISubscription>? subscriptions = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
return SubscriptionManagerSerializer.SaveAsync(
|
|
this,
|
|
stream,
|
|
messageContext,
|
|
subscriptions,
|
|
ct);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public ValueTask<IReadOnlyList<ISubscription>> LoadAsync(
|
|
System.IO.Stream stream,
|
|
IServiceMessageContext messageContext,
|
|
Func<string, ISubscriptionNotificationHandler> handlerFactory,
|
|
bool transferSubscriptions = false,
|
|
CancellationToken ct = default)
|
|
{
|
|
return SubscriptionManagerSerializer.LoadAsync(
|
|
this,
|
|
stream,
|
|
messageContext,
|
|
handlerFactory,
|
|
transferSubscriptions,
|
|
ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Resume subscriptions
|
|
/// </summary>
|
|
internal void Resume()
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
foreach (IManagedSubscription item in m_subscriptions)
|
|
{
|
|
item.NotifySubscriptionManagerPaused(false);
|
|
}
|
|
}
|
|
m_running.Set();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Pause subscriptions
|
|
/// </summary>
|
|
internal void Pause()
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
foreach (IManagedSubscription item in m_subscriptions)
|
|
{
|
|
item.NotifySubscriptionManagerPaused(true);
|
|
}
|
|
}
|
|
m_running.Reset();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Wait for all in-flight publish requests to complete or be
|
|
/// cancelled. Used together with <see cref="Pause"/> by callers
|
|
/// that need a hard quiesce (e.g. a session re-create) before
|
|
/// rebinding session state. <see cref="Pause"/> alone is a
|
|
/// soft signal — workers complete their current cycle before
|
|
/// observing the pause.
|
|
/// </summary>
|
|
/// <param name="ct">Cancellation token.</param>
|
|
internal Task DrainAsync(CancellationToken ct)
|
|
{
|
|
if (Volatile.Read(ref m_activePublishRequests) == 0)
|
|
{
|
|
return Task.CompletedTask;
|
|
}
|
|
return m_drainSignal.WaitAsync(ct);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Recreate subscriptions
|
|
/// </summary>
|
|
/// <param name="previousSessionId"></param>
|
|
/// <param name="ct"></param>
|
|
/// <returns></returns>
|
|
internal async Task RecreateSubscriptionsAsync(
|
|
NodeId? previousSessionId,
|
|
CancellationToken ct)
|
|
{
|
|
if (Count == 0)
|
|
{
|
|
// Nothing to do
|
|
return;
|
|
}
|
|
|
|
IReadOnlyList<IManagedSubscription> subscriptions = [.. m_subscriptions];
|
|
if (TransferSubscriptionsOnRecreate && previousSessionId != null)
|
|
{
|
|
subscriptions = await TransferSubscriptionsAsync(subscriptions,
|
|
false, ct).ConfigureAwait(false);
|
|
}
|
|
// Force creation of the subscriptions which were not transferred.
|
|
foreach (IManagedSubscription subscription in subscriptions)
|
|
{
|
|
bool force = previousSessionId != null && subscription.Created;
|
|
await subscription.RecreateAsync(ct).ConfigureAwait(false);
|
|
}
|
|
m_publishControl.Set();
|
|
|
|
// Helper to try and transfer the subscriptions
|
|
async Task<IReadOnlyList<IManagedSubscription>> TransferSubscriptionsAsync(
|
|
IReadOnlyList<IManagedSubscription> subscriptions,
|
|
bool sendInitialValues,
|
|
CancellationToken ct)
|
|
{
|
|
var remaining = subscriptions.Where(s => !s.Created).ToList();
|
|
subscriptions = [.. subscriptions.Where(s => s.Created)];
|
|
if (subscriptions.Count == 0)
|
|
{
|
|
return remaining;
|
|
}
|
|
var subscriptionIds = subscriptions.Select(s => s.Id).ToArrayOf();
|
|
TransferSubscriptionsResponse response = await m_session.TransferSubscriptionsAsync(
|
|
null,
|
|
subscriptionIds,
|
|
sendInitialValues,
|
|
ct).ConfigureAwait(false);
|
|
|
|
ResponseHeader responseHeader = response.ResponseHeader;
|
|
if (!StatusCode.IsGood(responseHeader.ServiceResult))
|
|
{
|
|
if (responseHeader.ServiceResult == StatusCodes.BadServiceUnsupported)
|
|
{
|
|
TransferSubscriptionsOnRecreate = false;
|
|
m_logger.LogWarning("Transfer subscription unsupported, " +
|
|
"TransferSubscriptionsOnReconnect set to false.");
|
|
}
|
|
else
|
|
{
|
|
m_logger.LogError(
|
|
"Transfer subscriptions failed with error {Error}.",
|
|
responseHeader.ServiceResult);
|
|
}
|
|
remaining.AddRange(subscriptions);
|
|
return remaining;
|
|
}
|
|
|
|
ArrayOf<TransferResult> transferResults = response.Results;
|
|
ArrayOf<DiagnosticInfo> diagnosticInfos = response.DiagnosticInfos;
|
|
ClientBase.ValidateResponse(transferResults, subscriptionIds);
|
|
ClientBase.ValidateDiagnosticInfos(diagnosticInfos, subscriptionIds);
|
|
for (int index = 0; index < subscriptions.Count; index++)
|
|
{
|
|
if (transferResults[index].StatusCode == StatusCodes.BadNothingToDo)
|
|
{
|
|
m_logger.LogDebug(
|
|
"Subscription {Id} is already member of the session.",
|
|
subscriptionIds[index]);
|
|
// Done
|
|
}
|
|
else if (!StatusCode.IsGood(transferResults[index].StatusCode))
|
|
{
|
|
m_logger.LogError(
|
|
"Subscription {Id} failed to transfer, StatusCode={Status}",
|
|
subscriptionIds[index],
|
|
transferResults[index].StatusCode);
|
|
remaining.Add(subscriptions[index]);
|
|
}
|
|
else
|
|
{
|
|
bool success = await subscriptions[index].TryCompleteTransferAsync(
|
|
transferResults[index].AvailableSequenceNumbers.ToList(),
|
|
ct).ConfigureAwait(false);
|
|
if (success)
|
|
{
|
|
continue;
|
|
}
|
|
//
|
|
// Recreate as we cannot sync the subscription. This happens
|
|
// when the GetMonitoredItems call fails and we cannot synchronize
|
|
// the subscription state
|
|
//
|
|
remaining.Add(subscriptions[index]);
|
|
}
|
|
}
|
|
return remaining;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Get subscription with the specified id
|
|
/// </summary>
|
|
/// <param name="subscriptionId"></param>
|
|
/// <returns></returns>
|
|
private IManagedSubscription? GetById(uint subscriptionId)
|
|
{
|
|
lock (m_subscriptionLock)
|
|
{
|
|
// find the subscription.
|
|
foreach (IManagedSubscription subscription in m_subscriptions)
|
|
{
|
|
if (subscription.Id == subscriptionId)
|
|
{
|
|
return subscription;
|
|
}
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Controls the publish workers.
|
|
/// </summary>
|
|
/// <param name="ct"></param>
|
|
/// <returns></returns>
|
|
private async Task PublishControllerAsync(CancellationToken ct)
|
|
{
|
|
var publishWorkers = new List<PublishWorker>();
|
|
try
|
|
{
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
int desiredWorkerCount = GetDesiredPublishWorkerCount();
|
|
if (publishWorkers.Count > desiredWorkerCount)
|
|
{
|
|
// Too many workers, reduce
|
|
#if NET8_0_OR_GREATER
|
|
foreach (PublishWorker worker in publishWorkers[desiredWorkerCount..])
|
|
#else
|
|
foreach (PublishWorker worker in publishWorkers.Skip(desiredWorkerCount))
|
|
#endif
|
|
{
|
|
m_logger.LogInformation("Removing publish worker {Index}",
|
|
worker.Index);
|
|
await worker.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
#if NET8_0_OR_GREATER
|
|
publishWorkers = publishWorkers[..desiredWorkerCount];
|
|
#else
|
|
publishWorkers = [.. publishWorkers.Take(desiredWorkerCount)];
|
|
#endif
|
|
}
|
|
else if (desiredWorkerCount > publishWorkers.Count)
|
|
{
|
|
// Not enough workers increase
|
|
publishWorkers.AddRange(Enumerable
|
|
.Range(publishWorkers.Count,
|
|
desiredWorkerCount - publishWorkers.Count)
|
|
.Select(index => new PublishWorker(this, index)));
|
|
}
|
|
PublishWorkerCount = publishWorkers.Count;
|
|
|
|
Task[] waiting = [.. publishWorkers
|
|
.Select(w => w.Task)
|
|
.Prepend(m_publishControl.WaitAsync(ct))];
|
|
await Task.WhenAny(waiting).ConfigureAwait(false);
|
|
PublishControlCycles++;
|
|
int index = 0;
|
|
foreach (Task? item in waiting.Skip(1)) // Skip wait handle
|
|
{
|
|
if (item.IsCompleted)
|
|
{
|
|
PublishWorker worker = publishWorkers[index];
|
|
m_logger.LogInformation(
|
|
"Publish worker {Index} exited",
|
|
worker.Index);
|
|
await worker.DisposeAsync().ConfigureAwait(false);
|
|
publishWorkers.RemoveAt(index);
|
|
continue;
|
|
}
|
|
index++;
|
|
}
|
|
|
|
// Now lower the max publish request if we got any too
|
|
// many requests errors
|
|
if (publishWorkers.Any(w => w.TooManyPublishRequests))
|
|
{
|
|
if (MaxPublishWorkerCount > 1)
|
|
{
|
|
MaxPublishWorkerCount--;
|
|
}
|
|
}
|
|
PublishWorkerCount = publishWorkers.Count;
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
}
|
|
finally
|
|
{
|
|
// Controller exits, clean up all workers
|
|
foreach (PublishWorker worker in publishWorkers)
|
|
{
|
|
await worker.DisposeAsync().ConfigureAwait(false);
|
|
PublishWorkerCount--;
|
|
}
|
|
}
|
|
|
|
int GetDesiredPublishWorkerCount()
|
|
{
|
|
int publishCount = CreatedCount;
|
|
if (publishCount != 0)
|
|
{
|
|
//
|
|
// Limit resulting to a number between min and max
|
|
// request count. If max is below min, we honor the
|
|
// min publish request count.
|
|
//
|
|
if (publishCount > MaxPublishWorkerCount)
|
|
{
|
|
publishCount = MaxPublishWorkerCount;
|
|
}
|
|
if (publishCount < MinPublishWorkerCount)
|
|
{
|
|
publishCount = MinPublishWorkerCount;
|
|
}
|
|
if (publishCount <= 0)
|
|
{
|
|
publishCount = 1;
|
|
}
|
|
}
|
|
return publishCount;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Worker object that manages the publish worker tasks inside
|
|
/// the controller.
|
|
/// </summary>
|
|
private sealed class PublishWorker : IAsyncDisposable
|
|
{
|
|
/// <summary>
|
|
/// Worker id
|
|
/// </summary>
|
|
public int Index { get; }
|
|
|
|
/// <summary>
|
|
/// Task to wait until worker exits
|
|
/// </summary>
|
|
public Task Task { get; }
|
|
|
|
/// <summary>
|
|
/// Signal too many publish requests running
|
|
/// </summary>
|
|
public bool TooManyPublishRequests { get; private set; }
|
|
|
|
/// <summary>
|
|
/// Create worker
|
|
/// </summary>
|
|
/// <param name="outer"></param>
|
|
/// <param name="index"></param>
|
|
public PublishWorker(SubscriptionManager outer, int index)
|
|
{
|
|
Index = index;
|
|
m_outer = outer;
|
|
m_cts = CancellationTokenSource.CreateLinkedTokenSource(outer.m_cts.Token);
|
|
m_logger = m_outer.m_loggerFactory.CreateLogger<PublishWorker>();
|
|
Task = PublishWorkerAsync(m_cts.Token);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask DisposeAsync()
|
|
{
|
|
try
|
|
{
|
|
await m_cts.CancelAsync().ConfigureAwait(false);
|
|
if (!Task.IsCompleted)
|
|
{
|
|
try
|
|
{
|
|
await Task.ConfigureAwait(false);
|
|
}
|
|
catch
|
|
{
|
|
} // Ignore
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
m_cts.Dispose();
|
|
GC.SuppressFinalize(this);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Represents a continously running publish forwarder that forwards
|
|
/// publish responses to the subscriptions contained in this session.
|
|
/// The publish worker tasks have a controller that reduces or
|
|
/// increases the number of workers as new subscriptions are added
|
|
/// or subscriptions are removed. Once the message has been delivered
|
|
/// to the subscription, the subscription will queue acknowledges
|
|
/// to the worker which it will send.
|
|
/// </summary>
|
|
/// <param name="ct"></param>
|
|
private async Task PublishWorkerAsync(CancellationToken ct)
|
|
{
|
|
long publishLatencyStart = m_outer.m_timeProvider.GetTimestamp();
|
|
TimeSpan publishLatency = TimeSpan.Zero;
|
|
bool publishLatencyRunning = false;
|
|
uint timeoutHint = 0u;
|
|
bool moreNotifications = true; // Dont wait first time we enter the loop.
|
|
m_logger.LogInformation("PUBLISH Worker #{Handle} - STARTED.", Index);
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
if (!m_outer.m_running.IsSet)
|
|
{
|
|
m_logger.LogInformation("PUBLISH Worker #{Handle} - PAUSED.", Index);
|
|
try
|
|
{
|
|
await m_outer.m_running.WaitAsync(ct).ConfigureAwait(false);
|
|
m_logger.LogInformation(
|
|
"PUBLISH Worker #{Handle} - RESUMED.",
|
|
Index);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
continue;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
m_logger.LogError(
|
|
ex,
|
|
"PUBLISH Worker #{Handle} - Unexpected exception while waiting to connect.",
|
|
Index);
|
|
break;
|
|
}
|
|
}
|
|
long currentLatencyMs = publishLatencyRunning
|
|
? (long)m_outer.m_timeProvider.GetElapsedTime(publishLatencyStart).TotalMilliseconds
|
|
: (long)publishLatency.TotalMilliseconds;
|
|
int ackWaitTimeout = CalculateTimeouts(
|
|
currentLatencyMs, ref timeoutHint);
|
|
ArrayOf<SubscriptionAcknowledgement> acks = GetAcksReadyToSend();
|
|
uint handle = Utils.IncrementIdentifier(ref m_outer.m_publishRequestCounter);
|
|
try
|
|
{
|
|
if (acks.Count == 0 && !moreNotifications && ackWaitTimeout != 0)
|
|
{
|
|
// Throttle publishing as we wait for acks to arrive
|
|
acks = await WaitForAcksAsync(ackWaitTimeout, ct).ConfigureAwait(false);
|
|
}
|
|
publishLatencyStart = m_outer.m_timeProvider.GetTimestamp();
|
|
publishLatencyRunning = true;
|
|
if (Interlocked.Increment(ref m_outer.m_activePublishRequests) == 1)
|
|
{
|
|
m_outer.m_drainSignal.Reset();
|
|
}
|
|
try
|
|
{
|
|
PublishResponse response = await m_outer.m_session.PublishAsync(new RequestHeader
|
|
{
|
|
TimeoutHint = timeoutHint,
|
|
ReturnDiagnostics = (uint)(int)m_outer.ReturnDiagnostics,
|
|
RequestHandle = handle
|
|
}, acks, ct).ConfigureAwait(false);
|
|
|
|
moreNotifications = response.MoreNotifications;
|
|
uint subscriptionId = response.SubscriptionId;
|
|
NotificationMessage notificationMessage = response.NotificationMessage;
|
|
ArrayOf<uint> availableSequenceNumbers = response.AvailableSequenceNumbers;
|
|
|
|
ArrayOf<StatusCode> acknowledgeResults = response.Results;
|
|
ArrayOf<DiagnosticInfo> acknowledgeDiagnosticInfos = response.DiagnosticInfos;
|
|
ClientBase.ValidateResponse(acknowledgeResults, acks);
|
|
ClientBase.ValidateDiagnosticInfos(acknowledgeDiagnosticInfos, acks);
|
|
TooManyPublishRequests = false;
|
|
|
|
// Get the subscription with the provided identifier
|
|
IManagedSubscription? subscription = m_outer.GetById(subscriptionId);
|
|
publishLatency = m_outer.m_timeProvider.GetElapsedTime(publishLatencyStart);
|
|
publishLatencyRunning = false;
|
|
if (subscription != null)
|
|
{
|
|
// deliver to subscription
|
|
await subscription.OnPublishReceivedAsync(
|
|
notificationMessage,
|
|
availableSequenceNumbers.ToList(),
|
|
response.ResponseHeader.StringTable.ToList()).ConfigureAwait(false);
|
|
Interlocked.Increment(ref m_outer.m_goodPublishRequestCount);
|
|
}
|
|
else if (!ct.IsCancellationRequested &&
|
|
!m_outer.m_subscriptionHistory.Contains(subscriptionId))
|
|
{
|
|
// ignore messages with a subscription that was deleted
|
|
// Do not delete publish requests of stale subscriptions
|
|
m_logger.LogInformation(
|
|
"PUBLISH Worker #{Handle}-{Id} - Received Publish Response " +
|
|
"for Unknown SubscriptionId={SubscriptionId}. Deleting...",
|
|
Index, handle, subscriptionId);
|
|
Interlocked.Increment(ref m_outer.m_badPublishRequestCount);
|
|
await m_outer.m_session.DeleteSubscriptionsAsync(
|
|
null,
|
|
[subscriptionId],
|
|
ct).ConfigureAwait(false);
|
|
moreNotifications = true;
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
int active = Interlocked.Decrement(ref m_outer.m_activePublishRequests);
|
|
if (active == 0)
|
|
{
|
|
m_outer.m_drainSignal.Set();
|
|
}
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
catch (Exception e)
|
|
{
|
|
// raise an error event.
|
|
publishLatency = m_outer.m_timeProvider.GetElapsedTime(publishLatencyStart);
|
|
publishLatencyRunning = false;
|
|
var error = new ServiceResult(e);
|
|
|
|
if (error.Code == StatusCodes.BadRequestInterrupted &&
|
|
ct.IsCancellationRequested)
|
|
{
|
|
break;
|
|
}
|
|
|
|
Interlocked.Increment(ref m_outer.m_badPublishRequestCount);
|
|
// Rollback acks we collected
|
|
acks.ForEach(ack => m_outer.m_acks.Writer.TryWrite(ack));
|
|
|
|
// ignore errors if paused.
|
|
if (!m_outer.m_running.IsSet)
|
|
{
|
|
m_logger.LogWarning("PUBLISH Worker #{Handle}-{Id} - Publish " +
|
|
"abandoned after error due to reconnect: {Message}",
|
|
Index, handle, e.Message);
|
|
continue;
|
|
}
|
|
|
|
// don't send another publish for these errors,
|
|
// or throttle to avoid server overload.
|
|
StatusCode statusCode = error.StatusCode;
|
|
if (statusCode == StatusCodes.BadTooManyPublishRequests)
|
|
{
|
|
TooManyPublishRequests = true;
|
|
}
|
|
else if (statusCode == StatusCodes.BadNoSubscription ||
|
|
statusCode == StatusCodes.BadSessionClosed ||
|
|
statusCode == StatusCodes.BadSecurityChecksFailed ||
|
|
statusCode == StatusCodes.BadCertificateInvalid ||
|
|
statusCode == StatusCodes.BadServerHalted)
|
|
{
|
|
// ignore
|
|
}
|
|
// may require a reconnect or activate to recover
|
|
else if (statusCode == StatusCodes.BadSessionIdInvalid ||
|
|
statusCode == StatusCodes.BadSecureChannelIdInvalid ||
|
|
statusCode == StatusCodes.BadSecureChannelClosed)
|
|
{
|
|
// TODO
|
|
// OnKeepAliveError(error);
|
|
}
|
|
// Servers may return this error when overloaded
|
|
else if (statusCode == StatusCodes.BadTooManyOperations ||
|
|
statusCode == StatusCodes.BadTcpServerTooBusy ||
|
|
statusCode == StatusCodes.BadServerTooBusy)
|
|
{
|
|
// throttle the next publish to reduce server load
|
|
m_logger.LogDebug("PUBLISH Worker #{Handle}-{Id} - " +
|
|
"Server busy, throttling worker.",
|
|
Index, handle);
|
|
moreNotifications = false; // throttle
|
|
}
|
|
else if (statusCode == StatusCodes.BadTimeout ||
|
|
statusCode == StatusCodes.BadRequestTimeout)
|
|
{
|
|
// Timed out - retry with larger timeout
|
|
timeoutHint += 1000; // Increase by seconds
|
|
m_logger.LogDebug("PUBLISH Worker #{Handle}-{Id} - " +
|
|
"Timed out, increasing timeout to {Timeout}.",
|
|
Index, handle, timeoutHint);
|
|
moreNotifications = true;
|
|
}
|
|
else
|
|
{
|
|
m_logger.LogError(e, "PUBLISH Worker #{Handle}-{Id} - " +
|
|
"Unhandled error {Status} during Publish.",
|
|
Index, handle, error.StatusCode);
|
|
}
|
|
}
|
|
}
|
|
m_logger.LogInformation("PUBLISH Worker #{Handle} - STOPPED.", Index);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Wait until acks arrive and return them
|
|
/// </summary>
|
|
/// <param name="maxWaitTime"></param>
|
|
/// <param name="ct"></param>
|
|
/// <returns></returns>
|
|
private async Task<ArrayOf<SubscriptionAcknowledgement>> WaitForAcksAsync(
|
|
int maxWaitTime,
|
|
CancellationToken ct)
|
|
{
|
|
Debug.Assert(maxWaitTime != 0, "Checked before entering");
|
|
long swStart = m_outer.m_timeProvider.GetTimestamp();
|
|
CancellationTokenSource? timeoutCts = null;
|
|
CancellationTokenSource? linkedCts = null;
|
|
CancellationToken waitToken;
|
|
if (maxWaitTime != Timeout.Infinite)
|
|
{
|
|
#if FALSE
|
|
var workers = m_outer.PublishWorkerCount;
|
|
if (workers == 0)
|
|
{
|
|
Debug.Fail("Must have at least this worker here.");
|
|
workers = 1;
|
|
}
|
|
maxWaitTime /= workers;
|
|
#endif
|
|
m_logger.LogDebug(
|
|
"PUBLISH Worker #{Handle} - Waiting max {Time}ms for acks to arrive.",
|
|
Index, maxWaitTime);
|
|
timeoutCts = m_outer.m_timeProvider
|
|
.CreateCancellationTokenSource(TimeSpan.FromMilliseconds(maxWaitTime));
|
|
linkedCts = CancellationTokenSource
|
|
.CreateLinkedTokenSource(ct, timeoutCts.Token);
|
|
waitToken = linkedCts.Token;
|
|
}
|
|
else
|
|
{
|
|
m_logger.LogDebug(
|
|
"PUBLISH Worker #{Handle} - Waiting for acks to arrive.",
|
|
Index);
|
|
waitToken = ct;
|
|
}
|
|
try
|
|
{
|
|
SubscriptionAcknowledgement firstAck = await m_outer.m_acks.Reader.ReadAsync(
|
|
waitToken).ConfigureAwait(false);
|
|
ArrayOf<SubscriptionAcknowledgement> restAcks = GetAcksReadyToSend();
|
|
var ackList = restAcks.ToList();
|
|
ackList.Insert(0, firstAck);
|
|
ArrayOf<SubscriptionAcknowledgement> acks = ackList;
|
|
m_logger.LogDebug(
|
|
"PUBLISH Worker #{Handle} - Publish {Count} acks after pausing {Duration}.",
|
|
Index,
|
|
acks.Count,
|
|
m_outer.m_timeProvider.GetElapsedTime(swStart));
|
|
return acks;
|
|
}
|
|
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
|
{
|
|
m_logger.LogInformation(
|
|
"PUBLISH Worker #{Handle} - Publish with no acks after waiting {Duration}.",
|
|
Index,
|
|
m_outer.m_timeProvider.GetElapsedTime(swStart));
|
|
return [];
|
|
}
|
|
finally
|
|
{
|
|
linkedCts?.Dispose();
|
|
timeoutCts?.Dispose();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Get acks that are ready to send
|
|
/// </summary>
|
|
/// <returns></returns>
|
|
private ArrayOf<SubscriptionAcknowledgement> GetAcksReadyToSend()
|
|
{
|
|
var acks = new List<SubscriptionAcknowledgement>();
|
|
|
|
// TODO: Is this something that we can get from ops limit?
|
|
int available = m_outer.m_acks.Reader.Count;
|
|
int maxAcks = available / Math.Max(m_outer.PublishWorkerCount, 1);
|
|
for (int i = 0; i < maxAcks &&
|
|
m_outer.m_acks.Reader.TryRead(out SubscriptionAcknowledgement? ack); i++)
|
|
{
|
|
acks.Add(ack);
|
|
}
|
|
if (acks.Count != 0)
|
|
{
|
|
m_logger.LogDebug(
|
|
"PUBLISH Worker #{Handle} - Acknoledging {Count} of {Total} messages.",
|
|
Index, acks.Count, available);
|
|
}
|
|
return acks;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Calculate the publish timeout to use
|
|
/// </summary>
|
|
/// <param name="latency"></param>
|
|
/// <param name="currentTimeout"></param>
|
|
/// <returns>Max time to throttle the publish</returns>
|
|
private int CalculateTimeouts(long latency, ref uint currentTimeout)
|
|
{
|
|
List<IManagedSubscription> created = m_outer.Created;
|
|
if (created.Count == 0)
|
|
{
|
|
return 0;
|
|
}
|
|
|
|
TimeSpan timeout = TimeSpan.Zero;
|
|
int minPublishInterval = Timeout.Infinite;
|
|
foreach (IManagedSubscription s in created)
|
|
{
|
|
TimeSpan publishingInterval = s.CurrentPublishingInterval;
|
|
TimeSpan keepAlive = publishingInterval.Multiply(s.CurrentKeepAliveCount);
|
|
if (timeout < keepAlive)
|
|
{
|
|
timeout = keepAlive;
|
|
}
|
|
|
|
int pi = (int)publishingInterval.TotalMilliseconds;
|
|
if (pi <= 0)
|
|
{
|
|
continue;
|
|
}
|
|
if (minPublishInterval > pi ||
|
|
minPublishInterval == Timeout.Infinite)
|
|
{
|
|
minPublishInterval = pi;
|
|
}
|
|
}
|
|
//
|
|
// The timeout while publishing should be twice the
|
|
// value for PublishingInterval * KeepAliveCount
|
|
// TODO: Validate this against spec
|
|
//
|
|
timeout = timeout.Multiply(2);
|
|
if (timeout < s_minOperationTimeout)
|
|
{
|
|
timeout = s_minOperationTimeout;
|
|
}
|
|
if (timeout > s_maxOperationTimeout)
|
|
{
|
|
timeout = s_maxOperationTimeout;
|
|
}
|
|
uint newTimeout = (uint)timeout.TotalMilliseconds;
|
|
if (newTimeout > currentTimeout)
|
|
{
|
|
currentTimeout = newTimeout;
|
|
}
|
|
if (minPublishInterval < latency)
|
|
{
|
|
return 0;
|
|
}
|
|
return minPublishInterval - (int)latency;
|
|
}
|
|
|
|
private readonly ILogger m_logger;
|
|
private readonly SubscriptionManager m_outer;
|
|
private readonly CancellationTokenSource m_cts;
|
|
}
|
|
|
|
private static readonly TimeSpan s_maxOperationTimeout = TimeSpan.FromMinutes(30);
|
|
private static readonly TimeSpan s_minOperationTimeout = TimeSpan.FromSeconds(1);
|
|
private const int kMaxSubscriptionHistory = 10;
|
|
private uint m_publishRequestCounter;
|
|
#pragma warning disable IDE0032 // Use auto property
|
|
private int m_badPublishRequestCount;
|
|
private int m_goodPublishRequestCount;
|
|
#pragma warning restore IDE0032 // Use auto property
|
|
private readonly Channel<SubscriptionAcknowledgement> m_acks;
|
|
private readonly AsyncManualResetEvent m_running = new();
|
|
private readonly AsyncAutoResetEvent m_publishControl = new();
|
|
private readonly AsyncManualResetEvent m_drainSignal = new(true);
|
|
private int m_activePublishRequests;
|
|
private int m_disposed;
|
|
private readonly ConcurrentQueue<uint> m_subscriptionHistory = new();
|
|
private readonly Task m_publishController;
|
|
private readonly Lock m_subscriptionLock = new();
|
|
// Dispatch registry: every partition subscription this manager
|
|
// owns, including the primaries of logical wrappers. Publish
|
|
// dispatch (GetById), acknowledgement routing, recreate, and
|
|
// transfer iterate this set so they remain partition-aware.
|
|
private readonly HashSet<IManagedSubscription> m_subscriptions = [];
|
|
// Public registry: logical wrappers returned to callers from
|
|
// ISubscriptionManager.Items / Add. One wrapper per logical
|
|
// subscription regardless of how many partitions back it.
|
|
private readonly HashSet<LogicalSubscription> m_logicals = [];
|
|
private readonly CancellationTokenSource m_cts = new();
|
|
private readonly ISubscriptionManagerContext m_session;
|
|
private readonly ILoggerFactory m_loggerFactory;
|
|
private readonly ILogger m_logger;
|
|
private readonly TimeProvider m_timeProvider;
|
|
}
|
|
}
|