Client Subscriptions

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

Reading and writing give you values when you ask for them. With a subscription, the server watches the nodes you name and notifies you when a value changes or an event fires, so your application reacts without polling. This page subscribes to the thermostat's Setpoint and follows that Subscription through to shutdown. You need a connected client from Connecting and a node the Session user may monitor.

How a subscription works

A subscription has two parts. The Subscription controls delivery, which covers the publishing interval, keep-alives, lifetime, priority, and the maximum number of notifications per publish. Each MonitoredItem watches one attribute of one node and controls its sampling, queueing, filtering, and monitoring mode. Sampling produces observations, and publishing delivers the queued notifications to you.

Milo treats both as desired state that you edit locally and then apply. Constructing an OpcUaSubscription, adding an OpcUaMonitoredItem, or changing an item's settings touches only local objects. create() creates the Subscription on the server, and synchronizeMonitoredItems() makes the server's MonitoredItems match your local list. Because the local objects hold the whole desired state, you can batch edits into one synchronization and rebuild the server side from them after the server loses it.

Watch the setpoint

The method below subscribes to Setpoint, prints every value the server reports, and returns the running Subscription so the caller can delete it at shutdown. Build setpointId from the tutorial namespace index as Your first client does for Temperature: new NodeId(namespaceIndex, "Setpoint").

static OpcUaSubscription watchSetpoint(OpcUaClient client, NodeId setpointId) throws UaException {
  var subscription = new OpcUaSubscription(client, 500.0);

  OpcUaMonitoredItem item = OpcUaMonitoredItem.newDataItem(setpointId);
  item.setSamplingInterval(100.0);
  item.setDataValueListener(
      (monitoredItem, value) -> {
        StatusCode status = value.statusCode();
        if (status.isGood()) {
          Object setpoint = value.value().value();
          System.out.println("Setpoint: " + setpoint);
        } else {
          System.out.println("Setpoint unavailable: " + status);
        }
      });
  subscription.addMonitoredItem(item);

  subscription.create();
  try {
    subscription.synchronizeMonitoredItems();
  } catch (MonitoredItemSynchronizationException e) {
    subscription.delete();
    throw e;
  }

  return subscription;
}

Everything before create() is local. If synchronizeMonitoredItems() cannot create the item, the method deletes the new Subscription so nothing is left publishing, and rethrows.

The listener goes on first because the server sends the item's current value as soon as the item exists, without waiting for a change. Against a freshly started tutorial server, the method prints Setpoint: 22.0 almost at once. After that, a write to Setpoint (Writing) or an AdjustSetpoint call (Methods) prints the new value. Two changes within one 500 ms publishing interval arrive as one line with the later value, because the item keeps the default queue size of 1.

The server can revise what you request. After create(), getRevisedPublishingInterval(), getRevisedLifetimeCount(), and getRevisedMaxKeepAliveCount() report what it granted, and after synchronization the item reports getRevisedSamplingInterval() and getRevisedQueueSize(). The two-argument constructor derives the requested lifetime and keep-alive counts from the publishing interval. Even a granted 100 ms sampling interval does not guarantee a 100 ms physical device acquisition or delivery deadline.

create(), modify(), delete(), and setPublishingMode(...) block until the server answers, so never call them on a transport executor thread, which is where Milo calls your listeners. Each has an async form, such as createAsync(), that returns a CompletionStage for use with continuations when a blocking call would stall a callback.

Change what you monitor

Changes to a running subscription follow the same pattern. addMonitoredItem and removeMonitoredItem change the local set, and setSamplingInterval, setQueueSize, setDiscardOldest, and setFilter change an item locally. synchronizeMonitoredItems() then deletes removed items, modifies changed ones, and creates new ones, so a removed item keeps running on the server until you call it. To sample Setpoint once a second, for example, call item.setSamplingInterval(1000.0) and synchronize. Each item is a separate server operation, so a synchronization can partially succeed (When things go wrong).

For Subscription parameters, call the setters and then modify(). To pause delivery without removing anything, setPublishingMode(boolean) turns publishing off or on for the whole Subscription, while setMonitoringMode(mode, items) changes whether individual items sample and report and returns a result per item to inspect.

Receive events

An event item watches a node's EventNotifier attribute instead of its Value. The select clauses of its EventFilter list the fields to return, in order, and an optional where clause restricts which events match. The node is usually the Server Object or an application notifier, and it must route the relevant event sources and permit the user to receive them.

The tutorial server fires no events yet, so this example works against any server that does. It adds an item on the Server Object to an existing Subscription, such as the one watchSetpoint returns. A Milo server treats the Server Object as the root notifier, so once Server events adds a "setpoint changed" event to the tutorial server, this item prints each one.

static OpcUaMonitoredItem watchServerEvents(OpcUaSubscription subscription) throws UaException {
  EventFilter filter =
      new EventFilterBuilder()
          .select(NodeIds.BaseEventType, new QualifiedName(0, "EventId"))
          .select(NodeIds.BaseEventType, new QualifiedName(0, "Message"))
          .build();

  OpcUaMonitoredItem item = OpcUaMonitoredItem.newEventItem(NodeIds.Server, filter);
  item.setEventValueListener(
      (monitoredItem, fields) -> {
        // Fields arrive in select-clause order: EventId, then Message.
        Object message = fields[1].value();
        if (message instanceof LocalizedText text) {
          System.out.println("Event: " + text.text());
        }
      });

  subscription.addMonitoredItem(item);
  subscription.synchronizeMonitoredItems();

  return item;
}

