mirror of
https://github.com/OPCFoundation/UA-.NETStandard.git
synced 2026-10-17 05:11:13 +08:00
b1c0cd20fd
# Description Persisted classic subscriptions retain their server/client handles but previously lost relative-path resolution metadata. Transfer could succeed, yet a later `ApplyChangesAsync` attempted to resolve an already-created item and failed. The SDK now owns this restoration lifecycle without exposing a public resolution-state setter. - Preserve the originally monitored target in internal snapshot state and restore both `ResolvedNodeId` and `Status.NodeId`. Transfer/reactivation use handles without re-browsing; fresh creation resolves paths again. - Version classic subscription/configuration files and read existing unversioned 2.0 files using frozen legacy wire layouts. Legacy missing targets remain unknown rather than being guessed. - Exclude existing items from creation-time resolution, reserve creation before asynchronous translation, and reject stale translation results. Failed deletion retains identity and pending retry state. - Support forward-only snapshot input using position tracking and bounded, per-ExtensionObject buffering in the binary decoder, including short reads. No whole-file buffering or reflection-based serialization is introduced. - Add unit/integration regressions and document the supported lifecycle and migration implications. **Compatibility:** Newly written files require the updated SDK. Existing 2.0 files remain readable. Assigning a different `StartNodeId` or `RelativePath` to a created/in-flight item now immediately throws `BadInvalidState`; unchanged assignments remain allowed. No V2 subscription-engine changes or 1.5 XML import are included. ## Checklist _Put an `x` in the boxes that apply. You can complete these step by step after opening the PR._ - [ ] I have signed the [CLA](https://opcfoundation.org/license/cla/ContributorLicenseAgreementv1.0.pdf) and read the [CONTRIBUTING](https://github.com/OPCFoundation/UA-.NETStandard/blob/master/CONTRIBUTING.md) doc. - [ ] I have added tests that prove my fix is effective or that my feature works and increased code coverage. - [x] I have added all necessary documentation. - [x] I have verified that my changes do not introduce (new) build or analyzer warnings. - [ ] I ran **all** tests locally using the **UA.slnx** solution against at least .net **framework** and .net **10**, and all passed. - [ ] I fixed **all** failing and flaky tests in the CI pipelines and **all** CodeQL warnings. - [ ] I have addressed **all** PR feedback received.
1062 lines
56 KiB
Markdown
1062 lines
56 KiB
Markdown
# Subscriptions and Monitored Items Service Set
|
|
|
|
The OPC UA Subscription and MonitoredItems service sets (Part 4 §5.13
|
|
and §5.14) are exposed through three layered client APIs in this
|
|
stack:
|
|
|
|
1. **Classic `Opc.Ua.Client.Subscription`** — the historical
|
|
event-driven API. Items added with
|
|
`subscription.AddItem(item); subscription.ApplyChangesAsync()`;
|
|
notifications delivered through the per-item `Notification` event
|
|
or the per-subscription `FastDataChangeCallback` / `FastEventCallback`.
|
|
2. **V2 `ISubscriptionManager` / `ISubscription`** — the options-based
|
|
callback API in `src/Opc.Ua.Client/Subscription/`. Items added
|
|
with `subscription.MonitoredItems.TryAdd(name, options, out _)`;
|
|
notifications delivered through `ISubscriptionNotificationHandler`
|
|
callbacks. The default engine on `ManagedSession`.
|
|
3. **`IStreamingSubscription`** — a thin `IAsyncEnumerable<T>`
|
|
abstraction on top of the V2 engine for state-machine waits and
|
|
short-lived monitoring.
|
|
|
|
**Recommendation:** new code should use the V2 surface
|
|
(`ISubscriptionManager` for long-lived application subscriptions,
|
|
`IStreamingSubscription` for short-lived / await-until-X scenarios).
|
|
The classic API stays supported for migration and continues to ship
|
|
in `Opc.Ua.Client.Subscription`; an internal bridge forwards classic
|
|
subscriptions onto the V2 publish pipeline when the V2 engine is
|
|
active, so existing classic-API code keeps working alongside V2.
|
|
|
|
For the full engine comparison (publish-pipeline ownership, worker
|
|
pools, when to pick which) see
|
|
[Sessions: subscription engines](Sessions.md#5-subscription-engines).
|
|
|
|
## Contents
|
|
|
|
- [Quick reference](#quick-reference)
|
|
- [Streaming subscriptions](#streaming-subscriptions)
|
|
- [Obtaining an `IStreamingSubscription`](#obtaining-an-istreamingsubscription)
|
|
- [Subscribing to data changes](#subscribing-to-data-changes)
|
|
- [Subscribing to events](#subscribing-to-events)
|
|
- [Composing streams](#composing-streams)
|
|
- [Lifecycle, cancellation, disposal](#lifecycle-cancellation-disposal)
|
|
- [Pairing with `AlarmClient`](#pairing-with-alarmclient)
|
|
- [Server retransmission queues](#server-retransmission-queues)
|
|
- [Publishing during session recovery](#publishing-during-session-recovery)
|
|
- [Triggering (SetTriggering)](#triggering-settriggering)
|
|
- [Declarative triggering](#declarative-triggering)
|
|
- [Imperative triggering](#imperative-triggering)
|
|
- [Cancellation semantics](#cancellation-semantics)
|
|
- [N:M relationships](#nm-relationships)
|
|
- [Save / load / restore behavior](#save--load--restore-behavior)
|
|
- [Error handling](#error-handling)
|
|
- [Unbounded monitored items](#unbounded-monitored-items)
|
|
- [How it works](#how-it-works)
|
|
- [Affinity — pinning items to the same partition](#affinity--pinning-items-to-the-same-partition)
|
|
- [Configuration](#configuration)
|
|
- [Snapshot, save / load, and transfer](#snapshot-save--load-and-transfer)
|
|
- [Durable subscriptions](#durable-subscriptions)
|
|
- [Performance](#performance)
|
|
- [Limitations and roadmap](#limitations-and-roadmap)
|
|
- [Classic → V2 surface mapping](#classic--v2-surface-mapping)
|
|
- [Subscription lifecycle](#subscription-lifecycle)
|
|
- [Notifications and callbacks](#notifications-and-callbacks)
|
|
- [Subscription-level service operations](#subscription-level-service-operations)
|
|
- [Monitored-item management](#monitored-item-management)
|
|
- [Per-item runtime](#per-item-runtime)
|
|
- [Persistence (save / load)](#persistence-save--load)
|
|
- [Tuning knobs](#tuning-knobs)
|
|
- [Engine wiring](#engine-wiring)
|
|
- [Per-item options mapping](#per-item-options-mapping)
|
|
- [Three-API summary](#three-api-summary)
|
|
- [Reference](#reference)
|
|
|
|
## Quick reference
|
|
|
|
| Concern | API |
|
|
|---|---|
|
|
| Add a V2 subscription | `session.AddSubscription(handler, SubscriptionOptions)` (or `ISubscriptionManager.Add`) |
|
|
| Add a monitored item | `subscription.MonitoredItems.TryAdd(name, IOptionsMonitor<MonitoredItemOptions>, out _)` (or `subscription.TryAddMonitoredItem(name, nodeId, configure, out _)` extension) |
|
|
| Set triggering — declarative | `MonitoredItemOptions.TriggeredByNames` (`IReadOnlyList<string>`) at item-add time |
|
|
| Set triggering — imperative | `ISubscription.SetTriggeringAsync(IMonitoredItem, add, remove, ct)` returning `SetTriggeringResult` |
|
|
| Triggering — name-based fluent | `ISubscription.SetTriggeringAsync(string trigName, params string[] tgtNames)` |
|
|
| Triggering — navigation | `IMonitoredItem.TriggeringItems` / `IMonitoredItem.TriggeredItems` (N:M) |
|
|
| Save / load subscriptions | `session.SaveSubscriptionsAsync(stream)` / `session.LoadSubscriptionsAsync(stream, factory)` |
|
|
| Stream — get streaming subscription | `ManagedSession.DefaultStreaming` |
|
|
| Stream — construct manually | `new StreamingSubscription(subscriptionManager, options?)` |
|
|
| Stream — subscribe one variable | `streaming.SubscribeDataChangesAsync(nodeId, options?, ct)` |
|
|
| Stream — subscribe many variables | `streaming.SubscribeDataChangesAsync(IReadOnlyList<NodeId>, ...)` |
|
|
| Stream — subscribe to events | `streaming.SubscribeEventsAsync(notifierId, EventFilter, options?, ct)` |
|
|
| Stream — wait until predicate | `.TakeUntilAsync(predicate)` |
|
|
| Stream — cap by wall-clock time | `.WithTimeoutAsync(timeout)` |
|
|
| Stream — take N items | `.TakeAsync(count)` |
|
|
| Stream — buffer first N | `.BufferedAsync(count)` |
|
|
| Stream — typed alarms | `streaming.SubscribeAlarmsAsync(notifierId, filter?)` |
|
|
|
|
## Streaming subscriptions
|
|
|
|
`IStreamingSubscription` exposes OPC UA subscription notifications as
|
|
`IAsyncEnumerable<T>` streams. Each `SubscribeXxxAsync` call adds a
|
|
monitored item to a shared, lazy-created OPC UA subscription and pipes
|
|
notifications through a `System.Threading.Channels.Channel<T>`.
|
|
Disposing the enumerator removes the monitored item.
|
|
|
|
This API sits on top of the V2 subscription engine
|
|
(`src/Opc.Ua.Client/Subscription`). It is **not** a replacement
|
|
for either the classic or V2 callback-based API — it is a thin
|
|
abstraction targeted at three concrete client scenarios:
|
|
|
|
- **State-machine waits** — "subscribe, observe transitions until X,
|
|
unsubscribe".
|
|
- **Short-lived monitoring** — "I need a sample for 30 seconds".
|
|
- **Typed alarm streaming** — Part 9 alarm records via
|
|
`AlarmStreamExtensions` (see [AlarmsAndConditions.md](AlarmsAndConditions.md)).
|
|
|
|
For long-lived application subscriptions, the callback-based
|
|
`ISubscriptionManager.Add` + `ISubscriptionNotificationHandler` API
|
|
remains the right choice (see
|
|
[Sessions: DefaultSubscriptionEngine (V2)](Sessions.md#defaultsubscriptionengine-v2)).
|
|
|
|
### Obtaining an `IStreamingSubscription`
|
|
|
|
`ManagedSession.DefaultStreaming` lazily constructs a shared instance
|
|
the first time it is accessed and disposes it when the session is
|
|
disposed.
|
|
|
|
```csharp
|
|
ManagedSession session = await new ManagedSessionBuilder(configuration, telemetry)
|
|
.UseEndpoint(endpoint)
|
|
.ConnectAsync(ct);
|
|
|
|
IStreamingSubscription streaming = session.DefaultStreaming;
|
|
```
|
|
|
|
If you need to bind to a specific `ISubscriptionManager` (for example
|
|
in tests or with the raw `Session`'s V2 engine), construct directly:
|
|
|
|
```csharp
|
|
if (!session.TryGetSubscriptionManager(out var manager))
|
|
{
|
|
throw new InvalidOperationException("A V2 subscription engine is required.");
|
|
}
|
|
await using var streaming = new StreamingSubscription(
|
|
subscriptionManager: manager,
|
|
subscriptionOptions: new SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromMilliseconds(250),
|
|
KeepAliveCount = 10,
|
|
LifetimeCount = 100,
|
|
});
|
|
```
|
|
|
|
The underlying OPC UA subscription is **not** created until the first
|
|
`SubscribeXxxAsync` call. Once created it is shared by every active
|
|
`SubscribeXxxAsync` enumerator on this `StreamingSubscription`.
|
|
|
|
### Subscribing to data changes
|
|
|
|
```csharp
|
|
await foreach (DataValueChange change in streaming
|
|
.SubscribeDataChangesAsync(VariableIds.Server_ServerStatus_CurrentTime, ct: ct)
|
|
.ConfigureAwait(false))
|
|
{
|
|
Console.WriteLine($"{change.MonitoredItem?.Name}: {change.Value}");
|
|
}
|
|
```
|
|
|
|
For multiple variables in a single stream:
|
|
|
|
```csharp
|
|
var nodes = new[]
|
|
{
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
VariableIds.Server_ServerStatus_SecondsTillShutdown,
|
|
};
|
|
|
|
await foreach (DataValueChange change in streaming
|
|
.SubscribeDataChangesAsync(nodes, options: new MonitoredItemOptions
|
|
{
|
|
SamplingInterval = TimeSpan.FromMilliseconds(500),
|
|
QueueSize = 5,
|
|
}, ct: ct).ConfigureAwait(false))
|
|
{
|
|
Console.WriteLine($"{change.MonitoredItem?.ClientHandle}: {change.Value}");
|
|
}
|
|
```
|
|
|
|
Each `MonitoredItemOptions` is treated as a template; only `StartNodeId`
|
|
is overridden internally per node.
|
|
|
|
### Subscribing to events
|
|
|
|
```csharp
|
|
EventFilter filter = new AlarmEventFilterBuilder()
|
|
.ForAlarms()
|
|
.Build();
|
|
|
|
await foreach (EventNotification evt in streaming
|
|
.SubscribeEventsAsync(ObjectIds.Server, filter, ct: ct)
|
|
.ConfigureAwait(false))
|
|
{
|
|
// Raw access; for typed records use SubscribeAlarmsAsync.
|
|
Console.WriteLine($"#{evt.Fields.Count} fields");
|
|
}
|
|
```
|
|
|
|
`SubscribeEventsAsync` does *not* validate the filter. Use
|
|
`AlarmEventFilterBuilder` for Part 9 events and any custom
|
|
`SimpleAttributeOperand`-based filter for other event types.
|
|
|
|
### Composing streams
|
|
|
|
`StreamingSubscriptionExtensions` ships four LINQ-style helpers
|
|
designed for the bounded-observation use cases. They wrap the source
|
|
`IAsyncEnumerable<T>` and stop iterating (which disposes the
|
|
underlying enumerator and the monitored item) when their condition
|
|
fires.
|
|
|
|
```csharp
|
|
// 1) Wait until a predicate matches, then return the matching item.
|
|
DataValueChange completed = await streaming
|
|
.SubscribeDataChangesAsync(buildProgressVariable)
|
|
.TakeUntilAsync(c => (double)c.Value.Value >= 100.0)
|
|
.LastAsync(ct);
|
|
|
|
// 2) Cap by wall-clock time. Completes silently when the timeout elapses.
|
|
List<DataValueChange> samples = await streaming
|
|
.SubscribeDataChangesAsync(sensor)
|
|
.WithTimeoutAsync(TimeSpan.FromSeconds(30))
|
|
.ToListAsync(ct);
|
|
|
|
// 3) Take exactly N values.
|
|
List<DataValueChange> firstThree = await streaming
|
|
.SubscribeDataChangesAsync(sensor)
|
|
.TakeAsync(3)
|
|
.ToListAsync(ct);
|
|
|
|
// 4) Buffer N items into a list — for one-shot snapshots.
|
|
IReadOnlyList<DataValueChange> snapshot = await streaming
|
|
.SubscribeDataChangesAsync(sensor)
|
|
.BufferedAsync(count: 10, ct);
|
|
```
|
|
|
|
`TakeUntilAsync` yields the matching item *last*, so callers can grab
|
|
the transition value with `LastAsync` / a terminating handler.
|
|
|
|
### Lifecycle, cancellation, disposal
|
|
|
|
Each stream has a bounded local buffer, including the default data-change
|
|
subscription. Its capacity is `max(1, QueueSize) * monitoredItemCount`, capped
|
|
at `int.MaxValue`; event streams default to ten entries per monitored item.
|
|
When full, `DiscardOldest` chooses between evicting the oldest buffered entry
|
|
and discarding the incoming entry. `StreamingSubscription.DroppedNotificationCount`
|
|
reports the cumulative local drops across data-change and event streams.
|
|
|
|
For a multi-node stream, this is one shared queue, not a per-node allocation.
|
|
With `DiscardOldest` enabled, a busy node can evict a quiet node's only queued
|
|
value. There is no local guarantee that each node's latest value remains buffered,
|
|
and overflow increments the drop counter rather than failing the stream. Use
|
|
separate single-node streams, consumed independently, when per-node retention is
|
|
required. Server-side per-monitored-item queues do not change this local policy.
|
|
|
|
The streaming subscription guarantees three invariants:
|
|
|
|
1. **Lazy subscription creation.** No OPC UA `CreateSubscription`
|
|
round-trip happens until your first `SubscribeXxxAsync` call.
|
|
2. **Reference-counted monitored items.** Each call adds a monitored
|
|
item; disposing the enumerator (end of the `await foreach`, an
|
|
exception, or explicit `await using`) removes that monitored item.
|
|
The underlying subscription stays alive for other enumerators.
|
|
3. **Disposal order.** Disposing the `StreamingSubscription` completes
|
|
all open channels and attempts to delete the underlying OPC UA subscription.
|
|
Disposing the `ManagedSession` calls this for you.
|
|
|
|
V2 subscription disposal gives each server-side delete a separate five-second
|
|
cancellation deadline. Local cleanup does not wait indefinitely for an unavailable
|
|
server. A failed or timed-out delete is logged; it does not prove remote deletion,
|
|
and the server may retain the subscription until its configured lifetime expires.
|
|
|
|
Publish workers survive recovery for subscriptions that the V2 engine actually
|
|
created. A never-created subscription does not inherit worker retention from a
|
|
classic subscription or a V2 subscription that has already been removed.
|
|
An intentionally disabled or deleted subscription likewise contributes no
|
|
retained demand until it is created again.
|
|
|
|
Cancellation propagates the natural way: pass a `CancellationToken` to
|
|
`SubscribeXxxAsync` *or* the outer `await foreach` (via
|
|
`WithCancellation`). When the token fires the enumerator stops, the
|
|
finally block removes the monitored item, and the loop body throws
|
|
`OperationCanceledException`.
|
|
|
|
```csharp
|
|
using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
|
|
try
|
|
{
|
|
await foreach (DataValueChange change in
|
|
streaming.SubscribeDataChangesAsync(sensor, ct: cts.Token))
|
|
{
|
|
Process(change);
|
|
}
|
|
}
|
|
catch (OperationCanceledException) { /* expected */ }
|
|
```
|
|
|
|
### Pairing with `AlarmClient`
|
|
|
|
The streaming subscription and `AlarmClient` are complementary:
|
|
|
|
- `IStreamingSubscription` delivers the *events* (raw or typed).
|
|
- `AlarmClient` calls the *methods* (acknowledge, shelve, suppress,
|
|
reset, etc.).
|
|
|
|
```csharp
|
|
AlarmClient alarms = session.GetAlarmClient();
|
|
IStreamingSubscription streaming = session.DefaultStreaming;
|
|
|
|
await foreach (ConditionTypeRecord rec in streaming
|
|
.SubscribeAlarmsAsync(ObjectIds.Server)
|
|
.TakeUntilAsync(r =>
|
|
r is AlarmConditionTypeRecord a &&
|
|
a.ConditionId == myAlarmId &&
|
|
a.ActiveStateId == true))
|
|
{
|
|
if (rec is AlarmConditionTypeRecord active && active.ActiveStateId == true)
|
|
{
|
|
await alarms.AcknowledgeAsync(
|
|
active.ConditionId,
|
|
active.EventId,
|
|
new LocalizedText("en", "Auto-acked")).ConfigureAwait(false);
|
|
}
|
|
}
|
|
```
|
|
|
|
See [AlarmsAndConditions.md](AlarmsAndConditions.md) for the typed
|
|
record hierarchy.
|
|
|
|
## Server retransmission queues
|
|
|
|
On the server, newly created monitored items remain provisional until
|
|
the subscription rechecks ownership and deletion after node-manager
|
|
creation. If transfer, session detachment, deletion, or cancellation wins,
|
|
normal node-manager deletion releases those unattached items.
|
|
|
|
Publish, queued Publish, Republish, and persistence consumers receive
|
|
deep notification snapshots. Acknowledgement, eviction, or queue disposal
|
|
can recycle the retained pooled graph without changing a response still
|
|
waiting for encoding or a captured retransmission/durable snapshot.
|
|
Pooling remains enabled for the retained notifications.
|
|
|
|
The server retains notification messages for acknowledgement and Republish
|
|
(Part 4 §5.14). When a new batch exceeds the configured retransmission capacity,
|
|
only the excess oldest retained messages are evicted. The remaining sequence
|
|
numbers, persistence removal deltas and discarded-unacknowledged-message
|
|
diagnostics describe the same retained queue. An oversized new batch is trimmed
|
|
to capacity before it displaces older messages.
|
|
|
|
The per-session Publish request limit counts only requests still awaiting a
|
|
response. Cancellation, timeout, normal completion and session/server teardown
|
|
release a request's admission exactly once; a completed request does not prevent
|
|
the client from queuing its replacement while a long keep-alive is pending.
|
|
|
|
Publish from a live session without subscriptions returns `BadNoSubscription`,
|
|
regardless of other sessions' subscriptions. A closing session retains
|
|
`BadSessionClosed`. The Republish request and requested-message diagnostic
|
|
counters remain equal, with one increment per authorized request; the
|
|
successful-message counter advances only when a message is returned.
|
|
|
|
## Publishing during session recovery
|
|
|
|
V2 session recreation pauses publishing and drains active Publish attempts
|
|
before replacing the session. The drain cancels each attempt, including an
|
|
attempt parked at the shared channel's ready gate, without terminating its
|
|
worker. A cancelled attempt rolls its acknowledgements back before releasing its
|
|
active count. Recreation must await this complete unwind because a timeout that skips
|
|
the drain could acknowledge notifications from the old subscription generation
|
|
against a reused subscription identifier.
|
|
|
|
Automatic subscription and monitored-item updates also pause during session
|
|
recreation. Recovery cancels an active update pass; its retry waits until the
|
|
recovery owner completes subscription restoration. If restoration fails, this
|
|
pause remains in effect across the outer-policy handoff. The state worker cannot
|
|
create a competing subscription on a retiring or replacement session. Pending
|
|
option changes resume after restoration, without changing explicit
|
|
`RecreateAsync` or transfer semantics. Cancellation remains client-side and does
|
|
not guarantee that an already-sent request was not processed by the server.
|
|
|
|
Once the session and subscriptions are restored, publishing resumes through the
|
|
same subscription-facing interface. Temporarily clearing server-side subscription
|
|
identifiers does not replace existing workers. Pool limits and actual subscription
|
|
removal still apply. Intentional deletion, including setting the V2 subscription's
|
|
`Disabled` option, retires its worker demand even if it remains registered.
|
|
Successful creation restores demand; recovery-only resets preserve it.
|
|
With transfer-on-recreate enabled, an invalid old subscription
|
|
falls back to recreation. A
|
|
[managed-session channel deadline](Sessions.md#shared-retry-budget-with-managedsession)
|
|
cancels recovery and hands control to the outer reconnect policy. After expiry,
|
|
any replacement attempt still waits for cancelled Publish attempts to roll back
|
|
their acknowledgements and release their active counts before replacing session
|
|
and subscription state.
|
|
|
|
## Triggering (SetTriggering)
|
|
|
|
Triggering (OPC UA Part 4 §5.13.5) links a *triggering* monitored
|
|
item to one or more *triggered* items. When the triggering item fires
|
|
a notification, the triggered items' queued notifications are
|
|
reported in the next publish even if their monitoring mode is
|
|
`Sampling` (which would otherwise suppress reporting). This is the
|
|
canonical pattern for "sample many items at high rate, report on
|
|
demand". See Part 4 §5.13.1.6 for the full triggering model.
|
|
|
|
Aggregate-filtered monitored items follow these same monitoring-mode rules
|
|
(Part 4 5.13.1.3). Completion of an aggregate interval does not enable reporting
|
|
for a `Disabled` item or an untriggered `Sampling` item. A legitimately triggered
|
|
`Sampling` item can publish its queued aggregates; after that trigger is consumed,
|
|
interval completion alone does not cause another report.
|
|
|
|
The V2 engine exposes triggering through a **hybrid API**:
|
|
|
|
- **Declarative** — set `MonitoredItemOptions.TriggeredByNames` when
|
|
the item is added.
|
|
- **Imperative** — call
|
|
`ISubscription.SetTriggeringAsync(IMonitoredItem triggeringItem,
|
|
add, remove, ct)` at any time.
|
|
|
|
Both paths flow through the same batched apply pipeline: multiple
|
|
operations targeting the same triggering item collapse into a single
|
|
`SetTriggering` request carrying both the merged `linksToAdd` and
|
|
`linksToRemove` lists. Per Part 4 §5.13.5.2 the server processes
|
|
`linksToRemove` before `linksToAdd`, which matches the engine's
|
|
last-intent-wins conflict resolution.
|
|
|
|
The V2 engine models the topology as a per-item **desired state**:
|
|
each monitored item carries a list of stable triggering-item names
|
|
(`DesiredTriggeredByNames`) that reflects what the caller wants. The
|
|
engine reconciles desired state against the server via `SetTriggering`
|
|
requests whenever:
|
|
|
|
- An options change pushes a different `TriggeredByNames` value.
|
|
- The imperative API mutates the desired set.
|
|
- An item is `Reset` (recreate scenario) — the engine replays the
|
|
desired set so transient server-side state matches the durable
|
|
client-side intent.
|
|
|
|
The OPC UA spec allows a triggered item to be linked to **multiple**
|
|
triggering items (N:M); the V2 API exposes this directly via the
|
|
plural `TriggeringItems` / `TriggeredItems` projections.
|
|
|
|
### Declarative triggering
|
|
|
|
The simplest way to declare triggering is via `MonitoredItemOptions`
|
|
at item-add time:
|
|
|
|
```csharp
|
|
ManagedSession session = await new ManagedSessionBuilder(config, telemetry)
|
|
.UseEndpoint(endpoint)
|
|
.ConnectAsync(ct);
|
|
|
|
ISubscription sub = session.AddSubscription(handler, new SubscriptionOptions
|
|
{
|
|
PublishingInterval = TimeSpan.FromSeconds(1)
|
|
});
|
|
|
|
// Triggering item — reports normally at its sampling interval.
|
|
sub.TryAddMonitoredItem("trig",
|
|
VariableIds.Server_ServerStatus_CurrentTime,
|
|
o => o with { MonitoringMode = MonitoringMode.Reporting },
|
|
out _);
|
|
|
|
// Triggered item — samples in the background; only reports when
|
|
// "trig" reports.
|
|
sub.TryAddMonitoredItem("sensor",
|
|
new NodeId("Sensor1", 2),
|
|
o => o with
|
|
{
|
|
MonitoringMode = MonitoringMode.Sampling,
|
|
TriggeredByNames = ["trig"]
|
|
},
|
|
out _);
|
|
```
|
|
|
|
The engine batches the underlying `CreateMonitoredItems` /
|
|
`SetTriggering` requests and applies them on the next subscription
|
|
apply pass. There is no need to explicitly call any method to
|
|
"finish" the triggering setup — once both items are `Created`, the
|
|
engine issues the `SetTriggering` call automatically.
|
|
|
|
Order of `TryAddMonitoredItem` calls does not matter; the engine
|
|
will keep the triggering operation queued until both items are
|
|
created on the server, then issue the request.
|
|
|
|
### Imperative triggering
|
|
|
|
Mutate triggering relationships at any time after items exist via
|
|
`SetTriggeringAsync`:
|
|
|
|
```csharp
|
|
SetTriggeringResult result = await sub.SetTriggeringAsync(
|
|
triggeringItem: trig,
|
|
linksToAdd: [sensor1, sensor2],
|
|
linksToRemove: null,
|
|
ct: ct);
|
|
|
|
foreach ((IMonitoredItem item, StatusCode status) in result.AddResults)
|
|
{
|
|
if (!StatusCode.IsGood(status))
|
|
{
|
|
// BadMonitoredItemIdInvalid, BadInvalidState, etc.
|
|
}
|
|
}
|
|
```
|
|
|
|
The call queues a single `TriggeringOperation` and returns a
|
|
`ValueTask` that completes when the next apply pass applies it. The
|
|
result's per-link entries are in the same order as the input lists
|
|
for easy pairing.
|
|
|
|
Convenience overloads use stable monitored-item names:
|
|
|
|
```csharp
|
|
await sub.SetTriggeringAsync("trig", "sensor1", "sensor2");
|
|
|
|
// add + remove in one call (server processes remove before add per §5.13.5.2):
|
|
await sub.SetTriggeringAsync(
|
|
"trig",
|
|
add: ["sensor3"],
|
|
remove: ["sensor1"],
|
|
ct: ct);
|
|
```
|
|
|
|
Unknown names throw `ArgumentException` synchronously.
|
|
|
|
### Cancellation semantics
|
|
|
|
The `CancellationToken` passed to `SetTriggeringAsync` controls the await **and** marks the queued operation as cancelled on a best-effort basis:
|
|
|
|
- Cancelling the token **aborts the await** and surfaces an `OperationCanceledException` to the caller.
|
|
- The engine attempts to **skip the queued operation on the next apply pass**. If cancellation fires before the apply pass picks the op off the queue, the server-side `SetTriggering` request is NOT issued for this op.
|
|
- This is **best effort only**: if the apply pass has already begun dispatching the RPC for the op's group, the server-side mutation cannot be cancelled — the in-flight RPC completes regardless.
|
|
- The desired-state mutations performed synchronously by the engine before the await (the updates that make `IMonitoredItem.TriggeringItems` / `IMonitoredItem.TriggeredItems` and any subsequent snapshot reflect the new topology) **already happened** when `SetTriggeringAsync` returned its `ValueTask`. They stand regardless of cancellation.
|
|
- To revert local intent, the caller must explicitly issue an opposing `SetTriggeringAsync` (or change the declarative `MonitoredItemOptions.TriggeredByNames`).
|
|
|
|
> [!WARNING]
|
|
> Cancelling the `CancellationToken` **cannot guarantee** the server-side `SetTriggering` call is skipped — it is best-effort; an op that has already begun its server request will still complete server-side, and local desired-state is never rolled back automatically.
|
|
|
|
```csharp
|
|
// Original intent: when "trig" fires, also report "sensor1" and "sensor2".
|
|
await sub.SetTriggeringAsync("trig", "sensor1", "sensor2");
|
|
|
|
// To undo, issue an opposing call with the items in `remove`. Note that
|
|
// even if the first call's cancellation token had fired, the server
|
|
// might still have applied the link (race with the apply pass) and the
|
|
// local desired-state intent would have stood — so an explicit opposing
|
|
// call is the only reliable revert path.
|
|
await sub.SetTriggeringAsync(
|
|
"trig",
|
|
add: null,
|
|
remove: ["sensor1", "sensor2"],
|
|
ct: ct);
|
|
```
|
|
|
|
### N:M relationships
|
|
|
|
A triggered item may be linked to many triggering items, and a
|
|
triggering item may have many triggered items. Use multiple
|
|
`SetTriggeringAsync` calls (or include the same triggered name under
|
|
multiple triggers' `TriggeredByNames`):
|
|
|
|
```csharp
|
|
sub.TryAddMonitoredItem("trigA", ..., out var trigA);
|
|
sub.TryAddMonitoredItem("trigB", ..., out var trigB);
|
|
sub.TryAddMonitoredItem("shared",
|
|
nodeId,
|
|
o => o with
|
|
{
|
|
MonitoringMode = MonitoringMode.Sampling,
|
|
TriggeredByNames = ["trigA", "trigB"]
|
|
},
|
|
out var shared);
|
|
|
|
// "shared" reports whenever EITHER trigA OR trigB fires:
|
|
Assert.That(shared.TriggeringItems, Is.EquivalentTo(new[] { trigA, trigB }));
|
|
Assert.That(trigA.TriggeredItems, Has.Member(shared));
|
|
Assert.That(trigB.TriggeredItems, Has.Member(shared));
|
|
```
|
|
|
|
`TriggeringItems` and `TriggeredItems` are read-only projections
|
|
resolved on demand against the subscription's monitored-item
|
|
collection. They reflect the *desired* topology (intent) — so they
|
|
work immediately on `TryAdd` before any request fires, and survive
|
|
restore-from-snapshot without depending on server state.
|
|
|
|
### Save / load / restore behavior
|
|
|
|
The triggering topology round-trips through the V2 snapshot path:
|
|
|
|
- `MonitoredItemStateSnapshot.TriggeredByNames` captures the runtime
|
|
desired state at snapshot time.
|
|
- `SubscriptionManager.SaveAsync` / `LoadAsync` (and the in-memory
|
|
`SnapshotSubscriptions` / `RestoreSubscriptionsAsync` extensions)
|
|
persist this field via the standard binary encoder.
|
|
|
|
On load, the engine takes one of three paths:
|
|
|
|
| Scenario | Behavior |
|
|
|---|---|
|
|
| **TransferSubscriptions success** | Local desired state is restored from the snapshot. NO `SetTriggering` request is issued — per Part 4 §5.13.5 conformant servers preserve server-side triggering relationships across a session transfer. |
|
|
| **Recreate-fallback** (transfer rejected or unsupported) | Items reset to "not created"; the saved `DesiredTriggeredByNames` is preserved on each item; the batched apply pass replays `SetTriggering` after items finish re-creating. |
|
|
| **In-session `RecreateAsync`** (forced recreate) | Same as recreate-fallback — desired state preserved through reset, replayed after re-creation. |
|
|
|
|
This closes the gap from issue
|
|
[#3834](https://github.com/OPCFoundation/UA-.NETStandard/issues/3834):
|
|
triggering links are now automatically restored on
|
|
`TransferSubscriptions` *and* on recreate/reconnect; no manual
|
|
re-issue of `SetTriggering` is required by callers.
|
|
|
|
If a server fails to preserve links across transfer (contrary to the
|
|
spec), the V2 client view will indicate the links exist while
|
|
server-side notifications stop firing. Callers detecting this can
|
|
manually re-issue `Subscription.SetTriggeringAsync` per relationship.
|
|
|
|
### Error handling
|
|
|
|
Per Part 4 §5.13.5.4 (Table 74) the only spec-specific per-link status
|
|
code is `Bad_MonitoredItemIdInvalid`. Service-level codes (Table 73)
|
|
include `Bad_NothingToDo`, `Bad_TooManyOperations`, and
|
|
`Bad_SubscriptionIdInvalid`.
|
|
|
|
The engine handles these as follows:
|
|
|
|
| Condition | Engine response |
|
|
|---|---|
|
|
| Per-link `Bad_MonitoredItemIdInvalid` on add | Surfaced via the returned `SetTriggeringResult.AddResults[i].Status`; the triggered item's desired-state entry for that triggering name is rolled back so it matches reality. |
|
|
| Per-link `Bad` on remove | Surfaced via `RemoveResults[i].Status`; the desired-state entry is re-added so the snapshot continues to reflect the still-existing server link. |
|
|
| Service-level `Bad_SubscriptionIdInvalid` | Operations are re-queued (not failed terminally); they replay after the subscription state machine recreates the subscription. |
|
|
| Service-level `Bad_TooManyOperations` | Surfaced as a fatal `ServiceResultException` on all contributing TCSs. Callers wishing to recover should chunk their `SetTriggeringAsync` calls manually; auto-chunking is a planned follow-up. |
|
|
| Communication errors (timeout, channel closed) | Routed through the existing retry policy; bounded by `MaxTriggeringRetryCount` to prevent infinite loops. |
|
|
| Triggering item never reaches `Created` | After `MaxTriggeringRetryCount` re-queues, the operation completes with the triggering item's propagated error status (or `Bad_MonitoredItemIdInvalid` as a fallback). |
|
|
|
|
## Unbounded monitored items
|
|
|
|
The V2 subscription engine
|
|
(`Opc.Ua.Client.Subscriptions.ISubscription`, returned from
|
|
`session.TryGetSubscriptionManager(out var manager)` then `manager.Add(...)`)
|
|
lets a single logical
|
|
subscription hold an effectively unlimited number of monitored items.
|
|
When the per-subscription cap (`MaxMonitoredItemsPerSubscription` from
|
|
the server's capabilities, OPC UA Part 4 §5.13.2) would be exceeded,
|
|
the engine transparently splits monitored items across additional
|
|
server-side **partition** subscriptions and presents one logical
|
|
collection to the caller. Single-partition workloads (the common
|
|
case) keep their pre-existing fast path; partitioning kicks in only
|
|
when needed.
|
|
|
|
### How it works
|
|
|
|
* `ISubscriptionManager.Add(handler, options)` returns a
|
|
`LogicalSubscription` wrapper that implements `ISubscription` and
|
|
`IPartitionedSubscription`. Pattern-match on the latter to
|
|
introspect the partition layout:
|
|
|
|
```csharp
|
|
if (!session.TryGetSubscriptionManager(out var manager))
|
|
{
|
|
throw new InvalidOperationException("A V2 subscription engine is required.");
|
|
}
|
|
ISubscription subscription = manager.Add(handler, optionsMonitor);
|
|
if (subscription is IPartitionedSubscription partitioned)
|
|
{
|
|
Console.WriteLine($"Spread over {partitioned.PartitionCount} partition(s)");
|
|
foreach (uint partitionId in partitioned.PartitionIds)
|
|
{
|
|
Console.WriteLine($" partition server id: {partitionId}");
|
|
}
|
|
}
|
|
```
|
|
|
|
* `ISubscription.MonitoredItems` is backed by a composite collection
|
|
that aggregates every partition's items behind one
|
|
`IMonitoredItemCollection`. `TryAdd` / `TryRemove` /
|
|
`TryGetMonitoredItemByName` / `TryGetMonitoredItemByClientHandle`
|
|
enumerate every partition; the caller never needs to know which
|
|
partition owns which item.
|
|
|
|
* When the placement policy decides a new partition is needed, the
|
|
engine constructs a sibling server-side subscription using the
|
|
same `SubscriptionOptions` and the same notification handler. The
|
|
new partition is registered in the manager's publish-dispatch
|
|
registry so publish responses route to it like any other
|
|
subscription.
|
|
|
|
* `ISubscriptionNotificationHandler` callbacks always receive the
|
|
logical wrapper as the `ISubscription subscription` parameter.
|
|
When more than one partition is active the engine serialises calls
|
|
through a per-wrapper semaphore so consumers observe one handler
|
|
invocation at a time, matching the single-partition behaviour they
|
|
came to rely on under V2.
|
|
|
|
* Notifications include the source partition's server-side id via
|
|
the `PartitionServerId` field on `DataValueChange` and
|
|
`EventNotification`. Sequence numbers stay per-partition
|
|
(`(PartitionServerId, sequenceNumber)` disambiguates across
|
|
partitions).
|
|
|
|
### Affinity — pinning items to the same partition
|
|
|
|
OPC UA `SetTriggering` is scoped to one server-side subscription
|
|
(`Part 4 §5.13.6`). Items that need to participate in a triggering
|
|
relationship (or any other per-subscription feature) must land in the
|
|
same partition. Pin them via
|
|
`MonitoredItemOptions.Affinity`:
|
|
|
|
```csharp
|
|
var optionsMonitor = new OptionsMonitor<MonitoredItemOptions>(new MonitoredItemOptions
|
|
{
|
|
StartNodeId = new NodeId("MyVariable", 2),
|
|
SamplingInterval = TimeSpan.FromMilliseconds(500),
|
|
Affinity = "alarms-group" // every item with this tag stays together
|
|
});
|
|
subscription.MonitoredItems.TryAdd("MyItem", optionsMonitor, out _);
|
|
```
|
|
|
|
* Items with the same non-null `Affinity` value are guaranteed to
|
|
share a partition.
|
|
* Once the partition reaches the per-partition cap the next
|
|
`TryAdd` with the same tag returns `false` — the contract is
|
|
**strict** so the group never splits. Callers must shrink the
|
|
group, raise `MaxMonitoredItemsPerPartition`, or pick a different
|
|
tag.
|
|
* `null` (the default) places no co-location constraint; items
|
|
without affinity fill partitions first-fit.
|
|
|
|
`LogicalSubscription.SetTriggeringAsync(triggering, linksToAdd,
|
|
linksToRemove)` validates that every linked item shares the same
|
|
partition as the triggering item and throws `ArgumentException`
|
|
otherwise. The exception message points callers at `Affinity` for
|
|
remediation.
|
|
|
|
### Configuration
|
|
|
|
All knobs live on `SubscriptionOptions`:
|
|
|
|
| Property | Default | Effect |
|
|
| --- | --- | --- |
|
|
| `DisableUnboundedItemMode` | `false` | When `true`, the wrapper is bound to one server-side subscription; `TryAdd` calls beyond the server cap surface `Bad_TooManyMonitoredItems` per-item like the pre-V2 engine. |
|
|
| `MaxMonitoredItemsPerPartition` | `null` | Per-partition upper bound. `null` means "let the reactive fallback discover the server's effective cap". Set to a smaller value to keep partitions small for snapshot/transfer scaling, or to a larger value to override an artificially low server limit. |
|
|
| `SecondaryPartitionIdleTimeout` | `30s` | Idle timeout after which an empty **secondary** partition is deleted from the server. The primary partition is never deleted while the logical subscription is alive so the wrapper's server-side identifier stays stable. Set to `TimeSpan.Zero` for immediate delete; set to `Timeout.InfiniteTimeSpan` to disable idle-delete. |
|
|
|
|
The reactive cap fallback is always on whenever
|
|
`DisableUnboundedItemMode` is `false`: the engine watches every
|
|
`CreateMonitoredItems` response and, on the first
|
|
`Bad_TooManyMonitoredItems` outcome, marks the offending partition
|
|
**no-grow** so subsequent placements fan out to a new partition. This
|
|
handles servers whose actual limit is lower than the advertised
|
|
capability and servers whose limit changes between session lifetimes.
|
|
|
|
### Snapshot, save / load, and transfer
|
|
|
|
Multi-partition logical subscriptions snapshot every partition. The
|
|
capture path is:
|
|
|
|
* `LogicalSubscription.SnapshotAllPartitions()` returns
|
|
`IReadOnlyList<SubscriptionStateSnapshot>` — one snapshot per
|
|
partition, ordered primary-first.
|
|
* Every snapshot in the list carries the wrapper's stable
|
|
`LogicalGroupId` (a lazy-generated GUID cached for the lifetime of
|
|
the wrapper) and an incrementing `PartitionIndex` (`0` for the
|
|
primary, `1+` for secondaries in mint order).
|
|
* `ManagedSession.SnapshotSubscriptions()` and
|
|
`ISubscriptionManager.SaveAsync(...)` flatten across partitions
|
|
automatically, so callers persisting the list see the full state.
|
|
* Single-partition wrappers continue to emit exactly one snapshot.
|
|
Old V1 snapshot files have `LogicalGroupId == null` and load via
|
|
the standalone path unchanged.
|
|
|
|
The restore path:
|
|
|
|
* `ISubscriptionManager.LoadAsync(...)` and
|
|
`ManagedSession.RestoreSubscriptionsAsync(...)` group incoming
|
|
snapshots by `LogicalGroupId`. Snapshots with a `null` group
|
|
restore as standalone subscriptions; non-null groups become one
|
|
multi-partition `LogicalSubscription` wrapper via
|
|
`SubscriptionManager.RestoreGroupAsync(...)` (internal).
|
|
* Restored secondary partitions go through a new internal
|
|
`LogicalSubscription.AppendPreloadedPartition(...)` hook so the
|
|
composite collection's placement policy + idle-delete timer + (if
|
|
applicable) durable hook are wired exactly as if the partition had
|
|
been minted on demand.
|
|
* When `transferSubscriptions: true` is passed, each partition's
|
|
saved server-side id is preserved via `TransferSubscriptions`;
|
|
partitions whose transfer fails fall back to recreate.
|
|
* **Strict grouping** is enforced: a non-null `LogicalGroupId` must
|
|
appear on a contiguous `0..N-1` sequence of `PartitionIndex`
|
|
values with no duplicates and exactly one primary. Malformed
|
|
snapshot groups throw `ServiceResultException(BadDecodingError)`
|
|
naming the offending `LogicalGroupId` — a corrupt or hand-edited
|
|
snapshot fails loudly rather than silently fragmenting state.
|
|
|
|
`MonitoredItemStateSnapshot.Affinity` round-trips so restored
|
|
subscriptions regroup items into the same affinity-pinned partition
|
|
the source had.
|
|
|
|
After a snapshot transfer, V2 rebinds the saved item handles and reapplies the
|
|
saved monitoring parameters with `ModifyMonitoredItems`. The response supplies
|
|
the actual revised sampling interval, queue size and filter result:
|
|
`GetMonitoredItems` returns handles only. This does not recreate the items or
|
|
change their monitoring mode or triggering links. Live items that already have
|
|
revised parameters do not require this extra snapshot-restoration step.
|
|
Refresh enqueueing is ordered with concurrent options changes, so a restored
|
|
snapshot cannot overwrite newer requested sampling intervals, queue sizes or
|
|
filters.
|
|
A successful initiated transfer also reports `PublishState.Transferred` to the
|
|
target handler after handle synchronization, before attempting retained-message
|
|
recovery; failed synchronization does not report transfer completion.
|
|
|
|
Republishing retained notifications and requesting initial values are independent.
|
|
If an origin session disconnects before acknowledging a message, a transfer can
|
|
recover that message from the server's retransmission queue. With
|
|
`SendInitialValuesOnTransfer` also enabled, the target can receive the same static
|
|
value again in a new notification message. Distinguish these batches by their
|
|
sequence numbers, not by value equality. On the classic API, retransmission is
|
|
controlled by `RepublishAfterTransfer`, while initial values are requested with
|
|
the transfer call's `sendInitialValues` argument. Deferring acknowledgements
|
|
retains messages for explicit `Republish` requests; it does not ask the server
|
|
to resend them unsolicited. See
|
|
[OPC UA Part 4, TransferSubscriptions](https://reference.opcfoundation.org/specs/OPC-10000-4/5.14.7).
|
|
The classic client acknowledges messages selected for replay only after receiving
|
|
them. Messages not selected for replay are acknowledged when transfer completes;
|
|
the advertised available-sequence list remains intact.
|
|
|
|
### Durable subscriptions
|
|
|
|
The server converts the configured durable lifetime in hours to milliseconds
|
|
before revising keep-alive and lifetime counts. The conversion uses wide
|
|
arithmetic, including lifetimes longer than the UInt32 millisecond range.
|
|
|
|
`SetAsDurableAsync(lifetime)` records the durable intent on the
|
|
wrapper and applies it synchronously to every partition that is
|
|
already `Created` (it returns the minimum revised lifetime across
|
|
those partitions). For partitions that have not yet completed
|
|
their initial `CreateSubscription` — both the primary on a brand
|
|
new subscription and any future secondary minted by the placement
|
|
policy under capacity pressure — the wrapper installs an
|
|
`OnAfterCreateAsync` hook on the partition's state machine. The
|
|
state machine awaits the hook exactly once between
|
|
`CreateSubscription` and the first `CreateMonitoredItems`,
|
|
satisfying the OPC UA Part 4 §5.13.9 ordering rule that
|
|
`SetSubscriptionDurable` must precede any monitored-item creation.
|
|
The hook is cleared after invocation; the wrapper re-installs it on
|
|
each `SetAsDurableAsync` call so it survives reconnect / recreate
|
|
cycles. Hook failures are logged as a warning and surfaced via
|
|
`SubscriptionState.Modified` rather than tearing the partition
|
|
down.
|
|
|
|
### Performance
|
|
|
|
Single-partition workloads observe **no overhead** from the wrapper
|
|
— the composite collection short-circuits to the primary's own
|
|
collection without taking any extra locks or building indexes.
|
|
Multi-partition workloads pay:
|
|
|
|
* One semaphore acquire/release per notification callback (small).
|
|
* O(P) partition scans for `TryGetMonitoredItemByName` /
|
|
`TryGetMonitoredItemByClientHandle` misses against the composite
|
|
index (P is typically single-digit).
|
|
* One extra `Subscription` instance per partition (each holds its
|
|
own state machine, ack queue position, and notification pipeline).
|
|
|
|
### Limitations and roadmap
|
|
|
|
Cross-partition `SetTriggering` is rejected by design: per OPC UA
|
|
Part 4 §5.13.6 the service is scoped to a single server-side
|
|
subscription, and re-grouping already-placed items would require
|
|
delete + recreate (losing per-item server ids and briefly stopping
|
|
publishing). Plan `Affinity` up-front for items that need to
|
|
participate in a triggering relationship; the engine will keep the
|
|
group co-located for the lifetime of the wrapper.
|
|
|
|
This is the only multi-partition feature gap — single-partition
|
|
workloads cover the vast majority of OPC UA client use cases
|
|
unchanged.
|
|
|
|
## Classic → V2 surface mapping
|
|
|
|
This section is the working reference for porting classic
|
|
(`Opc.Ua.Client.Subscription` / `Opc.Ua.Client.MonitoredItem`) code to
|
|
the V2 surface (`Opc.Ua.Client.Subscriptions.ISubscription` /
|
|
`Opc.Ua.Client.Subscriptions.MonitoredItems.IMonitoredItem`). The
|
|
tables describe today's APIs side-by-side; "not on V2" rows record
|
|
behaviour the V2 engine replaces by design (handler-centric,
|
|
channel-based pipeline, snapshot/restore, options-monitor-driven
|
|
reconciliation, etc.) along with the recommended V2 alternative.
|
|
|
|
### Subscription lifecycle
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `new Subscription(template)` + `Session.AddSubscription(s)` + `s.CreateAsync()` | `ISubscriptionManager.Add(handler, IOptionsMonitor<SubscriptionOptions>)` — V2 creates the subscription on the server asynchronously; callers poll `subscription.Created` (or attach an `OnSubscriptionStateChangedAsync` handler). |
|
|
| `Session.RemoveSubscriptionAsync(s)` | `await subscription.DisposeAsync()` — V2 removal is dispose-on-subscription. |
|
|
| `s.CreateAsync(ct)` / `s.ModifyAsync(ct)` / `s.DeleteAsync(silent, ct)` | Implicit via `Add` / options push / `DisposeAsync`. No explicit V2 calls — behaviour is driven by options + lifecycle. |
|
|
| `s.SetPublishingModeAsync(bool, ct)` | Push `SubscriptionOptions { PublishingEnabled = ... }` via the `IOptionsMonitor<SubscriptionOptions>`. The V2 manager picks up the change automatically. |
|
|
| `s.ChangesPending` / `s.ChangesCompleted()` | Not on V2 (the engine is fully push-driven; no "pending changes" concept). Callers wait on the side-effect (e.g. `IMonitoredItem.Created`, options-monitor change tokens). |
|
|
|
|
### Notifications and callbacks
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `s.FastDataChangeCallback` (delegate) | `ISubscriptionNotificationHandler.OnDataChangeNotificationAsync(...)` |
|
|
| `s.FastKeepAliveCallback` (delegate) | `ISubscriptionNotificationHandler.OnKeepAliveNotificationAsync(...)` |
|
|
| `s.FastEventCallback` | `ISubscriptionNotificationHandler.OnEventDataNotificationAsync(...)` |
|
|
| `item.Notification += handler` (per-item event) | Per-item dispatch through the handler with `DataValueChange.MonitoredItem` to identify the source. |
|
|
| `s.PublishStatusChanged` / `s.StateChanged` events | Unified into a single callback `ISubscriptionNotificationHandler.OnSubscriptionStateChangedAsync(ISubscription, SubscriptionState, PublishState, CancellationToken)` that surfaces lifecycle (Opened / Created / Modified / Deleted) and publish-state (Republish / Recovered / Transferred) transitions. |
|
|
| `item.DequeueValues()` (client-side cache) | Not on V2 — values stream into the handler. The handler is the cache; callers retain whatever they need. |
|
|
| `s.LastNotification` / `s.Notifications` / `s.LastNotificationTime` | Not on `ISubscription`. The manager exposes `MissingMessageCount` / `RepublishMessageCount`; handlers maintain their own derived state (e.g. via the `publishTime` arg on each callback). |
|
|
| `s.PublishingStopped` | Not exposed as a property. Handlers derive it from `OnSubscriptionStateChangedAsync` (the `PublishState` mask flips between `Republish` / `Recovered` / `Transferred`). |
|
|
|
|
### Subscription-level service operations
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `s.RepublishAsync(seq, ct)` | Not on V2. The V2 engine auto-republishes on gap detection via `MessageProcessor.TryRepublishAsync`; there is no user-driven variant. Callers needing raw access can call `session.RepublishAsync(null, subscriptionId, seq, ct)`. |
|
|
| `s.ResendDataAsync(ct)` | Not on V2. Callers needing raw access call `session.CallAsync(null, ResendData methodId, ...)`. |
|
|
| `s.ConditionRefreshAsync(ct)` | `s.ConditionRefreshAsync(ct)` — same shape. |
|
|
| `s.ConditionRefresh2Async(monitoredItemId, ct)` | `item.ConditionRefreshAsync(ct)` on `IMonitoredItem` (per-item; no `monitoredItemId` arg needed). |
|
|
| `s.SetTriggeringAsync(triggering, links, removes, ct)` returning `SetTriggeringResponse` | `ISubscription.SetTriggeringAsync(IMonitoredItem triggeringItem, IReadOnlyCollection<IMonitoredItem>? linksToAdd, IReadOnlyCollection<IMonitoredItem>? linksToRemove, CancellationToken ct)` returning `SetTriggeringResult` with per-link statuses; plus name-based fluent overloads; plus the declarative `MonitoredItemOptions.TriggeredByNames` option (see [Triggering](#triggering-settriggering)). Supports N:M via `IMonitoredItem.TriggeringItems` plural; batches per-triggering-item requests; replays automatically on recreate/reconnect. |
|
|
| `s.TransferAsync(target, sendInitialValues, ct)` | Configure transfer-on-recreate via `ManagedSessionBuilder.WithTransferSubscriptionsOnRecreate(true)`. The per-call `sendInitialValues` toggle lives on `SubscriptionOptions.SendInitialValuesOnTransfer` (default `false`). |
|
|
| `s.SetSubscriptionDurableAsync(...)` | `ISubscription.SetAsDurableAsync(TimeSpan lifetime, CancellationToken ct = default)` → revised lifetime hours. |
|
|
| `s.SaveMessageInCache(...)` | Not on V2 (classic internal). The V2 message pipeline is channel-based with no replay cache. |
|
|
|
|
### Monitored-item management
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `s.AddItem(item)` / `s.AddItems(IEnumerable)` | `s.MonitoredItems.TryAdd(name, IOptionsMonitor<MonitoredItemOptions>, out IMonitoredItem)` — V2 keys items by a caller-supplied stable string `name`. Fluent helper: `s.TryAddMonitoredItem(name, nodeId, configure, out _)`. |
|
|
| `s.RemoveItem(item)` / `s.RemoveItems(...)` | `s.MonitoredItems.TryRemove(clientHandle)`. |
|
|
| `s.ApplyChangesAsync(ct)` | Not needed — V2 batches automatically via the options monitor. |
|
|
| `s.CreateItemsAsync(ct)` / `s.ModifyItemsAsync(ct)` / `s.DeleteItemsAsync(...)` | Implicit via `TryAdd` / options push / `TryRemove`. |
|
|
| `s.SetMonitoringModeAsync(mode, ids, ct)` | Push `MonitoredItemOptions { MonitoringMode = ... }` per item. |
|
|
| `s.ResolveItemNodeIdsAsync(ct)` | Not on V2 — callers resolve `RelativePath` to `NodeId` ahead of time via `Browse` / `TranslateBrowsePathsToNodeIds`. |
|
|
| `s.MonitoredItems` / `s.MonitoredItemCount` | `s.MonitoredItems.Items` / `s.MonitoredItems.Count`. |
|
|
|
|
### Per-item runtime
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `item.ClientHandle` | `IMonitoredItem.ClientHandle`. |
|
|
| `item.ServerId` | `IMonitoredItem.ServerId`. |
|
|
| `item.Status.Error` / `item.Status.Created` / `item.Status.Id` | `IMonitoredItem.Error` / `Created` / `ServerId`. |
|
|
| `item.AttributesModified` | Not on V2 — reconciliation is driven by `IOptionsMonitor<MonitoredItemOptions>` change tokens; there is no "modified" flag to query. |
|
|
| `item.Filter` round-trip | `IMonitoredItem.FilterResult`. |
|
|
| `item.DequeueValues()` / `item.LastValue` | Not on V2 — values flow through `ISubscriptionNotificationHandler.OnDataChangeNotificationAsync(...)`; the handler decides what to retain. |
|
|
| `item.Notification += ...` (event) | Per-item dispatch through `OnDataChangeNotificationAsync` with `DataValueChange.MonitoredItem`. |
|
|
| `item.GetEventTypeAsync` / `GetFieldValue` / `GetEventTime` / `GetFieldName` | Not on V2 `IMonitoredItem` — event-field helpers are caller-side. For typed Part 9 alarm records use [`AlarmStreamExtensions.SubscribeAlarmsAsync`](AlarmsAndConditions.md). |
|
|
| `item.TriggeringItemId` / `item.TriggeredItems` (1:1) | `IMonitoredItem.TriggeringItems` (plural, N:M) and `IMonitoredItem.TriggeredItems` (reverse "items I trigger") — both on-demand projections by stable name. Matches OPC UA Part 4 §5.13.5 N:M; runtime desired-state mutations from imperative `SetTriggeringAsync` are immediately visible. |
|
|
|
|
### Persistence (save / load)
|
|
|
|
Classic snapshots preserve the original resolved target of relative-path items
|
|
as SDK-owned metadata. `Session.Load(..., transferSubscriptions: true)` restores
|
|
`ResolvedNodeId` and `Status.NodeId` on the local monitored-item objects before
|
|
`TransferSubscriptionsAsync` or `ReactivateSubscriptionsAsync` is called.
|
|
These operations use the saved handles without translating the path again, so
|
|
a changed browse relationship or lost browse permission does not retarget an
|
|
existing server monitored item.
|
|
|
|
```csharp
|
|
// Before deliberately closing the origin session:
|
|
origin.DeleteSubscriptionsOnClose = false;
|
|
origin.Save(snapshotStream, origin.Subscriptions);
|
|
|
|
// With a connected target session and the snapshot stream positioned at its start:
|
|
var restored = new SubscriptionCollection(
|
|
target.Load(snapshotStream, transferSubscriptions: true));
|
|
if (!await target.TransferSubscriptionsAsync(restored, sendInitialValues: true, ct))
|
|
{
|
|
throw new ServiceResultException(StatusCodes.BadSubscriptionIdInvalid);
|
|
}
|
|
|
|
foreach (Subscription subscription in restored)
|
|
{
|
|
await subscription.ApplyChangesAsync(ct);
|
|
}
|
|
```
|
|
|
|
Attach notification handlers after loading and before transfer. Loading only
|
|
restores local objects; it does not perform the server transfer. For a restored
|
|
session, use `ReactivateSubscriptionsAsync` after reactivating that session
|
|
instead. No call to `ResolveItemNodeIdsAsync` or a resolution-state setter is
|
|
needed for transfer, reactivation, or subsequent maintenance.
|
|
|
|
Loading for recreation or creating a replacement server subscription resolves
|
|
paths afresh instead of reusing the saved target. Changing `StartNodeId` or
|
|
`RelativePath` on a created item, or while its creation is in flight, throws
|
|
`BadInvalidState` immediately; assigning the same value is allowed. Delete the
|
|
server item successfully before changing its target, or replace the item.
|
|
Failed deletion retains its handle and target for retry.
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `session.Save(Stream, IEnumerable<Subscription>)` (BinaryEncoder + `SubscriptionState.Encode`) | `ISubscriptionManager.SaveAsync(Stream, IServiceMessageContext, IEnumerable<ISubscription>?, CancellationToken)`, with fluent `ManagedSession.SaveSubscriptionsAsync(stream)` extension. |
|
|
| `session.Load(Stream, bool transferSubscriptions)` | `ISubscriptionManager.LoadAsync(Stream, IServiceMessageContext, handlerFactory, transferSubscriptions, CancellationToken)`, with fluent `ManagedSession.LoadSubscriptionsAsync(stream, factory, transfer, ct)`. Recreate (`false`) and transfer (`true` via TransferSubscriptions; falls back to recreate on `Bad_SubscriptionIdInvalid` / `Bad_ServiceUnsupported`) both work end-to-end. |
|
|
| `s.Snapshot(out SubscriptionState)` / `s.Restore(SubscriptionState)` | `ISubscription.Snapshot()` → `SubscriptionStateSnapshot`; `ISubscriptionManager.RestoreAsync(handler, state, transfer, ct)`; per-item `IMonitoredItem.Snapshot()` → `MonitoredItemStateSnapshot`. Fluent: `ManagedSession.SnapshotSubscriptions()` / `RestoreSubscriptionsAsync(states, factory, transfer, ct)`. |
|
|
| Triggering survives save/load | Classic: `MonitoredItemState.TriggeredItems` (server-id based, 1:1). V2: `MonitoredItemStateSnapshot.TriggeredByNames` (name based, N:M; round-trips through the binary encoder; replays automatically on recreate, see [Save / load / restore behavior](#save--load--restore-behavior)). |
|
|
|
|
### Tuning knobs
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `s.MaxMessageCount` | Not on V2 — the channel-based pipeline uses an unbounded queue with backpressure. |
|
|
| `s.MinLifetimeInterval` (property) | `SubscriptionOptions.MinLifetimeInterval`. |
|
|
| `s.DisableMonitoredItemCache` | Not on V2 — there is no per-item cache to disable (the handler is the cache). |
|
|
| `s.SequentialPublishing` | Always-on. The V2 prioritized publish-ack channel guarantees per-subscription in-order delivery; documented on `ISubscriptionNotificationHandler`. |
|
|
| `s.RepublishAfterTransfer` | Implicit and always-on; no opt-out. When `TransferSubscriptions` succeeds, the V2 engine immediately republishes every sequence number the server reports as still available in its retransmission queue (`MessageProcessor.RecoverTransferredMessagesAsync`), so a subscription that stays quiet - or only emits keep-alives - after the transfer still recovers the pending notifications. Gaps detected later are filled by `MessageProcessor.TryRepublishAsync`. |
|
|
| `s.OutstandingMessageWorkers` (per-subscription) | Manager-wide `PublishWorkerCount`. |
|
|
| `s.Id` / `s.TransferId` | `ISubscription.ServerId` (`uint`). |
|
|
| `s.Handle` (caller bookkeeping) | Not on `ISubscription`. Callers keep a side dictionary keyed by the item `Name`. |
|
|
|
|
### Engine wiring
|
|
|
|
| Classic | V2 |
|
|
|---|---|
|
|
| `Session.SubscriptionEngineFactory` defaulted to `ClassicSubscriptionEngineFactory.Instance` | Defaulted to `DefaultSubscriptionEngineFactory.Instance` on `ManagedSession`. Raw `Session` constructed with `DefaultSessionFactory` defaults to classic for backwards compatibility; opt in by passing `SubscriptionEngineFactory = DefaultSubscriptionEngineFactory.Instance`. |
|
|
| `Session.AddSubscription(Subscription)` (classic-typed) | Unchanged — classic subscriptions are still added via this API on classic-engine sessions. On a V2-engine `ManagedSession`, classic subscriptions are bridged onto the V2 publish pipeline by an internal `SubscriptionBridge`. |
|
|
| `ISession.TryGetSubscriptionManager` (V2) | Returns the V2 manager alongside `ManagedSession.Subscriptions` (classic API), so both surfaces coexist on the same session. |
|
|
|
|
### Per-item options mapping
|
|
|
|
Every classic configurable property maps directly to a field on
|
|
`MonitoredItemOptions`: `StartNodeId`, `AttributeId`, `IndexRange`,
|
|
`Encoding`, `MonitoringMode`, `SamplingInterval`, `Filter`,
|
|
`QueueSize`, `DiscardOldest`, `TimestampsToReturn`. The `Order`
|
|
and `Name` keys are V2-only (the latter is the dictionary key
|
|
used by `IMonitoredItemCollection`). `item.RelativePath` /
|
|
`item.ResolveItemNodeIdsAsync` / `item.DisplayName` are classic
|
|
caller conventions with no V2 equivalent — V2 uses a stable
|
|
per-item `Name` string instead.
|
|
|
|
## Three-API summary
|
|
|
|
| Aspect | Classic `Subscription` | V2 `ISubscriptionManager` | `IStreamingSubscription` |
|
|
|---|---|---|---|
|
|
| Notification delivery | `MonitoredItem.Notification` event | `ISubscriptionNotificationHandler` callback | `IAsyncEnumerable<T>` |
|
|
| Add monitored item | `subscription.AddItem(item); ApplyChanges()` | `subscription.MonitoredItems.TryAdd(name, options, out _)` | `streaming.SubscribeXxxAsync(...)` |
|
|
| Remove monitored item | `subscription.RemoveItem(item); ApplyChanges()` | `subscription.MonitoredItems.TryRemove(handle)` | Dispose enumerator |
|
|
| Triggering API | `SetTriggeringAsync(MonitoredItem, ArrayOf<MonitoredItem>, ArrayOf<MonitoredItem>, ct)` (1:1 from triggered side via `MonitoredItem.TriggeringItemId`) | `SetTriggeringAsync(IMonitoredItem, IReadOnlyCollection<IMonitoredItem>?, IReadOnlyCollection<IMonitoredItem>?, ct)` returning `SetTriggeringResult`; declarative `MonitoredItemOptions.TriggeredByNames`; N:M | n/a (use the underlying V2 subscription) |
|
|
| Save / restore triggering | `MonitoredItemState.TriggeredItems` (server-id based) | `MonitoredItemStateSnapshot.TriggeredByNames` (name based, N:M, round-trips through binary encoder) | n/a |
|
|
| Reconnect / transfer triggering replay | `RestoreTriggeringAsync` re-issues `SetTriggering` after every reconnect | `TransferSubscriptions`: no replay (server preserves links per spec); Recreate: automatic replay | n/a |
|
|
| Cancellation | Manual | Per call (CT in handler) | Built-in (CT on stream) |
|
|
| Composition | Hand-rolled | Hand-rolled | LINQ-style helpers |
|
|
| Backpressure | Custom logic | Custom logic | Bounded channel internally |
|
|
| Best for | Legacy callers | Long-lived, multi-item subscriptions | Short-lived / wait-for-X scenarios |
|
|
|
|
All three APIs can coexist on the same session — pick the one that
|
|
matches the use case.
|
|
|
|
## Reference
|
|
|
|
- Subscription source: `src/Opc.Ua.Client/Subscription/`
|
|
- Streaming source: `src/Opc.Ua.Client/Subscription/Streaming/`
|
|
- Helpers: `src/Opc.Ua.Client/Subscription/Streaming/StreamingSubscriptionExtensions.cs`
|
|
- Alarm streaming: `src/Opc.Ua.Client/Alarms/AlarmStreamExtensions.cs`
|
|
- Sessions architecture and engine choice: [Sessions.md](Sessions.md)
|
|
- Reference client sample:
|
|
`samples/Reference/ConsoleReferenceClient/AlarmClientSample.cs`
|