davsclaus commented on code in PR #27780:
URL: https://github.com/apache/camel/pull/27780#discussion_r4266919108
##########
components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java:
##########
@@ -219,17 +223,29 @@ public void putSubscriptions(final Map<UInteger,
Subscription> subscriptions) th
this.manager.addMonitoredItem(item);
}
- // Create monitored items on the server - this assigns client
handles
- this.manager.createMonitoredItems();
+ // Create the monitored items of this call on the server
+ final List<MonitoredItemServiceOperationResult> results
+ =
this.manager.createMonitoredItems(itemToClientHandle::containsKey);
- // 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));
+ 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 {
+ // reported once, the item is created again when the
consumer restarts or the connection is rebuilt
+ failedItems.add(item);
+ final StatusCode status =
result.operationResult().orElse(result.serviceResult());
+
subscriptions.get(clientHandle).getValueConsumer().accept(new
DataValue(status));
Review Comment:
`putSubscriptions` runs while `SubscriptionManager.lock` is held: from
`registerItems` through `whenConnected`, and from `performAndEvalConnect` on a
reconnect. This `accept` calls `handleValueUpdate`, so with
`omitNullValues=false` the route processes the bad-status exchange on this
thread, inside the lock.
With a long `nodes` list and many failed nodes (for example the
server-restart case from @allthingssecurity's point 2), this is one routed
exchange per failed node while the lock is held. Every `write`/`read`/`call` on
the shared connection from other threads waits on the lock meanwhile. On the
reconnect path `this.connected` is not set yet, so a `milo-client` producer
called from this route gets `No connected`.
`handleSubscriptionError` on `main` does the same for an unknown namespace,
so this is not a regression. An option is to collect the failed (consumer,
status) pairs and deliver them after the lock is released. If you prefer to
keep it as it is, that is fine for this PR.
##########
components/camel-milo/src/main/docs/milo-client-component.adoc:
##########
@@ -127,6 +127,26 @@ 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.
Review Comment:
Suggest one more sentence here: all nodes of the list share the consumer's
`samplingInterval`, `queueSize`, `omitNullValues` and the monitor filter
options (`monitorFilterType`, `dataChangeFilter*`). Tags that need a different
sampling interval or deadband need a consumer of their own. With thousands of
tags from one placeholder, users will ask this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
