Skip to content

CAMEL-25690: camel-milo - answer a producer node with an unknown namespace URI instead of reconnecting #28045

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 1 commit into from
Oct 19, 2026
Merged

CAMEL-25690: camel-milo - answer a producer node with an unknown namespace URI instead of reconnecting #28045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions ...rc/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,11 @@ 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)
----

A producer request for a node whose namespace URI (`nsu=`) is not in the namespace table of the server is answered
like a request for a node that the server does not have. A write returns the status code `Bad_NodeIdUnknown`. A read,
or a browse of the xref:milo-browse-component.adoc[Milo Browse] component, returns that status code for this node and
the results of the other nodes. The connection and the consumers that share it are not affected.

=== Monitoring and writing a node

The following routes monitor one node and write another, on the server of the
Expand Down
5 changes: 5 additions & 0 deletions components/camel-milo/src/main/docs/milo-client-component.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,11 @@ 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)
----

A producer request for a node whose namespace URI (`nsu=`) is not in the namespace table of the server is answered
like a request for a node that the server does not have. A write returns the status code `Bad_NodeIdUnknown`. A read,
or a browse of the xref:milo-browse-component.adoc[Milo Browse] component, returns that status code for this node and
the results of the other nodes. The connection and the consumers that share it are not affected.

=== Monitoring and writing a node

