mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-22 01:26:20 +08:00
Marc Schier
d8a95b4e9e
# Description Full-repo `dotnet format` sweep across all 73 projects in `UA.slnx` — no behavioral changes.
1321 lines
55 KiB
C#
1321 lines
55 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;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using Opc.Ua.Client.Subscriptions.MonitoredItems;
|
|
|
|
namespace Opc.Ua.Client.Subscriptions
|
|
{
|
|
/// <summary>
|
|
/// A managed subscription inside a subscription manager. Can be
|
|
/// extended to provide custom subscription implementations on
|
|
/// top to route the received data appropriately per application.
|
|
/// The subscription itself is based on top of the message processor
|
|
/// implementation that routes messages to subscribers, but adds
|
|
/// state management of the subscription on the server using the
|
|
/// subscription and monitored item service set provided as context.
|
|
/// </summary>
|
|
internal abstract class Subscription : MessageProcessor, IManagedSubscription,
|
|
IMonitoredItemManagerContext
|
|
{
|
|
/// <inheritdoc/>
|
|
public byte CurrentPriority { get; private set; }
|
|
|
|
/// <summary>
|
|
/// Server-assigned subscription id. <c>0</c> when the
|
|
/// subscription has not been created on the server yet.
|
|
/// </summary>
|
|
public uint ServerId => Id;
|
|
|
|
/// <inheritdoc/>
|
|
public TimeSpan CurrentPublishingInterval { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
public uint CurrentKeepAliveCount { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
public uint CurrentLifetimeCount { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
public bool CurrentPublishingEnabled { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
public uint CurrentMaxNotificationsPerPublish { get; private set; }
|
|
|
|
/// <inheritdoc/>
|
|
public IMonitoredItemCollection MonitoredItems => m_monitoredItems;
|
|
|
|
/// <summary>
|
|
/// <para>
|
|
/// Optional callback invoked when the server rejects a
|
|
/// monitored-item creation on this partition with
|
|
/// <see cref="StatusCodes.BadTooManyMonitoredItems"/>. The
|
|
/// V2 reactive fallback (see
|
|
/// <see cref="PartitionPlacementPolicy.OnPartitionCapReached"/>)
|
|
/// uses this hook to discover the server's effective
|
|
/// per-subscription cap and mark the partition no-grow so
|
|
/// subsequent placements skip it.
|
|
/// </para>
|
|
/// <para>
|
|
/// Wired by <see cref="SubscriptionManager.Add"/> and
|
|
/// <see cref="SubscriptionManager.MintPartition"/> when the
|
|
/// owning <see cref="LogicalSubscription"/> has placement
|
|
/// enabled. <c>null</c> otherwise; the partition then surfaces
|
|
/// rejections through normal item-level error reporting only.
|
|
/// </para>
|
|
/// </summary>
|
|
internal Action<ServiceResult>? OnPartitionCapReached { get; set; }
|
|
|
|
/// <summary>
|
|
/// <para>
|
|
/// Optional callback invoked by <see cref="StateManagerAsync"/>
|
|
/// exactly once per partition lifetime, immediately after a
|
|
/// successful <see cref="CreateAsync"/> and before the first
|
|
/// <see cref="MonitoredItemManager.ApplyChangesAsync"/>
|
|
/// in the same iteration of the state-manager loop. Used by
|
|
/// <see cref="LogicalSubscription"/> to satisfy the OPC UA
|
|
/// Part 4 §5.13.9 rule that
|
|
/// <c>SetSubscriptionDurable</c> must precede any
|
|
/// monitored-item creation: the wrapper installs a hook that
|
|
/// awaits <see cref="SetAsDurableAsync"/> here so the call
|
|
/// is interleaved between the
|
|
/// <c>CreateSubscription</c> response and the first
|
|
/// <c>CreateMonitoredItems</c> request.
|
|
/// </para>
|
|
/// <para>
|
|
/// The state machine clears the property after invoking it so
|
|
/// modify cycles do not re-run the hook. The owning wrapper
|
|
/// re-installs the hook before the next Create pass (e.g.
|
|
/// after reconnect / recreate) when durable intent has been
|
|
/// recorded.
|
|
/// </para>
|
|
/// <para>
|
|
/// Exceptions thrown by the hook are logged as a warning and
|
|
/// surfaced via a <see cref="SubscriptionState.Modified"/>
|
|
/// state change rather than tearing the partition down — a
|
|
/// failed durable-apply is reported but does not prevent the
|
|
/// subscription from continuing.
|
|
/// </para>
|
|
/// </summary>
|
|
internal Func<CancellationToken, ValueTask>? OnAfterCreateAsync
|
|
{
|
|
get => m_onAfterCreateAsync;
|
|
set => m_onAfterCreateAsync = value;
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public bool Created => Id != 0;
|
|
|
|
/// <inheritdoc/>
|
|
public IMonitoredItemServiceSetClientMethods MonitoredItemServiceSet
|
|
=> m_context.MonitoredItemServiceSet;
|
|
|
|
/// <inheritdoc/>
|
|
public IMethodServiceSetClientMethods MethodServiceSet
|
|
=> m_context.MethodServiceSet;
|
|
|
|
/// <summary>
|
|
/// Returns true if the subscription is not receiving publishes.
|
|
/// </summary>
|
|
internal bool PublishingStopped
|
|
{
|
|
get
|
|
{
|
|
long lastNotificationTimestamp = LastNotificationTimestamp;
|
|
if (lastNotificationTimestamp == 0)
|
|
{
|
|
return false;
|
|
}
|
|
TimeSpan timeSinceLastNotification = TimeProvider
|
|
.GetElapsedTime(lastNotificationTimestamp);
|
|
return timeSinceLastNotification >
|
|
m_keepAliveInterval + s_keepAliveTimerMargin;
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Current subscription options
|
|
/// </summary>
|
|
protected internal SubscriptionOptions Options { get; protected set; } = new();
|
|
|
|
/// <summary>
|
|
/// Create subscription
|
|
/// </summary>
|
|
/// <param name="context"></param>
|
|
/// <param name="handler"></param>
|
|
/// <param name="completion"></param>
|
|
/// <param name="options"></param>
|
|
/// <param name="telemetry"></param>
|
|
/// <param name="loadState">Optional snapshot of a previously
|
|
/// active subscription. When supplied, the V2 state machine is
|
|
/// constructed in "loaded" mode: <see cref="MessageProcessor.Id"/>
|
|
/// is pre-set to the saved server-side subscription id, the
|
|
/// supplied monitored items are pre-bound to their saved
|
|
/// server/client handles, and the initial signal that would
|
|
/// otherwise trigger <c>CreateSubscription</c> is suppressed.
|
|
/// The owning <see cref="SubscriptionManager.RestoreAsync"/>
|
|
/// flow then issues TransferSubscriptions and binds runtime
|
|
/// state via <see cref="TryCompleteTransferAsync"/>.</param>
|
|
/// <param name="timeProvider">Optional <see cref="TimeProvider"/>
|
|
/// for elapsed-time and timer calculations. Defaults to
|
|
/// <see cref="TimeProvider.System"/> when <c>null</c>.</param>
|
|
protected Subscription(
|
|
ISubscriptionContext context,
|
|
ISubscriptionNotificationHandler handler,
|
|
IMessageAckQueue completion,
|
|
IOptionsMonitor<SubscriptionOptions> options,
|
|
ITelemetryContext telemetry,
|
|
SubscriptionLoadState? loadState = null,
|
|
TimeProvider? timeProvider = null)
|
|
: base(context.SubscriptionServiceSet, completion, telemetry, timeProvider)
|
|
{
|
|
m_handler = handler;
|
|
m_context = context;
|
|
m_monitoredItems = new MonitoredItemManager(this, telemetry);
|
|
m_publishTimer = TimeProvider.CreateTimer(OnKeepAlive,
|
|
null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
|
|
if (loadState != null)
|
|
{
|
|
// Pre-install server-assigned identifiers + items
|
|
// before hooking change tracking, so the V2 state
|
|
// machine sees a coherent "loaded" snapshot when it
|
|
// first wakes.
|
|
Id = loadState.ServerId;
|
|
m_createdEvent.Set();
|
|
Options = options.CurrentValue;
|
|
foreach (MonitoredItemLoadState item in loadState.MonitoredItems)
|
|
{
|
|
m_monitoredItems.AddLoaded(item);
|
|
}
|
|
// Hook options change tracking but do NOT signal
|
|
// m_stateControl — the manager's RestoreAsync drives
|
|
// transfer + completion explicitly.
|
|
m_changeTracking = options.OnChange((o, _) => OnOptionsChanged(o));
|
|
}
|
|
else
|
|
{
|
|
OnOptionsChanged(options.CurrentValue);
|
|
m_changeTracking = options.OnChange((o, _) => OnOptionsChanged(o));
|
|
}
|
|
m_stateManagement = StateManagerAsync(m_cts.Token);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public override string ToString()
|
|
{
|
|
return $"{m_context}:{Id}";
|
|
}
|
|
|
|
/// <summary>
|
|
/// Capture an immutable snapshot of this subscription's
|
|
/// configuration + identifiers + the per-item state.
|
|
/// </summary>
|
|
public SubscriptionStateSnapshot Snapshot()
|
|
{
|
|
var items = new List<MonitoredItemStateSnapshot>();
|
|
foreach (IMonitoredItem item in m_monitoredItems.Items)
|
|
{
|
|
if (item is MonitoredItems.MonitoredItem concrete)
|
|
{
|
|
items.Add(concrete.Snapshot());
|
|
}
|
|
}
|
|
uint[] available = AvailableInRetransmissionQueue == null
|
|
? []
|
|
: [.. AvailableInRetransmissionQueue];
|
|
return SubscriptionStateSnapshot.AsOptions(
|
|
Options,
|
|
Id,
|
|
available.ToArrayOf(),
|
|
items.ToArrayOf());
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask ConditionRefreshAsync(CancellationToken ct)
|
|
{
|
|
if (!Created)
|
|
{
|
|
throw ServiceResultException.Create(StatusCodes.BadSubscriptionIdInvalid,
|
|
"Subscription has not been created.");
|
|
}
|
|
var methodsToCall = new CallMethodRequest[]
|
|
{
|
|
new()
|
|
{
|
|
MethodId = MethodIds.ConditionType_ConditionRefresh,
|
|
InputArguments = [new Variant(Id)]
|
|
}
|
|
};
|
|
await m_context.MethodServiceSet.CallAsync(null, methodsToCall,
|
|
ct).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask<SetTriggeringResult> SetTriggeringAsync(
|
|
IMonitoredItem triggeringItem,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToAdd = null,
|
|
IReadOnlyCollection<IMonitoredItem>? linksToRemove = null,
|
|
CancellationToken ct = default)
|
|
{
|
|
if (triggeringItem == null)
|
|
{
|
|
throw new ArgumentNullException(nameof(triggeringItem));
|
|
}
|
|
IReadOnlyList<IMonitoredItem> add =
|
|
linksToAdd is null
|
|
? []
|
|
: linksToAdd as IReadOnlyList<IMonitoredItem>
|
|
?? [.. linksToAdd];
|
|
IReadOnlyList<IMonitoredItem> remove =
|
|
linksToRemove is null
|
|
? []
|
|
: linksToRemove as IReadOnlyList<IMonitoredItem>
|
|
?? [.. linksToRemove];
|
|
|
|
// Eager validation by reference identity under the manager
|
|
// lock — rejects items from other subscriptions and items
|
|
// that have already been removed. Per Part 4 §5.13.5.1
|
|
// both ends of every link must belong to the same
|
|
// subscription.
|
|
m_monitoredItems.ValidateBelongsAndUpdateDesired(
|
|
triggeringItem, add, remove);
|
|
|
|
// Enqueue a single operation that the batched
|
|
// ApplyMonitoredItemChangesAsync pass will pick up, merge with
|
|
// other operations targeting the same triggering item, and
|
|
// turn into one SetTriggering RPC.
|
|
var tcs = new TaskCompletionSource<SetTriggeringResult>(
|
|
TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var op = new MonitoredItemManager.TriggeringOperation(
|
|
triggeringItem, add, remove, tcs);
|
|
m_monitoredItems.EnqueueTriggeringOperation(op);
|
|
// Wake the state manager so the operations queue gets
|
|
// drained on the next ApplyChangesAsync pass.
|
|
m_stateControl.Set();
|
|
// Cancellation semantics (best-effort): aborting the await
|
|
// stops the caller from observing the eventual result AND
|
|
// marks the queued operation as cancelled so the next
|
|
// ApplyTriggeringOperationsAsync pass will skip it (no
|
|
// server-side SetTriggering request issued for this op).
|
|
// Two important caveats:
|
|
// 1. If the apply pass has already dispatched the RPC for
|
|
// this op's group, the server-side mutation cannot be
|
|
// undone — the cancel is best-effort only.
|
|
// 2. Desired-state mutations performed synchronously by
|
|
// ValidateBelongsAndUpdateDesired stand regardless of
|
|
// cancellation; to revert local intent the caller must
|
|
// issue an explicit opposing SetTriggeringAsync (or
|
|
// change MonitoredItemOptions.TriggeredByNames).
|
|
// The token is captured in the closure and passed to
|
|
// TrySetCanceled for correct cancellation metadata on the
|
|
// propagated exception (the
|
|
// Register(Action<object?, CancellationToken>, object?)
|
|
// overload was only added in .NET 5 and is unavailable on
|
|
// net48/net472/netstandard2.1, so the closure capture is
|
|
// the only portable form).
|
|
using CancellationTokenRegistration reg = ct.Register(() =>
|
|
{
|
|
op.MarkCancelled();
|
|
tcs.TrySetCanceled(ct);
|
|
});
|
|
return await tcs.Task.ConfigureAwait(false);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask<TimeSpan> SetAsDurableAsync(
|
|
TimeSpan lifetime,
|
|
CancellationToken ct = default)
|
|
{
|
|
if (!Created)
|
|
{
|
|
throw ServiceResultException.Create(
|
|
StatusCodes.BadSubscriptionIdInvalid,
|
|
"Subscription has not been created on the server.");
|
|
}
|
|
// SetSubscriptionDurable uses whole-hour granularity (UInt32);
|
|
// round up so requesting 90 minutes asks the server for 2 hours.
|
|
uint lifetimeInHours = (uint)Math.Max(
|
|
1,
|
|
Math.Ceiling(lifetime.TotalHours));
|
|
ArrayOf<CallMethodRequest> methodsToCall =
|
|
[
|
|
new CallMethodRequest
|
|
{
|
|
ObjectId = ObjectIds.Server,
|
|
MethodId = MethodIds.Server_SetSubscriptionDurable,
|
|
InputArguments =
|
|
[
|
|
new Variant(Id),
|
|
new Variant(lifetimeInHours)
|
|
]
|
|
}
|
|
];
|
|
CallResponse response = await m_context.MethodServiceSet
|
|
.CallAsync(null, methodsToCall, ct).ConfigureAwait(false);
|
|
ArrayOf<CallMethodResult> results = response.Results;
|
|
ArrayOf<DiagnosticInfo> diagnosticInfos = response.DiagnosticInfos;
|
|
ClientBase.ValidateResponse(results, methodsToCall);
|
|
ClientBase.ValidateDiagnosticInfos(diagnosticInfos, methodsToCall);
|
|
if (StatusCode.IsBad(results[0].StatusCode))
|
|
{
|
|
throw new ServiceResultException(ClientBase.GetResult(
|
|
results[0].StatusCode, 0, diagnosticInfos,
|
|
response.ResponseHeader));
|
|
}
|
|
ArrayOf<Variant> outputs = results[0].OutputArguments;
|
|
if (outputs.Count == 0 ||
|
|
!outputs[0].TryGetValue(out uint revised))
|
|
{
|
|
throw ServiceResultException.Create(StatusCodes.BadUnexpectedError,
|
|
"Server.SetSubscriptionDurable returned no revised lifetime.");
|
|
}
|
|
return TimeSpan.FromHours(revised);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask RecreateAsync(CancellationToken ct)
|
|
{
|
|
await m_stateLock.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
LastSequenceNumberProcessed = 0;
|
|
LastNotificationTimestamp = 0;
|
|
|
|
Id = 0;
|
|
CurrentPublishingInterval = TimeSpan.Zero;
|
|
CurrentKeepAliveCount = 0;
|
|
CurrentPublishingEnabled = false;
|
|
CurrentPriority = 0;
|
|
|
|
// Recreate subscription
|
|
await CreateAsync(Options, ct).ConfigureAwait(false);
|
|
|
|
await m_monitoredItems.ApplyChangesAsync(true, true,
|
|
ct).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
m_stateLock.Release();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public void NotifySubscriptionManagerPaused(bool paused)
|
|
{
|
|
m_monitoredItems.NotifySubscriptionManagerPaused(paused);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public async ValueTask<bool> TryCompleteTransferAsync(
|
|
IReadOnlyList<uint> availableSequenceNumbers, CancellationToken ct)
|
|
{
|
|
await m_stateLock.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
StopKeepAliveTimer();
|
|
bool success = await m_monitoredItems.TrySynchronizeHandlesAsync(
|
|
ct).ConfigureAwait(false);
|
|
if (!success)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
// save available sequence numbers
|
|
AvailableInRetransmissionQueue = availableSequenceNumbers;
|
|
|
|
await m_monitoredItems.ApplyChangesAsync(true, false,
|
|
ct).ConfigureAwait(false);
|
|
StartKeepAliveTimer();
|
|
return true;
|
|
}
|
|
finally
|
|
{
|
|
m_stateLock.Release();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reset this loaded subscription back into the create-fresh
|
|
/// pipeline. Called by
|
|
/// <see cref="SubscriptionManager.RestoreAsync"/> when
|
|
/// <c>TransferSubscriptions</c> rejects the saved server-side
|
|
/// id (typically <c>BadSubscriptionIdInvalid</c>): the V2
|
|
/// engine then drops the loaded identifiers and creates a fresh
|
|
/// server-side subscription via the normal
|
|
/// <see cref="StateManagerAsync"/> path.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// Triggering links captured in the load state are intentionally
|
|
/// not replayed on recreate — the saved server-side triggering
|
|
/// relationships are tied to the (now-stale) server item ids.
|
|
/// Callers that want triggering preserved across a fallback
|
|
/// recreate must re-issue
|
|
/// <see cref="SetTriggeringAsync"/> after the items finish
|
|
/// re-creating.
|
|
/// </remarks>
|
|
internal async ValueTask ResetToRecreateAsync(CancellationToken ct)
|
|
{
|
|
await m_stateLock.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
|
{
|
|
Id = 0;
|
|
m_createdEvent.Reset();
|
|
CurrentPublishingInterval = TimeSpan.Zero;
|
|
CurrentKeepAliveCount = 0;
|
|
CurrentLifetimeCount = 0;
|
|
CurrentPublishingEnabled = false;
|
|
CurrentPriority = 0;
|
|
CurrentMaxNotificationsPerPublish = 0;
|
|
LastSequenceNumberProcessed = 0;
|
|
LastNotificationTimestamp = 0;
|
|
AvailableInRetransmissionQueue = [];
|
|
|
|
// Reset every loaded item so it queues a fresh
|
|
// CreateMonitoredItem on the next ApplyChanges pass.
|
|
foreach (IMonitoredItem item in m_monitoredItems.Items)
|
|
{
|
|
if (item is MonitoredItems.MonitoredItem mi)
|
|
{
|
|
mi.Reset();
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
m_stateLock.Release();
|
|
}
|
|
// Wake the state manager so the next pass observes
|
|
// !Created and runs CreateAsync.
|
|
m_stateControl.Set();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Wait until the subscription becomes <see cref="Created"/>
|
|
/// (the server has assigned an Id and
|
|
/// <see cref="OnSubscriptionUpdateComplete"/> has run).
|
|
/// Used by <see cref="SubscriptionManager.RestoreTransferAsync"/>
|
|
/// to await the recreate-fallback before returning to the
|
|
/// caller of <see cref="SubscriptionManager.LoadAsync"/>.
|
|
/// </summary>
|
|
/// <param name="ct"></param>
|
|
internal Task WaitForCreatedAsync(CancellationToken ct)
|
|
{
|
|
return m_createdEvent.WaitAsync(ct);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public override ValueTask OnPublishReceivedAsync(NotificationMessage message,
|
|
IReadOnlyList<uint>? availableSequenceNumbers,
|
|
IReadOnlyList<string> stringTable)
|
|
{
|
|
// Reset the keep alive timer
|
|
m_publishTimer.Change(m_keepAliveInterval, m_keepAliveInterval);
|
|
|
|
// send notification that publishing received a keep alive
|
|
// or has to republish.
|
|
if (PublishingStopped)
|
|
{
|
|
OnPublishStateChanged(PublishState.Recovered);
|
|
}
|
|
return base.OnPublishReceivedAsync(message, availableSequenceNumbers,
|
|
stringTable);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override async ValueTask DisposeAsync(bool disposing)
|
|
{
|
|
if (disposing && !Disposed)
|
|
{
|
|
try
|
|
{
|
|
await m_cts.CancelAsync().ConfigureAwait(false);
|
|
await m_stateManagement.ConfigureAwait(false);
|
|
|
|
await m_monitoredItems.DisposeAsync().ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
m_publishTimer.Dispose();
|
|
m_changeTracking?.Dispose();
|
|
m_cts.Dispose();
|
|
m_stateLock.Dispose();
|
|
}
|
|
}
|
|
await base.DisposeAsync(disposing).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public void Update()
|
|
{
|
|
m_stateControl.Set();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
public MonitoredItems.MonitoredItem CreateMonitoredItem(string name,
|
|
IOptionsMonitor<MonitoredItems.MonitoredItemOptions> options, IMonitoredItemContext context)
|
|
{
|
|
return CreateMonitoredItem(name, options, context, Observability);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Create monitored item
|
|
/// </summary>
|
|
/// <param name="name"></param>
|
|
/// <param name="options"></param>
|
|
/// <param name="context"></param>
|
|
/// <param name="telemetry"></param>
|
|
/// <returns></returns>
|
|
protected abstract MonitoredItems.MonitoredItem CreateMonitoredItem(string name,
|
|
IOptionsMonitor<MonitoredItems.MonitoredItemOptions> options, IMonitoredItemContext context,
|
|
ITelemetryContext telemetry);
|
|
|
|
/// <summary>
|
|
/// Called when the options changed
|
|
/// </summary>
|
|
/// <param name="options"></param>
|
|
protected virtual void OnOptionsChanged(SubscriptionOptions options)
|
|
{
|
|
SubscriptionOptions currentOptions = Options;
|
|
if (currentOptions == options)
|
|
{
|
|
return;
|
|
}
|
|
Options = options;
|
|
m_stateControl.Set();
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override ValueTask OnKeepAliveNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, PublishState publishStateMask)
|
|
{
|
|
return m_handler.OnKeepAliveNotificationAsync(this, sequenceNumber,
|
|
publishTime, publishStateMask);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override async ValueTask OnDataChangeNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, DataChangeNotification notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
try
|
|
{
|
|
await m_handler.OnDataChangeNotificationAsync(this, sequenceNumber,
|
|
publishTime, m_monitoredItems.CreateNotification(notification),
|
|
publishStateMask, stringTable).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
if (PoolNotifications)
|
|
{
|
|
ReuseDataChangeNotification(notification);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override async ValueTask OnEventDataNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, EventNotificationList notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
try
|
|
{
|
|
await m_handler.OnEventDataNotificationAsync(this, sequenceNumber,
|
|
publishTime, m_monitoredItems.CreateNotification(notification),
|
|
publishStateMask, stringTable).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
if (PoolNotifications)
|
|
{
|
|
ReuseEventNotificationList(notification);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Walk a dispatched <see cref="DataChangeNotification"/> after the
|
|
/// handler returns and release pooled payload instances back to
|
|
/// their activators. Items that do not implement
|
|
/// <see cref="IPooledEncodeable"/> are skipped silently — this
|
|
/// allows the walk to be safe before source-gen learns to emit
|
|
/// the interface for tagged types, and serves as the universal
|
|
/// dispatch path for any future pooled payload variant.
|
|
/// </summary>
|
|
private static void ReuseDataChangeNotification(DataChangeNotification notification)
|
|
{
|
|
ReadOnlySpan<MonitoredItemNotification> items =
|
|
notification.MonitoredItems.Span;
|
|
for (int i = 0; i < items.Length; i++)
|
|
{
|
|
if (items[i] is IPooledEncodeable item)
|
|
{
|
|
item.Reuse();
|
|
}
|
|
}
|
|
if (notification is IPooledEncodeable container)
|
|
{
|
|
container.Reuse();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Event-side companion to <see cref="ReuseDataChangeNotification"/>.
|
|
/// </summary>
|
|
private static void ReuseEventNotificationList(EventNotificationList notification)
|
|
{
|
|
ReadOnlySpan<EventFieldList> events = notification.Events.Span;
|
|
for (int i = 0; i < events.Length; i++)
|
|
{
|
|
if (events[i] is IPooledEncodeable evt)
|
|
{
|
|
evt.Reuse();
|
|
}
|
|
}
|
|
if (notification is IPooledEncodeable container)
|
|
{
|
|
container.Reuse();
|
|
}
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override ValueTask OnStatusChangeNotificationAsync(uint sequenceNumber,
|
|
DateTime publishTime, StatusChangeNotification notification,
|
|
PublishState publishStateMask, IReadOnlyList<string> stringTable)
|
|
{
|
|
// The base MessageProcessor does not automatically raise
|
|
// OnPublishStateChanged for status notifications, so do it
|
|
// here. This is what surfaces the Transferred / Timeout
|
|
// flags to user-visible publish-state handlers.
|
|
OnPublishStateChanged(publishStateMask);
|
|
|
|
// Recovery on unsolicited Good_SubscriptionTransferred.
|
|
// Per OPC UA Part 4 §5.14.7 this notification is sent to
|
|
// the *old* session when its subscription was transferred
|
|
// away — receiving it on a freshly created subscription
|
|
// here is a server quirk (e.g. Kepware leaking
|
|
// pre-restart state). Opt-in policy lets the caller ask
|
|
// for an in-place recreate instead of leaving the
|
|
// subscription dark.
|
|
if (notification.Status == StatusCodes.GoodSubscriptionTransferred &&
|
|
Options.RecoveryPolicy
|
|
.HasFlag(SubscriptionRecoveryPolicy.RecreateOnUnsolicitedTransfer) &&
|
|
Created &&
|
|
!Disposed &&
|
|
Interlocked.CompareExchange(
|
|
ref m_recreateAfterTransferInProgress, 1, 0) == 0)
|
|
{
|
|
_ = Task.Run(RecoverAfterUnsolicitedTransferAsync);
|
|
}
|
|
return default;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Run an in-place recreate on the same session after an
|
|
/// unsolicited <c>Good_SubscriptionTransferred</c>. Drops
|
|
/// queued acknowledgements for the dead subscription id
|
|
/// before invoking <see cref="ResetToRecreateAsync"/> so the
|
|
/// state-manager loop sees a coherent reset and the ack
|
|
/// queue cannot leak <c>BadSubscriptionIdInvalid</c>s on
|
|
/// servers that re-use subscription identifiers. Idempotent:
|
|
/// concurrent dispatches collapse through
|
|
/// <see cref="m_recreateAfterTransferInProgress"/>.
|
|
/// </summary>
|
|
private async Task RecoverAfterUnsolicitedTransferAsync()
|
|
{
|
|
uint deadId = Id;
|
|
try
|
|
{
|
|
Logger.LogWarning(
|
|
"{Subscription}: unsolicited Good_SubscriptionTransferred received — " +
|
|
"auto-recreating on the same session " +
|
|
"(RecoveryPolicy=RecreateOnUnsolicitedTransfer).",
|
|
this);
|
|
|
|
if (deadId != 0)
|
|
{
|
|
int dropped = AckQueue.DropPendingForSubscription(deadId);
|
|
if (dropped > 0)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: dropped {Count} stale acknowledgement(s) " +
|
|
"before recovery recreate.",
|
|
this,
|
|
dropped);
|
|
}
|
|
}
|
|
|
|
if (Disposed)
|
|
{
|
|
return;
|
|
}
|
|
|
|
await ResetToRecreateAsync(CancellationToken.None)
|
|
.ConfigureAwait(false);
|
|
|
|
Logger.LogInformation(
|
|
"{Subscription}: recreate signalled after unsolicited " +
|
|
"Good_SubscriptionTransferred (old SubscriptionId={OldId}).",
|
|
this,
|
|
deadId);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Logger.LogWarning(
|
|
ex,
|
|
"{Subscription}: recovery after unsolicited " +
|
|
"Good_SubscriptionTransferred failed (old SubscriptionId={OldId}); " +
|
|
"the subscription stays dark until the next reconnect or " +
|
|
"manual recreate.",
|
|
this,
|
|
deadId);
|
|
}
|
|
finally
|
|
{
|
|
Interlocked.Exchange(ref m_recreateAfterTransferInProgress, 0);
|
|
}
|
|
}
|
|
|
|
private int m_recreateAfterTransferInProgress;
|
|
|
|
/// <summary>
|
|
/// Called when the subscription state changed
|
|
/// </summary>
|
|
/// <param name="state"></param>
|
|
protected virtual void OnSubscriptionStateChanged(SubscriptionState state)
|
|
{
|
|
Logger.LogInformation("{Subscription}: {State}.", this, state);
|
|
FireStateChangedToHandler(state, default);
|
|
}
|
|
|
|
/// <inheritdoc/>
|
|
protected override void OnPublishStateChanged(PublishState stateMask)
|
|
{
|
|
base.OnPublishStateChanged(stateMask);
|
|
FireStateChangedToHandler(default, stateMask);
|
|
}
|
|
|
|
private void FireStateChangedToHandler(SubscriptionState state,
|
|
PublishState publishStateMask)
|
|
{
|
|
try
|
|
{
|
|
// Fire-and-forget: handler is allowed to block, but we
|
|
// intentionally don't await here because OnSubscriptionStateChanged
|
|
// is invoked from inside StateManagerAsync (under m_stateLock)
|
|
// and OnPublishStateChanged is invoked from the publish dispatch
|
|
// path. Either await would risk a deadlock if the handler
|
|
// re-enters the engine. Handlers that need backpressure
|
|
// should buffer in OnSubscriptionStateChangedAsync and
|
|
// process on a worker.
|
|
_ = m_handler.OnSubscriptionStateChangedAsync(this, state,
|
|
publishStateMask).AsTask();
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Logger.LogWarning(ex,
|
|
"{Subscription}: OnSubscriptionStateChangedAsync handler threw.",
|
|
this);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Controls the state changes of the subscriptions and the contained monitored
|
|
/// items.
|
|
/// </summary>
|
|
/// <param name="ct"></param>
|
|
/// <returns></returns>
|
|
private async Task StateManagerAsync(CancellationToken ct)
|
|
{
|
|
OnSubscriptionStateChanged(SubscriptionState.Opened);
|
|
try
|
|
{
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
await m_stateControl.WaitAsync(ct).ConfigureAwait(false);
|
|
await m_stateLock.WaitAsync(ct).ConfigureAwait(false);
|
|
SubscriptionOptions options = Options;
|
|
try
|
|
{
|
|
while (!ct.IsCancellationRequested)
|
|
{
|
|
if (options.Disabled)
|
|
{
|
|
await DeleteAsync(ct).ConfigureAwait(false);
|
|
break; // Wait for changes while disabled
|
|
}
|
|
|
|
if (!Created)
|
|
{
|
|
await CreateAsync(options, ct).ConfigureAwait(false);
|
|
// Run the post-create hook exactly
|
|
// once per partition lifetime to
|
|
// satisfy ordering-sensitive callers
|
|
// (e.g. SetSubscriptionDurable, which
|
|
// per OPC UA Part 4 §5.13.9 must
|
|
// precede any monitored-item
|
|
// creation). Clear after invocation
|
|
// so modify cycles do not re-run it;
|
|
// the owning wrapper re-installs the
|
|
// hook before the next Create pass.
|
|
Func<CancellationToken, ValueTask>? hook
|
|
= Interlocked.Exchange(
|
|
ref m_onAfterCreateAsync, null);
|
|
if (hook != null)
|
|
{
|
|
try
|
|
{
|
|
await hook(ct).ConfigureAwait(false);
|
|
}
|
|
catch (Exception hookEx)
|
|
{
|
|
Logger.LogWarning(hookEx,
|
|
"{Subscription}: OnAfterCreateAsync " +
|
|
"hook threw; partition continues but " +
|
|
"the post-create operation did not " +
|
|
"complete.", this);
|
|
OnSubscriptionStateChanged(
|
|
SubscriptionState.Modified);
|
|
}
|
|
}
|
|
}
|
|
else
|
|
{
|
|
await ModifyAsync(options, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
bool modified = await m_monitoredItems.ApplyChangesAsync(
|
|
false, false, ct).ConfigureAwait(false);
|
|
if (modified)
|
|
{
|
|
OnSubscriptionStateChanged(SubscriptionState.Modified);
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
|
{
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Logger.LogError(ex, "Failed to apply subscription changes.");
|
|
}
|
|
finally
|
|
{
|
|
m_stateLock.Release();
|
|
}
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
}
|
|
|
|
// Delete subscription on server on dispose
|
|
await DeleteAsync(default).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Deletes a subscription on the server, but keeps the subscription in the session.
|
|
/// </summary>
|
|
/// <param name="ct"></param>
|
|
///
|
|
/// <exception cref="ServiceResultException"></exception>
|
|
internal async ValueTask DeleteAsync(CancellationToken ct)
|
|
{
|
|
// nothing to do if not created.
|
|
if (!Created)
|
|
{
|
|
return;
|
|
}
|
|
try
|
|
{
|
|
// delete the subscription.
|
|
ArrayOf<uint> subscriptionIds = new uint[] { Id };
|
|
DeleteSubscriptionsResponse response = await m_context.SubscriptionServiceSet.DeleteSubscriptionsAsync(null,
|
|
subscriptionIds, ct).ConfigureAwait(false);
|
|
// validate response.
|
|
ClientBase.ValidateResponse(response.Results, subscriptionIds);
|
|
ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, subscriptionIds);
|
|
|
|
if (StatusCode.IsBad(response.Results[0]))
|
|
{
|
|
throw new ServiceResultException(ClientBase.GetResult(
|
|
response.Results[0], 0, response.DiagnosticInfos,
|
|
response.ResponseHeader));
|
|
}
|
|
}
|
|
// suppress exception if silent flag is set.
|
|
catch (Exception e)
|
|
{
|
|
Logger.LogInformation(e, "Deleting subscription on server failed.");
|
|
}
|
|
OnSubscriptionDeleteCompleted();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates a subscription on the server and adds all monitored items.
|
|
/// </summary>
|
|
/// <param name="options"></param>
|
|
/// <param name="ct"></param>
|
|
internal async ValueTask CreateAsync(SubscriptionOptions options, CancellationToken ct)
|
|
{
|
|
// create the subscription.
|
|
AdjustCounts(options, out uint revisedMaxKeepAliveCount, out uint revisedLifetimeCount);
|
|
|
|
CreateSubscriptionResponse response = await m_context.SubscriptionServiceSet.CreateSubscriptionAsync(null,
|
|
options.PublishingInterval.TotalMilliseconds, revisedLifetimeCount,
|
|
revisedMaxKeepAliveCount, options.MaxNotificationsPerPublish,
|
|
options.PublishingEnabled, options.Priority, ct).ConfigureAwait(false);
|
|
|
|
OnSubscriptionUpdateComplete(true, response.SubscriptionId,
|
|
TimeSpan.FromMilliseconds(response.RevisedPublishingInterval),
|
|
response.RevisedMaxKeepAliveCount, response.RevisedLifetimeCount,
|
|
options.Priority, options.MaxNotificationsPerPublish,
|
|
options.PublishingEnabled);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Modifies a subscription on the server.
|
|
/// </summary>
|
|
/// <param name="options"></param>
|
|
/// <param name="ct"></param>
|
|
/// <exception cref="ServiceResultException"></exception>
|
|
internal async ValueTask ModifyAsync(SubscriptionOptions options, CancellationToken ct)
|
|
{
|
|
// modify the subscription.
|
|
AdjustCounts(options, out uint revisedMaxKeepAliveCount, out uint revisedLifetimeCount);
|
|
|
|
if (revisedMaxKeepAliveCount != CurrentKeepAliveCount ||
|
|
revisedLifetimeCount != CurrentLifetimeCount ||
|
|
options.Priority != CurrentPriority ||
|
|
options.MaxNotificationsPerPublish != CurrentMaxNotificationsPerPublish ||
|
|
options.PublishingInterval != CurrentPublishingInterval)
|
|
{
|
|
ModifySubscriptionResponse response = await m_context.SubscriptionServiceSet.ModifySubscriptionAsync(null, Id,
|
|
options.PublishingInterval.TotalMilliseconds, revisedLifetimeCount,
|
|
revisedMaxKeepAliveCount, options.MaxNotificationsPerPublish, options.Priority,
|
|
ct).ConfigureAwait(false);
|
|
|
|
if (options.PublishingEnabled != CurrentPublishingEnabled)
|
|
{
|
|
await SetPublishingModeAsync(options, ct).ConfigureAwait(false);
|
|
}
|
|
|
|
OnSubscriptionUpdateComplete(false, 0,
|
|
TimeSpan.FromMilliseconds(response.RevisedPublishingInterval),
|
|
response.RevisedMaxKeepAliveCount, response.RevisedLifetimeCount,
|
|
options.Priority, options.MaxNotificationsPerPublish,
|
|
options.PublishingEnabled);
|
|
}
|
|
else if (options.PublishingEnabled != CurrentPublishingEnabled)
|
|
{
|
|
await SetPublishingModeAsync(options, ct).ConfigureAwait(false);
|
|
|
|
// update current state.
|
|
CurrentPublishingEnabled = options.PublishingEnabled;
|
|
OnSubscriptionStateChanged(SubscriptionState.Modified);
|
|
}
|
|
|
|
async Task SetPublishingModeAsync(SubscriptionOptions options, CancellationToken ct)
|
|
{
|
|
// modify the subscription.
|
|
ArrayOf<uint> subscriptionIds = new uint[] { Id };
|
|
SetPublishingModeResponse response = await m_context.SubscriptionServiceSet.SetPublishingModeAsync(
|
|
null, options.PublishingEnabled, subscriptionIds, ct).ConfigureAwait(false);
|
|
|
|
// validate response.
|
|
ClientBase.ValidateResponse(response.Results, subscriptionIds);
|
|
ClientBase.ValidateDiagnosticInfos(response.DiagnosticInfos, subscriptionIds);
|
|
|
|
if (StatusCode.IsBad(response.Results[0]))
|
|
{
|
|
throw new ServiceResultException(
|
|
ClientBase.GetResult(response.Results[0], 0,
|
|
response.DiagnosticInfos, response.ResponseHeader));
|
|
}
|
|
|
|
Logger.LogInformation(
|
|
"{Subscription}: Modified - Publishing is now {New}.",
|
|
this, options.PublishingEnabled ? "Enabled" : "Disabled");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Update the subscription with the given revised settings.
|
|
/// </summary>
|
|
/// <param name="created"></param>
|
|
/// <param name="subscriptionId"></param>
|
|
/// <param name="revisedPublishingInterval"></param>
|
|
/// <param name="revisedKeepAliveCount"></param>
|
|
/// <param name="revisedLifetimeCount"></param>
|
|
/// <param name="priority"></param>
|
|
/// <param name="maxNotificationsPerPublish"></param>
|
|
/// <param name="publishingEnabled"></param>
|
|
internal void OnSubscriptionUpdateComplete(bool created,
|
|
uint subscriptionId, TimeSpan revisedPublishingInterval,
|
|
uint revisedKeepAliveCount, uint revisedLifetimeCount,
|
|
byte priority, uint maxNotificationsPerPublish,
|
|
bool publishingEnabled)
|
|
{
|
|
if (CurrentPublishingEnabled != publishingEnabled)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Created - Publishing is {New}.",
|
|
this, publishingEnabled ? "Enabled" : "Disabled");
|
|
CurrentPublishingEnabled = publishingEnabled;
|
|
}
|
|
|
|
if (CurrentKeepAliveCount != revisedKeepAliveCount)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Changed KeepAliveCount to {New}.",
|
|
this, revisedKeepAliveCount);
|
|
|
|
CurrentKeepAliveCount = revisedKeepAliveCount;
|
|
}
|
|
|
|
if (CurrentPublishingInterval != revisedPublishingInterval)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Changed PublishingInterval to {New}.",
|
|
this, revisedPublishingInterval);
|
|
CurrentPublishingInterval = revisedPublishingInterval;
|
|
}
|
|
|
|
if (CurrentMaxNotificationsPerPublish != maxNotificationsPerPublish)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Change MaxNotificationsPerPublish to {New}.",
|
|
this, maxNotificationsPerPublish);
|
|
CurrentMaxNotificationsPerPublish = maxNotificationsPerPublish;
|
|
}
|
|
|
|
if (CurrentLifetimeCount != revisedLifetimeCount)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Changed LifetimeCount to {New}.",
|
|
this, revisedLifetimeCount);
|
|
CurrentLifetimeCount = revisedLifetimeCount;
|
|
}
|
|
|
|
if (CurrentPriority != priority)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Changed Priority to {New}.",
|
|
this, priority);
|
|
CurrentPriority = priority;
|
|
}
|
|
|
|
if (created)
|
|
{
|
|
Id = subscriptionId;
|
|
StartKeepAliveTimer();
|
|
NotifyManagerOfCreation();
|
|
m_createdEvent.Set();
|
|
}
|
|
|
|
// Notify all monitored items of the changes
|
|
SubscriptionState state = created ?
|
|
SubscriptionState.Created : SubscriptionState.Modified;
|
|
m_monitoredItems.OnSubscriptionStateChange(state, CurrentPublishingInterval);
|
|
OnSubscriptionStateChanged(state);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Delete the subscription.
|
|
/// Ignore errors, always reset all parameter.
|
|
/// </summary>
|
|
internal void OnSubscriptionDeleteCompleted()
|
|
{
|
|
LastSequenceNumberProcessed = 0;
|
|
LastNotificationTimestamp = 0;
|
|
|
|
Id = 0;
|
|
m_createdEvent.Reset();
|
|
CurrentPublishingInterval = TimeSpan.Zero;
|
|
CurrentKeepAliveCount = 0;
|
|
CurrentPublishingEnabled = false;
|
|
CurrentPriority = 0;
|
|
|
|
m_deletedItems.Clear();
|
|
|
|
// Notify all monitored items of the changes
|
|
m_monitoredItems.OnSubscriptionStateChange(SubscriptionState.Deleted,
|
|
CurrentPublishingInterval);
|
|
OnSubscriptionStateChanged(SubscriptionState.Deleted);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Stop the keep alive timer for the subscription.
|
|
/// </summary>
|
|
private void StopKeepAliveTimer()
|
|
{
|
|
m_publishTimer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Starts a timer to ensure publish requests are sent frequently
|
|
/// enough to detect network interruptions.
|
|
/// </summary>
|
|
private void StartKeepAliveTimer()
|
|
{
|
|
SubscriptionOptions options = Options;
|
|
LastNotificationTimestamp = TimeProvider.GetTimestamp();
|
|
m_keepAliveInterval = CurrentPublishingInterval.Multiply(CurrentKeepAliveCount + 1);
|
|
if (m_keepAliveInterval < s_minKeepAliveTimerInterval)
|
|
{
|
|
m_keepAliveInterval = options.PublishingInterval.Multiply(options.KeepAliveCount + 1);
|
|
}
|
|
if (m_keepAliveInterval > Timeout.InfiniteTimeSpan)
|
|
{
|
|
m_keepAliveInterval = Timeout.InfiniteTimeSpan;
|
|
}
|
|
if (m_keepAliveInterval < s_minKeepAliveTimerInterval)
|
|
{
|
|
m_keepAliveInterval = s_minKeepAliveTimerInterval;
|
|
}
|
|
m_publishTimer.Change(m_keepAliveInterval, m_keepAliveInterval);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Checks if a notification has arrived in time.
|
|
/// </summary>
|
|
/// <param name="state"></param>
|
|
private void OnKeepAlive(object? state)
|
|
{
|
|
// check if a publish has arrived.
|
|
if (PublishingStopped)
|
|
{
|
|
Interlocked.Increment(ref m_publishLateCount);
|
|
OnPublishStateChanged(PublishState.Stopped);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Ensures sensible values for the counts.
|
|
/// </summary>
|
|
/// <param name="options"></param>
|
|
/// <param name="keepAliveCount"></param>
|
|
/// <param name="lifetimeCount"></param>
|
|
private void AdjustCounts(SubscriptionOptions options, out uint keepAliveCount,
|
|
out uint lifetimeCount)
|
|
{
|
|
const uint kDefaultKeepAlive = 10;
|
|
const uint kDefaultLifeTime = 1000;
|
|
|
|
keepAliveCount = options.KeepAliveCount;
|
|
lifetimeCount = options.LifetimeCount;
|
|
// keep alive count must be at least 1, 10 is a good default.
|
|
if (keepAliveCount == 0)
|
|
{
|
|
Logger.LogInformation("{Subscription}: Adjusted KeepAliveCount " +
|
|
"from {Old} to {New}.", this, keepAliveCount, kDefaultKeepAlive);
|
|
keepAliveCount = kDefaultKeepAlive;
|
|
}
|
|
|
|
// ensure the lifetime is sensible given the sampling interval.
|
|
if (options.PublishingInterval > TimeSpan.Zero)
|
|
{
|
|
if (options.MinLifetimeInterval > TimeSpan.Zero &&
|
|
options.MinLifetimeInterval < m_context.SessionTimeout)
|
|
{
|
|
Logger.LogWarning(
|
|
"{Subscription}: A smaller minimum LifetimeInterval " +
|
|
"{Counter}ms than session timeout {Timeout}ms configured.",
|
|
this, options.MinLifetimeInterval, m_context.SessionTimeout);
|
|
}
|
|
|
|
uint minLifetimeInterval = (uint)options.MinLifetimeInterval.TotalMilliseconds;
|
|
uint publishingInterval = (uint)options.PublishingInterval.TotalMilliseconds;
|
|
uint minLifetimeCount = minLifetimeInterval / publishingInterval;
|
|
if (lifetimeCount < minLifetimeCount)
|
|
{
|
|
lifetimeCount = minLifetimeCount;
|
|
|
|
if (minLifetimeInterval % publishingInterval != 0)
|
|
{
|
|
lifetimeCount++;
|
|
}
|
|
Logger.LogInformation(
|
|
"{Subscription}: Adjusted LifetimeCount to value={New}.",
|
|
this, lifetimeCount);
|
|
}
|
|
|
|
if (options.PublishingInterval.Multiply(lifetimeCount) < m_context.SessionTimeout)
|
|
{
|
|
Logger.LogWarning(
|
|
"{Subscription}: Lifetime {LifeTime}ms configured is less " +
|
|
"than session timeout {Timeout}ms.", this,
|
|
options.PublishingInterval.Multiply(lifetimeCount), m_context.SessionTimeout);
|
|
}
|
|
}
|
|
else if (lifetimeCount == 0)
|
|
{
|
|
// don't know what the sampling interval will be - use something large
|
|
// enough to ensure the user does not experience unexpected drop outs.
|
|
Logger.LogInformation(
|
|
"{Subscription}: Adjusted LifetimeCount from {Old} to {New}. ",
|
|
this, lifetimeCount, kDefaultLifeTime);
|
|
lifetimeCount = kDefaultLifeTime;
|
|
}
|
|
|
|
// validate spec: lifetimecount shall be at least 3*keepAliveCount
|
|
uint minLifeTimeCount = 3 * keepAliveCount;
|
|
if (lifetimeCount < minLifeTimeCount)
|
|
{
|
|
Logger.LogInformation(
|
|
"{Subscription}: Adjusted LifetimeCount from {Old} to {New}.",
|
|
this, lifetimeCount, minLifeTimeCount);
|
|
lifetimeCount = minLifeTimeCount;
|
|
}
|
|
}
|
|
|
|
private static readonly TimeSpan s_minKeepAliveTimerInterval = TimeSpan.FromSeconds(1);
|
|
private static readonly TimeSpan s_keepAliveTimerMargin = TimeSpan.FromSeconds(1);
|
|
private TimeSpan m_keepAliveInterval;
|
|
private int m_publishLateCount;
|
|
private Func<CancellationToken, ValueTask>? m_onAfterCreateAsync;
|
|
private readonly AsyncAutoResetEvent m_stateControl = new();
|
|
private readonly AsyncManualResetEvent m_createdEvent = new();
|
|
private readonly CancellationTokenSource m_cts = new();
|
|
private readonly Task m_stateManagement;
|
|
private readonly SemaphoreSlim m_stateLock = new(1, 1);
|
|
private readonly List<uint> m_deletedItems = [];
|
|
private readonly ITimer m_publishTimer;
|
|
private readonly IDisposable? m_changeTracking;
|
|
private readonly ISubscriptionNotificationHandler m_handler;
|
|
private readonly ISubscriptionContext m_context;
|
|
private readonly MonitoredItemManager m_monitoredItems;
|
|
}
|
|
}
|