This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 69856151bcf3 CAMEL-25691: camel-milo - recreate the subscription when
it stops publishing (#27910)
69856151bcf3 is described below
commit 69856151bcf366aba832c9a88de7bb4889f6544c
Author: Serdar Gökay <[email protected]>
AuthorDate: Sun Oct 18 10:50:51 2026 +0300
CAMEL-25691: camel-milo - recreate the subscription when it stops
publishing (#27910)
Co-authored-by: Claude Fable 5.1 <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../camel/catalog/docs/milo-client-component.adoc | 23 +++
.../src/main/docs/milo-client-component.adoc | 23 +++
.../milo/client/internal/SubscriptionManager.java | 209 ++++++++++++++++++++-
.../component/milo/MonitorItemLifecycleTest.java | 205 +++++++++++++++++++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 +
5 files changed, 459 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
index 02055945a739..6b9e7e4b13e5 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
@@ -204,6 +204,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),
diff --git a/components/camel-milo/src/main/docs/milo-client-component.adoc
b/components/camel-milo/src/main/docs/milo-client-component.adoc
index 02055945a739..6b9e7e4b13e5 100644
--- a/components/camel-milo/src/main/docs/milo-client-component.adoc
+++ b/components/camel-milo/src/main/docs/milo-client-component.adoc
@@ -204,6 +204,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),
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
index ed096e646d49..e8dab0b93336 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
@@ -31,6 +31,7 @@ import java.util.Objects;
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;
@@ -49,6 +50,8 @@ import
org.apache.camel.component.milo.client.MonitorFilterConfiguration;
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;
@@ -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
@@ -298,6 +352,29 @@ public class SubscriptionManager {
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) {
@@ -676,6 +753,129 @@ public class SubscriptionManager {
}
}
+ /**
+ * 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 {
@@ -850,6 +1050,9 @@ public class SubscriptionManager {
}
});
+ final SubscriptionListenerImpl listener = new
SubscriptionListenerImpl();
+ client.addSessionActivityListener(listener);
+
try {
// connect is called inside this try block to ensure that
disconnect is called on exceptions
@@ -858,7 +1061,7 @@ public class SubscriptionManager {
// 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);
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
index cec8822d6eb0..f6797ff65d19 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.milo;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import org.apache.camel.EndpointInject;
@@ -31,21 +32,29 @@ import
org.apache.camel.component.milo.client.MiloClientEndpoint;
import org.apache.camel.component.milo.server.MiloServerComponent;
import org.apache.camel.component.mock.MockEndpoint;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
+import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
+import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
+import org.eclipse.milo.opcua.sdk.server.Session;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import static org.awaitility.Awaitility.await;
+import static
org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * Testing the monitored items of consumers sharing a connection, when a
consumer is restarted, its node does not exist
- * or the server rejects the CreateMonitoredItems service call
+ * Testing the monitored items of consumers sharing a connection, when a
consumer is restarted, its node does not exist,
+ * the server rejects the CreateMonitoredItems service call or the
subscription stops publishing or times out
*/
public class MonitorItemLifecycleTest extends AbstractMiloServerTest {
@@ -96,6 +105,11 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
private static final String MOCK_ITEMS_1_2 = "mock:items12";
+ // the route of the first item takes longer than the watchdog of a
shortened subscription for this value, once
+ private static final String SLOW_VALUE = "Slow";
+
+ private final AtomicBoolean slowValueProcessed = new AtomicBoolean();
+
@EndpointInject(MOCK_ITEM_1)
protected MockEndpoint item1Endpoint;
@@ -128,7 +142,12 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
from(resolve(MILO_CLIENT_UNKNOWN_ITEM)).routeId("unknownItem").to(MOCK_UNKNOWN_ITEM);
// the consumers deliver the current values while they start,
so the tests start these routes
-
from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").autoStartup(false).to(MOCK_ITEM_1);
+
from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").autoStartup(false)
+ // the slow route under test: a synchronous delay
holds the delivery thread, once
+ .filter(exchange -> isSlowValueOnce(exchange))
+ .delay(3000).syncDelayed().end() // the delay is a
block of its own
+ .end()
+ .to(MOCK_ITEM_1);
from(resolve(MILO_CLIENT_ITEM_2)).routeId("item2").autoStartup(false).to(MOCK_ITEM_2);
from(resolve(MILO_CLIENT_ITEMS_1_2)).routeId("items12").autoStartup(false).to(MOCK_ITEMS_1_2);
@@ -142,6 +161,11 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
};
}
+ private boolean isSlowValueOnce(Exchange exchange) {
+ final Object value =
exchange.getMessage().getBody(DataValue.class).getValue().getValue();
+ return SLOW_VALUE.equals(value) &&
slowValueProcessed.compareAndSet(false, true);
+ }
+
@Test
void restartedConsumerReceivesEachValueOnce() throws Exception {
producer1.sendBody("Foo");
@@ -264,6 +288,181 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
.getBody(DataValue.class)));
}
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatStopsPublishingIsRecreatedInTheSameSession() throws
Exception {
+ producer1.sendBody("Foo");
+
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final OpcUaSubscription subscription =
client.getSubscriptions().get(0);
+ // shortens the watchdog of the subscription to a few publishing
intervals
+ subscription.setMaxKeepAliveCount(uint(1));
+ subscription.setLifetimeCount(uint(3));
+ subscription.modify();
+
+ // the server deletes the subscription and answers the pending
publish requests without one
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElseThrow();
+
assertTrue(client.deleteSubscriptions(List.of(subscriptionId)).getResults()[0].isGood());
+
+ producer1.sendBody("Bar");
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+
+ assertSame(client, connection.getOpcUaClient());
+ assertNotEquals(subscriptionId,
subscription.getSubscriptionId().orElseThrow());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatTimesOutOnTheServerIsRecreatedInTheSameSession()
throws Exception {
+ producer1.sendBody("Foo");
+
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final UInteger subscriptionId =
client.getSubscriptions().get(0).getSubscriptionId().orElseThrow();
+
+ timeOutSubscription(subscriptionId);
+
+ producer1.sendBody("Bar");
+ await().atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+
+ assertSame(client, connection.getOpcUaClient());
+ assertNotEquals(subscriptionId,
client.getSubscriptions().get(0).getSubscriptionId().orElseThrow());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void subscriptionThatASlowRouteHeldUpIsKept() throws Exception {
+ producer1.sendBody("Foo");
+
+ item1Endpoint.expectedMinimumMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final OpcUaSubscription subscription =
client.getSubscriptions().get(0);
+ // shortens the watchdog of the subscription to a few publishing
intervals
+ subscription.setMaxKeepAliveCount(uint(1));
+ subscription.modify();
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElseThrow();
+ // replaced when the server confirms the subscription
+ final OpcUaSubscription.ServerState stateBefore =
subscription.getServerState().orElseThrow();
+
+ // the client sends no publish request while the route processes
the value, so the watchdog elapses
+ producer1.sendBody(SLOW_VALUE);
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
countValue(item1Endpoint, SLOW_VALUE) >= 1);
+
+ // a recreated subscription would deliver the current value again
+ await().during(3, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .until(() -> countValue(item1Endpoint, SLOW_VALUE) == 1);
+ assertEquals(subscriptionId,
subscription.getSubscriptionId().orElse(null));
+ assertNotSame(stateBefore,
subscription.getServerState().orElseThrow(), "the subscription was not
checked");
+ producer1.sendBody("Bar");
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Bar").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void
nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws
Exception {
+ producer1.sendBody("Foo");
+ producer2.sendBody("Bar");
+ context.getRouteController().startRoute("item1");
+
+ items12Endpoint.expectedMessageCount(2);
+ context.getRouteController().startRoute("items12");
+ items12Endpoint.assertIsSatisfied();
+
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final OpcUaSubscription subscription =
client.getSubscriptions().get(0);
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElseThrow();
+
+ // the second node is removed on the server before the
subscription times out there
+
getServer().getAddressSpaceManager().getManagedNode(ExpandedNodeId.parse(NODE_ITEM_2)).orElseThrow().delete();
+ timeOutSubscription(subscriptionId);
+
+ // the subscription is recreated in the same session, with the
monitored items of the first node only
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(() -> subscription.getSubscriptionId().filter(id ->
!id.equals(subscriptionId)).isPresent());
+ assertSame(client, connection.getOpcUaClient());
+ // the items are added one by one before they are created, so the
count alone passes 2 on the way to 3
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
subscription.getMonitoredItems().size() == 2
+ && subscription.isMonitoredItemsSynchronized());
+
+ // the consumer of the second node is told that it is gone, the
consumers of the first node keep receiving
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
items12Endpoint.getReceivedExchanges().stream()
+ .anyMatch(exchange ->
NODE_ITEM_2.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID))
+ && new StatusCode(StatusCodes.Bad_NodeIdUnknown)
+
.equals(exchange.getMessage().getBody(DataValue.class).getStatusCode())));
+ producer1.sendBody("Baz");
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Baz").accept(lastExchange(item1Endpoint).getMessage()
+ .getBody(DataValue.class)));
+ assertSame(client, connection.getOpcUaClient());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ private OpcUaServer getServer() {
+ return context.getComponent("milo-server",
MiloServerComponent.class).getServer();
+ }
+
+ /**
+ * The server reports the lifetime of the subscription as expired and
removes it
+ */
+ private void timeOutSubscription(final UInteger subscriptionId) {
+ for (final Session session :
getServer().getSessionManager().getAllSessions()) {
+ final var subscriptions = session.getSubscriptionManager();
+ final var subscription =
subscriptions.getSubscription(subscriptionId);
+ if (subscription != null) {
+ subscriptions.sendStatusChangeNotification(subscription, new
StatusCode(StatusCodes.Bad_Timeout));
+ subscriptions.removeSubscription(subscriptionId);
+ }
+ }
+ }
+
+ private static long countValue(final MockEndpoint endpoint, final Object
value) {
+ return endpoint.getReceivedExchanges().stream()
+ .filter(exchange ->
value.equals(exchange.getMessage().getBody(DataValue.class).getValue().getValue()))
+ .count();
+ }
+
private static Exchange lastExchange(final MockEndpoint endpoint) {
final List<Exchange> exchanges = endpoint.getReceivedExchanges();
return exchanges.get(exchanges.size() - 1);
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 85e983b9376f..e11786cc9c92 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1551,6 +1551,11 @@ Stopping a consumer now deletes its monitored items.
Previously they stayed on t
the shared connection open, so a restarted consumer received each value twice.
While the server is unreachable, the
stop waits until the client's next attempt to reconnect has failed.
+When the server removes the subscription that the consumers of a connection
share, or the subscription times out on the
+server, it is now created again in the existing session; previously the
consumers stopped receiving values until the
+connection was rebuilt for another reason. The monitored items deliver their
current values again after the
+subscription is recreated.
+
=== camel-protobuf - a bytes field in a Map
When a protobuf message is converted to a `Map` (for example
`getBody(Map.class)` after unmarshalling), the value of