The following routes monitor one node and write another, on the server of the
Expand Down
93 changes: 71 additions & 22 deletions ...lo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
Expand Down Expand Up @@ -66,6 +67,7 @@
import org.eclipse.milo.opcua.stack.core.security.CertificateValidator;
import org.eclipse.milo.opcua.stack.core.types.builtin.ByteString;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.DiagnosticInfo;
import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
Expand Down Expand Up @@ -266,7 +268,7 @@ public List<PendingValue> putSubscriptions(final Map<UInteger, Subscription> sub
for (final Map.Entry<UInteger, Subscription> entry : subscriptions.entrySet()) {
final Subscription s = entry.getValue();

final NodeId node = lookupMonitoredNode(s.getNodeId());
final NodeId node = lookupNamespace(s.getNodeId()).get();

if (node == null) {
// reported once like a node the server rejects, the other nodes and the connection are not affected
Expand Down Expand Up @@ -342,16 +344,6 @@ public List<PendingValue> putSubscriptions(final Map<UInteger, Subscription> sub
return pendingValues;
}

/**
* @return the node ID with the namespace index of the server, or null if its namespace URI is not in the
* namespace table of the server
*/
private NodeId lookupMonitoredNode(final ExpandedNodeId nodeId) throws Exception {
final String uri = nodeId.getNamespaceUri();
final UShort index = uri == null ? nodeId.getNamespaceIndex() : lookupNamespaceIndex(uri).get();
return index == null ? null : toNodeId(index, nodeId);
}

/**
* Creates the subscription on the server again, with the monitored items it had, after the server lost it.
*
Expand Down Expand Up @@ -445,14 +437,24 @@ public void dispose() {
}
}

/**
* @return the node ID with the namespace index of the server, completed with null if its namespace URI is not
* in the namespace table of the server: such a node does not exist on the server
*/
private CompletableFuture<NodeId> lookupNamespace(final ExpandedNodeId nodeId) {
LOG.trace("Expanded Node Id: {}", nodeId);

final String uri = nodeId.getNamespaceUri();

if (uri != null) {
LOG.trace("Looking up namespace: {}", uri);
return lookupNamespaceIndex(uri).thenApply(index -> toNodeId(index, nodeId));
return lookupNamespaceIndex(uri).thenApply(index -> {
if (index == null) {
Comment thread
davsclaus marked this conversation as resolved.
Resolved
LOG.debug("The namespace URI of node {} is unknown to the server", nodeId.toParseableString());
return null;
}
return toNodeId(index, nodeId);
});
} else {
final UShort index = nodeId.getNamespaceIndex();
LOG.trace("Using provided index: {}", index);
Expand All @@ -479,6 +481,11 @@ public CompletableFuture<StatusCode> write(final ExpandedNodeId nodeId, final Da

LOG.debug("Node - expanded: {}, full: {}", nodeId, node);

if (node == null) {
// answered like a node the server does not have
return completedFuture(new StatusCode(StatusCodes.Bad_NodeIdUnknown));
}

return this.client.writeValuesAsync(List.of(node), List.of(value))
.thenApply(r -> r.get(0))
.whenComplete((status, error) -> {
Expand All @@ -502,6 +509,14 @@ public CompletableFuture<CallMethodResult> call(

LOG.debug("Method - expanded: {}, full: {}", methodId, method);

if (node == null || method == null) {
// answered like the server: an unknown object is Bad_NodeIdUnknown, an unknown method Bad_MethodInvalid
final StatusCode status
= new StatusCode(node == null ? StatusCodes.Bad_NodeIdUnknown : StatusCodes.Bad_MethodInvalid);
return completedFuture(
new CallMethodResult(status, new StatusCode[0], new DiagnosticInfo[0], new Variant[0]));
}

final CallMethodRequest cmr = new CallMethodRequest(node, method, inputArguments);

/*
Expand Down Expand Up @@ -535,7 +550,23 @@ public CompletableFuture<List<DataValue>> readValues(List<ExpandedNodeId> expand

return CompletableFuture.allOf(nodeIdFutures).thenCompose(param -> {
List<NodeId> nodeIds = Stream.of(nodeIdFutures).map(CompletableFuture::join).collect(Collectors.toList());
return this.client.readValuesAsync(0, TimestampsToReturn.Server, nodeIds);
final List<NodeId> known = nodeIds.stream().filter(Objects::nonNull).collect(Collectors.toList());
if (known.size() == nodeIds.size()) {
return this.client.readValuesAsync(0, TimestampsToReturn.Server, nodeIds);
}
// a node the server does not have gets the value the server returns for such a node
final CompletableFuture<List<DataValue>> read = known.isEmpty()
? completedFuture(List.of()) : this.client.readValuesAsync(0, TimestampsToReturn.Server, known);
return read.thenApply(values -> {
final Iterator<DataValue> knownValues = values.iterator();
final List<DataValue> result = new ArrayList<>(nodeIds.size());
for (final NodeId nodeId : nodeIds) {
result.add(nodeId == null
? new DataValue(new StatusCode(StatusCodes.Bad_NodeIdUnknown))
: knownValues.next());
}
return result;
});
});
}

Expand Down Expand Up @@ -704,15 +735,33 @@ public CompletableFuture<Map<ExpandedNodeId, BrowseResult>> browse(

.thenCompose(__ -> {

final List<NodeId> nodeIds = Stream.of(futures).map(CompletableFuture::join)
.collect(Collectors.toList());

return completedFuture(nodeIds.stream().map(nodeId -> new BrowseDescription(
nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses),
uint(BrowseResultMask.All.getValue()))).collect(Collectors.toList()));
})

.thenCompose(descriptions -> browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest));
// a node the server does not have gets the result the server returns for such a node
final Map<ExpandedNodeId, BrowseResult> unknown = new HashMap<>();
final List<BrowseDescription> descriptions = new ArrayList<>();
for (int i = 0; i < futures.length; i++) {
final NodeId nodeId = futures[i].join();
if (nodeId == null) {
unknown.put(expandedNodeIds.get(i), new BrowseResult(
new StatusCode(StatusCodes.Bad_NodeIdUnknown), ByteString.NULL_VALUE,
new ReferenceDescription[0]));
} else {
descriptions.add(new BrowseDescription(
nodeId, direction, NodeIds.References, includeSubTypes, uint(nodeClasses),
uint(BrowseResultMask.All.getValue())));
}
}
if (unknown.isEmpty()) {
return browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest);
}
final CompletableFuture<Map<ExpandedNodeId, BrowseResult>> knownResults = descriptions.isEmpty()
? completedFuture(Map.of())
: browse(descriptions, depth, maxDepth, pattern, maxNodesPerRequest);
return knownResults.thenApply(results -> {
final Map<ExpandedNodeId, BrowseResult> all = new HashMap<>(results);
all.putAll(unknown);
return all;
});
});
}
}

Expand Down
Loading
Loading