Skip to content

CAMEL-25691: camel-milo - recreate the subscription when it stops publishing #27910

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 5 commits into from
Oct 18, 2026
Merged

CAMEL-25691: camel-milo - recreate the subscription when it stops publishing #27910

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
675dec4
CAMEL-25691: camel-milo - recreate the subscription when it stops pub…
etzel-mes Oct 15, 2026
26050d4
CAMEL-25691: camel-milo - recreate the subscription after a confirmed…
etzel-mes Oct 15, 2026
a00e7f6
CAMEL-25691: camel-milo - keep a subscription that the server still h…
etzel-mes Oct 16, 2026
202f1e2
CAMEL-25691: camel-milo - the slow route of the test uses a synchrono…
davsclaus Oct 17, 2026
6f24e05
CAMEL-25691: camel-milo - end the delay block before the filter in th…
davsclaus Oct 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions ...rc/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,29 @@ The default `omitNullValues=true` also omits a null value with a Good status tha
does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a
`String`.

=== Subscriptions that stop publishing

The consumers of a connection share one subscription on the server. When the server stops publishing for it (no
notification or keep-alive within about one and a half keep-alive intervals), Camel checks whether the server still
has the subscription. If the server does not confirm it, or when the server reports that the subscription timed out
(`Bad_Timeout`), Camel deletes it and, once the server has confirmed that it is gone, creates it again in the existing
session. A node that can no longer be monitored is reported as described above and does not affect the other nodes.

A route that is slow to process a value stops the publishing as well, because the client asks the server for more
notifications only as the routes finish processing the ones it received. The subscription is then kept, unless the
route takes longer than the lifetime of the subscription, which lets it time out on the server.

A check or a delete that fails because the connection was lost is left to the session recovery of the client, which
restores the session once the server can be reached again. The subscription then resumes publishing, or it is created
again as above when the server no longer has it. If the server no longer has the session either, for example after a
restart, the subscription cannot be transferred to a new session and the connection is rebuilt. A network that drops
silently can stop the publishing before the client notices that the connection is lost; if the network is back before
that, the check reaches the server and is handled as above.

Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server.
After the subscription is recreated, every monitored item delivers its current value again, and value changes in
between are not delivered.

=== Method ID

It is possible to perform methods calls on OPC UA nodes. If the parameter `method` is set to the Node ID of a method call (the node ID must be set to the parent object in this case),
Expand Down
23 changes: 23 additions & 0 deletions components/camel-milo/src/main/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,29 @@ The default `omitNullValues=true` also omits a null value with a Good status tha
does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a
`String`.

=== Subscriptions that stop publishing

The consumers of a connection share one subscription on the server. When the server stops publishing for it (no
notification or keep-alive within about one and a half keep-alive intervals), Camel checks whether the server still
has the subscription. If the server does not confirm it, or when the server reports that the subscription timed out
(`Bad_Timeout`), Camel deletes it and, once the server has confirmed that it is gone, creates it again in the existing
session. A node that can no longer be monitored is reported as described above and does not affect the other nodes.

A route that is slow to process a value stops the publishing as well, because the client asks the server for more
notifications only as the routes finish processing the ones it received. The subscription is then kept, unless the
route takes longer than the lifetime of the subscription, which lets it time out on the server.

A check or a delete that fails because the connection was lost is left to the session recovery of the client, which
restores the session once the server can be reached again. The subscription then resumes publishing, or it is created
again as above when the server no longer has it. If the server no longer has the session either, for example after a
restart, the subscription cannot be transferred to a new session and the connection is rebuilt. A network that drops
silently can stop the publishing before the client notices that the connection is lost; if the network is back before
that, the check reaches the server and is handled as above.

Any other failure of the delete, or a subscription that cannot be created again, rebuilds the connection to the server.
After the subscription is recreated, every monitored item delivers its current value again, and value changes in
between are not delivered.

=== Method ID

