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 3ba0ced7902eb..4f08ac6de8d6d 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 @@ -192,8 +192,13 @@ monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a diff need a consumer of their own. A node that does not exist on the server does not affect the other nodes. A node that cannot be monitored is -reported once with its status code and is not retried; restarting the route or a rebuilt connection to the server creates the monitored items again. -A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported in the same way. +reported once with its status code, and Camel tries to create its monitored item again every 10 seconds while the +connection lasts, for all such nodes of the connection in one request. A failed attempt delivers nothing; once the item +is created, the node delivers its current value as its first value. This covers a node that is added to the server +later, for example with a new PLC program, and a server that refused the item because it had too many. A rebuilt +connection creates the monitored items of all nodes again, and a stopped consumer is not retried. +A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported and retried in the same +way, so a namespace that the server adds later is found. The consumer delivers one exchange with the bad status code for such a node, and the `CamelMiloNodeId` header names it. === Status codes and null values 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 3ba0ced7902eb..4f08ac6de8d6d 100644 --- a/components/camel-milo/src/main/docs/milo-client-component.adoc +++ b/components/camel-milo/src/main/docs/milo-client-component.adoc @@ -192,8 +192,13 @@ monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a diff need a consumer of their own. A node that does not exist on the server does not affect the other nodes. A node that cannot be monitored is -reported once with its status code and is not retried; restarting the route or a rebuilt connection to the server creates the monitored items again. -A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported in the same way. +reported once with its status code, and Camel tries to create its monitored item again every 10 seconds while the +connection lasts, for all such nodes of the connection in one request. A failed attempt delivers nothing; once the item +is created, the node delivers its current value as its first value. This covers a node that is added to the server +later, for example with a new PLC program, and a server that refused the item because it had too many. A rebuilt +connection creates the monitored items of all nodes again, and a stopped consumer is not retried. +A node whose namespace URI (`nsu=`) is not in the namespace table of the server is reported and retried in the same +way, so a namespace that the server adds later is found. The consumer delivers one exchange with the bad status code for such a node, and the `CamelMiloNodeId` header names it. === Status codes and null values 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 88aae74851140..1ea704e01576c 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 @@ -23,8 +23,10 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -255,7 +257,8 @@ private static class Subscription { /** * Whether the server refused to monitor the node: its status is that refusal, whatever the communication, until - * its item is created + * its item is created. Set under the lock of the manager when a call hands the refusal out for delivery, and + * cleared under it when a call creates the item. */ private volatile boolean refused; @@ -293,11 +296,13 @@ synchronized void deliver(final long generation, final DataValue value) { } /** - * Delivers the status with which the server refused to monitor the node. + * Delivers the status with which the server refused to monitor the node, unless its item has been created + * since: the node then delivers its values instead. */ - void refuse(final long generation, final DataValue status) { - this.refused = true; - deliver(generation, status); + synchronized void refuse(final long generation, final DataValue status) { + if (this.refused) { + deliver(generation, status); + } } /** @@ -326,7 +331,8 @@ public MonitoringFilter createMonitoringFilter() { } /** - * A status value for a node that could not be monitored, delivered to its consumer outside of the lock + * A status value for a node that could not be monitored, delivered to its consumer outside of the lock while the + * node is still refused */ private record PendingValue(Subscription subscription, long generation, DataValue value) { void deliver() { @@ -344,6 +350,17 @@ private class Connected { private final Map namespaceCache = new ConcurrentHashMap<>(); + /** + * The client handles of the nodes that the server refused to monitor on this connection, which are created + * again every {@code reconnectTimeout} milliseconds, guarded by {@code lock} + */ + private final Set refused = new LinkedHashSet<>(); + + /** + * The next attempt to create the refused nodes again, if one is scheduled, guarded by {@code lock} + */ + private Future retryJob; + Connected(final OpcUaClient client, final OpcUaSubscription manager, final SubscriptionListenerImpl listener) { this.client = client; this.manager = manager; @@ -352,15 +369,21 @@ private class Connected { } /** - * Creates the monitored items of the subscriptions on the server. + * Creates the monitored items of the subscriptions on the server. A node that the server refuses to monitor is + * created again on this connection every {@code reconnectTimeout} milliseconds, until its item is created or it + * is unregistered. * * @param subscriptions the subscriptions to create, by client handle + * @param retry whether this is an attempt to create refused nodes again: their refusal has been queued + * for delivery already, so a node refused again is neither logged above DEBUG nor + * delivered * @return the bad status values of the nodes that could not be monitored, to be delivered to * their consumers after the lock of the manager has been released - * @throws UaException if the server rejected the CreateMonitoredItems service call, the connection has to be - * rebuilt + * @throws UaException if the server rejected the CreateMonitoredItems service call; unless {@code retry} is + * set, the connection has to be rebuilt */ - public List putSubscriptions(final Map subscriptions) throws Exception { + public List putSubscriptions(final Map subscriptions, final boolean retry) + throws Exception { final List pendingValues = new ArrayList<>(); @@ -373,18 +396,24 @@ public List putSubscriptions(final Map sub // Milo 1.0.5 API: create monitored items and set up listeners final List items = new ArrayList<>(subscriptions.size()); final Map itemToClientHandle = new HashMap<>(); + // the namespace URIs that the server does not have, read once per call: all nodes of a missing namespace + final Set unknownNamespaces = new HashSet<>(); for (final Map.Entry entry : subscriptions.entrySet()) { final Subscription s = entry.getValue(); - final NodeId node = lookupNamespace(s.getNodeId()).get(); + final String namespaceUri = s.getNodeId().getNamespaceUri(); + final NodeId node = namespaceUri != null && unknownNamespaces.contains(namespaceUri) + ? null : lookupNamespace(s.getNodeId()).get(); if (node == null) { - // reported once like a node the server rejects, the other nodes and the connection are not affected - final StatusCode status = new StatusCode(StatusCodes.Bad_NodeIdUnknown); - LOG.warn("Monitored item for node {} could not be created, the namespace URI is unknown to the server: {}", - s.getNodeId().toParseableString(), status); - pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); + if (namespaceUri != null) { + unknownNamespaces.add(namespaceUri); + } + // refused like a node the server rejects, the other nodes and the connection are not affected; the + // namespace table is read again on the next attempt, so a namespace added later is found + refuse(entry.getKey(), s, new StatusCode(StatusCodes.Bad_NodeIdUnknown), + "the namespace URI is unknown to the server", retry, pendingValues); } else { final ReadValueId itemId = new ReadValueId(node, AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE); final OpcUaMonitoredItem item = new OpcUaMonitoredItem(itemId, MonitoringMode.Reporting); @@ -425,17 +454,18 @@ public List putSubscriptions(final Map sub if (result.isGood()) { this.goodSubscriptions.put(clientHandle, item); subscriptions.get(clientHandle).refused = false; + if (this.refused.remove(clientHandle) && retry) { + LOG.info("Monitored item for node {} created on retry", + subscriptions.get(clientHandle).getNodeId().toParseableString()); + } } else if (result.serviceResult().isBad()) { failedItems.add(item); rejectedServiceResult = result.serviceResult(); } else { - // reported once, the item is created again when the consumer restarts or the connection is rebuilt + // reported once, the item is created again later on this connection failedItems.add(item); - final Subscription s = subscriptions.get(clientHandle); - final StatusCode status = result.operationResult().orElse(result.serviceResult()); - LOG.warn("Monitored item for node {} could not be created: {}", s.getNodeId().toParseableString(), - status); - pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); + refuse(clientHandle, subscriptions.get(clientHandle), + result.operationResult().orElse(result.serviceResult()), null, retry, pendingValues); } } @@ -451,9 +481,58 @@ public List putSubscriptions(final Map sub } } + // marked under the lock once the call has succeeded: a refusal that the creation of its item overtakes, by + // an attempt or a new connection, is then dropped, and a call that fails leaves the nodes to the loss report + // of the connection + pendingValues.forEach(pending -> pending.subscription().refused = true); + scheduleRetry(); return pendingValues; } + /** + * Records that the server refused to monitor a node. The first refusal is logged at WARN and queued for + * delivery to the consumer, a refusal on a later attempt only logged at DEBUG. + */ + private void refuse( + final UInteger clientHandle, final Subscription s, final StatusCode status, final String reason, + final boolean retry, final List pendingValues) { + this.refused.add(clientHandle); + final String node = s.getNodeId().toParseableString(); + final String because = reason == null ? "" : ", " + reason; + if (retry) { + LOG.debug("Monitored item for node {} could not be created on retry{}: {}", node, because, status); + } else { + LOG.warn("Monitored item for node {} could not be created{}, trying again every {} ms: {}", node, because, + SubscriptionManager.this.reconnectTimeout, status); + pendingValues.add(new PendingValue(s, this.generation, new DataValue(status))); + } + } + + /** + * Schedules the next attempt to create the refused nodes again, unless there is none or one is scheduled. The + * timer runs on the reconnect thread, the attempt, which blocks until the server answers, on the executor of + * the connection, so that a slow attempt does not hold up a reconnect. + */ + private void scheduleRetry() { + if (this.refused.isEmpty() || this.retryJob != null || SubscriptionManager.this.disposed) { + return; + } + this.retryJob = SubscriptionManager.this.reconnectExecutor.schedule( + () -> SubscriptionManager.this.executor.execute(() -> retryRefused(this)), + SubscriptionManager.this.reconnectTimeout, TimeUnit.MILLISECONDS); + } + + /** + * Cancels the next attempt to create the refused nodes again, if one is scheduled: a connection that is + * replaced or disposed is not retried + */ + private void cancelRetry() { + if (this.retryJob != null) { + this.retryJob.cancel(false); + this.retryJob = null; + } + } + /** * Creates the subscription on the server again, with the monitored items it had, after the server lost it. * @@ -474,12 +553,13 @@ public List recreate(final Map subscriptio this.manager.deleteMonitoredItems(); this.manager.create(); - return putSubscriptions(monitored); + return putSubscriptions(monitored, false); } public void deactivate(final List clientHandles) throws Exception { final List items = new ArrayList<>(clientHandles.size()); for (final UInteger clientHandle : clientHandles) { + this.refused.remove(clientHandle); final OpcUaMonitoredItem item = this.goodSubscriptions.remove(clientHandle); if (item != null) { items.add(item); @@ -1040,6 +1120,49 @@ private void recreateSubscription(final OpcUaSubscription subscription) { pendingValues.forEach(PendingValue::deliver); } + /** + * Creates the nodes that the server refused to monitor on a connection again, in one CreateMonitoredItems call, + * while the connection is the current one. Nothing is delivered to the consumers: a node created now delivers its + * first value from the server, and its refusal, if not delivered yet, is dropped. While the session is not active, + * the attempt is skipped. A call that fails is made again with the next attempt and never rebuilds the connection, + * which would interrupt the nodes that are monitored: a lost session or subscription is recovered without it. + *

+ * Like the registration of a consumer, the attempt holds {@code lock} until the server answers, so that a consumer + * that stops or a subscription that is recreated in the meantime cannot interleave with it, and the producers and + * {@link #getOpcUaClient()} wait for it. The wait is bounded: the attempt is only sent while the session is active, + * a call in flight when the connection drops silently fails once the keep-alive of the client declares the session + * lost, and a call that the server does not answer fails after the request timeout. + */ + private void retryRefused(final Connected connected) { + lock.lock(); + try { + connected.retryJob = null; + if (this.disposed || this.connected != connected || connected.refused.isEmpty()) { + return; + } + if (!connected.listener.sessionActive) { + LOG.debug("Session not active, the refused monitored items are created again later"); + connected.scheduleRetry(); + return; + } + final Map refused = new LinkedHashMap<>(); + for (final UInteger clientHandle : connected.refused) { + final Subscription subscription = this.subscriptions.get(clientHandle); + if (subscription != null) { + refused.put(clientHandle, subscription); + } + } + try { + connected.putSubscriptions(refused, true); + } catch (final Exception e) { + LOG.debug("Failed to create the refused monitored items again, trying again later", e); + connected.scheduleRetry(); + } + } finally { + lock.unlock(); + } + } + private boolean isCurrent(final OpcUaSubscription subscription) { return !this.disposed && this.connected != null && this.connected.manager == subscription; } @@ -1114,6 +1237,7 @@ private void handleConnectionFailure(final Throwable e) { if (this.connected != null) { // the nodes lose their communication with the connection, after the values it still delivers reportLoss(this.connected.manager, this.connected.generation, this.subscriptions); + this.connected.cancelRetry(); this.connected.dispose(); this.connected = null; } @@ -1163,7 +1287,7 @@ private void performAndEvalConnect() { try { LOG.debug("Setting subscriptions: {}", this.subscriptions.size()); - pendingValues.addAll(connected.putSubscriptions(this.subscriptions)); + pendingValues.addAll(connected.putSubscriptions(this.subscriptions, false)); LOG.debug("Update state : {} -> {}", this.connected, connected); final Connected oldConnected = this.connected; @@ -1172,11 +1296,13 @@ private void performAndEvalConnect() { if (oldConnected != null) { LOG.debug("Dispose old state"); + oldConnected.cancelRetry(); oldConnected.dispose(); } } catch (final Exception e) { LOG.info("Failed to set subscriptions", e); + connected.cancelRetry(); connected.dispose(); throw e; } @@ -1336,6 +1462,10 @@ public void dispose() { } this.disposed = true; connected = this.connected; + if (connected != null) { + // the next attempt is cancelled under the lock that guards it, the connection is disposed outside of it + connected.cancelRetry(); + } if (this.reconnectJob != null) { // a waiting reconnect would keep the thread of the reconnect executor until it is due this.reconnectJob.cancel(false); @@ -1504,7 +1634,7 @@ public List registerItems( reportLoss(this.connected.manager, this.connected.generation, subscriptions); } - whenConnected(connected -> pendingValues.addAll(connected.putSubscriptions(subscriptions))); + whenConnected(connected -> pendingValues.addAll(connected.putSubscriptions(subscriptions, false))); } finally { lock.unlock(); } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java index 958faad13dd47..d89133882a330 100644 --- a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java @@ -23,6 +23,8 @@ import java.net.Socket; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.camel.EndpointInject; import org.apache.camel.Produce; @@ -40,6 +42,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; @@ -195,6 +198,63 @@ void lostSessionReportsTheNodeAndResendsItsValue() throws Exception { } } + @Test + void refusedNodeIsCreatedOnceTheSessionIsActiveAgain() throws Exception { + producer1.sendBody("Foo"); + + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + unknownEndpoint.expectedMessageCount(1); + testBody(unknownEndpoint.message(0), + value -> assertEquals(StatusCodes.Bad_NodeIdUnknown, value.getStatusCode().getValue())); + context.getRouteController().startRoute("unknown"); + unknownEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint( + resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@", Integer.toString(relayPort.getPort())), + MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + final NodeId sessionId = client.getSession().getSessionId(); + + // the refused node is not reported, and delivers its value once it is created after the session is back + unknownEndpoint.reset(); + unknownEndpoint.expectedMessageCount(1); + unknownEndpoint.setResultWaitTime(60_000); + unknownEndpoint.setAssertPeriod(2_000); + testBody(unknownEndpoint.message(0), assertGoodValue("Baz")); + item1Endpoint.reset(); + item1Endpoint.expectedMessageCount(2); + item1Endpoint.setResultWaitTime(60_000); + item1Endpoint.setAssertPeriod(2_000); + testBody(item1Endpoint.message(0), assertNoCommunication()); + testBody(item1Endpoint.message(1), assertGoodValue("Foo")); + + // the server stays unreachable for longer than the retry interval of the refused node + relay.stop(); + await().atMost(10, TimeUnit.SECONDS).until(() -> item1Endpoint.getReceivedCounter() >= 1); + await().during(12, TimeUnit.SECONDS).atMost(15, TimeUnit.SECONDS) + .until(() -> unknownEndpoint.getReceivedCounter() == 0); + // an attempt while the session is lost is skipped: it does not hold the connection until the session is back + assertSame(client, CompletableFuture.supplyAsync(connection::getOpcUaClient).get(2, TimeUnit.SECONDS)); + template.sendBody("milo-server:unknownItem", "Baz"); + relay.start(relayPort.getPort(), getServerPort()); + + unknownEndpoint.assertIsSatisfied(); + item1Endpoint.assertIsSatisfied(); + + // the attempts while the session was lost did not rebuild the connection + assertSame(client, connection.getOpcUaClient()); + assertEquals(sessionId, client.getSession().getSessionId()); + } finally { + endpoint.releaseConnection(connection); + } + } + @Test void consumerStartedWithoutTheServerReportsTheNodeThenReceivesItsValue() throws Exception { producer1.sendBody("Foo"); 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 a770a93c20d87..c67ec3365ac30 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 @@ -35,6 +35,7 @@ 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.sdk.server.nodes.UaNode; 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; @@ -402,8 +403,8 @@ void subscriptionThatASlowRouteHeldUpIsKept() throws Exception { } @Test - @Timeout(value = 60, unit = TimeUnit.SECONDS) - void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() throws Exception { + @Timeout(value = 90, unit = TimeUnit.SECONDS) + void nodeThatCannotBeMonitoredAgainIsReportedAndCreatedOnceItIsBackWithoutRebuildingTheConnection() throws Exception { producer1.sendBody("Foo"); producer2.sendBody("Bar"); context.getRouteController().startRoute("item1"); @@ -420,7 +421,9 @@ void nodeThatCannotBeMonitoredAgainIsReportedWithoutRebuildingTheConnection() th 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(); + final UaNode node2 = getServer().getAddressSpaceManager().getManagedNode(ExpandedNodeId.parse(NODE_ITEM_2)) + .orElseThrow(); + node2.delete(); timeOutSubscription(subscriptionId); // the subscription is recreated in the same session, with the monitored items of the first node only @@ -441,6 +444,13 @@ && new StatusCode(StatusCodes.Bad_NodeIdUnknown) .untilAsserted(() -> assertGoodValue("Baz").accept(lastExchange(item1Endpoint).getMessage() .getBody(DataValue.class))); assertSame(client, connection.getOpcUaClient()); + + // the node is added again on the server: its item is created again and delivers its current value + node2.getNodeManager().addNode(node2); + await().atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertGoodValue("Bar").accept(lastExchange(items12Endpoint, NODE_ITEM_2).getMessage() + .getBody(DataValue.class))); + assertSame(client, connection.getOpcUaClient()); } finally { endpoint.releaseConnection(connection); } diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java new file mode 100644 index 0000000000000..468d4b918042e --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitoredItemRetryTest.java @@ -0,0 +1,270 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.milo; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +import org.apache.camel.EndpointInject; +import org.apache.camel.Exchange; +import org.apache.camel.Produce; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.milo.client.MiloClientConnection; +import org.apache.camel.component.milo.client.MiloClientEndpoint; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.milo.server.internal.CamelNamespace; +import org.apache.camel.component.mock.MockEndpoint; +import org.eclipse.milo.opcua.sdk.client.OpcUaClient; +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.StatusCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; + +/** + * A node that the server refuses to monitor is reported once and created again on the same connection, every 10 + * seconds, until the server has it or its consumer stops + */ +public class MonitoredItemRetryTest extends AbstractMiloServerTest { + + private static final String DIRECT_START_1 = "direct:start1"; + + private static final String MILO_SERVER_ITEM_1 = "milo-server:myitem1"; + + // a namespace that the server does not have until a test adds it + private static final String LATE_NAMESPACE_URI = "urn:org:apache:camel:late"; + + // the client endpoints only differ in their nodes, so they share one connection + private static final String MILO_CLIENT_BASE = "milo-client:opc.tcp://foo:bar@localhost:@@port@@" + + "?allowedSecurityPolicies=None&overrideHost=true" + + "&samplingInterval=50&requestedPublishingInterval=100"; + + private static final String MILO_CLIENT = MILO_CLIENT_BASE + "&node="; + + // a node that the server never has, and one that it gets while the refusal of the first is being delivered + private static final String MISSING_NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=missingItem"; + + private static final String OVERTAKEN_NODE = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=overtakenItem"; + + private static final String MILO_CLIENT_OVERTAKEN + = MILO_CLIENT_BASE + "&nodes=RAW(" + MISSING_NODE + "," + OVERTAKEN_NODE + ")"; + + private static final String MILO_CLIENT_ITEM_1 + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1"); + + private static final String MILO_CLIENT_LATE_ITEM + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "lateItem"); + + private static final String MILO_CLIENT_LATE_NAMESPACE_ITEM + = MILO_CLIENT + NodeIds.nodeValue(LATE_NAMESPACE_URI, "lateNamespaceItem"); + + private static final String MILO_CLIENT_STOPPED_ITEM + = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "stoppedItem"); + + @EndpointInject("mock:item1") + protected MockEndpoint item1Endpoint; + + @EndpointInject("mock:lateItem") + protected MockEndpoint lateItemEndpoint; + + @EndpointInject("mock:lateNamespaceItem") + protected MockEndpoint lateNamespaceItemEndpoint; + + @EndpointInject("mock:stoppedItem") + protected MockEndpoint stoppedItemEndpoint; + + @EndpointInject("mock:overtaken") + protected MockEndpoint overtakenEndpoint; + + private final CountDownLatch overtakenCreated = new CountDownLatch(1); + + @Produce(DIRECT_START_1) + protected ProducerTemplate producer1; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from(DIRECT_START_1).to(MILO_SERVER_ITEM_1); + + // 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:item1"); + from(resolve(MILO_CLIENT_LATE_ITEM)).routeId("lateItem").autoStartup(false).to("mock:lateItem"); + from(resolve(MILO_CLIENT_LATE_NAMESPACE_ITEM)).routeId("lateNamespaceItem").autoStartup(false) + .to("mock:lateNamespaceItem"); + from(resolve(MILO_CLIENT_STOPPED_ITEM)).routeId("stoppedItem").autoStartup(false) + .to("mock:stoppedItem"); + from(resolve(MILO_CLIENT_OVERTAKEN)).routeId("overtaken").autoStartup(false) + .process(exchange -> holdRefusalOfMissingNode(exchange)).to("mock:overtaken"); + } + }; + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void nodeCreatedOnTheServerLaterDeliversItsValue() throws Exception { + lateItemEndpoint.expectedMessageCount(1); + testBody(lateItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("lateItem"); + lateItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_LATE_ITEM), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + // the node is created on the server: its first value follows the refusal, without another bad status + lateItemEndpoint.reset(); + lateItemEndpoint.expectedMessageCount(1); + lateItemEndpoint.setResultWaitTime(30_000); + lateItemEndpoint.setAssertPeriod(2_000); + testBody(lateItemEndpoint.message(0), assertGoodValue("Foo")); + template.sendBody("milo-server:lateItem", "Foo"); + lateItemEndpoint.assertIsSatisfied(); + + // on the same connection + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void nodeOfANamespaceAddedLaterDeliversItsValue() throws Exception { + lateNamespaceItemEndpoint.expectedMessageCount(1); + testBody(lateNamespaceItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("lateNamespaceItem"); + lateNamespaceItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint + = context.getEndpoint(resolve(MILO_CLIENT_LATE_NAMESPACE_ITEM), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + // the server adds the namespace URI to its namespace table and the node to the namespace + final CamelNamespace namespace = new CamelNamespace(LATE_NAMESPACE_URI, getServer().getServer()); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + lateNamespaceItemEndpoint.reset(); + lateNamespaceItemEndpoint.expectedMessageCount(1); + lateNamespaceItemEndpoint.setResultWaitTime(30_000); + lateNamespaceItemEndpoint.setAssertPeriod(2_000); + testBody(lateNamespaceItemEndpoint.message(0), assertGoodValue("Foo")); + namespace.startup(); + namespace.getOrAddItem("lateNamespaceItem").update("Foo"); + lateNamespaceItemEndpoint.assertIsSatisfied(); + + assertSame(client, connection.getOpcUaClient()); + } finally { + endpoint.releaseConnection(connection); + namespace.shutdown(); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void stoppedConsumerIsNotCreatedAgain() throws Exception { + // the consumer of another node keeps the connection open + producer1.sendBody("Foo"); + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Foo")); + context.getRouteController().startRoute("item1"); + item1Endpoint.assertIsSatisfied(); + + stoppedItemEndpoint.expectedMessageCount(1); + testBody(stoppedItemEndpoint.message(0), assertStatus(StatusCodes.Bad_NodeIdUnknown)); + context.getRouteController().startRoute("stoppedItem"); + stoppedItemEndpoint.assertIsSatisfied(); + + final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class); + final MiloClientConnection connection = endpoint.createConnection(); + try { + final OpcUaClient client = connection.getOpcUaClient(); + + context.getRouteController().stopRoute("stoppedItem"); + stoppedItemEndpoint.reset(); + stoppedItemEndpoint.expectedMessageCount(0); + template.sendBody("milo-server:stoppedItem", "Foo"); + + // the next attempt would have created it by now + await().during(15, TimeUnit.SECONDS).atMost(20, TimeUnit.SECONDS) + .until(() -> client.getSubscriptions().get(0).getMonitoredItems().stream() + .noneMatch(item -> "stoppedItem" + .equals(item.getReadValueId().getNodeId().getIdentifier()))); + stoppedItemEndpoint.assertIsSatisfied(); + assertSame(client, connection.getOpcUaClient()); + + // the other consumer keeps receiving + item1Endpoint.reset(); + item1Endpoint.expectedMessageCount(1); + testBody(item1Endpoint.message(0), assertGoodValue("Bar")); + producer1.sendBody("Bar"); + item1Endpoint.assertIsSatisfied(); + } finally { + endpoint.releaseConnection(connection); + } + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void refusalOvertakenByTheNextAttemptIsNotDelivered() throws Exception { + // the missing node delivers its refusal, the other node its first value and no refusal after it + overtakenEndpoint.expectedHeaderValuesReceivedInAnyOrder(MiloConstants.HEADER_NODE_ID, MISSING_NODE, + OVERTAKEN_NODE); + overtakenEndpoint.setAssertPeriod(2_000); + context.getRouteController().startRoute("overtaken"); + overtakenEndpoint.assertIsSatisfied(); + + for (final Exchange exchange : overtakenEndpoint.getReceivedExchanges()) { + final DataValue value = exchange.getMessage().getBody(DataValue.class); + if (OVERTAKEN_NODE.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID))) { + assertGoodValue("Foo").accept(value); + } else { + assertStatus(StatusCodes.Bad_NodeIdUnknown).accept(value); + } + } + } + + /** + * Holds the refusal of the missing node, which the consumer delivers first while it starts, until the server has + * the other node and the next attempt has created it, so that the first value of that node overtakes its refusal + */ + private void holdRefusalOfMissingNode(final Exchange exchange) throws InterruptedException { + if (MISSING_NODE.equals(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID))) { + template.sendBody("milo-server:overtakenItem", "Foo"); + overtakenCreated.await(30, TimeUnit.SECONDS); + } else if (exchange.getMessage().getBody(DataValue.class).getStatusCode().isGood()) { + overtakenCreated.countDown(); + } + } + + private MiloServerComponent getServer() { + return context.getComponent("milo-server", MiloServerComponent.class); + } + + private static Consumer assertStatus(final long statusCode) { + return value -> assertEquals(new StatusCode(statusCode), value.getStatusCode()); + } +} 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 3fda251c39ed4..c188339981e4c 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 @@ -1970,8 +1970,9 @@ Bad status of a failed sensor. Since 4.16, a `milo-client` consumer whose monitored item could not be created, for example because the node does not exist, received nothing. Now it receives one `DataValue` with the bad status code. The -monitored item is created again when the route is restarted or when Camel rebuilds the connection to the server; unlike -Camel 3.x, it is not retried every 10 seconds. +monitored item is then created again every 10 seconds on the same connection, without delivering anything for a failed +attempt, until the server accepts it; its first value is the current value of the node. A route restart or a +connection that Camel rebuilds creates it again too. A node ID with a namespace URI (`nsu=`) that is not in the namespace table of the server no longer takes the whole client connection down, including the consumers of other nodes. Its consumer receives one `DataValue` with