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 6bf60f0cf610 CAMEL-25692: camel-milo - create a refused monitored item
again instead of dropping it (#28212)
6bf60f0cf610 is described below
commit 6bf60f0cf610044b2c7f6aba7bd9d4f723a8bff0
Author: Serdar Gökay <[email protected]>
AuthorDate: Wed Oct 21 22:29:39 2026 +0300
CAMEL-25692: camel-milo - create a refused monitored item again instead of
dropping it (#28212)
* CAMEL-25692: camel-milo - create a refused monitored item again instead
of dropping it
A node whose monitored item the server refuses is still reported once with
the status of the refusal. While the
connection lasts, Camel now creates the refused items of the connection
again every 10 seconds in one request,
quietly, until the server accepts them; the first value of a created item
is the current value of the node. This
covers nodes and namespace URIs that the server gets later; the namespace
table is read once per attempt. The
attempt is skipped while the session is not active, a failed attempt is
made again with the next one and never
rebuilds the connection, and a stopped consumer is not retried.
* CAMEL-25692: camel-milo - cancel a retry under the lock and name the
retry interval in the warning
Review of the retry of refused monitored items. The next attempt of a
connection is cancelled under the lock that
guards it wherever the connection is replaced or disposed, also in the
dispose of the manager, which disposes the
connection itself outside of the lock. The warning for a refused node says
that Camel tries again and how often. The
javadoc of retryRefused says that the producers wait for an attempt and why
that wait is bounded. CommunicationLossTest
waits with during() and checks that the refused node is not reported while
the server is unreachable, instead of
sleeping.
* CAMEL-25692: camel-milo - deliver a refusal only while the node is still
refused
The refusals of a registration or of a recreated subscription are delivered
outside the lock, one node after another.
When the next attempt, or a new connection, created a node before its
refusal was delivered, the route received the
first value of the node and then its refusal, and the node stayed marked as
refused, so a later loss of communication
was no longer reported for it. The node is now marked as refused under the
lock once the call that refused it has
succeeded, and its refusal is delivered only while that still holds. A call
that fails leaves the node unmarked, so that
it still receives the loss report of the connection.
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../camel/catalog/docs/milo-client-component.adoc | 9 +-
.../src/main/docs/milo-client-component.adoc | 9 +-
.../milo/client/internal/SubscriptionManager.java | 180 ++++++++++++--
.../component/milo/CommunicationLossTest.java | 60 +++++
.../component/milo/MonitorItemLifecycleTest.java | 16 +-
.../component/milo/MonitoredItemRetryTest.java | 270 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 5 +-
7 files changed, 515 insertions(+), 34 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 9de57764a695..2620d3733dc2 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 9de57764a695..2620d3733dc2 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 88aae7485114..1ea704e01576 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.ArrayList;
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 @@ public class SubscriptionManager {
/**
* 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 @@ public class SubscriptionManager {
}
/**
- * 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 class SubscriptionManager {
}
/**
- * 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 @@ public class SubscriptionManager {
private final Map<String, UShort> 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<UInteger> 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 @@ public class SubscriptionManager {
}
/**
- * 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<PendingValue> putSubscriptions(final Map<UInteger,
Subscription> subscriptions) throws Exception {
+ public List<PendingValue> putSubscriptions(final Map<UInteger,
Subscription> subscriptions, final boolean retry)
+ throws Exception {
final List<PendingValue> pendingValues = new ArrayList<>();
@@ -373,18 +396,24 @@ public class SubscriptionManager {
// Milo 1.0.5 API: create monitored items and set up listeners
final List<OpcUaMonitoredItem> items = new
ArrayList<>(subscriptions.size());
final Map<OpcUaMonitoredItem, UInteger> itemToClientHandle = new
HashMap<>();
+ // the namespace URIs that the server does not have, read once per
call: all nodes of a missing namespace
+ final Set<String> unknownNamespaces = new HashSet<>();
for (final Map.Entry<UInteger, Subscription> 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 class SubscriptionManager {
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 class SubscriptionManager {
}
}
+ // 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<PendingValue> 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 class SubscriptionManager {
this.manager.deleteMonitoredItems();
this.manager.create();
- return putSubscriptions(monitored);
+ return putSubscriptions(monitored, false);
}
public void deactivate(final List<UInteger> clientHandles) throws
Exception {
final List<OpcUaMonitoredItem> 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 @@ public class SubscriptionManager {
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.
+ * <p>
+ * 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<UInteger, Subscription> 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 @@ public class SubscriptionManager {
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 @@ public class SubscriptionManager {
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 @@ public class SubscriptionManager {
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 class SubscriptionManager {
}
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 class SubscriptionManager {
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 02a3faa341e2..0fda3a88b4f8 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.ServerSocket;
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.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
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;
@@ -211,6 +214,63 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
}
}
+ @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 a770a93c20d8..c67ec3365ac3 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.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.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 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
}
@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 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
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 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
.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 000000000000..468d4b918042
--- /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<DataValue> 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 697208b5c676..8fad0e784a69 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
@@ -2167,8 +2167,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
