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 bf6f7bd1252b6..d7bfda50f74f1 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 @@ -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), 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 bf6f7bd1252b6..d7bfda50f74f1 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -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), 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 ed096e646d49b..e8dab0b933369 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.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.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 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 @@ 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 recreate(final Map subscriptions) throws Exception { + final Map 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 clientHandles) throws Exception { final List items = new ArrayList<>(clientHandles.size()); for (final UInteger clientHandle : clientHandles) { @@ -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. + *

+ * 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. + *

+ * 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. + *

+ * 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. + *

+ * 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. + *

+ * 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 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 @@ 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 @@ -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); 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 cec8822d6eb0f..f6797ff65d196 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 @@ 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.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 void configure() { 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 void configure() { }; } + 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 @@ void failedCreateMonitoredItemsCallRebuildsTheConnection() throws Exception { .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 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 1e5463becddd1..204aba3951f85 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 @@ -1544,6 +1544,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