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 900cfc883ab0 CAMEL-25566: camel-milo - monitor several nodes with one
consumer (#27780)
900cfc883ab0 is described below
commit 900cfc883ab00e89ce414e729151e2b80777241b
Author: Serdar Gökay <[email protected]>
AuthorDate: Thu Oct 15 14:59:39 2026 +0300
CAMEL-25566: camel-milo - monitor several nodes with one consumer (#27780)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../camel/catalog/components/milo-client.json | 52 +++--
.../camel/catalog/docs/milo-client-component.adoc | 25 ++
.../milo/client/MiloClientEndpointConfigurer.java | 3 +
.../milo/client/MiloClientEndpointUriFactory.java | 3 +-
.../camel/component/milo/client/milo-client.json | 52 +++--
.../src/main/docs/milo-client-component.adoc | 25 ++
.../apache/camel/component/milo/MiloConstants.java | 5 +
.../milo/client/MiloClientConnection.java | 23 +-
.../component/milo/client/MiloClientConsumer.java | 53 ++++-
.../component/milo/client/MiloClientEndpoint.java | 14 ++
.../milo/client/internal/SubscriptionManager.java | 152 +++++++++----
.../component/milo/MonitorItemLifecycleTest.java | 252 +++++++++++++++++++++
.../camel/component/milo/MonitorNodesTest.java | 240 ++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
.../dsl/MiloClientEndpointBuilderFactory.java | 29 +++
15 files changed, 836 insertions(+), 105 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/milo-client.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/milo-client.json
index 58b851de167e..f592a62caebd 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/milo-client.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/milo-client.json
@@ -55,7 +55,8 @@
},
"headers": {
"CamelMiloNodeIds": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"List", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The node ids.", "constantName":
"org.apache.camel.component.milo.MiloConstants#HEADER_NODE_IDS" },
- "CamelMiloAwait": { "index": 1, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Boolean", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The await setting for writes.",
"constantName": "org.apache.camel.component.milo.MiloConstants#HEADER_AWAIT" }
+ "CamelMiloAwait": { "index": 1, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Boolean", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The await setting for writes.",
"constantName": "org.apache.camel.component.milo.MiloConstants#HEADER_AWAIT" },
+ "CamelMiloNodeId": { "index": 2, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The node ID of the value, as configured in the
node or nodes option. Not to be confused with CamelMiloNodeIds, the list of
nodes that the producer reads.", "constantName":
"org.apache.camel.component.milo.MiloConstants#HEADER_NODE_ID" }
},
"properties": {
"endpointUri": { "index": 0, "kind": "path", "displayName": "Endpoint
Uri", "group": "common", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The OPC UA server
endpoint" },
@@ -72,29 +73,30 @@
"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" },
"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" },
- "bridgeErrorHandler": { "index": 14, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
- "exceptionHandler": { "index": 15, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
- "exchangePattern": { "index": 16, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
- "lazyStartProducer": { "index": 17, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
- "allowedSecurityPolicies": { "index": 18, "kind": "parameter",
"displayName": "Allowed Security Policies", "group": "client", "label":
"client", "required": false, "type": "string", "javaType": "java.lang.String",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "A set of allowed
security policy URIs. Default is to accept all and use t [...]
- "applicationName": { "index": 19, "kind": "parameter", "displayName":
"Application Name", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "Apache Camel adapter for
Eclipse Milo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application name" },
- "applicationUri": { "index": 20, "kind": "parameter", "displayName":
"Application Uri", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo\/Client", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application URI" },
- "channelLifetime": { "index": 21, "kind": "parameter", "displayName":
"Channel Lifetime", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Channel lifetime in
milliseconds" },
- "keyAlias": { "index": 22, "kind": "parameter", "displayName": "Key
Alias", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The name of the key in
the keystore file" },
- "keyPassword": { "index": 23, "kind": "parameter", "displayName": "Key
Password", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key password" },
- "keyStorePassword": { "index": 24, "kind": "parameter", "displayName":
"Key Store Password", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The keystore password" },
- "keyStoreType": { "index": 25, "kind": "parameter", "displayName": "Key
Store Type", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key store type" },
- "keyStoreUrl": { "index": 26, "kind": "parameter", "displayName": "Key
Store Url", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The URL where the key
should be loaded from" },
- "maxPendingPublishRequests": { "index": 27, "kind": "parameter",
"displayName": "Max Pending Publish Requests", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending publish requests" },
- "maxResponseMessageSize": { "index": 28, "kind": "parameter",
"displayName": "Max Response Message Size", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
bytes a response message may have" },
- "overrideHost": { "index": 29, "kind": "parameter", "displayName":
"Override Host", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint host with the host from the endpoint URI." },
- "overridePort": { "index": 30, "kind": "parameter", "displayName":
"Override Port", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint port with the port from the endpoint URI." },
- "productUri": { "index": 31, "kind": "parameter", "displayName": "Product
Uri", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The product URI" },
- "requestedPublishingInterval": { "index": 32, "kind": "parameter",
"displayName": "Requested Publishing Interval", "group": "client", "label":
"client", "required": false, "type": "number", "javaType": "java.lang.Double",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"1_000.0", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The requested publishing
interval in m [...]
- "requestTimeout": { "index": 33, "kind": "parameter", "displayName":
"Request Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Request timeout in
milliseconds" },
- "sessionName": { "index": 34, "kind": "parameter", "displayName": "Session
Name", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session name" },
- "sessionTimeout": { "index": 35, "kind": "parameter", "displayName":
"Session Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session timeout in
milliseconds" },
- "password": { "index": 36, "kind": "parameter", "displayName": "Password",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The password for
authentication. Use this instead of embedding credentials in th [...]
- "username": { "index": 37, "kind": "parameter", "displayName": "Username",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The username for
authentication. Use this instead of embedding credentials in the endpoint URI
when t [...]
+ "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)" },
+ "bridgeErrorHandler": { "index": 15, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
+ "exceptionHandler": { "index": 16, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
+ "exchangePattern": { "index": 17, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
+ "lazyStartProducer": { "index": 18, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
+ "allowedSecurityPolicies": { "index": 19, "kind": "parameter",
"displayName": "Allowed Security Policies", "group": "client", "label":
"client", "required": false, "type": "string", "javaType": "java.lang.String",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "A set of allowed
security policy URIs. Default is to accept all and use t [...]
+ "applicationName": { "index": 20, "kind": "parameter", "displayName":
"Application Name", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "Apache Camel adapter for
Eclipse Milo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application name" },
+ "applicationUri": { "index": 21, "kind": "parameter", "displayName":
"Application Uri", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo\/Client", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application URI" },
+ "channelLifetime": { "index": 22, "kind": "parameter", "displayName":
"Channel Lifetime", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Channel lifetime in
milliseconds" },
+ "keyAlias": { "index": 23, "kind": "parameter", "displayName": "Key
Alias", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The name of the key in
the keystore file" },
+ "keyPassword": { "index": 24, "kind": "parameter", "displayName": "Key
Password", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key password" },
+ "keyStorePassword": { "index": 25, "kind": "parameter", "displayName":
"Key Store Password", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The keystore password" },
+ "keyStoreType": { "index": 26, "kind": "parameter", "displayName": "Key
Store Type", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key store type" },
+ "keyStoreUrl": { "index": 27, "kind": "parameter", "displayName": "Key
Store Url", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The URL where the key
should be loaded from" },
+ "maxPendingPublishRequests": { "index": 28, "kind": "parameter",
"displayName": "Max Pending Publish Requests", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending publish requests" },
+ "maxResponseMessageSize": { "index": 29, "kind": "parameter",
"displayName": "Max Response Message Size", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
bytes a response message may have" },
+ "overrideHost": { "index": 30, "kind": "parameter", "displayName":
"Override Host", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint host with the host from the endpoint URI." },
+ "overridePort": { "index": 31, "kind": "parameter", "displayName":
"Override Port", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint port with the port from the endpoint URI." },
+ "productUri": { "index": 32, "kind": "parameter", "displayName": "Product
Uri", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The product URI" },
+ "requestedPublishingInterval": { "index": 33, "kind": "parameter",
"displayName": "Requested Publishing Interval", "group": "client", "label":
"client", "required": false, "type": "number", "javaType": "java.lang.Double",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"1_000.0", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The requested publishing
interval in m [...]
+ "requestTimeout": { "index": 34, "kind": "parameter", "displayName":
"Request Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Request timeout in
milliseconds" },
+ "sessionName": { "index": 35, "kind": "parameter", "displayName": "Session
Name", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session name" },
+ "sessionTimeout": { "index": 36, "kind": "parameter", "displayName":
"Session Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session timeout in
milliseconds" },
+ "password": { "index": 37, "kind": "parameter", "displayName": "Password",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The password for
authentication. Use this instead of embedding credentials in th [...]
+ "username": { "index": 38, "kind": "parameter", "displayName": "Username",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The username for
authentication. Use this instead of embedding credentials in the endpoint URI
when t [...]
}
}
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 3c5b1aecdb95..e04f72f92737 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
@@ -127,6 +127,31 @@ However Camel allows to wrap the actual value inside
`RAW(…)`, which makes esc
milo-client:opc.tcp://user:password@localhost:12345?node=RAW(nsu=http://foo.bar;s=foo/bar)
----
+=== Monitoring several nodes
+
+A consumer can monitor several nodes with the `nodes` option instead of `node`
(the two options cannot be combined).
+The consumer creates the monitored items of all nodes together and deletes
them together when it stops. Every value
+arrives in its own exchange, and the `CamelMiloNodeId` header holds the node
ID as it was configured.
+
+The `nodes` option takes a comma-separated list of <<nodeid, Node IDs>>. The
entries are trimmed, empty entries are
+skipped and a node that is listed twice is monitored once:
+
+----
+milo-client:opc.tcp://localhost:12345?nodes=RAW(ns=2;s=line1/temperature,ns=2;s=line1/pressure)
+----
+
+A string node ID that contains a comma cannot be used in `nodes`; monitor it
with the `node` option instead. Long lists
+can come from a property placeholder, for example `+nodes=RAW({{tags}})+`.
+
+All nodes of the list share the `samplingInterval`, `queueSize` and
`omitNullValues` options of the consumer and its
+monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a
different sampling interval or deadband
+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.
+
=== Method ID
It is possible to perform methods calls on OPC UA nodes. If the parameter
`method` is set to the Node ID of a method call (the node ID must be set to the
parent object in this case),
diff --git
a/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointConfigurer.java
b/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointConfigurer.java
index c415e53a4812..921685c5fb79 100644
---
a/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointConfigurer.java
+++
b/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointConfigurer.java
@@ -71,6 +71,7 @@ public class MiloClientEndpointConfigurer extends
PropertyConfigurerSupport impl
case "monitorfiltertype":
case "monitorFilterType":
target.getMonitorFilterConfiguration().setMonitorFilterType(property(camelContext,
org.apache.camel.component.milo.client.MonitorFilterType.class, value));
return true;
case "node": target.setNode(property(camelContext,
java.lang.String.class, value)); return true;
+ case "nodes": target.setNodes(property(camelContext,
java.lang.String.class, value)); return true;
case "omitnullvalues":
case "omitNullValues": target.setOmitNullValues(property(camelContext,
boolean.class, value)); return true;
case "overridehost":
@@ -148,6 +149,7 @@ public class MiloClientEndpointConfigurer extends
PropertyConfigurerSupport impl
case "monitorfiltertype":
case "monitorFilterType": return
org.apache.camel.component.milo.client.MonitorFilterType.class;
case "node": return java.lang.String.class;
+ case "nodes": return java.lang.String.class;
case "omitnullvalues":
case "omitNullValues": return boolean.class;
case "overridehost":
@@ -226,6 +228,7 @@ public class MiloClientEndpointConfigurer extends
PropertyConfigurerSupport impl
case "monitorfiltertype":
case "monitorFilterType": return
target.getMonitorFilterConfiguration().getMonitorFilterType();
case "node": return target.getNode();
+ case "nodes": return target.getNodes();
case "omitnullvalues":
case "omitNullValues": return target.isOmitNullValues();
case "overridehost":
diff --git
a/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointUriFactory.java
b/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointUriFactory.java
index 15e5b54e0ab5..678e4ed402d7 100644
---
a/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointUriFactory.java
+++
b/components/camel-milo/src/generated/java/org/apache/camel/component/milo/client/MiloClientEndpointUriFactory.java
@@ -24,7 +24,7 @@ public class MiloClientEndpointUriFactory extends
org.apache.camel.support.compo
private static final Set<String> ENDPOINT_IDENTITY_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(38);
+ Set<String> props = new HashSet<>(39);
props.add("allowedSecurityPolicies");
props.add("applicationName");
props.add("applicationUri");
@@ -51,6 +51,7 @@ public class MiloClientEndpointUriFactory extends
org.apache.camel.support.compo
props.add("method");
props.add("monitorFilterType");
props.add("node");
+ props.add("nodes");
props.add("omitNullValues");
props.add("overrideHost");
props.add("overridePort");
diff --git
a/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/component/milo/client/milo-client.json
b/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/component/milo/client/milo-client.json
index 58b851de167e..f592a62caebd 100644
---
a/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/component/milo/client/milo-client.json
+++
b/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/component/milo/client/milo-client.json
@@ -55,7 +55,8 @@
},
"headers": {
"CamelMiloNodeIds": { "index": 0, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"List", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The node ids.", "constantName":
"org.apache.camel.component.milo.MiloConstants#HEADER_NODE_IDS" },
- "CamelMiloAwait": { "index": 1, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Boolean", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The await setting for writes.",
"constantName": "org.apache.camel.component.milo.MiloConstants#HEADER_AWAIT" }
+ "CamelMiloAwait": { "index": 1, "kind": "header", "displayName": "",
"group": "producer", "label": "producer", "required": false, "javaType":
"Boolean", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The await setting for writes.",
"constantName": "org.apache.camel.component.milo.MiloConstants#HEADER_AWAIT" },
+ "CamelMiloNodeId": { "index": 2, "kind": "header", "displayName": "",
"group": "consumer", "label": "consumer", "required": false, "javaType":
"String", "deprecated": false, "deprecationNote": "", "autowired": false,
"secret": false, "description": "The node ID of the value, as configured in the
node or nodes option. Not to be confused with CamelMiloNodeIds, the list of
nodes that the producer reads.", "constantName":
"org.apache.camel.component.milo.MiloConstants#HEADER_NODE_ID" }
},
"properties": {
"endpointUri": { "index": 0, "kind": "path", "displayName": "Endpoint
Uri", "group": "common", "label": "", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The OPC UA server
endpoint" },
@@ -72,29 +73,30 @@
"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" },
"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" },
- "bridgeErrorHandler": { "index": 14, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
- "exceptionHandler": { "index": 15, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
- "exchangePattern": { "index": 16, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
- "lazyStartProducer": { "index": 17, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
- "allowedSecurityPolicies": { "index": 18, "kind": "parameter",
"displayName": "Allowed Security Policies", "group": "client", "label":
"client", "required": false, "type": "string", "javaType": "java.lang.String",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "A set of allowed
security policy URIs. Default is to accept all and use t [...]
- "applicationName": { "index": 19, "kind": "parameter", "displayName":
"Application Name", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "Apache Camel adapter for
Eclipse Milo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application name" },
- "applicationUri": { "index": 20, "kind": "parameter", "displayName":
"Application Uri", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo\/Client", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application URI" },
- "channelLifetime": { "index": 21, "kind": "parameter", "displayName":
"Channel Lifetime", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Channel lifetime in
milliseconds" },
- "keyAlias": { "index": 22, "kind": "parameter", "displayName": "Key
Alias", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The name of the key in
the keystore file" },
- "keyPassword": { "index": 23, "kind": "parameter", "displayName": "Key
Password", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key password" },
- "keyStorePassword": { "index": 24, "kind": "parameter", "displayName":
"Key Store Password", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The keystore password" },
- "keyStoreType": { "index": 25, "kind": "parameter", "displayName": "Key
Store Type", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key store type" },
- "keyStoreUrl": { "index": 26, "kind": "parameter", "displayName": "Key
Store Url", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The URL where the key
should be loaded from" },
- "maxPendingPublishRequests": { "index": 27, "kind": "parameter",
"displayName": "Max Pending Publish Requests", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending publish requests" },
- "maxResponseMessageSize": { "index": 28, "kind": "parameter",
"displayName": "Max Response Message Size", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
bytes a response message may have" },
- "overrideHost": { "index": 29, "kind": "parameter", "displayName":
"Override Host", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint host with the host from the endpoint URI." },
- "overridePort": { "index": 30, "kind": "parameter", "displayName":
"Override Port", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint port with the port from the endpoint URI." },
- "productUri": { "index": 31, "kind": "parameter", "displayName": "Product
Uri", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The product URI" },
- "requestedPublishingInterval": { "index": 32, "kind": "parameter",
"displayName": "Requested Publishing Interval", "group": "client", "label":
"client", "required": false, "type": "number", "javaType": "java.lang.Double",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"1_000.0", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The requested publishing
interval in m [...]
- "requestTimeout": { "index": 33, "kind": "parameter", "displayName":
"Request Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Request timeout in
milliseconds" },
- "sessionName": { "index": 34, "kind": "parameter", "displayName": "Session
Name", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session name" },
- "sessionTimeout": { "index": 35, "kind": "parameter", "displayName":
"Session Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session timeout in
milliseconds" },
- "password": { "index": 36, "kind": "parameter", "displayName": "Password",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The password for
authentication. Use this instead of embedding credentials in th [...]
- "username": { "index": 37, "kind": "parameter", "displayName": "Username",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The username for
authentication. Use this instead of embedding credentials in the endpoint URI
when t [...]
+ "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)" },
+ "bridgeErrorHandler": { "index": 15, "kind": "parameter", "displayName":
"Bridge Error Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Allows for bridging the consumer to the
Camel routing Error Handler, which mean any exceptions (if possible) occurred
while the Camel consumer is trying to pickup incoming [...]
+ "exceptionHandler": { "index": 16, "kind": "parameter", "displayName":
"Exception Handler", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "object", "javaType":
"org.apache.camel.spi.ExceptionHandler", "optionalPrefix": "consumer.",
"deprecated": false, "autowired": false, "secret": false, "description": "To
let the consumer use a custom ExceptionHandler. Notice if the option
bridgeErrorHandler is enabled then this option is not in use. By de [...]
+ "exchangePattern": { "index": 17, "kind": "parameter", "displayName":
"Exchange Pattern", "group": "consumer (advanced)", "label":
"consumer,advanced", "required": false, "type": "enum", "javaType":
"org.apache.camel.ExchangePattern", "enum": [ "InOnly", "InOut" ],
"deprecated": false, "autowired": false, "secret": false, "description": "Sets
the exchange pattern when the consumer creates an exchange." },
+ "lazyStartProducer": { "index": 18, "kind": "parameter", "displayName":
"Lazy Start Producer", "group": "producer (advanced)", "label":
"producer,advanced", "required": false, "type": "boolean", "javaType":
"boolean", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": false, "description": "Whether the producer should be started
lazy (on the first message). By starting lazy you can use this to allow
CamelContext and routes to startup in situations where a produ [...]
+ "allowedSecurityPolicies": { "index": 19, "kind": "parameter",
"displayName": "Allowed Security Policies", "group": "client", "label":
"client", "required": false, "type": "string", "javaType": "java.lang.String",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "A set of allowed
security policy URIs. Default is to accept all and use t [...]
+ "applicationName": { "index": 20, "kind": "parameter", "displayName":
"Application Name", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue": "Apache Camel adapter for
Eclipse Milo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application name" },
+ "applicationUri": { "index": 21, "kind": "parameter", "displayName":
"Application Uri", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo\/Client", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The application URI" },
+ "channelLifetime": { "index": 22, "kind": "parameter", "displayName":
"Channel Lifetime", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Channel lifetime in
milliseconds" },
+ "keyAlias": { "index": 23, "kind": "parameter", "displayName": "Key
Alias", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The name of the key in
the keystore file" },
+ "keyPassword": { "index": 24, "kind": "parameter", "displayName": "Key
Password", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key password" },
+ "keyStorePassword": { "index": 25, "kind": "parameter", "displayName":
"Key Store Password", "group": "client", "label": "client", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The keystore password" },
+ "keyStoreType": { "index": 26, "kind": "parameter", "displayName": "Key
Store Type", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The key store type" },
+ "keyStoreUrl": { "index": 27, "kind": "parameter", "displayName": "Key
Store Url", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The URL where the key
should be loaded from" },
+ "maxPendingPublishRequests": { "index": 28, "kind": "parameter",
"displayName": "Max Pending Publish Requests", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
pending publish requests" },
+ "maxResponseMessageSize": { "index": 29, "kind": "parameter",
"displayName": "Max Response Message Size", "group": "client", "label":
"client", "required": false, "type": "integer", "javaType": "java.lang.Long",
"deprecated": false, "autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The maximum number of
bytes a response message may have" },
+ "overrideHost": { "index": 30, "kind": "parameter", "displayName":
"Override Host", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint host with the host from the endpoint URI." },
+ "overridePort": { "index": 31, "kind": "parameter", "displayName":
"Override Port", "group": "client", "label": "client", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Override the server
reported endpoint port with the port from the endpoint URI." },
+ "productUri": { "index": 32, "kind": "parameter", "displayName": "Product
Uri", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "defaultValue":
"http:\/\/camel.apache.org\/EclipseMilo", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The product URI" },
+ "requestedPublishingInterval": { "index": 33, "kind": "parameter",
"displayName": "Requested Publishing Interval", "group": "client", "label":
"client", "required": false, "type": "number", "javaType": "java.lang.Double",
"deprecated": false, "autowired": false, "secret": false, "defaultValue":
"1_000.0", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The requested publishing
interval in m [...]
+ "requestTimeout": { "index": 34, "kind": "parameter", "displayName":
"Request Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Request timeout in
milliseconds" },
+ "sessionName": { "index": 35, "kind": "parameter", "displayName": "Session
Name", "group": "client", "label": "client", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session name" },
+ "sessionTimeout": { "index": 36, "kind": "parameter", "displayName":
"Session Timeout", "group": "client", "label": "client", "required": false,
"type": "integer", "javaType": "java.lang.Long", "deprecated": false,
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "Session timeout in
milliseconds" },
+ "password": { "index": 37, "kind": "parameter", "displayName": "Password",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": true, "security": "secret", "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The password for
authentication. Use this instead of embedding credentials in th [...]
+ "username": { "index": 38, "kind": "parameter", "displayName": "Username",
"group": "security", "label": "security", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.milo.client.MiloClientConfiguration",
"configurationField": "configuration", "description": "The username for
authentication. Use this instead of embedding credentials in the endpoint URI
when t [...]
}
}
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 3c5b1aecdb95..e04f72f92737 100644
--- a/components/camel-milo/src/main/docs/milo-client-component.adoc
+++ b/components/camel-milo/src/main/docs/milo-client-component.adoc
@@ -127,6 +127,31 @@ However Camel allows to wrap the actual value inside
`RAW(…)`, which makes esc
milo-client:opc.tcp://user:password@localhost:12345?node=RAW(nsu=http://foo.bar;s=foo/bar)
----
+=== Monitoring several nodes
+
+A consumer can monitor several nodes with the `nodes` option instead of `node`
(the two options cannot be combined).
+The consumer creates the monitored items of all nodes together and deletes
them together when it stops. Every value
+arrives in its own exchange, and the `CamelMiloNodeId` header holds the node
ID as it was configured.
+
+The `nodes` option takes a comma-separated list of <<nodeid, Node IDs>>. The
entries are trimmed, empty entries are
+skipped and a node that is listed twice is monitored once:
+
+----
+milo-client:opc.tcp://localhost:12345?nodes=RAW(ns=2;s=line1/temperature,ns=2;s=line1/pressure)
+----
+
+A string node ID that contains a comma cannot be used in `nodes`; monitor it
with the `node` option instead. Long lists
+can come from a property placeholder, for example `+nodes=RAW({{tags}})+`.
+
+All nodes of the list share the `samplingInterval`, `queueSize` and
`omitNullValues` options of the consumer and its
+monitor filter (`monitorFilterType`, `dataChangeFilter*`). Tags that need a
different sampling interval or deadband
+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.
+
=== Method ID
It is possible to perform methods calls on OPC UA nodes. If the parameter
`method` is set to the Node ID of a method call (the node ID must be set to the
parent object in this case),
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/MiloConstants.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/MiloConstants.java
index b4c11dc57e24..82e27ac52757 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/MiloConstants.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/MiloConstants.java
@@ -30,6 +30,11 @@ public final class MiloConstants {
@Metadata(label = "producer", description = "The \"await\" setting for
writes.", javaType = "Boolean",
applicableFor = SCHEME_CLIENT)
public static final String HEADER_AWAIT = "CamelMiloAwait";
+ @Metadata(label = "consumer",
+ description = "The node ID of the value, as configured in the
node or nodes option. Not to be confused with"
+ + " CamelMiloNodeIds, the list of nodes that the
producer reads.",
+ javaType = "String", applicableFor = SCHEME_CLIENT)
+ public static final String HEADER_NODE_ID = "CamelMiloNodeId";
private MiloConstants() {
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java
index b8222164a8f6..73fbe7ae6a20 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConnection.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.milo.client;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@@ -118,16 +119,28 @@ public class MiloClientConnection implements
AutoCloseable {
final ExpandedNodeId nodeId, Double samplingInterval, Integer
queueSize,
MonitorFilterConfiguration monitorFilterConfiguration,
Consumer<DataValue> valueConsumer) {
- requireNonNull(configuration);
requireNonNull(valueConsumer);
+ return monitorValues(Collections.singletonMap(nodeId, valueConsumer),
samplingInterval, queueSize,
+ monitorFilterConfiguration);
+ }
+
+ /**
+ * Monitors several nodes, each with its own value consumer. The monitored
items of all nodes are created together,
+ * and the returned handle deletes them together.
+ */
+ public MonitorHandle monitorValues(
+ final Map<ExpandedNodeId, Consumer<DataValue>> valueConsumers,
final Double samplingInterval,
+ final Integer queueSize, final MonitorFilterConfiguration
monitorFilterConfiguration) {
+
+ requireNonNull(valueConsumers);
+
checkInit();
- final UInteger handle
- = this.manager.registerItem(nodeId, samplingInterval,
queueSize, valueConsumer,
- monitorFilterConfiguration);
+ final List<UInteger> handles
+ = this.manager.registerItems(valueConsumers, samplingInterval,
queueSize, monitorFilterConfiguration);
- return () -> MiloClientConnection.this.manager.unregisterItem(handle);
+ return () ->
MiloClientConnection.this.manager.unregisterItems(handles);
}
public String getConnectionId() {
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java
index 5f66a3a823fa..a6b6e0c2cd1c 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientConsumer.java
@@ -16,10 +16,16 @@
*/
package org.apache.camel.component.milo.client;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.component.milo.Messages;
+import org.apache.camel.component.milo.MiloConstants;
import
org.apache.camel.component.milo.client.MiloClientConnection.MonitorHandle;
import org.apache.camel.support.DefaultConsumer;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
@@ -33,14 +39,15 @@ public class MiloClientConsumer extends DefaultConsumer {
private MiloClientConnection connection;
private MonitorHandle handle;
- private ExpandedNodeId node;
private Double samplingInterval;
private Integer queueSize;
private boolean omitNullValues;
public MiloClientConsumer(final MiloClientEndpoint endpoint, final
Processor processor) {
super(endpoint, processor);
- this.node = endpoint.getNodeId();
+ if (endpoint.getNode() != null && endpoint.getNodes() != null) {
+ throw new IllegalArgumentException("Configure either the node or
the nodes option, not both");
+ }
this.samplingInterval = endpoint.getSamplingInterval();
this.queueSize = endpoint.getQueueSize();
this.omitNullValues = endpoint.isOmitNullValues();
@@ -54,9 +61,44 @@ public class MiloClientConsumer extends DefaultConsumer {
@Override
protected void doStart() throws Exception {
super.doStart();
+ final Map<ExpandedNodeId, Consumer<DataValue>> valueConsumers = new
LinkedHashMap<>();
+ resolveNodes().forEach((nodeId, node) -> valueConsumers.put(nodeId,
value -> handleValueUpdate(node, value)));
this.connection = getEndpoint().createConnection();
- this.handle = this.connection.monitorValue(this.node,
this.samplingInterval, this.queueSize,
- getEndpoint().getMonitorFilterConfiguration(),
this::handleValueUpdate);
+ this.handle = this.connection.monitorValues(valueConsumers,
this.samplingInterval, this.queueSize,
+ getEndpoint().getMonitorFilterConfiguration());
+ }
+
+ /**
+ * The node IDs of the node or nodes option, in the configured order and
without duplicates. A node that is listed
+ * in several spellings is monitored once, the value is the first spelling
of the node ID.
+ */
+ private Map<ExpandedNodeId, String> resolveNodes() {
+ final Map<ExpandedNodeId, String> result = new LinkedHashMap<>();
+
+ final String node = getEndpoint().getNode();
+ if (node != null) {
+ result.put(parseNodeId(node), node);
+ return result;
+ }
+
+ final String nodes = getEndpoint().getNodes();
+ if (nodes != null) {
+ Arrays.stream(nodes.split(",")).map(String::trim).filter(entry ->
!entry.isEmpty())
+ .forEach(entry -> result.putIfAbsent(parseNodeId(entry),
entry));
+ }
+
+ if (result.isEmpty()) {
+ throw new IllegalArgumentException("No node to monitor, configure
the node or the nodes option");
+ }
+ return result;
+ }
+
+ private static ExpandedNodeId parseNodeId(final String node) {
+ try {
+ return ExpandedNodeId.parse(node);
+ } catch (final Exception e) {
+ throw new IllegalArgumentException("Invalid node ID: " + node, e);
+ }
}
@Override
@@ -71,7 +113,7 @@ public class MiloClientConsumer extends DefaultConsumer {
super.doStop();
}
- private void handleValueUpdate(final DataValue value) {
+ 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)) {
@@ -82,6 +124,7 @@ public class MiloClientConsumer extends DefaultConsumer {
final Exchange exchange = createExchange(true);
try {
mapToMessage(value, exchange.getMessage());
+ exchange.getMessage().setHeader(MiloConstants.HEADER_NODE_ID,
node);
getProcessor().process(exchange);
} catch (final Exception e) {
getExceptionHandler().handleException("Error processing exchange",
e);
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java
index 50e32811074a..752e4adb114d 100644
---
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java
@@ -54,6 +54,12 @@ public class MiloClientEndpoint extends DefaultEndpoint {
@UriParam
private String node;
+ /**
+ * Comma-separated node definitions to monitor with one consumer, instead
of node (see Node ID)
+ */
+ @UriParam(label = "consumer")
+ private String nodes;
+
/**
* The method definition (see Method ID)
*/
@@ -170,6 +176,14 @@ public class MiloClientEndpoint extends DefaultEndpoint {
}
}
+ public void setNodes(final String nodes) {
+ this.nodes = nodes;
+ }
+
+ public String getNodes() {
+ return nodes;
+ }
+
ExpandedNodeId getMethodId() {
if (this.method != null) {
return ExpandedNodeId.parse(this.method);
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 65bd86614470..32957dfaf103 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
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -51,6 +52,7 @@ import
org.eclipse.milo.opcua.sdk.client.identity.AnonymousProvider;
import org.eclipse.milo.opcua.sdk.client.identity.CompositeProvider;
import org.eclipse.milo.opcua.sdk.client.identity.IdentityProvider;
import org.eclipse.milo.opcua.sdk.client.identity.UsernameProvider;
+import
org.eclipse.milo.opcua.sdk.client.subscriptions.MonitoredItemServiceOperationResult;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaMonitoredItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.OpcUaSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
@@ -160,6 +162,15 @@ public class SubscriptionManager {
}
}
+ /**
+ * A status value for a node that could not be monitored, delivered to its
consumer outside of the lock
+ */
+ private record PendingValue(Consumer<DataValue> consumer, DataValue value)
{
+ void deliver() {
+ consumer.accept(value);
+ }
+ }
+
private class Connected {
private OpcUaClient client;
private final OpcUaSubscription manager;
@@ -175,10 +186,21 @@ public class SubscriptionManager {
this.manager = manager;
}
- public void putSubscriptions(final Map<UInteger, Subscription>
subscriptions) throws Exception {
+ /**
+ * Creates the monitored items of the subscriptions on the server.
+ *
+ * @param subscriptions the subscriptions to create, by client handle
+ * @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
+ */
+ public List<PendingValue> putSubscriptions(final Map<UInteger,
Subscription> subscriptions) throws Exception {
+
+ final List<PendingValue> pendingValues = new ArrayList<>();
if (subscriptions.isEmpty()) {
- return;
+ return pendingValues;
}
// convert to requests
@@ -193,7 +215,7 @@ public class SubscriptionManager {
final NodeId node = lookupNamespace(s.getNodeId()).get();
if (node == null) {
- handleSubscriptionError(new
StatusCode(StatusCodes.Bad_InvalidArgument), entry.getKey(), s);
+ handleSubscriptionError(new
StatusCode(StatusCodes.Bad_InvalidArgument), entry.getKey(), s, pendingValues);
} else {
final ReadValueId itemId = new ReadValueId(node,
AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE);
final OpcUaMonitoredItem item = new
OpcUaMonitoredItem(itemId, MonitoringMode.Reporting);
@@ -207,6 +229,8 @@ public class SubscriptionManager {
if (filter != null) {
item.setFilter(filter);
}
+ // set the listener before the item is created, values
without a listener are dropped
+ item.setDataValueListener((monitoredItem, dataValue) ->
s.getValueConsumer().accept(dataValue));
items.add(item);
// Keep track of which subscription this item belongs to
itemToClientHandle.put(item, entry.getKey());
@@ -219,54 +243,82 @@ public class SubscriptionManager {
this.manager.addMonitoredItem(item);
}
- // Create monitored items on the server - this assigns client
handles
- this.manager.createMonitoredItems();
-
- // Now set up data value listeners after items are created and
have client handles
- for (OpcUaMonitoredItem item : items) {
- UInteger originalClientHandle =
itemToClientHandle.get(item);
- final Subscription s =
subscriptions.get(originalClientHandle);
- if (s != null) {
- item.setDataValueListener((monitoredItem, dataValue)
-> s.getValueConsumer().accept(dataValue));
+ // Create the monitored items of this call on the server
+ final List<MonitoredItemServiceOperationResult> results
+ =
this.manager.createMonitoredItems(itemToClientHandle::containsKey);
+
+ // the service call is partitioned, a rejected partition
reports its service result for each of its items
+ StatusCode rejectedServiceResult = null;
+ final List<OpcUaMonitoredItem> failedItems = new ArrayList<>();
+ for (final MonitoredItemServiceOperationResult result :
results) {
+ final OpcUaMonitoredItem item = result.monitoredItem();
+ final UInteger clientHandle = itemToClientHandle.get(item);
+ if (result.isGood()) {
+ this.goodSubscriptions.put(clientHandle, item);
+ } 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
+ 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.getValueConsumer(), new DataValue(status)));
}
}
+
+ if (!failedItems.isEmpty()) {
+ // remove the failed items from the subscription, they
were not created on the server
+ this.manager.removeMonitoredItems(failedItems);
+ this.manager.deleteMonitoredItems();
+ }
+
+ if (rejectedServiceResult != null) {
+ // the subscription is unusable, the caller tears the
connection down and builds it up again
+ throw new UaException(rejectedServiceResult);
+ }
}
if (!this.badSubscriptions.isEmpty()) {
SubscriptionManager.this.executor.schedule(this::resubscribe,
SubscriptionManager.this.reconnectTimeout,
TimeUnit.MILLISECONDS);
}
+
+ return pendingValues;
}
- private void handleSubscriptionError(final StatusCode statusCode,
final UInteger clientHandle, final Subscription s) {
+ private void handleSubscriptionError(
+ final StatusCode statusCode, final UInteger clientHandle,
final Subscription s,
+ final List<PendingValue> pendingValues) {
this.badSubscriptions.put(clientHandle, s);
- s.getValueConsumer().accept(new DataValue(statusCode));
+ pendingValues.add(new PendingValue(s.getValueConsumer(), new
DataValue(statusCode)));
}
private void resubscribe() {
final Map<UInteger, Subscription> subscriptions = new
HashMap<>(this.badSubscriptions);
this.badSubscriptions.clear();
try {
- putSubscriptions(subscriptions);
+ putSubscriptions(subscriptions).forEach(PendingValue::deliver);
} catch (final Exception e) {
handleConnectionFailure(e);
}
}
- public void activate(final UInteger clientHandle, final Subscription
subscription) throws Exception {
- putSubscriptions(Collections.singletonMap(clientHandle,
subscription));
- }
-
- public void deactivate(final UInteger clientHandle) throws Exception {
- final OpcUaMonitoredItem item =
this.goodSubscriptions.remove(clientHandle);
- if (item != null) {
- // TODO is this migrated correctly?
- this.manager.removeMonitoredItem(item);
- // this.manager.synchronizeMonitoredItems();
+ public void deactivate(final List<UInteger> clientHandles) throws
Exception {
+ final List<OpcUaMonitoredItem> items = new
ArrayList<>(clientHandles.size());
+ for (final UInteger clientHandle : clientHandles) {
+ final OpcUaMonitoredItem item =
this.goodSubscriptions.remove(clientHandle);
+ if (item != null) {
+ items.add(item);
+ } else {
+ this.badSubscriptions.remove(clientHandle);
+ }
+ }
+ if (!items.isEmpty()) {
+ this.manager.removeMonitoredItems(items);
this.manager.deleteMonitoredItems();
- //
this.manager.deleteMonitoredItems(Collections.singletonList(item)).get();
- } else {
- this.badSubscriptions.remove(clientHandle);
}
}
@@ -670,6 +722,7 @@ public class SubscriptionManager {
}
private void performAndEvalConnect() {
+ final List<PendingValue> pendingValues = new ArrayList<>();
try {
final Connected connected = performConnect();
LOG.debug("Connect call done");
@@ -682,7 +735,7 @@ public class SubscriptionManager {
try {
LOG.debug("Setting subscriptions: {}",
this.subscriptions.size());
- connected.putSubscriptions(this.subscriptions);
+
pendingValues.addAll(connected.putSubscriptions(this.subscriptions));
LOG.debug("Update state : {} -> {}", this.connected,
connected);
final Connected oldConnected = this.connected;
@@ -705,6 +758,9 @@ public class SubscriptionManager {
LOG.info("Failed to connect", e);
triggerReconnect(false);
}
+
+ // deliver outside of the lock, the routes may use the connection
+ pendingValues.forEach(PendingValue::deliver);
}
private Connected performConnect() throws Exception {
@@ -955,35 +1011,43 @@ public class SubscriptionManager {
return result;
}
- public UInteger registerItem(
- final ExpandedNodeId nodeId, final Double samplingInterval,
Integer queueSize,
- final Consumer<DataValue> valueConsumer,
- final MonitorFilterConfiguration monitorFilterConfiguration) {
+ public List<UInteger> registerItems(
+ final Map<ExpandedNodeId, Consumer<DataValue>> valueConsumers,
final Double samplingInterval,
+ final Integer queueSize, final MonitorFilterConfiguration
monitorFilterConfiguration) {
- final UInteger clientHandle =
Unsigned.uint(this.clientHandleCounter.incrementAndGet());
- final Subscription subscription
- = new Subscription(nodeId, samplingInterval, queueSize,
valueConsumer, monitorFilterConfiguration);
+ final Map<UInteger, Subscription> subscriptions = new
LinkedHashMap<>();
+ valueConsumers.forEach((nodeId, valueConsumer) -> subscriptions.put(
+ Unsigned.uint(this.clientHandleCounter.incrementAndGet()),
+ new Subscription(nodeId, samplingInterval, queueSize,
valueConsumer, monitorFilterConfiguration)));
+ final List<PendingValue> pendingValues = new ArrayList<>();
lock.lock();
try {
- this.subscriptions.put(clientHandle, subscription);
+ this.subscriptions.putAll(subscriptions);
- whenConnected(connected -> {
- connected.activate(clientHandle, subscription);
- });
+ whenConnected(connected ->
pendingValues.addAll(connected.putSubscriptions(subscriptions)));
} finally {
lock.unlock();
}
- return clientHandle;
+ // deliver outside of the lock, the routes may use the connection
+ pendingValues.forEach(PendingValue::deliver);
+
+ return new ArrayList<>(subscriptions.keySet());
}
- public void unregisterItem(final UInteger clientHandle) {
+ public void unregisterItems(final List<UInteger> clientHandles) {
lock.lock();
try {
- if (this.subscriptions.remove(clientHandle) != null) {
+ final List<UInteger> removed = new
ArrayList<>(clientHandles.size());
+ for (final UInteger clientHandle : clientHandles) {
+ if (this.subscriptions.remove(clientHandle) != null) {
+ removed.add(clientHandle);
+ }
+ }
+ if (!removed.isEmpty()) {
whenConnected(connected -> {
- connected.deactivate(clientHandle);
+ connected.deactivate(removed);
});
}
} finally {
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
new file mode 100644
index 000000000000..b5f50b241691
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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.List;
+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.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.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Testing the monitored items of consumers sharing a connection, when a
consumer is restarted, its node does not exist
+ * or the server rejects the CreateMonitoredItems service call
+ */
+public class MonitorItemLifecycleTest extends AbstractMiloServerTest {
+
+ private static final String DIRECT_START_1 = "direct:start1";
+
+ private static final String DIRECT_START_2 = "direct:start2";
+
+ private static final String MILO_SERVER_ITEM_1 = "milo-server:myitem1";
+
+ private static final String MILO_SERVER_ITEM_2 = "milo-server:myitem2";
+
+ private static final String NODE_ITEM_1 = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=myitem1";
+
+ private static final String NODE_ITEM_2 = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=myitem2";
+
+ private static final String NODE_UNKNOWN_ITEM_WRITE_BACK
+ = "nsu=" + MiloServerComponent.DEFAULT_NAMESPACE_URI +
";s=unknownItemWriteBack";
+
+ // the client endpoints only differ in the node and the consumer options,
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=";
+
+ private static final String MILO_CLIENT_ITEM_1
+ = MILO_CLIENT +
NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1");
+
+ private static final String MILO_CLIENT_ITEM_2
+ = 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";
+
+ private static final String MILO_CLIENT_UNKNOWN_ITEM_WRITE_BACK
+ = MILO_CLIENT +
NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI,
"unknownItemWriteBack")
+ + "&omitNullValues=false";
+
+ private static final String MILO_CLIENT_ITEMS_1_2
+ = MILO_CLIENT_BASE + "&omitNullValues=false&nodes=RAW(" +
NODE_ITEM_1 + "," + NODE_ITEM_2 + ")";
+
+ private static final String MOCK_ITEM_1 = "mock:item1";
+
+ private static final String MOCK_ITEM_2 = "mock:item2";
+
+ private static final String MOCK_UNKNOWN_ITEM = "mock:unknownItem";
+
+ private static final String MOCK_WRITE_BACK = "mock:writeBack";
+
+ private static final String MOCK_ITEMS_1_2 = "mock:items12";
+
+ @EndpointInject(MOCK_ITEM_1)
+ protected MockEndpoint item1Endpoint;
+
+ @EndpointInject(MOCK_ITEM_2)
+ protected MockEndpoint item2Endpoint;
+
+ @EndpointInject(MOCK_UNKNOWN_ITEM)
+ protected MockEndpoint unknownItemEndpoint;
+
+ @EndpointInject(MOCK_WRITE_BACK)
+ protected MockEndpoint writeBackEndpoint;
+
+ @EndpointInject(MOCK_ITEMS_1_2)
+ protected MockEndpoint items12Endpoint;
+
+ @Produce(DIRECT_START_1)
+ protected ProducerTemplate producer1;
+
+ @Produce(DIRECT_START_2)
+ protected ProducerTemplate producer2;
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ 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_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);
+
+ // the bad status of the unknown node is processed on another
thread, which writes to the same connection
+
from(resolve(MILO_CLIENT_UNKNOWN_ITEM_WRITE_BACK)).routeId("writeBack").autoStartup(false)
+ .threads(1)
+ .setBody(constant("Foo"))
+ .to(resolve(MILO_CLIENT_ITEM_1 +
"&defaultAwaitWrites=true"))
+ .to(MOCK_WRITE_BACK);
+ }
+ };
+ }
+
+ @Test
+ void restartedConsumerReceivesEachValueOnce() throws Exception {
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ item1Endpoint.assertIsSatisfied();
+
+ // the other consumers keep the shared connection open
+ context.getRouteController().stopRoute("item1");
+
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(0);
+ item1Endpoint.setAssertPeriod(1000);
+ producer1.sendBody("Bar");
+ item1Endpoint.assertIsSatisfied();
+
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Bar"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(1);
+ item1Endpoint.setAssertPeriod(1000);
+ testBody(item1Endpoint.message(0), assertGoodValue("Baz"));
+ producer1.sendBody("Baz");
+ item1Endpoint.assertIsSatisfied();
+ }
+
+ @Test
+ void unknownNodeReceivesBadStatus() throws Exception {
+ unknownItemEndpoint.expectedMinimumMessageCount(1);
+ testBody(unknownItemEndpoint.message(0),
assertStatus(StatusCodes.Bad_NodeIdUnknown));
+ unknownItemEndpoint.assertIsSatisfied();
+
+ // the other consumers of the connection are not affected
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ producer1.sendBody("Foo");
+ item1Endpoint.assertIsSatisfied();
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void badStatusIsDeliveredWithoutHoldingTheConnectionLock() throws
Exception {
+ // the route of the bad status writes to the shared connection from
another thread while the consumer starts
+ writeBackEndpoint.expectedMessageCount(1);
+ writeBackEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID,
NODE_UNKNOWN_ITEM_WRITE_BACK);
+ testBody(writeBackEndpoint.message(0), StatusCode.class, status ->
assertTrue(status.isGood(), status.toString()));
+
+ context.getRouteController().startRoute("writeBack");
+
+ writeBackEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ void failedCreateMonitoredItemsCallRebuildsTheConnection() throws
Exception {
+ producer1.sendBody("Foo");
+ producer2.sendBody("Bar");
+
+ item2Endpoint.expectedMessageCount(1);
+ testBody(item2Endpoint.message(0), assertGoodValue("Bar"));
+ context.getRouteController().startRoute("item2");
+ item2Endpoint.assertIsSatisfied();
+
+ // the server deletes the subscription, so the next
CreateMonitoredItems service call fails as a whole
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_ITEM_1), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final UInteger subscriptionId =
client.getSubscriptions().get(0).getSubscriptionId().orElseThrow();
+
assertTrue(client.deleteSubscriptions(List.of(subscriptionId)).getResults()[0].isGood());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+
+ // the connection is rebuilt and the new consumer receives the current
values, not a bad status
+ items12Endpoint.expectedMessageCount(2);
+ items12Endpoint.allMessages().predicate(exchange ->
exchange.getMessage().getBody(DataValue.class).getStatusCode()
+ .isGood());
+ context.getRouteController().startRoute("items12");
+ items12Endpoint.assertIsSatisfied();
+
+ producer1.sendBody("Baz");
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final Exchange last = lastExchange(items12Endpoint);
+ assertEquals(NODE_ITEM_1,
last.getMessage().getHeader(MiloConstants.HEADER_NODE_ID, String.class));
+
assertGoodValue("Baz").accept(last.getMessage().getBody(DataValue.class));
+ });
+
+ // the consumer that was running before keeps working on the rebuilt
connection
+ producer2.sendBody("Qux");
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() ->
assertGoodValue("Qux").accept(lastExchange(item2Endpoint).getMessage()
+ .getBody(DataValue.class)));
+ }
+
+ private static Exchange lastExchange(final MockEndpoint endpoint) {
+ final List<Exchange> exchanges = endpoint.getReceivedExchanges();
+ return exchanges.get(exchanges.size() - 1);
+ }
+
+ private static Consumer<DataValue> assertStatus(final long statusCode) {
+ return value -> assertEquals(new StatusCode(statusCode),
value.getStatusCode());
+ }
+}
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorNodesTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorNodesTest.java
new file mode 100644
index 000000000000..18460c74164a
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorNodesTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+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.mock.MockEndpoint;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
+import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
+import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Testing one consumer that monitors several nodes with the nodes option
+ */
+public class MonitorNodesTest extends AbstractMiloServerTest {
+
+ private static final String NODE_1 = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=node1";
+
+ private static final String NODE_2 = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=node2";
+
+ private static final String UNKNOWN_NODE = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=unknownNode";
+
+ // the same node as NODE_1 with the default server index spelled out
+ private static final String NODE_1_WITH_SERVER_INDEX = "svr=0;" + NODE_1;
+
+ // the client endpoints only differ in the node options, so they share one
connection
+ private static final String MILO_CLIENT =
"milo-client:opc.tcp://foo:bar@localhost:@@port@@"
+ +
"?allowedSecurityPolicies=None&overrideHost=true"
+ +
"&samplingInterval=50&requestedPublishingInterval=100";
+
+ private static final String MILO_CLIENT_NODES = MILO_CLIENT +
"&nodes=RAW(" + NODE_1 + "," + NODE_2 + ")";
+
+ private static final String MILO_CLIENT_UNKNOWN_NODE
+ = MILO_CLIENT + "&omitNullValues=false&nodes=RAW(" + NODE_1 + ","
+ UNKNOWN_NODE + "," + NODE_2 + ")";
+
+ private static final String MILO_CLIENT_DUPLICATES
+ = MILO_CLIENT + "&nodes=RAW( " + NODE_1 + " ,, " + NODE_2 + " ," +
NODE_1 + ")";
+
+ private static final String MILO_CLIENT_NODE = MILO_CLIENT + "&node=RAW("
+ NODE_1 + ")";
+
+ private static final String MILO_CLIENT_SAME_NODE
+ = MILO_CLIENT + "&nodes=RAW(" + NODE_1 + "," +
NODE_1_WITH_SERVER_INDEX + ")";
+
+ @EndpointInject("mock:nodes")
+ protected MockEndpoint nodesEndpoint;
+
+ @EndpointInject("mock:unknownNode")
+ protected MockEndpoint unknownNodeEndpoint;
+
+ @EndpointInject("mock:duplicates")
+ protected MockEndpoint duplicatesEndpoint;
+
+ @EndpointInject("mock:node")
+ protected MockEndpoint nodeEndpoint;
+
+ @EndpointInject("mock:sameNode")
+ protected MockEndpoint sameNodeEndpoint;
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // the consumers deliver the current values while they start,
so each test starts its route
+
from(resolve(MILO_CLIENT_NODES)).routeId("nodes").autoStartup(false).to("mock:nodes");
+
from(resolve(MILO_CLIENT_UNKNOWN_NODE)).routeId("unknownNode").autoStartup(false).to("mock:unknownNode");
+
from(resolve(MILO_CLIENT_DUPLICATES)).routeId("duplicates").autoStartup(false).to("mock:duplicates");
+
from(resolve(MILO_CLIENT_NODE)).routeId("node").autoStartup(false).to("mock:node");
+
from(resolve(MILO_CLIENT_SAME_NODE)).routeId("sameNode").autoStartup(false).to("mock:sameNode");
+ }
+ };
+ }
+
+ @Test
+ void consumerMonitorsSeveralNodes() throws Exception {
+ writeServerValues();
+
+ nodesEndpoint.expectedMessageCount(2);
+ context.getRouteController().startRoute("nodes");
+ nodesEndpoint.assertIsSatisfied();
+
+ final Map<String, DataValue> values = valuesByNode(nodesEndpoint);
+ assertEquals(Set.of(NODE_1, NODE_2), values.keySet());
+ assertGoodValue("Foo").accept(values.get(NODE_1));
+ assertGoodValue("Bar").accept(values.get(NODE_2));
+
+ nodesEndpoint.reset();
+ nodesEndpoint.expectedMessageCount(1);
+ nodesEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID,
NODE_2);
+ testBody(nodesEndpoint.message(0), assertGoodValue("Baz"));
+ template.sendBody("milo-server:node2", "Baz");
+ nodesEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ void unknownNodeReceivesBadStatusAndTheOtherNodesKeepWorking() throws
Exception {
+ writeServerValues();
+
+ unknownNodeEndpoint.expectedMessageCount(3);
+ context.getRouteController().startRoute("unknownNode");
+ unknownNodeEndpoint.assertIsSatisfied();
+
+ final Map<String, DataValue> values =
valuesByNode(unknownNodeEndpoint);
+ assertEquals(Set.of(NODE_1, UNKNOWN_NODE, NODE_2), values.keySet());
+ assertEquals(new StatusCode(StatusCodes.Bad_NodeIdUnknown),
values.get(UNKNOWN_NODE).getStatusCode());
+ assertGoodValue("Foo").accept(values.get(NODE_1));
+ assertGoodValue("Bar").accept(values.get(NODE_2));
+
+ unknownNodeEndpoint.reset();
+ unknownNodeEndpoint.expectedMessageCount(1);
+
unknownNodeEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID,
NODE_1);
+ testBody(unknownNodeEndpoint.message(0), assertGoodValue("Baz"));
+ template.sendBody("milo-server:node1", "Baz");
+ unknownNodeEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ void duplicateNodesAreMonitoredOnce() throws Exception {
+ writeServerValues();
+
+ // entries are trimmed and empty entries are skipped
+ duplicatesEndpoint.expectedMessageCount(2);
+ duplicatesEndpoint.setAssertPeriod(1000);
+ context.getRouteController().startRoute("duplicates");
+ duplicatesEndpoint.assertIsSatisfied();
+
+ final Map<String, DataValue> values = valuesByNode(duplicatesEndpoint);
+ assertEquals(Set.of(NODE_1, NODE_2), values.keySet());
+ assertGoodValue("Foo").accept(values.get(NODE_1));
+ assertGoodValue("Bar").accept(values.get(NODE_2));
+ }
+
+ @Test
+ void spellingsOfOneNodeAreMonitoredOnce() throws Exception {
+ // the entries are de-duplicated by the parsed node ID, not by their
spelling
+ assertEquals(ExpandedNodeId.parse("ns=0;i=2258"),
ExpandedNodeId.parse("i=2258"));
+ assertEquals(ExpandedNodeId.parse(NODE_1),
ExpandedNodeId.parse(NODE_1_WITH_SERVER_INDEX));
+
+ writeServerValues();
+
+ // one exchange per value, the header carries the first spelling
+ sameNodeEndpoint.expectedMessageCount(1);
+ sameNodeEndpoint.setAssertPeriod(1000);
+ sameNodeEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID,
NODE_1);
+ testBody(sameNodeEndpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("sameNode");
+ sameNodeEndpoint.assertIsSatisfied();
+
+ // and one monitored item on the server
+ final MiloClientEndpoint endpoint =
context.getEndpoint(resolve(MILO_CLIENT_SAME_NODE), MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ assertEquals(1,
connection.getOpcUaClient().getSubscriptions().get(0).getMonitoredItems().size());
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ void nodeAndNodesTogetherAreRejected() {
+ final Endpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_NODE
+ "&nodes=RAW(" + NODE_2 + ")"));
+
+ assertThrows(IllegalArgumentException.class, () ->
endpoint.createConsumer(exchange -> {
+ }));
+ }
+
+ @Test
+ void nodesWithoutNodeIdsAreRejected() throws Exception {
+ final Consumer consumer = context.getEndpoint(resolve(MILO_CLIENT +
"&nodes=RAW( , )")).createConsumer(exchange -> {
+ });
+
+ assertThrows(IllegalArgumentException.class, consumer::start);
+ }
+
+ @Test
+ void invalidNodeIdIsNamed() throws Exception {
+ final Consumer consumer = context.getEndpoint(resolve(MILO_CLIENT +
"&nodes=RAW(" + NODE_1 + ",ns=2;x=foo)"))
+ .createConsumer(exchange -> {
+ });
+
+ final IllegalArgumentException e =
assertThrows(IllegalArgumentException.class, consumer::start);
+ assertEquals("Invalid node ID: ns=2;x=foo", e.getMessage());
+ }
+
+ @Test
+ void singleNodeConsumerSetsTheNodeHeader() throws Exception {
+ writeServerValues();
+
+ nodeEndpoint.expectedMessageCount(1);
+ nodeEndpoint.expectedHeaderReceived(MiloConstants.HEADER_NODE_ID,
NODE_1);
+ testBody(nodeEndpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("node");
+ nodeEndpoint.assertIsSatisfied();
+ }
+
+ private void writeServerValues() {
+ // the server creates the nodes when the server endpoints start
+ template.sendBody("milo-server:node1", "Foo");
+ template.sendBody("milo-server:node2", "Bar");
+ }
+
+ private static Map<String, DataValue> valuesByNode(final MockEndpoint
endpoint) {
+ final Map<String, DataValue> values = new HashMap<>();
+ for (final Exchange exchange : endpoint.getReceivedExchanges()) {
+
values.put(exchange.getMessage().getHeader(MiloConstants.HEADER_NODE_ID,
String.class),
+ exchange.getMessage().getBody(DataValue.class));
+ }
+ return values;
+ }
+}
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 5d1d302fc766..2e119511a5ee 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
@@ -797,3 +797,16 @@ Semantic audit backends can opt into the additive
`inputType` extension. Readers
support it keep their existing projection. The `camel semantic audit --json`
display omits image
base64 and reports metadata instead. Use `--output=<new-file>` to export the
full returned JSON,
with `--include-input` for case/bulk requests that need captured input.
Existing files are not overwritten.
+
+=== camel-milo - creating and deleting the monitored items of a consumer
+
+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 (when `omitNullValues=false`). 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.
+
+The consumer now sets the `CamelMiloNodeId` header on every exchange,
including a consumer with a single `node`.
+
+Stopping a consumer now deletes its monitored items. Previously they stayed on
the server while another endpoint kept
+the shared connection open, so a restarted consumer received each value twice.
While the server is unreachable, the
+stop waits until the client's next attempt to reconnect has failed.
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MiloClientEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MiloClientEndpointBuilderFactory.java
index d6e3c02ca735..fdde075f0dce 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MiloClientEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/MiloClientEndpointBuilderFactory.java
@@ -361,6 +361,21 @@ public interface MiloClientEndpointBuilderFactory {
doSetProperty("samplingInterval", samplingInterval);
return this;
}
+ /**
+ * Comma-separated node definitions to monitor with one consumer,
+ * instead of node (see Node ID).
+ *
+ * The option is a: <code>java.lang.String</code> type.
+ *
+ * Group: consumer
+ *
+ * @param nodes the value to set
+ * @return the dsl builder
+ */
+ default MiloClientEndpointConsumerBuilder nodes(String nodes) {
+ doSetProperty("nodes", nodes);
+ return this;
+ }
/**
* A set of allowed security policy URIs. Default is to accept all and
* use the highest.
@@ -2542,6 +2557,20 @@ public interface MiloClientEndpointBuilderFactory {
public String miloAwait() {
return "CamelMiloAwait";
}
+ /**
+ * The node ID of the value, as configured in the node or nodes option.
+ * Not to be confused with CamelMiloNodeIds, the list of nodes that the
+ * producer reads.
+ *
+ * The option is a: {@code String} type.
+ *
+ * Group: consumer
+ *
+ * @return the name of the header {@code MiloNodeId}.
+ */
+ public String miloNodeId() {
+ return "CamelMiloNodeId";
+ }
}
static MiloClientEndpointBuilder endpointBuilder(String componentName,
String path) {
class MiloClientEndpointBuilderImpl extends AbstractEndpointBuilder
implements MiloClientEndpointBuilder, AdvancedMiloClientEndpointBuilder {