It is possible to perform methods calls on OPC UA nodes. If the parameter `method` is set to the Node ID of a method call (the node ID must be set to the parent object in this case),
Expand Down
209 changes: 206 additions & 3 deletions ...lo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
Expand All @@ -49,6 +50,8 @@
import org.eclipse.milo.opcua.sdk.client.DiscoveryClient;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.OpcUaClientConfigBuilder;
import org.eclipse.milo.opcua.sdk.client.SessionActivityListener;
import org.eclipse.milo.opcua.sdk.client.UaSession;
import org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
import org.eclipse.milo.opcua.sdk.client.identity.CompositeProvider;
import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider;
Expand Down Expand Up @@ -97,13 +100,64 @@ public class SubscriptionManager {

private static final Logger LOG = LoggerFactory.getLogger(SubscriptionManager.class);

/**
* The status codes of a request that failed because the connection or the session was lost, which the session
* recovery of the client deals with: a closed connection, and the service faults on which the client leaves the
* active session because its session or its secure channel is no longer valid. The faults are those of
* {@code SESSION_ERROR} and {@code SECURE_CHANNEL_ERROR} in the session state machine of Milo 1.1
* ({@code SessionFsmFactory}).
*/
private static final Set<Long> SESSION_LOST = Set.of(
StatusCodes.Bad_ConnectionClosed, StatusCodes.Bad_SessionClosed, StatusCodes.Bad_SessionIdInvalid,
StatusCodes.Bad_SessionNotActivated, StatusCodes.Bad_SecureChannelIdInvalid,
StatusCodes.Bad_SecurityChecksFailed, StatusCodes.Bad_TcpSecureChannelUnknown,
StatusCodes.Bad_RequestTypeInvalid);

private final AtomicLong clientHandleCounter = new AtomicLong();

private final class SubscriptionListenerImpl implements OpcUaSubscription.SubscriptionListener {
/**
* Listens to the subscription of a connection and to the state of its session. It is registered on the client
* before the client connects, so that it sees every change of the session state.
*/
private final class SubscriptionListenerImpl
implements OpcUaSubscription.SubscriptionListener, SessionActivityListener {

private volatile boolean sessionActive;

@Override
public void onSessionActive(final UaSession session) {
this.sessionActive = true;
}

@Override
public void onSessionInactive(final UaSession session) {
this.sessionActive = false;
}

/**
* Whether a request failed because the connection or the session was lost, which the session recovery of the
* client deals with. A request that fails while the session is not active counts too, whatever its status: one
* that was sent after the loss waits for that recovery and fails with the error of a failed reconnect attempt.
*/
boolean isSessionLost(final StatusCode status) {
return !this.sessionActive || SESSION_LOST.contains(status.getValue());
}

@Override
public void onStatusChanged(final OpcUaSubscription subscription, final StatusCode status) {
LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status);
if (status.getValue() == StatusCodes.Bad_Timeout) {
LOG.warn("Subscription timed out on the server, recreating it");
recoverSubscription(subscription, this);
} else {
LOG.info("Subscription status changed {} : {}", subscription.getSubscriptionId(), status);
}
}

@Override
public void onWatchdogTimerElapsed(final OpcUaSubscription subscription) {
LOG.debug("Subscription {} stopped publishing, checking it on the server",
subscription.getSubscriptionId().orElse(null));
checkSubscription(subscription, this);
}

@Override
Expand Down Expand Up @@ -298,6 +352,29 @@ private NodeId lookupMonitoredNode(final ExpandedNodeId nodeId) throws Exception
return index == null ? null : toNodeId(index, nodeId);
}

/**
* Creates the subscription on the server again, with the monitored items it had, after the server lost it.
*
* @param subscriptions the registered subscriptions, by client handle
* @return the bad status values of the nodes that could not be monitored again, to be delivered
* to their consumers after the lock of the manager has been released
* @throws Exception if the subscription could not be created or the server rejected the
* CreateMonitoredItems service call, the connection has to be rebuilt
*/
public List<PendingValue> recreate(final Map<UInteger, Subscription> subscriptions) throws Exception {
final Map<UInteger, Subscription> monitored = new HashMap<>();
this.goodSubscriptions.keySet()
.forEach(clientHandle -> monitored.put(clientHandle, subscriptions.get(clientHandle)));
this.goodSubscriptions.clear();

// the monitored items were reset together with the subscription, so they are dropped without a service call
this.manager.removeMonitoredItems(this.manager.getMonitoredItems());
this.manager.deleteMonitoredItems();

this.manager.create();
return putSubscriptions(monitored);
}

public void deactivate(final List<UInteger> clientHandles) throws Exception {
final List<OpcUaMonitoredItem> items = new ArrayList<>(clientHandles.size());
for (final UInteger clientHandle : clientHandles) {
Expand Down Expand Up @@ -676,6 +753,129 @@ public OpcUaClient getOpcUaClient() {
}
}

/**
* Checks the subscription of a connection on the server after it stopped publishing, and recovers it unless the
* server confirms that it still has it.
* <p>
* A subscription that the server still has stops publishing too while a route is slow: the client replaces each of
* its outstanding publish requests only once the notifications of that request have been processed, so the server
* has none left to answer. The check is a SetPublishingMode request that enables publishing; the subscription is
* always created with publishing enabled, so the request does not change its settings.
* <p>
* If the client has reset the subscription in the meantime, it is already being recovered: the client resets it
* when the server reports that it timed out, when it was deleted and when it could not be transferred, and each of
* these recovers it. A check that fails because the connection or the session was lost is left to the session
* recovery of the client, as a failed delete in
* {@link #recoverSubscription(OpcUaSubscription, SubscriptionListenerImpl)} is. Any other answer goes on to
* {@link #recoverSubscription(OpcUaSubscription, SubscriptionListenerImpl)}.
*/
private void checkSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) {
final UInteger subscriptionId = subscription.getSubscriptionId().orElse(null);
subscription.setPublishingModeAsync(true).whenCompleteAsync((unit, e) -> {
final StatusCode status = e == null
? StatusCode.GOOD
: UaException.extractStatusCode(e).orElse(new StatusCode(StatusCodes.Bad_UnexpectedError));
if (status.isGood()) {
LOG.info("Subscription {} stopped publishing but still exists on the server, keeping it",
subscriptionId);
} else if (status.getValue() == StatusCodes.Bad_InvalidState) {
LOG.debug("Subscription {} was reset while it was checked, it is already being recovered",
subscriptionId);
} else if (listener.isSessionLost(status)) {
LOG.info("Session lost while checking subscription {}, leaving it to the session recovery: {}",
subscriptionId, status);
} else {
LOG.warn("Subscription {} stopped publishing and was not confirmed by the server, recreating it: {}",
subscriptionId, status);
recoverSubscription(subscription, listener);
}
}, clientExecutor(subscription));
}

/**
* Recovers the subscription of a connection after the server did not confirm it or reported that it timed out.
* <p>
* The subscription is deleted first, and it is only created again, inside the existing session, once the server has
* confirmed that it no longer exists; one that the client has already reset because it timed out is deleted without
* a service call. A delete that fails because the connection or the session was lost, or that fails while the
* session is not active, is left to the session recovery of the client: it resumes the subscription, whose watchdog
* is armed again once publishing resumes, or reports that the subscription could not be transferred, which rebuilds
* the connection. Any other failure rebuilds the connection, which closes the session together with its
* subscriptions.
* <p>
* The delete neither blocks nor holds {@code lock}: a request sent before the connection was lost only fails once
* the client notices the loss, and holding {@code lock} until then would hold up the producers and
* {@link #getOpcUaClient()} as well.
*/
private void recoverSubscription(final OpcUaSubscription subscription, final SubscriptionListenerImpl listener) {
subscription.deleteAsync().whenCompleteAsync((unit, e) -> {
final StatusCode status = e == null
? StatusCode.GOOD
: UaException.extractStatusCode(e).orElse(new StatusCode(StatusCodes.Bad_UnexpectedError));
if (status.isGood() || status.getValue() == StatusCodes.Bad_SubscriptionIdInvalid) {
recreateSubscription(subscription);
} else if (listener.isSessionLost(status)) {
LOG.info("Session lost while deleting subscription {}, leaving it to the session recovery: {}",
subscription.getSubscriptionId().orElse(null), status);
} else {
lock.lock();
try {
if (isCurrent(subscription)) {
handleConnectionFailure(e);
}
} finally {
lock.unlock();
}
}
}, clientExecutor(subscription));
}

/**
* The executor that the client of a subscription delivers its values on, which handles the outcome of a check or a
* delete. Creating the subscription again and rebuilding the connection block until the server answers, and a
* request sent while the connection is lost waits for the channel until a reconnect attempt restores it or fails.
* The client schedules those attempts on the shared scheduled executor of Milo, which {@link #executor} is as well
* and which has a single thread on a machine with one processor, so the wait must not hold one of its threads.
* <p>
* Milo asks that blocking calls are not made on the threads of this executor either, because they complete the
* responses. The executor of the transport is not set here, so this is the shared executor of Milo, a cached pool
* that starts another thread for a response while one of its threads waits, as it does while a route processes a
* value.
*/
private static Executor clientExecutor(final OpcUaSubscription subscription) {
return subscription.getClient().getTransport().getConfig().getExecutor();
}

/**
* Creates the subscription of the current connection again, inside the existing session, together with the
* monitored items it had. Does nothing if the subscription no longer belongs to the current connection or has been
* created again already. A node that cannot be monitored again is reported to its consumer, as when the consumer
* starts; if the subscription itself cannot be created again, the connection is rebuilt.
*/
private void recreateSubscription(final OpcUaSubscription subscription) {
final List<PendingValue> pendingValues = new ArrayList<>();
lock.lock();
try {
if (!isCurrent(subscription) || subscription.getSubscriptionId().isPresent()) {
LOG.debug("Subscription is stale or already recreated");
return;
}
pendingValues.addAll(this.connected.recreate(this.subscriptions));
LOG.info("Subscription recreated as {}", subscription.getSubscriptionId().orElse(null));
} catch (final Exception e) {
handleConnectionFailure(e);
} finally {
lock.unlock();
}

// deliver outside of the lock, the routes may use the connection
pendingValues.forEach(PendingValue::deliver);
}

private boolean isCurrent(final OpcUaSubscription subscription) {
return !this.disposed && this.connected != null && this.connected.manager == subscription;
}

private void handleConnectionFailure(final Throwable e) {
lock.lock();
try {
Expand Down Expand Up @@ -850,6 +1050,9 @@ private Connected performConnect() throws Exception {
}
});

final SubscriptionListenerImpl listener = new SubscriptionListenerImpl();
client.addSessionActivityListener(listener);

try {

// connect is called inside this try block to ensure that disconnect is called on exceptions
Expand All @@ -858,7 +1061,7 @@ private Connected performConnect() throws Exception {
// Create subscription synchronously - the create() method blocks until complete
OpcUaSubscription subscription = new OpcUaSubscription(client);
subscription.setPublishingInterval(this.configuration.getRequestedPublishingInterval());
subscription.setSubscriptionListener(new SubscriptionListenerImpl());
subscription.setSubscriptionListener(listener);
subscription.create();

return new Connected(client, subscription);
Expand Down
Loading
Loading