CAMEL-25626: camel-milo - deliver a data change with a Bad status even when its value is null

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/milo-client.json

@@ -70,7 +70,7 @@
     "method": { "index": 8, "kind": "parameter", "displayName": "Method", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The method definition (see Method ID)" },
     "monitorFilterType": { "index": 9, "kind": "parameter", "displayName": "Monitor Filter Type", "group": "common", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.milo.client.MonitorFilterType", "enum": [ "dataChangeFilter" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.milo.client.MonitorFilterConfiguration", "configurationField": "monitorFilterConfiguration", "description": "Monitor Filter Type for MonitoredItems." },
     "node": { "index": 10, "kind": "parameter", "displayName": "Node", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The node definition (see Node ID)" },
-    "omitNullValues": { "index": 11, "kind": "parameter", "displayName": "Omit Null Values", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Omit notifications in case of null values" },
+    "omitNullValues": { "index": 11, "kind": "parameter", "displayName": "Omit Null Values", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Omit a data change whose value is null and whose status is Good. A data change with a Bad or Uncertain status is always delivered." },
     "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The queue size used for OPC UA subscriptions. If not set, a queue size of 1 is requested." },
     "samplingInterval": { "index": 13, "kind": "parameter", "displayName": "Sampling Interval", "group": "common", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000.0", "description": "The sampling interval in milliseconds" },
     "nodes": { "index": 14, "kind": "parameter", "displayName": "Nodes", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Comma-separated node definitions to monitor with one consumer, instead of node (see Node ID)" },

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc

@@ -149,8 +149,20 @@ 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.
-To see such nodes, set `omitNullValues=false`: the consumer then delivers one exchange with the bad status code for that
-node, and the `CamelMiloNodeId` header names it.
+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
+
+A data change with a Bad or Uncertain status is always delivered, even when its value is null; `omitNullValues` only
+omits a null value with a Good status. This includes the first notification of a monitored item: it carries the current
+value of the node, at startup and again after a reconnect, so it can be a Bad status such as `Bad_WaitingForInitialData`
+or the status of a `milo-server` item that was not written yet. To skip such a notification, a route filters on its
+status code (`getStatusCode()` of the `DataValue` body); a filter on `isGood()` would also drop the Bad status of a
+failed sensor.
+
+The default `omitNullValues=true` also omits a null value with a Good status that follows a Bad status, so the route
+does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a
+`String`.
 
 === Method ID
 

components/camel-milo/src/generated/resources/META-INF/org/apache/camel/component/milo/client/milo-client.json

@@ -70,7 +70,7 @@
     "method": { "index": 8, "kind": "parameter", "displayName": "Method", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The method definition (see Method ID)" },
     "monitorFilterType": { "index": 9, "kind": "parameter", "displayName": "Monitor Filter Type", "group": "common", "label": "", "required": false, "type": "enum", "javaType": "org.apache.camel.component.milo.client.MonitorFilterType", "enum": [ "dataChangeFilter" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.milo.client.MonitorFilterConfiguration", "configurationField": "monitorFilterConfiguration", "description": "Monitor Filter Type for MonitoredItems." },
     "node": { "index": 10, "kind": "parameter", "displayName": "Node", "group": "common", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The node definition (see Node ID)" },
-    "omitNullValues": { "index": 11, "kind": "parameter", "displayName": "Omit Null Values", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Omit notifications in case of null values" },
+    "omitNullValues": { "index": 11, "kind": "parameter", "displayName": "Omit Null Values", "group": "common", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Omit a data change whose value is null and whose status is Good. A data change with a Bad or Uncertain status is always delivered." },
     "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue Size", "group": "common", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "description": "The queue size used for OPC UA subscriptions. If not set, a queue size of 1 is requested." },
     "samplingInterval": { "index": 13, "kind": "parameter", "displayName": "Sampling Interval", "group": "common", "label": "", "required": false, "type": "number", "javaType": "java.lang.Double", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "1000.0", "description": "The sampling interval in milliseconds" },
     "nodes": { "index": 14, "kind": "parameter", "displayName": "Nodes", "group": "consumer", "label": "consumer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Comma-separated node definitions to monitor with one consumer, instead of node (see Node ID)" },

components/camel-milo/src/main/docs/milo-client-component.adoc

@@ -149,8 +149,20 @@ 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.
-To see such nodes, set `omitNullValues=false`: the consumer then delivers one exchange with the bad status code for that
-node, and the `CamelMiloNodeId` header names it.
+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
+
+A data change with a Bad or Uncertain status is always delivered, even when its value is null; `omitNullValues` only
+omits a null value with a Good status. This includes the first notification of a monitored item: it carries the current
+value of the node, at startup and again after a reconnect, so it can be a Bad status such as `Bad_WaitingForInitialData`
+or the status of a `milo-server` item that was not written yet. To skip such a notification, a route filters on its
+status code (`getStatusCode()` of the `DataValue` body); a filter on `isGood()` would also drop the Bad status of a
+failed sensor.
+
+The default `omitNullValues=true` also omits a null value with a Good status that follows a Bad status, so the route
+does not learn that the node recovered. Set `omitNullValues=false` for a node whose value can be null, such as a
+`String`.
 
 === Method ID
 

components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java

@@ -116,8 +116,8 @@ protected void doStop() throws Exception {
     private void handleValueUpdate(final String node, final DataValue value) {
         LOG.debug("Handle item update - {} = {}", node, value);
 
-        if (omitNullValues && (value == null || value.getValue() == null || value.getValue().getValue() == null)) {
-            LOG.debug("Handle item update omitted due to null values (see omitNullValues parameter)");
+        if (omitNullValues && isNullValueWithGoodStatus(value)) {
+            LOG.debug("Handle item update omitted due to a null value with a Good status (see omitNullValues parameter)");
             return;
         }
 
@@ -131,6 +131,15 @@ private void handleValueUpdate(final String node, final DataValue value) {
         }
     }
 
+    /**
+     * The data change that omitNullValues omits: a null value with a Good status. A data change with a Bad or Uncertain
+     * status is delivered even when its value is null.
+     */
+    private static boolean isNullValueWithGoodStatus(final DataValue value) {
+        return value == null
+                || (value.getValue().getValue() == null && value.getStatusCode().isGood());
+    }
+
     private void mapToMessage(final DataValue value, final Message message) {
         if (value != null) {
             Messages.fillFromDataValue(value, message);

components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java

@@ -91,7 +91,8 @@ public class MiloClientEndpoint extends DefaultEndpoint {
     private boolean defaultAwaitWrites;
 
     /**
-     * Omit notifications in case of null values
+     * Omit a data change whose value is null and whose status is Good. A data change with a Bad or Uncertain status is
+     * always delivered.
      */
     @UriParam(defaultValue = "true")
     private boolean omitNullValues = true;

components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java

@@ -40,6 +40,7 @@
 
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
@@ -77,15 +78,13 @@ public class MonitorItemLifecycleTest extends AbstractMiloServerTest {
             = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem2");
 
     private static final String MILO_CLIENT_UNKNOWN_ITEM
-            = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "unknownItem")
-              + "&omitNullValues=false";
+            = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "unknownItem");
 
     private static final String MILO_CLIENT_UNKNOWN_ITEM_WRITE_BACK
-            = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "unknownItemWriteBack")
-              + "&omitNullValues=false";
+            = MILO_CLIENT + NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "unknownItemWriteBack");
 
     private static final String MILO_CLIENT_ITEMS_1_2
-            = MILO_CLIENT_BASE + "&omitNullValues=false&nodes=RAW(" + NODE_ITEM_1 + "," + NODE_ITEM_2 + ")";
+            = MILO_CLIENT_BASE + "&nodes=RAW(" + NODE_ITEM_1 + "," + NODE_ITEM_2 + ")";
 
     private static final String MOCK_ITEM_1 = "mock:item1";
 
@@ -126,10 +125,10 @@ public void configure() {
                 from(DIRECT_START_1).to(MILO_SERVER_ITEM_1);
                 from(DIRECT_START_2).to(MILO_SERVER_ITEM_2);
 
-                from(resolve(MILO_CLIENT_ITEM_1)).routeId("item1").to(MOCK_ITEM_1);
                 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_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);
 
@@ -145,9 +144,11 @@ public void configure() {
 
     @Test
     void restartedConsumerReceivesEachValueOnce() throws Exception {
+        producer1.sendBody("Foo");
+
         item1Endpoint.expectedMessageCount(1);
         testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
-        producer1.sendBody("Foo");
+        context.getRouteController().startRoute("item1");
         item1Endpoint.assertIsSatisfied();
 
         // the other consumers keep the shared connection open
@@ -180,12 +181,34 @@ void unknownNodeReceivesBadStatus() throws Exception {
         unknownItemEndpoint.assertIsSatisfied();
 
         // the other consumers of the connection are not affected
+        producer1.sendBody("Foo");
+
         item1Endpoint.expectedMessageCount(1);
         testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
-        producer1.sendBody("Foo");
+        context.getRouteController().startRoute("item1");
         item1Endpoint.assertIsSatisfied();
     }
 
+    @Test
+    void itemThatWasNotWrittenYetReportsABadStatusAtStartup() throws Exception {
+        // the first notification carries the current value of the item, which is a Bad status without a value until
+        // the item is written; the consumer delivers it like any other Bad status
+        item2Endpoint.expectedMessageCount(1);
+        item2Endpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID, NODE_ITEM_2);
+        testBody(item2Endpoint.message(0), value -> {
+            assertTrue(value.getStatusCode().isBad(), value.getStatusCode().toString());
+            assertNull(value.getValue().getValue());
+        });
+        context.getRouteController().startRoute("item2");
+        item2Endpoint.assertIsSatisfied();
+
+        item2Endpoint.reset();
+        item2Endpoint.expectedMessageCount(1);
+        testBody(item2Endpoint.message(0), assertGoodValue("Bar"));
+        producer2.sendBody("Bar");
+        item2Endpoint.assertIsSatisfied();
+    }
+
     @Test
     @Timeout(value = 60, unit = TimeUnit.SECONDS)
     void badStatusIsDeliveredWithoutHoldingTheConnectionLock() throws Exception {

components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemMultiConnectionsTest.java

@@ -86,9 +86,10 @@ protected RoutesBuilder createRouteBuilder() {
             public void configure() {
                 from(DIRECT_START_1).to(MILO_SERVER_ITEM_1);
 
-                from(resolve(MILO_CLIENT_ITEM_C1_1)).to(MOCK_TEST_1);
-                from(resolve(MILO_CLIENT_ITEM_C2_1)).to(MOCK_TEST_2);
-                from(resolve(MILO_CLIENT_ITEM_C3_1)).to(MOCK_TEST_3);
+                // the consumers deliver the current value while they start, so the test starts these routes
+                from(resolve(MILO_CLIENT_ITEM_C1_1)).autoStartup(false).to(MOCK_TEST_1);
+                from(resolve(MILO_CLIENT_ITEM_C2_1)).autoStartup(false).to(MOCK_TEST_2);
+                from(resolve(MILO_CLIENT_ITEM_C3_1)).autoStartup(false).to(MOCK_TEST_3);
             }
         };
     }
@@ -110,8 +111,9 @@ public void testMonitorItem1() throws Exception {
         this.test3Endpoint.setExpectedCount(0);
         this.test3Endpoint.setSleepForEmptyTest(5_000);
 
-        // set server value
+        // set server value, then start the consumers
         this.producer1.sendBody("Foo");
+        context.getRouteController().startAllRoutes();
 
         // assert
         MockEndpoint.assertIsSatisfied(context);