The Variant[] positions match the select clauses, not Java property names, so the listener reads fields by position. A field can be null or unavailable for a particular event type, so check each value's type before using it, as the instanceof test does. To stop watching, remove the returned item and synchronize again.

EventId identifies an event but gives no durable replay of missed events. Conditions also need their model-specific acknowledgement and refresh protocol, described in Alarms and Conditions.

Handle notifications

Item listeners suit independent streams like Setpoint. For batched delivery and lifecycle notifications, set a SubscriptionListener with setSubscriptionListener(...). Its onDataReceived and onEventReceived callbacks receive a list of items and a list of values or event fields, paired by index. Its other callbacks report keep-alives, status changes, lost notifications, watchdog timeouts, and failed transfers.

Milo delivers notifications synchronously as backpressure. It sends the Publish request that replaces the one just answered only after your listeners return, so a blocking listener delays later notifications and holds back new Publish requests. Keep listeners short, and hand expensive work to a bounded application queue with defined overflow behavior.

Report items together with triggering

Triggering lets one item's report pull in samples from others. Create a reporting item and one or more sampling-only items in the same Subscription, for example with OpcUaMonitoredItem.newDataItem(nodeId, MonitoringMode.Sampling), and synchronize them. Then call client.setTriggering(...) with the server IDs of the Subscription, the triggering item, and the items to link, and inspect the add and remove results. When the triggering item reports, the linked items report their queued samples, though not as an atomic device snapshot. Reestablish the links whenever recreation changes the server IDs.

When things go wrong

Subscription problems surface in synchronization results, in the status of delivered values, and in SubscriptionListener callbacks.

What you see	What it means	What to do
synchronizeMonitoredItems() throws MonitoredItemSynchronizationException	At least one create, modify, or delete failed. getCreateResults(), getModifyResults(), and getDeleteResults() return one result per item, naming the item and separating the service result from the operation result.	Keep the items that succeeded. Fix or remove the failed ones and synchronize again instead of recreating the whole set.
A delivered DataValue has a Bad statusCode()	The node became unreadable or its source failed. A successful create does not promise Good data later.	Check every delivered status, as the Setpoint listener does.
The server rejects an event filter clause	The item's create or modify operation result is Bad, or the filter result reports a problem with a clause.	Read the operation result from the exception. getFilterResult() holds the filter result from the last successful create or modify only. For the server's diagnostics on a rejected filter, call client.createMonitoredItems(...) or modifyMonitoredItems(...) directly and read each result's getFilterResult().
onStatusChanged with Bad_Timeout	The server deleted the Subscription. Milo unregisters it and resets its server state.	Recreate the Subscription and synchronize its retained items.
onTransferFailed	A reconnect could not transfer the Subscription to the new Session, so Milo resets its server state.	Recreate the Subscription and synchronize its retained items.
onNotificationDataLost	Republish could not recover missing notifications. It returns only messages the server still retains, so it cannot guarantee lossless history.	Reconcile from current values, backfill from history, or record a gap.
onWatchdogTimerElapsed	No Publish response for this Subscription arrived within the watchdog multiplier (1.5 by default) times the keep-alive interval. Silence no longer means values are unchanged.	Consider deleting and recreating the Subscription.

When onTransferFailed or a Bad_Timeout status change resets a Subscription, the OpcUaSubscription still holds its MonitoredItems as desired state, so you recover by calling create() and synchronizeMonitoredItems() on the same object. Serialize that work with shutdown and configuration changes, so a late recovery cannot recreate a Subscription you are deleting. Recovery and limits shows a listener that does this on an application executor, explains the reconnect stages, and covers rebuilding triggering links.

Delete the subscription at shutdown

At shutdown, delete each Subscription you own and then disconnect, so you get an explicit result for each Subscription instead of relying on the Session close.

static void stopWatching(OpcUaClient client, OpcUaSubscription subscription) throws UaException {
  try {
    subscription.delete();
  } finally {
    client.disconnect();
  }
}

A Good response or Bad_SubscriptionIdInvalid resets the local state. A timeout or connection failure throws and leaves the local registration intact, which is why the example still disconnects in finally. Outside shutdown, retry the delete after reconnecting, or call reset() and let the server expire the Subscription when its lifetime runs out. reset() only discards local knowledge and does not replace remote deletion.

A normal disconnect() closes the Session with deleteSubscriptions=true, so a reachable server also deletes any Subscription that remains. Local shutdown deletes nothing on an unavailable server, so record the cleanup failure and reconcile when the server returns. Also stop the application consumers and executors that your listeners feed.

Examples and reference

• SubscriptionDataExample: Receive data changes and inspect subscription delivery.

• SubscriptionEventExample: Select and receive event fields.

• Sample source: ClientSnippets

• OpcUaSubscription lifecycle and listeners

• OpcUaMonitoredItem configuration and state

• EventFilterBuilder

• Triggering service example

Next steps

You can now watch Setpoint change as it happens. To change it through the thermostat's own logic, which checks the result against its 0 to 100 range, continue with Methods.

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