mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-29 15:11:53 +08:00
Client: install the durable hook before a new partition creates itself
A secondary partition minted on demand starts its state machine in its constructor, but the partition factory installed the durable OnAfterCreateAsync hook only after the constructor returned. When the server answered CreateSubscription first, the partition was created without SetSubscriptionDurable and its monitored items were created on a non-durable subscription. The internal Subscription constructor and ISubscriptionManagerContext.CreateSubscription now take an optional initializer that runs before the state machine starts. SubscriptionManager uses it to attach the reactive fallback and the durable hook to on-demand partitions.
This commit is contained in:
Wolfgang Colsman
@@ -252,13 +252,14 @@ namespace Opc.Ua.Client
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<Subscriptions.SubscriptionOptions> options,
|
||||
IMessageAckQueue queue,
|
||||
SubscriptionLoadState? loadState = null)
|
||||
SubscriptionLoadState? loadState = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
{
|
||||
var subscriptionContext =
|
||||
new SubscriptionContextAdapter(m_context);
|
||||
return new DefaultSubscription(
|
||||
subscriptionContext, handler, queue,
|
||||
options, m_context.Telemetry, loadState, m_timeProvider);
|
||||
options, m_context.Telemetry, loadState, m_timeProvider, initialize);
|
||||
}
|
||||
|
||||
/// <inheritdoc/>
|
||||
@@ -599,9 +600,10 @@ namespace Opc.Ua.Client
|
||||
IOptionsMonitor<Subscriptions.SubscriptionOptions> options,
|
||||
ITelemetryContext telemetry,
|
||||
SubscriptionLoadState? loadState = null,
|
||||
TimeProvider? timeProvider = null)
|
||||
TimeProvider? timeProvider = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
: base(context, handler, completion,
|
||||
options, telemetry, loadState, timeProvider)
|
||||
options, telemetry, loadState, timeProvider, initialize)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
@@ -27,6 +27,7 @@
|
||||
* http://opcfoundation.org/License/MIT/1.00/
|
||||
* ======================================================================*/
|
||||
|
||||
using System;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.Options;
|
||||
@@ -63,12 +64,17 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
/// server-side identifiers and pre-populated with the saved
|
||||
/// monitored items; the caller is responsible for issuing the
|
||||
/// take-over via <c>TransferSubscriptions</c>.</param>
|
||||
/// <param name="initialize">Optional callback that must be
|
||||
/// invoked with the new subscription before it starts its first
|
||||
/// create pass, so hooks it installs are in place before
|
||||
/// <c>CreateSubscription</c> can complete on the server.</param>
|
||||
/// <returns></returns>
|
||||
IManagedSubscription CreateSubscription(
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
IMessageAckQueue queue,
|
||||
SubscriptionLoadState? loadState = null);
|
||||
SubscriptionLoadState? loadState = null,
|
||||
Action<IManagedSubscription>? initialize = null);
|
||||
|
||||
/// <summary>
|
||||
/// Publish service
|
||||
|
||||
@@ -197,6 +197,10 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
/// <param name="timeProvider">Optional <see cref="TimeProvider"/>
|
||||
/// for elapsed-time and timer calculations. Defaults to
|
||||
/// <see cref="TimeProvider.System"/> when <c>null</c>.</param>
|
||||
/// <param name="initialize">Optional callback invoked with the
|
||||
/// new subscription before its state machine starts, so hooks
|
||||
/// such as <see cref="OnAfterCreateAsync"/> are in place before
|
||||
/// the first <c>CreateSubscription</c> can complete.</param>
|
||||
protected Subscription(
|
||||
ISubscriptionContext context,
|
||||
ISubscriptionNotificationHandler handler,
|
||||
@@ -204,7 +208,8 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
ITelemetryContext telemetry,
|
||||
SubscriptionLoadState? loadState = null,
|
||||
TimeProvider? timeProvider = null)
|
||||
TimeProvider? timeProvider = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
: base(context.SubscriptionServiceSet, completion, telemetry, timeProvider)
|
||||
{
|
||||
m_handler = handler;
|
||||
@@ -242,6 +247,7 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
m_changeTracking = options.OnChange((o, _) => OnOptionsChanged(o));
|
||||
m_stateControl.Set();
|
||||
}
|
||||
initialize?.Invoke(this);
|
||||
m_stateManagement = StateManagerAsync(m_cts.Token);
|
||||
}
|
||||
|
||||
|
||||
@@ -611,13 +611,8 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
// 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;
|
||||
};
|
||||
partitionFactory = () => MintPartition(effectiveHandler, options,
|
||||
part => InitializePartition(part, created));
|
||||
wrapper = new LogicalSubscription(primary, policy, partitionFactory,
|
||||
m_timeProvider,
|
||||
NormalizeIdleTimeout(snapshot.SecondaryPartitionIdleTimeout),
|
||||
@@ -673,6 +668,22 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wire a freshly constructed secondary partition to its owning
|
||||
/// logical wrapper. Runs before the partition starts its first
|
||||
/// create pass, so a durable logical subscription sends
|
||||
/// <c>SetSubscriptionDurable</c> on the partition even when the
|
||||
/// server answers <c>CreateSubscription</c> before the partition
|
||||
/// factory returns.
|
||||
/// </summary>
|
||||
private static void InitializePartition(
|
||||
IManagedSubscription partition,
|
||||
LogicalSubscription? wrapper)
|
||||
{
|
||||
AttachReactiveFallback(partition, wrapper);
|
||||
wrapper?.TryApplyDurableToNewPartition(partition);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Normalize a caller-supplied
|
||||
/// <see cref="SubscriptionOptions.SecondaryPartitionIdleTimeout"/>
|
||||
@@ -786,9 +797,10 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
/// </summary>
|
||||
private IManagedSubscription MintPartition(
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<SubscriptionOptions> options)
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
Action<IManagedSubscription> initialize)
|
||||
{
|
||||
return MintPartitionCore(handler, options, loadState: null);
|
||||
return MintPartitionCore(handler, options, loadState: null, initialize);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -805,16 +817,17 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
SubscriptionLoadState loadState)
|
||||
{
|
||||
return MintPartitionCore(handler, options, loadState);
|
||||
return MintPartitionCore(handler, options, loadState, initialize: null);
|
||||
}
|
||||
|
||||
private IManagedSubscription MintPartitionCore(
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
SubscriptionLoadState? loadState)
|
||||
SubscriptionLoadState? loadState,
|
||||
Action<IManagedSubscription>? initialize)
|
||||
{
|
||||
IManagedSubscription partition = m_session.CreateSubscription(
|
||||
handler, options, this, loadState);
|
||||
handler, options, this, loadState, initialize);
|
||||
lock (m_subscriptionLock)
|
||||
{
|
||||
if (!m_subscriptions.Add(partition))
|
||||
@@ -972,13 +985,8 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
options.MaxMonitoredItemsPerPartition ?? 0,
|
||||
options.MaxPartitionCount);
|
||||
LogicalSubscription? created = null;
|
||||
partitionFactory = () =>
|
||||
{
|
||||
IManagedSubscription part = MintPartition(effectiveHandler, optionsMonitor);
|
||||
AttachReactiveFallback(part, created);
|
||||
created?.TryApplyDurableToNewPartition(part);
|
||||
return part;
|
||||
};
|
||||
partitionFactory = () => MintPartition(effectiveHandler, optionsMonitor,
|
||||
part => InitializePartition(part, created));
|
||||
wrapper = new LogicalSubscription(subscription, policy, partitionFactory,
|
||||
m_timeProvider,
|
||||
NormalizeIdleTimeout(options.SecondaryPartitionIdleTimeout),
|
||||
|
||||
@@ -74,6 +74,17 @@ namespace Opc.Ua.Client.Subscriptions.Fakes
|
||||
= (_, _, _) => throw new InvalidOperationException(
|
||||
"CreateSubscriptionFactory not set on FakeSubscriptionManagerContext.");
|
||||
|
||||
/// <summary>
|
||||
/// Optional factory that receives the initializer of
|
||||
/// <see cref="CreateSubscription"/> and must run it before the
|
||||
/// subscription starts. When null, <see cref="CreateSubscriptionFactory"/>
|
||||
/// is used and the initializer runs after it returns.
|
||||
/// </summary>
|
||||
public Func<ISubscriptionNotificationHandler,
|
||||
IOptionsMonitor<SubscriptionOptions>, IMessageAckQueue,
|
||||
Action<IManagedSubscription>?, IManagedSubscription>? CreateInitializedSubscriptionFactory
|
||||
{ get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Optional override for <see cref="PublishAsync"/>. If null,
|
||||
/// returns a default <see cref="PublishResponse"/>.
|
||||
@@ -100,11 +111,18 @@ namespace Opc.Ua.Client.Subscriptions.Fakes
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
IMessageAckQueue queue,
|
||||
SubscriptionLoadState? loadState = null)
|
||||
SubscriptionLoadState? loadState = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
{
|
||||
Record(m_createSubscriptionCalls,
|
||||
new CreateSubscriptionCall(handler, options, queue, loadState));
|
||||
return CreateSubscriptionFactory(handler, options, queue);
|
||||
if (CreateInitializedSubscriptionFactory != null)
|
||||
{
|
||||
return CreateInitializedSubscriptionFactory(handler, options, queue, initialize);
|
||||
}
|
||||
IManagedSubscription subscription = CreateSubscriptionFactory(handler, options, queue);
|
||||
initialize?.Invoke(subscription);
|
||||
return subscription;
|
||||
}
|
||||
|
||||
public ValueTask<PublishResponse> PublishAsync(
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
/* ========================================================================
|
||||
* 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.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Moq;
|
||||
using NUnit.Framework;
|
||||
using Opc.Ua.Client.Subscriptions.Fakes;
|
||||
using Opc.Ua.Tests;
|
||||
|
||||
namespace Opc.Ua.Client.Subscriptions
|
||||
{
|
||||
public sealed partial class SubscriptionTests
|
||||
{
|
||||
[Test]
|
||||
[CancelAfter(30_000)]
|
||||
public async Task DurableIsSetOnNewPartitionCreatedBeforeFactoryReturnsAsync(
|
||||
CancellationToken ct)
|
||||
{
|
||||
var calls = new ConcurrentQueue<string>();
|
||||
SetupDurablePartitionServer(calls);
|
||||
int created = 0;
|
||||
var context = new FakeSubscriptionManagerContext
|
||||
{
|
||||
CreateInitializedSubscriptionFactory = (handler, _, queue, initialize) =>
|
||||
{
|
||||
var subscription = new TestSubscription(
|
||||
m_session, handler, queue, m_options, m_telemetry, initialize: initialize);
|
||||
if (Interlocked.Increment(ref created) == 2)
|
||||
{
|
||||
// The server answers CreateSubscription of the new
|
||||
// partition before the partition factory returns.
|
||||
Assert.That(SpinWait.SpinUntil(() => subscription.Created,
|
||||
TimeSpan.FromSeconds(10)), Is.True);
|
||||
}
|
||||
return subscription;
|
||||
}
|
||||
};
|
||||
m_options.Configure(_ => TestSubscription.SubscriptionOptions with
|
||||
{
|
||||
MaxMonitoredItemsPerPartition = 1
|
||||
});
|
||||
await using var manager = new SubscriptionManager(
|
||||
context, m_telemetry.LoggerFactory, DiagnosticsMasks.None);
|
||||
ISubscription subscription = manager.Add(m_mockNotificationDataHandler.Object, m_options);
|
||||
await WaitForAsync(() => subscription.Created, TimeSpan.FromSeconds(10))
|
||||
.ConfigureAwait(false);
|
||||
Assert.That(subscription.Created, Is.True);
|
||||
|
||||
TimeSpan revised = await subscription.SetAsDurableAsync(TimeSpan.FromHours(10), ct)
|
||||
.ConfigureAwait(false);
|
||||
Assert.That(revised, Is.EqualTo(TimeSpan.FromHours(10)));
|
||||
Assert.That(subscription.MonitoredItems.TryAdd("First",
|
||||
OptionsFactory.Create(TestMonitoredItem.CreatedOptions), out _), Is.True);
|
||||
await WaitForAsync(() => calls.Contains("items:22"), TimeSpan.FromSeconds(10))
|
||||
.ConfigureAwait(false);
|
||||
|
||||
Assert.That(subscription.MonitoredItems.TryAdd("Second",
|
||||
OptionsFactory.Create(TestMonitoredItem.CreatedOptions), out _), Is.True);
|
||||
await WaitForAsync(() => calls.Contains("items:23"), TimeSpan.FromSeconds(10))
|
||||
.ConfigureAwait(false);
|
||||
|
||||
Assert.That(((IPartitionedSubscription)subscription).PartitionCount, Is.EqualTo(2));
|
||||
Assert.That(calls, Is.EqualTo(s_durableOnBothPartitions));
|
||||
}
|
||||
|
||||
[Test]
|
||||
[CancelAfter(10_000)]
|
||||
public async Task InitializeRunsBeforeTheFirstCreatePassAsync(CancellationToken ct)
|
||||
{
|
||||
var calls = new ConcurrentQueue<string>();
|
||||
SetupDurablePartitionServer(calls);
|
||||
m_options.Configure(_ => TestSubscription.SubscriptionOptions);
|
||||
bool createdWhenInitialized = true;
|
||||
var afterCreate = new TaskCompletionSource<bool>(
|
||||
TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
await using var sut = new TestSubscription(
|
||||
m_session, m_mockNotificationDataHandler.Object, m_completion, m_options, m_telemetry,
|
||||
initialize: subscription =>
|
||||
{
|
||||
createdWhenInitialized = subscription.Created;
|
||||
((Subscription)subscription).OnAfterCreateAsync = _ =>
|
||||
{
|
||||
afterCreate.TrySetResult(true);
|
||||
return default;
|
||||
};
|
||||
});
|
||||
|
||||
await afterCreate.Task.WaitAsync(ct).ConfigureAwait(false);
|
||||
Assert.That(createdWhenInitialized, Is.False);
|
||||
Assert.That(sut.Id, Is.EqualTo(22));
|
||||
Assert.That(calls, Is.EqualTo(s_createOnly));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Server double that answers every call immediately, assigns
|
||||
/// subscription ids 22, 23, ... and records CreateSubscription,
|
||||
/// SetSubscriptionDurable (with the requested hours) and
|
||||
/// CreateMonitoredItems in call order.
|
||||
/// </summary>
|
||||
private void SetupDurablePartitionServer(ConcurrentQueue<string> calls)
|
||||
{
|
||||
int nextSubscriptionId = 21;
|
||||
int nextItemId = 100;
|
||||
m_mockSubscriptionServices
|
||||
.Setup(s => s.CreateSubscriptionAsync(
|
||||
It.IsAny<RequestHeader>(), It.IsAny<double>(), It.IsAny<uint>(),
|
||||
It.IsAny<uint>(), It.IsAny<uint>(), It.IsAny<bool>(),
|
||||
It.IsAny<byte>(), It.IsAny<CancellationToken>()))
|
||||
.Returns((RequestHeader _, double interval, uint lifetimeCount, uint keepAliveCount,
|
||||
uint _, bool _, byte _, CancellationToken _) =>
|
||||
{
|
||||
uint id = (uint)Interlocked.Increment(ref nextSubscriptionId);
|
||||
calls.Enqueue($"create:{id}");
|
||||
return new ValueTask<CreateSubscriptionResponse>(new CreateSubscriptionResponse
|
||||
{
|
||||
SubscriptionId = id,
|
||||
RevisedPublishingInterval = interval,
|
||||
RevisedLifetimeCount = lifetimeCount,
|
||||
RevisedMaxKeepAliveCount = keepAliveCount
|
||||
});
|
||||
});
|
||||
m_mockSubscriptionServices
|
||||
.Setup(s => s.DeleteSubscriptionsAsync(
|
||||
It.IsAny<RequestHeader>(), It.IsAny<ArrayOf<uint>>(), It.IsAny<CancellationToken>()))
|
||||
.Returns((RequestHeader _, ArrayOf<uint> ids, CancellationToken _) =>
|
||||
{
|
||||
var results = new List<StatusCode>();
|
||||
for (int i = 0; i < ids.Count; i++)
|
||||
{
|
||||
results.Add(StatusCodes.Good);
|
||||
}
|
||||
return new ValueTask<DeleteSubscriptionsResponse>(
|
||||
new DeleteSubscriptionsResponse { Results = results.ToArrayOf() });
|
||||
});
|
||||
m_mockMethodServices
|
||||
.Setup(s => s.CallAsync(
|
||||
It.IsAny<RequestHeader>(),
|
||||
It.Is<ArrayOf<CallMethodRequest>>(r =>
|
||||
r.Count == 1 && r[0].MethodId == MethodIds.Server_SetSubscriptionDurable),
|
||||
It.IsAny<CancellationToken>()))
|
||||
.Returns((RequestHeader _, ArrayOf<CallMethodRequest> requests, CancellationToken _) =>
|
||||
{
|
||||
Assert.That(requests[0].InputArguments[0].TryGetValue(out uint id), Is.True);
|
||||
Assert.That(requests[0].InputArguments[1].TryGetValue(out uint hours), Is.True);
|
||||
calls.Enqueue($"durable:{id}:{hours}h");
|
||||
return new ValueTask<CallResponse>(new CallResponse
|
||||
{
|
||||
Results =
|
||||
[
|
||||
new CallMethodResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
OutputArguments = [new Variant(hours)]
|
||||
}
|
||||
]
|
||||
});
|
||||
});
|
||||
m_mockMonitoredItemServices
|
||||
.Setup(s => s.CreateMonitoredItemsAsync(
|
||||
It.IsAny<RequestHeader>(), It.IsAny<uint>(), It.IsAny<TimestampsToReturn>(),
|
||||
It.IsAny<ArrayOf<MonitoredItemCreateRequest>>(), It.IsAny<CancellationToken>()))
|
||||
.Returns((RequestHeader _, uint id, TimestampsToReturn _,
|
||||
ArrayOf<MonitoredItemCreateRequest> requests, CancellationToken _) =>
|
||||
{
|
||||
calls.Enqueue($"items:{id}");
|
||||
var results = new List<MonitoredItemCreateResult>();
|
||||
for (int i = 0; i < requests.Count; i++)
|
||||
{
|
||||
results.Add(new MonitoredItemCreateResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
MonitoredItemId = (uint)Interlocked.Increment(ref nextItemId),
|
||||
RevisedSamplingInterval = 1000,
|
||||
RevisedQueueSize = 5
|
||||
});
|
||||
}
|
||||
return new ValueTask<CreateMonitoredItemsResponse>(
|
||||
new CreateMonitoredItemsResponse { Results = results.ToArrayOf() });
|
||||
});
|
||||
}
|
||||
|
||||
private static readonly string[] s_durableOnBothPartitions =
|
||||
[
|
||||
"create:22", "durable:22:10h", "items:22",
|
||||
"create:23", "durable:23:10h", "items:23"
|
||||
];
|
||||
|
||||
private static readonly string[] s_createOnly = ["create:22"];
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,7 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
[Category("SubscriptionManager")]
|
||||
[SetCulture("en-us")]
|
||||
[SetUICulture("en-us")]
|
||||
public sealed class SubscriptionTests
|
||||
public sealed partial class SubscriptionTests
|
||||
{
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
@@ -3282,9 +3282,11 @@ namespace Opc.Ua.Client.Subscriptions
|
||||
IMessageAckQueue completion, OptionsMonitor<SubscriptionOptions> options,
|
||||
ITelemetryContext telemetry, uint? subscriptionIdForAlreadyCreatedState = null,
|
||||
SubscriptionRecoveryPolicy recoveryPolicy = SubscriptionRecoveryPolicy.ReportOnly,
|
||||
TimeProvider? timeProvider = null)
|
||||
TimeProvider? timeProvider = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
: base(session, handler, completion, !subscriptionIdForAlreadyCreatedState.HasValue ?
|
||||
options : options.Configure(o => o with { Disabled = true }), telemetry, timeProvider: timeProvider)
|
||||
options : options.Configure(o => o with { Disabled = true }), telemetry,
|
||||
timeProvider: timeProvider, initialize: initialize)
|
||||
{
|
||||
// Let the subscription create itself
|
||||
if (subscriptionIdForAlreadyCreatedState.HasValue)
|
||||
|
||||
@@ -599,7 +599,8 @@ namespace Opc.Ua.Subscriptions.Tests
|
||||
ISubscriptionNotificationHandler handler,
|
||||
IOptionsMonitor<SubscriptionOptions> options,
|
||||
IMessageAckQueue queue,
|
||||
SubscriptionLoadState? loadState = null)
|
||||
SubscriptionLoadState? loadState = null,
|
||||
Action<IManagedSubscription>? initialize = null)
|
||||
{
|
||||
CreateSubscriptionCallCount++;
|
||||
LastCreateQueue = queue;
|
||||
@@ -608,7 +609,9 @@ namespace Opc.Ua.Subscriptions.Tests
|
||||
throw new InvalidOperationException(
|
||||
"CreateSubscriptionFactory was not configured for this test.");
|
||||
}
|
||||
return CreateSubscriptionFactory(handler, options, queue);
|
||||
IManagedSubscription subscription = CreateSubscriptionFactory(handler, options, queue);
|
||||
initialize?.Invoke(subscription);
|
||||
return subscription;
|
||||
}
|
||||
|
||||
public ValueTask<PublishResponse> PublishAsync(
|
||||
|
||||
Reference in New Issue
Block a user