Client Subscriptions

Use OpcUaSubscription and OpcUaMonitoredItem for data-change and event delivery. Start with a connected client and a node the Session user may monitor.

Milo 1.2.0-SNAPSHOT, Java 17. Source baseline 7faaf4f0516c.

Create a subscription and receive data

A Subscription controls publishing cadence, keep-alives, lifetime, priority, and notification limits. A MonitoredItem controls sampling, queueing, filtering, and monitoring mode for one attribute. Sampling and publishing are distinct: sampling produces observations, while publishing delivers queued notifications. Requested intervals and queue sizes can be revised by the server.

This bounded example installs its listener before synchronizing the item, waits for one notification, checks its quality, and deletes the Subscription in finally. It returns only the first notification, normally the initial value. A server sends the initial value without waiting for a change.

static DataValue firstNotification(OpcUaClient client, NodeId nodeId) throws Exception {
  var subscription = new OpcUaSubscription(client);
  subscription.setPublishingInterval(100.0);
  var firstValue = new CompletableFuture<DataValue>();
  var item = OpcUaMonitoredItem.newDataItem(nodeId);
  item.setSamplingInterval(100.0);
  item.setDataValueListener((ignored, value) -> firstValue.complete(value));
  subscription.addMonitoredItem(item);
  try {
    subscription.create();
    subscription.synchronizeMonitoredItems();
    DataValue value = firstValue.get(10, TimeUnit.SECONDS);
    if (!value.statusCode().isGood()) throw new UaException(value.statusCode());
    return value;
  } finally {
    subscription.delete();
  }
}

create() allocates the server Subscription. Its async form returns a CompletionStage; use asynchronous continuations when a blocking lifecycle call would stall a callback. Inspect getRevisedPublishingInterval(), revised lifetime/keep-alive counts, and the item's getRevisedSamplingInterval() / getRevisedQueueSize() after successful creation. Requesting 100 ms does not guarantee a 100 ms physical device acquisition or delivery deadline.

Maintain desired MonitoredItems

addMonitoredItem and removeMonitoredItem change the desired local set. Changes to sampling interval, queue size, discard-oldest behavior, or filter are also local until synchronized. Call synchronizeMonitoredItems() to delete removed items, modify changed items, and create new items. The operation can partially succeed.

On MonitoredItemSynchronizationException, inspect its create, modify, and delete result lists. Each result identifies the item and separates service result from operation result. Keep successfully created items instead of blindly recreating the whole set. A successful create also does not guarantee Good data forever: a monitored node can later become unreadable or its source can fail. Inspect every delivered DataValue.statusCode().

After changing Subscription parameters, call modify(); use setPublishingMode(...) to change publishing enablement. Use setMonitoringMode(...) for item monitoring modes and inspect its per-item results. Disabling publishing and disabling sampling are different operations.

Receive events with a filter

An event item monitors a node's EventNotifier attribute, usually the Server Object or an application notifier, with an EventFilter. Its select clauses define the ordered fields; the where clause can restrict matching events. The node must route the relevant event sources and permit the user to receive them.

This example selects EventId followed by Message. The local test server emits events and the fixture checks the two returned types and their order.

static Variant[] firstEvent(OpcUaClient client, NodeId notifierId) throws Exception {
  var subscription = new OpcUaSubscription(client);
  subscription.setPublishingInterval(100.0);
  EventFilter filter =
      new EventFilterBuilder()
          .select(NodeIds.BaseEventType, new QualifiedName(0, "EventId"))
          .select(NodeIds.BaseEventType, new QualifiedName(0, "Message"))
          .build();
  var item = OpcUaMonitoredItem.newEventItem(notifierId, filter);
  var firstFields = new CompletableFuture<Variant[]>();
  item.setEventValueListener((ignored, fields) -> firstFields.complete(fields));
  subscription.addMonitoredItem(item);
  try {
    subscription.create();
    subscription.synchronizeMonitoredItems();
    return firstFields.get(10, TimeUnit.SECONDS);
  } finally {
    subscription.delete();
  }
}

Variant[] positions correspond to select-clause positions, not Java property names. Fields can be null or unavailable for an event type; validate each before casting. Inspect event-filter create/modify diagnostics when a server rejects a clause. EventId is useful for identifying an event but is not a durable replay guarantee. Condition handling also needs the model-specific acknowledgement/refresh protocol; see Alarms and Conditions.

Delivery, recovery, and triggering

Use item listeners for individual streams or a Subscription listener for batched data/events and lifecycle notifications. Item lists and value lists are paired by index. Keep listeners short: blocked delivery can delay subsequent notifications and the Publish pipeline. Dispatch expensive work to bounded application queues and define overflow behavior rather than accumulating data indefinitely.

Monitor onStatusChanged, onNotificationDataLost, onWatchdogTimerElapsed, and onTransferFailed. Session recovery attempts to preserve or transfer subscriptions, but unsuccessful transfer resets the local server state. A Bad_Timeout StatusChangeNotification also means the server deleted the Subscription; Milo unregisters and resets it. In either case, the application must recreate the Subscription and synchronize its retained desired items. Schedule recovery outside delivery callbacks and serialize it with shutdown/configuration changes. Republish can recover retained notification messages; it cannot guarantee lossless history once the server has discarded them.

For triggering, create a reporting item and one or more sampling-only items in the same Subscription, synchronize them, then call client.setTriggering with their assigned server IDs. Inspect the response's add/remove results. Trigger links cause queued samples from linked items to be reported when the triggering item reports; they are not atomic device snapshots. Reestablish links when server IDs change after recreation.

Delete and release resources

Removing an item locally does not delete its server MonitoredItem until synchronization. delete() blocks and must not run on a transport executor thread. A Good response, or Bad_SubscriptionIdInvalid, resets tracked local state. A timeout or connection failure throws and leaves local registration intact; retry after reconnect, or call reset() and let the remote lifetime expire. Normal client disconnect closes the Session with deleteSubscriptions=true. reset() only discards local knowledge and is not a substitute for remote deletion. Delete subscriptions before disconnecting when possible, and stop application consumers/executors you own. If the server is unavailable, record cleanup failure and reconcile on recovery; do not infer immediate remote deletion from local shutdown.

Examples and reference

• SubscriptionDataExample: Receive data changes and inspect subscription delivery.

• SubscriptionEventExample: Select and receive event fields.

• Runnable fixture: WikiClientGuideTest

• OpcUaSubscription lifecycle and listeners

• OpcUaMonitoredItem configuration and state

• EventFilterBuilder

• Triggering service example

Related pages

Client SDK · Recovery and limits · Server sampling · Server events