CAMEL-25692: camel-milo - create a refused monitored item again instead of dropping it

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

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

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");

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);
         }