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 3b2931e3ab99 CAMEL-25913: camel-milo - add a dev console that lists
the connections and their refused monitored items (#28272)
3b2931e3ab99 is described below
commit 3b2931e3ab9935f8b61e81fc5385c6aa19c532b9
Author: Serdar Gökay <[email protected]>
AuthorDate: Thu Oct 22 23:49:37 2026 +0300
CAMEL-25913: camel-milo - add a dev console that lists the connections and
their refused monitored items (#28272)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/dev-consoles-openapi.json | 78 +++
.../apache/camel/catalog/dev-consoles.properties | 1 +
.../camel/catalog/dev-consoles/milo-client.json | 78 +++
.../camel/catalog/docs/milo-client-component.adoc | 5 +
.../org/apache/camel/dev-console/milo-client.json | 78 +++
.../org/apache/camel/dev-console/milo-client | 2 +
.../org/apache/camel/dev-consoles.properties | 7 +
.../src/main/docs/milo-client-component.adoc | 5 +
.../milo/client/MiloClientConnection.java | 37 +-
.../component/milo/client/MiloClientConsumer.java | 7 +
.../milo/client/MiloClientDevConsole.java | 133 +++++
.../milo/client/internal/SubscriptionManager.java | 56 +-
.../component/milo/CommunicationLossTest.java | 99 +---
.../component/milo/MiloClientDevConsoleTest.java | 588 +++++++++++++++++++++
.../org/apache/camel/component/milo/TcpRelay.java | 111 ++++
15 files changed, 1178 insertions(+), 107 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
index 6c30c57da90d..c6f2f1b1939f 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
@@ -4486,6 +4486,84 @@
}
}
},
+ "\/q\/dev\/milo-client": {
+ "get": {
+ "summary": "OPC UA Client",
+ "description": "OPC UA client connections and
their refused monitored items",
+ "operationId": "milo-client",
+ "responses": {
+ "200": {
+ "description": "OPC UA Client
output",
+ "content": {
+ "application\/json": {
+ "schema": {
+ "type":
"object",
+
"properties": {
+
"connections": {
+
"type": "array",
+
"items": {
+
"type": "object",
+
"properties": {
+
"endpointUri": {
+
"type": "string",
+
"description": "The OPC UA server endpoint,
with its password masked"
+
},
+
"clientId": {
+
"type": "string",
+
"description": "The client id of the connection
(only present when one is set)"
+
},
+
"routeIds": {
+
"type": "array",
+
"items": {
+
"type": "string"
+
},
+
"description": "The ids of the routes whose
consumers use the connection"
+
},
+
"sessionActive": {
+
"type": "boolean",
+
"description": "Whether the session with the
server is active"
+
},
+
"monitoredNodes": {
+
"type": "integer",
+
"description": "The number of nodes that have a
monitored item on the server"
+
},
+
"refusedNodes": {
+
"type": "array",
+
"items": {
+
"type": "object",
+
"properties": {
+
"nodeId": {
+
"type": "string",
+
"description": "The
node ID"
+
},
+
"status": {
+
"type": "string",
+
"description": "The
status of the latest attempt to create the monitored item of the node: the
refusal, or the failure of the attempt"
+
},
+
"reason": {
+
"type": "string",
+
"description": "Why the
client refused the node without asking the server (only present when it did)"
+
}
+
}
+
},
+
"description": "The nodes whose monitored item
was refused, which are created again every reconnectTimeout milliseconds"
+
}
+
},
+
"required": [
+
"sessionActive",
+
"monitoredNodes"
+
]
+
},
+
"description": "The connections of the consumers"
+
}
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ },
"\/q\/dev\/opentelemetry": {
"get": {
"summary": "OpenTelemetry Spans",
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
index d1ffc72f9789..d694988169b9 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles.properties
@@ -43,6 +43,7 @@ main-http-server
memory
message-history
micrometer
+milo-client
opentelemetry
platform-http
processor
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/milo-client.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/milo-client.json
new file mode 100644
index 000000000000..7b4768636f30
--- /dev/null
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/milo-client.json
@@ -0,0 +1,78 @@
+{
+ "console": {
+ "kind": "console",
+ "group": "camel",
+ "name": "milo-client",
+ "title": "OPC UA Client",
+ "description": "OPC UA client connections and their refused monitored
items",
+ "deprecated": false,
+ "readOnly": true,
+ "javaType": "org.apache.camel.component.milo.client.MiloClientDevConsole",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-milo",
+ "version": "4.24.0-SNAPSHOT"
+ },
+ "response": {
+ "type": "object",
+ "properties": {
+ "connections": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "endpointUri": {
+ "type": "string",
+ "description": "The OPC UA server endpoint, with its password
masked"
+ },
+ "clientId": {
+ "type": "string",
+ "description": "The client id of the connection (only present
when one is set)"
+ },
+ "routeIds": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ },
+ "description": "The ids of the routes whose consumers use the
connection"
+ },
+ "sessionActive": {
+ "type": "boolean",
+ "description": "Whether the session with the server is active"
+ },
+ "monitoredNodes": {
+ "type": "integer",
+ "description": "The number of nodes that have a monitored item
on the server"
+ },
+ "refusedNodes": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "nodeId": {
+ "type": "string",
+ "description": "The node ID"
+ },
+ "status": {
+ "type": "string",
+ "description": "The status of the latest attempt to create
the monitored item of the node: the refusal, or the failure of the attempt"
+ },
+ "reason": {
+ "type": "string",
+ "description": "Why the client refused the node without
asking the server (only present when it did)"
+ }
+ }
+ },
+ "description": "The nodes whose monitored item was refused,
which are created again every reconnectTimeout milliseconds"
+ }
+ },
+ "required": [
+ "sessionActive",
+ "monitoredNodes"
+ ]
+ },
+ "description": "The connections of the consumers"
+ }
+ }
+ }
+}
+
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 9164dfc23803..e12de6d7a3b7 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
@@ -201,6 +201,11 @@ A node whose namespace URI (`nsu=`) is not in the
namespace table of the server
way, so a namespace that the server adds later is found.
The consumer delivers one exchange with the bad status code for such a node,
and the `CamelMiloNodeId` header names it.
+The `milo-client` developer console (`/q/dev/milo-client`, for example with
`camel run --console`) shows which nodes are
+not monitored: for each connection of the consumers, it lists the routes that
use it, whether its session is active, the
+number of monitored nodes, and the refused nodes, each with the status of the
latest attempt to create its monitored
+item. Reading it does not wait for a request to the server in progress.
+
=== Status codes and null values
A data change with a Bad or Uncertain status is always delivered, even when
its value is null; `omitNullValues` only
diff --git
a/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/dev-console/milo-client.json
b/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/dev-console/milo-client.json
new file mode 100644
index 000000000000..7b4768636f30
--- /dev/null
+++
b/components/camel-milo/src/generated/resources/META-INF/org/apache/camel/dev-console/milo-client.json
@@ -0,0 +1,78 @@
+{
+ "console": {
+ "kind": "console",
+ "group": "camel",
+ "name": "milo-client",
+ "title": "OPC UA Client",
+ "description": "OPC UA client connections and their refused monitored
items",
+ "deprecated": false,
+ "readOnly": true,
+ "javaType": "org.apache.camel.component.milo.client.MiloClientDevConsole",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-milo",
+ "version": "4.24.0-SNAPSHOT"
+ },
+ "response": {
+ "type": "object",
+ "properties": {
+ "connections": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "endpointUri": {
+ "type": "string",
+ "description": "The OPC UA server endpoint, with its password
masked"
+ },
+ "clientId": {
+ "type": "string",
+ "description": "The client id of the connection (only present
when one is set)"
+ },
+ "routeIds": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ },
+ "description": "The ids of the routes whose consumers use the
connection"
+ },
+ "sessionActive": {
+ "type": "boolean",
+ "description": "Whether the session with the server is active"
+ },
+ "monitoredNodes": {
+ "type": "integer",
+ "description": "The number of nodes that have a monitored item
on the server"
+ },
+ "refusedNodes": {
+ "type": "array",
+ "items": {
+ "type": "object",
+ "properties": {
+ "nodeId": {
+ "type": "string",
+ "description": "The node ID"
+ },
+ "status": {
+ "type": "string",
+ "description": "The status of the latest attempt to create
the monitored item of the node: the refusal, or the failure of the attempt"
+ },
+ "reason": {
+ "type": "string",
+ "description": "Why the client refused the node without
asking the server (only present when it did)"
+ }
+ }
+ },
+ "description": "The nodes whose monitored item was refused,
which are created again every reconnectTimeout milliseconds"
+ }
+ },
+ "required": [
+ "sessionActive",
+ "monitoredNodes"
+ ]
+ },
+ "description": "The connections of the consumers"
+ }
+ }
+ }
+}
+
diff --git
a/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-console/milo-client
b/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-console/milo-client
new file mode 100644
index 000000000000..286b5696975b
--- /dev/null
+++
b/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-console/milo-client
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.milo.client.MiloClientDevConsole
diff --git
a/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
b/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
new file mode 100644
index 000000000000..1e245f56026c
--- /dev/null
+++
b/components/camel-milo/src/generated/resources/META-INF/services/org/apache/camel/dev-consoles.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+dev-consoles=milo-client
+groupId=org.apache.camel
+artifactId=camel-milo
+version=4.24.0-SNAPSHOT
+projectName=Camel :: Milo
+projectDescription=Camel OPC UA support
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 9164dfc23803..e12de6d7a3b7 100644
--- a/components/camel-milo/src/main/docs/milo-client-component.adoc
+++ b/components/camel-milo/src/main/docs/milo-client-component.adoc
@@ -201,6 +201,11 @@ A node whose namespace URI (`nsu=`) is not in the
namespace table of the server
way, so a namespace that the server adds later is found.
The consumer delivers one exchange with the bad status code for such a node,
and the `CamelMiloNodeId` header names it.
+The `milo-client` developer console (`/q/dev/milo-client`, for example with
`camel run --console`) shows which nodes are
+not monitored: for each connection of the consumers, it lists the routes that
use it, whether its session is active, the
+number of monitored nodes, and the refused nodes, each with the status of the
latest attempt to create its monitored
+item. Reading it does not wait for a request to the server in progress.
+
=== Status codes and null values
A data change with a Bad or Uncertain status is always delivered, even when
its value is null; `omitNullValues` only
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 64168c9197fa..3b276156dd81 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
@@ -48,12 +48,36 @@ import static java.util.Objects.requireNonNull;
public class MiloClientConnection implements AutoCloseable, CamelContextAware {
+ /**
+ * What a connection monitors on its server.
+ *
+ * @param sessionActive whether the session with the server is active
+ * @param monitoredNodes the number of nodes that have a monitored item on
the server
+ * @param refusedNodes the nodes whose monitored item was refused, which
are created again every
+ * {@code reconnectTimeout} milliseconds
+ */
+ public record MonitoringState(boolean sessionActive, int monitoredNodes,
List<RefusedNode> refusedNodes) {
+ }
+
+ /**
+ * A node whose monitored item was refused.
+ *
+ * @param nodeId the node ID
+ * @param status the status of the latest attempt to create the monitored
item of the node: the refusal, or the
+ * failure of the attempt
+ * @param reason why the client refused the node without asking the
server, or {@code null}
+ */
+ public record RefusedNode(String nodeId, StatusCode status, String reason)
{
+ }
+
private final Lock lock = new ReentrantLock();
private final MiloClientConfiguration configuration;
private volatile CamelContext camelContext;
private ExecutorService executor;
private ScheduledExecutorService reconnectExecutor;
- private SubscriptionManager manager;
+ // read by getMonitoringState without the lock, which the first connect
and the close hold while they wait for the
+ // server
+ private volatile SubscriptionManager manager;
private volatile boolean initialized;
public MiloClientConnection(final MiloClientConfiguration configuration) {
@@ -100,6 +124,17 @@ public class MiloClientConnection implements
AutoCloseable, CamelContextAware {
return this.manager.getOpcUaClient();
}
+ /**
+ * Returns what this connection monitors on its server. It does not wait
for a service call in progress and does not
+ * connect.
+ *
+ * @return the monitoring state, or {@code null} if the connection is not
used yet or is closed
+ */
+ MonitoringState getMonitoringState() {
+ final SubscriptionManager manager = this.manager;
+ return manager != null ? manager.getMonitoringState() : null;
+ }
+
protected void init() {
ObjectHelper.notNull(this.camelContext, "camelContext");
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 59e0c90aabc1..706ff217b909 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
@@ -58,6 +58,13 @@ public class MiloClientConsumer extends DefaultConsumer {
return (MiloClientEndpoint) super.getEndpoint();
}
+ /**
+ * The connection that monitors the nodes of this consumer while it is
started
+ */
+ MiloClientConnection getConnection() {
+ return this.connection;
+ }
+
@Override
protected void doStart() throws Exception {
super.doStart();
diff --git
a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientDevConsole.java
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientDevConsole.java
new file mode 100644
index 000000000000..8670f893dc72
--- /dev/null
+++
b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientDevConsole.java
@@ -0,0 +1,133 @@
+/*
+ * 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.client;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.Route;
+import
org.apache.camel.component.milo.client.MiloClientConnection.MonitoringState;
+import org.apache.camel.component.milo.client.MiloClientConnection.RefusedNode;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.DevConsole;
+import org.apache.camel.support.console.AbstractDevConsole;
+import org.apache.camel.util.URISupport;
+import org.apache.camel.util.json.JsonRecordSupport;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
+import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
+
+/**
+ * The connections of the OPC UA client consumers, with their refused
monitored items. It reads the state of the
+ * connections without waiting for a service call in progress.
+ */
+@DevConsole(name = "milo-client", displayName = "OPC UA Client",
+ description = "OPC UA client connections and their refused
monitored items")
+public class MiloClientDevConsole extends AbstractDevConsole {
+
+ public record RefusedNodeEntry(
+ @Metadata(description = "The node ID") String nodeId,
+ @Metadata(description = "The status of the latest attempt to
create the monitored item of the node: the refusal, or the failure of the
attempt") String status,
+ @Metadata(description = "Why the client refused the node without
asking the server (only present when it did)") String reason) {
+ }
+
+ public record ConnectionEntry(
+ @Metadata(description = "The OPC UA server endpoint, with its
password masked") String endpointUri,
+ @Metadata(description = "The client id of the connection (only
present when one is set)") String clientId,
+ @Metadata(description = "The ids of the routes whose consumers use
the connection") List<String> routeIds,
+ @Metadata(description = "Whether the session with the server is
active") boolean sessionActive,
+ @Metadata(description = "The number of nodes that have a monitored
item on the server") int monitoredNodes,
+ @Metadata(description = "The nodes whose monitored item was
refused, which are created again every reconnectTimeout milliseconds")
List<RefusedNodeEntry> refusedNodes) {
+ }
+
+ public record Response(@Metadata(description = "The connections of the
consumers") List<ConnectionEntry> connections) {
+ }
+
+ public MiloClientDevConsole() {
+ super("camel", "milo-client", "OPC UA Client",
+ "OPC UA client connections and their refused monitored items");
+ }
+
+ @Override
+ protected String doCallText(Map<String, Object> options) {
+ StringBuilder sb = new StringBuilder();
+ for (ConnectionEntry connection : connections()) {
+ sb.append(String.format("%n Endpoint: %s",
connection.endpointUri()));
+ if (connection.clientId() != null) {
+ sb.append(String.format("%n Client Id: %s",
connection.clientId()));
+ }
+ sb.append(String.format("%n Routes: %s", String.join(", ",
connection.routeIds())));
+ sb.append(String.format("%n Session Active: %b",
connection.sessionActive()));
+ sb.append(String.format("%n Monitored Nodes: %d",
connection.monitoredNodes()));
+ sb.append(String.format("%n Refused Nodes: %d",
connection.refusedNodes().size()));
+ for (RefusedNodeEntry node : connection.refusedNodes()) {
+ sb.append(String.format("%n %s: %s", node.nodeId(),
node.status()));
+ if (node.reason() != null) {
+ sb.append(String.format(" (%s)", node.reason()));
+ }
+ }
+ sb.append("\n");
+ }
+ return sb.toString();
+ }
+
+ @Override
+ protected Map<String, Object> doCallJson(Map<String, Object> options) {
+ Response response = new Response(connections());
+ return JsonRecordSupport.toJsonObject(response);
+ }
+
+ /**
+ * The connections of the started consumers, each with the routes that use
it, in the order of the routes
+ */
+ private List<ConnectionEntry> connections() {
+ Map<MiloClientConnection, List<String>> routeIds = new
LinkedHashMap<>();
+ for (Route route : getCamelContext().getRoutes()) {
+ if (route.getConsumer() instanceof MiloClientConsumer consumer &&
consumer.isStarted()) {
+ routeIds.computeIfAbsent(consumer.getConnection(), connection
-> new ArrayList<>())
+ .add(route.getRouteId());
+ }
+ }
+
+ List<ConnectionEntry> connections = new ArrayList<>();
+ for (Map.Entry<MiloClientConnection, List<String>> entry :
routeIds.entrySet()) {
+ MonitoringState state = entry.getKey().getMonitoringState();
+ if (state == null) {
+ // closed while the consumers stop
+ continue;
+ }
+ List<RefusedNodeEntry> refusedNodes = new ArrayList<>();
+ for (RefusedNode node : state.refusedNodes()) {
+ refusedNodes.add(new RefusedNodeEntry(node.nodeId(),
statusName(node.status()), node.reason()));
+ }
+ MiloClientConfiguration configuration =
entry.getKey().getConfiguration();
+ connections.add(new ConnectionEntry(
+ URISupport.sanitizeUri(configuration.getEndpointUri()),
configuration.getClientId(), entry.getValue(),
+ state.sessionActive(), state.monitoredNodes(),
refusedNodes));
+ }
+ return connections;
+ }
+
+ /**
+ * The symbolic name of a status code, or its hexadecimal value if it has
none
+ */
+ private static String statusName(StatusCode status) {
+ return StatusCodes.lookup(status.getValue()).map(nameAndDescription ->
nameAndDescription[0])
+ .orElseGet(() -> String.format("0x%08X", status.getValue()));
+ }
+}
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 1ea704e01576..7ff69a4ec4bc 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
@@ -26,7 +26,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -34,6 +33,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
@@ -49,6 +49,8 @@ import java.util.stream.Stream;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.milo.client.MiloClientConfiguration;
+import
org.apache.camel.component.milo.client.MiloClientConnection.MonitoringState;
+import org.apache.camel.component.milo.client.MiloClientConnection.RefusedNode;
import org.apache.camel.component.milo.client.MonitorFilterConfiguration;
import org.eclipse.milo.opcua.sdk.client.DiscoveryClient;
import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
@@ -346,15 +348,20 @@ public class SubscriptionManager {
private final SubscriptionListenerImpl listener;
private final long generation;
- private final Map<UInteger, OpcUaMonitoredItem> goodSubscriptions =
new HashMap<>();
+ /**
+ * The monitored items by client handle, written under {@code lock}
and counted by the monitoring state without
+ * it
+ */
+ private final Map<UInteger, OpcUaMonitoredItem> goodSubscriptions =
new ConcurrentHashMap<>();
private final Map<String, UShort> namespaceCache = new
ConcurrentHashMap<>();
/**
- * The client handles of the nodes that the server refused to monitor
on this connection, which are created
- * again every {@code reconnectTimeout} milliseconds, guarded by
{@code lock}
+ * The nodes whose monitored item was refused on this connection, by
client handle, each with the status of its
+ * latest attempt, which are created again every {@code
reconnectTimeout} milliseconds. Written under
+ * {@code lock} and read by the monitoring state without it.
*/
- private final Set<UInteger> refused = new LinkedHashSet<>();
+ private final Map<UInteger, RefusedNode> refused = new
ConcurrentSkipListMap<>();
/**
* The next attempt to create the refused nodes again, if one is
scheduled, guarded by {@code lock}
@@ -454,7 +461,7 @@ public class SubscriptionManager {
if (result.isGood()) {
this.goodSubscriptions.put(clientHandle, item);
subscriptions.get(clientHandle).refused = false;
- if (this.refused.remove(clientHandle) && retry) {
+ if (this.refused.remove(clientHandle) != null &&
retry) {
LOG.info("Monitored item for node {} created on
retry",
subscriptions.get(clientHandle).getNodeId().toParseableString());
}
@@ -496,8 +503,8 @@ public class SubscriptionManager {
private void refuse(
final UInteger clientHandle, final Subscription s, final
StatusCode status, final String reason,
final boolean retry, final List<PendingValue> pendingValues) {
- this.refused.add(clientHandle);
final String node = s.getNodeId().toParseableString();
+ this.refused.put(clientHandle, new RefusedNode(node, status,
reason));
final String because = reason == null ? "" : ", " + reason;
if (retry) {
LOG.debug("Monitored item for node {} could not be created on
retry{}: {}", node, because, status);
@@ -964,7 +971,10 @@ public class SubscriptionManager {
private final ServerCertificateTrust serverCertificateTrust;
private final Lock lock = new ReentrantLock();
- private Connected connected;
+ /**
+ * The current connection, written under {@code lock} and read by the
monitoring state without it
+ */
+ private volatile Connected connected;
private volatile Throwable lastConnectFailure;
private volatile boolean disposed;
private Future<?> reconnectJob;
@@ -1001,6 +1011,23 @@ public class SubscriptionManager {
}
}
+ /**
+ * Returns what the current connection monitors. It is read without {@code
lock}, which a service call holds until
+ * the server answers, so its values are read one after the other and need
not be consistent with each other.
+ *
+ * @return the monitoring state, with no session and no nodes while there
is no connection or it is being closed
+ */
+ public MonitoringState getMonitoringState() {
+ final Connected connected = this.connected;
+ // a connection that is being closed can wait for the server, its
nodes are reported lost already
+ if (connected == null || this.disposed || connected.listener.closed) {
+ return new MonitoringState(false, 0, List.of());
+ }
+ return new MonitoringState(
+ connected.listener.sessionActive,
connected.goodSubscriptions.size(),
+ List.copyOf(connected.refused.values()));
+ }
+
/**
* Checks the subscription of a connection on the server after it stopped
publishing, and recovers it unless the
* server confirms that it still has it.
@@ -1125,7 +1152,8 @@ public class SubscriptionManager {
* while the connection is the current one. Nothing is delivered to the
consumers: a node created now delivers its
* first value from the server, and its refusal, if not delivered yet, is
dropped. While the session is not active,
* the attempt is skipped. A call that fails is made again with the next
attempt and never rebuilds the connection,
- * which would interrupt the nodes that are monitored: a lost session or
subscription is recovered without it.
+ * which would interrupt the nodes that are monitored: a lost session or
subscription is recovered without it. Its
+ * failure is the status of the nodes until the next attempt.
* <p>
* Like the registration of a consumer, the attempt holds {@code lock}
until the server answers, so that a consumer
* that stops or a subscription that is recreated in the meantime cannot
interleave with it, and the producers and
@@ -1146,16 +1174,24 @@ public class SubscriptionManager {
return;
}
final Map<UInteger, Subscription> refused = new LinkedHashMap<>();
- for (final UInteger clientHandle : connected.refused) {
+ for (final UInteger clientHandle : connected.refused.keySet()) {
final Subscription subscription =
this.subscriptions.get(clientHandle);
if (subscription != null) {
refused.put(clientHandle, subscription);
}
}
+ final Map<UInteger, RefusedNode> beforeAttempt = new
HashMap<>(connected.refused);
try {
connected.putSubscriptions(refused, true);
} catch (final Exception e) {
LOG.debug("Failed to create the refused monitored items again,
trying again later", e);
+ // the call was rejected or failed, or the namespace table
could not be read: a node that the attempt
+ // did not answer, whose entry is still the one from before
it, has the failure of the attempt
+ final StatusCode failure
+ = UaException.extractStatusCode(e).orElse(new
StatusCode(StatusCodes.Bad_UnexpectedError));
+ refused.keySet().forEach(clientHandle ->
connected.refused.computeIfPresent(clientHandle,
+ (handle, node) -> node == beforeAttempt.get(handle)
+ ? new RefusedNode(node.nodeId(), failure,
null) : node));
connected.scheduleRetry();
}
} finally {
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
index d311681e54e5..8996efb3d0d2 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
@@ -16,13 +16,6 @@
*/
package org.apache.camel.component.milo;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.util.ArrayList;
-import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -99,11 +92,11 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
@RegisterExtension
AvailablePortFinder.Port reconnectTimeoutRelayPort =
AvailablePortFinder.find();
- private final Relay relay = new Relay();
+ private final TcpRelay relay = new TcpRelay();
- private final Relay lateRelay = new Relay();
+ private final TcpRelay lateRelay = new TcpRelay();
- private final Relay reconnectTimeoutRelay = new Relay();
+ private final TcpRelay reconnectTimeoutRelay = new TcpRelay();
@Override
protected RoutesBuilder createRouteBuilder() {
@@ -314,90 +307,4 @@ public class CommunicationLossTest extends
AbstractMiloServerTest {
reconnectTimeoutRelay.start(reconnectTimeoutRelayPort.getPort(),
getServerPort());
reconnectTimeoutEndpoint.assertIsSatisfied();
}
-
- /**
- * A TCP relay to the server, whose connections the test can drop while it
keeps listening, and which can be down
- * while the client tries to connect.
- */
- private static final class Relay {
-
- private volatile ServerSocket listener;
-
- private final List<Socket> sockets = new ArrayList<>();
-
- void start(final int port, final int serverPort) throws IOException {
- final ServerSocket listener = new ServerSocket(port);
- this.listener = listener;
- final Thread acceptor = new Thread(() -> {
- while (!listener.isClosed()) {
- Socket client = null;
- try {
- client = listener.accept();
- final Socket server = new Socket("localhost",
serverPort);
- synchronized (sockets) {
- sockets.add(client);
- sockets.add(server);
- }
- pump(client, server);
- pump(server, client);
- } catch (final IOException e) {
- // the listener was closed, or the server refused:
nothing to relay
- if (client != null) {
- close(client);
- }
- }
- }
- }, "relay-" + port);
- acceptor.setDaemon(true);
- acceptor.start();
- }
-
- private static void pump(final Socket from, final Socket to) {
- final Thread thread = new Thread(() -> {
- final byte[] buffer = new byte[8192];
- try (InputStream in = from.getInputStream(); OutputStream out
= to.getOutputStream()) {
- int read;
- while ((read = in.read(buffer)) >= 0) {
- out.write(buffer, 0, read);
- out.flush();
- }
- } catch (final IOException e) {
- // the connection was dropped
- } finally {
- close(from);
- close(to);
- }
- }, "relay-pump");
- thread.setDaemon(true);
- thread.start();
- }
-
- void dropConnections() {
- synchronized (sockets) {
- sockets.forEach(Relay::close);
- sockets.clear();
- }
- }
-
- void stop() {
- final ServerSocket listener = this.listener;
- this.listener = null;
- if (listener != null) {
- try {
- listener.close();
- } catch (final IOException e) {
- // nothing left to close
- }
- }
- dropConnections();
- }
-
- private static void close(final Socket socket) {
- try {
- socket.close();
- } catch (final IOException e) {
- // already closed
- }
- }
- }
}
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientDevConsoleTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientDevConsoleTest.java
new file mode 100644
index 000000000000..632121397aba
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientDevConsoleTest.java
@@ -0,0 +1,588 @@
+/*
+ * 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.time.Duration;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.milo.server.MiloServerComponent;
+import org.apache.camel.component.milo.server.internal.CamelNamespace;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.console.DevConsole;
+import org.apache.camel.support.PluginHelper;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.util.json.JsonArray;
+import org.apache.camel.util.json.JsonObject;
+import org.eclipse.milo.opcua.sdk.server.items.DataItem;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The milo-client dev console lists the connections of the consumers, with
their refused monitored items
+ */
+public class MiloClientDevConsoleTest extends AbstractMiloServerTest {
+
+ // the client endpoints only differ in their nodes, so they share one
connection, which creates the refused nodes
+ // again every second
+ private static final String MILO_CLIENT_BASE =
"milo-client:opc.tcp://foo:bar@localhost:@@port@@"
+ +
"?allowedSecurityPolicies=None&overrideHost=true"
+ +
"&samplingInterval=50&requestedPublishingInterval=100"
+ + "&reconnectTimeout=1000";
+
+ // namespaces that the server does not have until a test adds them
+ private static final String LATE_NAMESPACE_URI =
"urn:org:apache:camel:late";
+
+ private static final String HELD_NAMESPACE_URI =
"urn:org:apache:camel:held";
+
+ private static final String MONITORED_NODE = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=monitoredItem";
+
+ private static final String MISSING_NODE = "nsu=" +
MiloServerComponent.DEFAULT_NAMESPACE_URI + ";s=missingItem";
+
+ private static final String LATE_NAMESPACE_NODE = "nsu=" +
LATE_NAMESPACE_URI + ";s=lateNamespaceItem";
+
+ private static final String HELD_NODE = "nsu=" + HELD_NAMESPACE_URI +
";s=heldItem";
+
+ private static final String MILO_CLIENT_NODES
+ = MILO_CLIENT_BASE + "&nodes=RAW(" + MONITORED_NODE + "," +
MISSING_NODE + "," + LATE_NAMESPACE_NODE + ")";
+
+ private static final String MILO_CLIENT_OTHER
+ = MILO_CLIENT_BASE + "&node=" +
NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "otherItem");
+
+ private static final String MILO_CLIENT_HELD = MILO_CLIENT_BASE +
"&node=RAW(" + HELD_NODE + ")";
+
+ // a connection of its own, whose requests time out after a second
+ private static final String MILO_CLIENT_TIMEOUT_BASE = MILO_CLIENT_BASE +
"&clientId=timeout&requestTimeout=1000";
+
+ private static final String MILO_CLIENT_TIMEOUT = MILO_CLIENT_TIMEOUT_BASE
+ "&node=RAW(" + HELD_NODE + ")";
+
+ private static final String MILO_CLIENT_TIMEOUT_MONITORED
+ = MILO_CLIENT_TIMEOUT_BASE + "&node=RAW(" + MONITORED_NODE + ")";
+
+ private static final String MILO_CLIENT_TIMEOUT_LATE_NAMESPACE
+ = MILO_CLIENT_TIMEOUT_BASE + "&node=RAW(" + LATE_NAMESPACE_NODE +
")";
+
+ // a connection of its own, through the relay: the client uses the port of
the relay, not the one of the server
+ private static final String MILO_CLIENT_RELAYED
+ = MILO_CLIENT_BASE + "&overridePort=true&nodes=RAW(" +
MONITORED_NODE + "," + MISSING_NODE + ")";
+
+ // nothing listens on this port
+ @RegisterExtension
+ AvailablePortFinder.Port unusedPortHolder = AvailablePortFinder.find();
+
+ @RegisterExtension
+ AvailablePortFinder.Port relayPortHolder = AvailablePortFinder.find();
+
+ private final TcpRelay relay = new TcpRelay();
+
+ @EndpointInject("mock:nodes")
+ protected MockEndpoint nodesEndpoint;
+
+ @EndpointInject("mock:other")
+ protected MockEndpoint otherEndpoint;
+
+ @EndpointInject("mock:timeout")
+ protected MockEndpoint timeoutEndpoint;
+
+ @EndpointInject("mock:timeoutMonitored")
+ protected MockEndpoint timeoutMonitoredEndpoint;
+
+ @EndpointInject("mock:timeoutLateNamespace")
+ protected MockEndpoint timeoutLateNamespaceEndpoint;
+
+ @EndpointInject("mock:relayed")
+ protected MockEndpoint relayedEndpoint;
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // the consumers deliver the current values while they start,
so the tests start these routes
+
from(resolve(MILO_CLIENT_NODES)).routeId("nodes").autoStartup(false).to("mock:nodes");
+
from(resolve(MILO_CLIENT_OTHER)).routeId("other").autoStartup(false).to("mock:other");
+
from(resolve(MILO_CLIENT_HELD)).routeId("held").autoStartup(false).to("mock:held");
+
from(resolve(MILO_CLIENT_TIMEOUT)).routeId("timeout").autoStartup(false).to("mock:timeout");
+
from(resolve(MILO_CLIENT_TIMEOUT_MONITORED)).routeId("timeoutMonitored").autoStartup(false)
+ .to("mock:timeoutMonitored");
+
from(resolve(MILO_CLIENT_TIMEOUT_LATE_NAMESPACE)).routeId("timeoutLateNamespace").autoStartup(false)
+ .to("mock:timeoutLateNamespace");
+ from("milo-client:opc.tcp://localhost:" +
unusedPortHolder.getPort()
+ + "?allowedSecurityPolicies=None&node=RAW(" +
MONITORED_NODE + ")")
+
.routeId("unreachable").autoStartup(false).to("mock:unreachable");
+ from(MILO_CLIENT_RELAYED.replace("@@port@@",
Integer.toString(relayPortHolder.getPort())))
+
.routeId("relayed").autoStartup(false).to("mock:relayed");
+ }
+ };
+ }
+
+ @Override
+ protected void doPostTearDown() throws Exception {
+ // after the context, so that its client disconnects through the relay
+ relay.stop();
+ }
+
+ @Test
+ void consoleWithoutConsumersListsNoConnection() {
+ final DevConsole console = resolveConsole();
+ assertEquals("camel", console.getGroup());
+ assertEquals("milo-client", console.getId());
+
+ assertEquals("", console.call(DevConsole.MediaType.TEXT));
+ final JsonObject out = (JsonObject)
console.call(DevConsole.MediaType.JSON);
+ final JsonArray connections = out.getCollection("connections");
+ assertTrue(connections.isEmpty());
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void consoleListsTheRefusedNodesOfAConnection() throws Exception {
+ startNodes();
+
+ final DevConsole console = resolveConsole();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject connection = singleConnection(console);
+ assertEquals("opc.tcp://foo:xxxxxx@localhost:" + getServerPort(),
connection.getString("endpointUri"));
+ assertEquals(List.of("nodes"),
connection.getCollection("routeIds"));
+ assertTrue(connection.getBoolean("sessionActive"));
+ assertEquals(1, connection.getInteger("monitoredNodes"));
+
+ final JsonArray refusedNodes =
connection.getCollection("refusedNodes");
+ assertEquals(2, refusedNodes.size());
+ // refused by the server
+ assertRefusedNode(refusedNodes.get(0), MISSING_NODE,
"Bad_NodeIdUnknown", null);
+ // refused by the client, without asking the server
+ assertRefusedNode(refusedNodes.get(1), LATE_NAMESPACE_NODE,
"Bad_NodeIdUnknown",
+ "the namespace URI is unknown to the server");
+ });
+
+ assertEquals(String.format("%n Endpoint:
opc.tcp://foo:xxxxxx@localhost:%d", getServerPort())
+ + String.format("%n Routes: nodes")
+ + String.format("%n Session Active: true")
+ + String.format("%n Monitored Nodes: 1")
+ + String.format("%n Refused Nodes: 2")
+ + String.format("%n %s: Bad_NodeIdUnknown",
MISSING_NODE)
+ + String.format("%n %s: Bad_NodeIdUnknown (the
namespace URI is unknown to the server)",
+ LATE_NAMESPACE_NODE)
+ + "\n",
+ console.call(DevConsole.MediaType.TEXT));
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void nodeCreatedOnTheServerLaterLeavesTheRefusedNodes() throws Exception {
+ startNodes();
+ final DevConsole console = resolveConsole();
+
+ // the server gets the missing node, which the next attempt creates
+ template.sendBody("milo-server:missingItem", "Foo");
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject connection = singleConnection(console);
+ assertEquals(2, connection.getInteger("monitoredNodes"));
+ final JsonArray refusedNodes =
connection.getCollection("refusedNodes");
+ assertEquals(1, refusedNodes.size());
+ assertRefusedNode(refusedNodes.get(0), LATE_NAMESPACE_NODE,
"Bad_NodeIdUnknown",
+ "the namespace URI is unknown to the server");
+ });
+
+ // the server gets the namespace without the node: the next attempt
asks the server, which refuses it
+ final CamelNamespace namespace = new
CamelNamespace(LATE_NAMESPACE_URI, getServer().getServer());
+ try {
+ namespace.startup();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonArray refusedNodes =
singleConnection(console).getCollection("refusedNodes");
+ assertEquals(1, refusedNodes.size());
+ assertRefusedNode(refusedNodes.get(0), LATE_NAMESPACE_NODE,
"Bad_NodeIdUnknown", null);
+ });
+
+ // and then the node
+ namespace.getOrAddItem("lateNamespaceItem").update("Foo");
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject connection = singleConnection(console);
+ assertEquals(3, connection.getInteger("monitoredNodes"));
+ assertTrue(connection.<JsonArray>
getCollection("refusedNodes").isEmpty());
+ });
+ } finally {
+ namespace.shutdown();
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void failedAttemptIsTheStatusOfItsNodes() throws Exception {
+ // refused by the client, the server does not have the namespace yet
+ timeoutEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("timeout");
+ timeoutEndpoint.assertIsSatisfied();
+ // and a node of a namespace that the server does not get in this test
+ timeoutLateNamespaceEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("timeoutLateNamespace");
+ timeoutLateNamespaceEndpoint.assertIsSatisfied();
+
+ // the server gets the node, but holds the CreateMonitoredItems calls
of the next attempts until they time out
+ final CountDownLatch release = new CountDownLatch(1);
+ final CamelNamespace namespace = holdingNamespace(new
CountDownLatch(1), new CountDownLatch(1), release);
+ try {
+ namespace.startup();
+ namespace.getOrAddItem("heldItem").update("Foo");
+
+ final DevConsole console = resolveConsole();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertRefusedNode(
+ singleConnection(console).<JsonArray>
getCollection("refusedNodes").get(0), HELD_NODE, "Bad_Timeout",
+ null));
+ // the session stays active, and the other node keeps the refusal
of the same attempt, which did not send it
+ // to the server
+ final JsonObject connection = singleConnection(console);
+ assertTrue(connection.getBoolean("sessionActive"));
+ final JsonArray refusedNodes =
connection.getCollection("refusedNodes");
+ assertEquals(2, refusedNodes.size());
+ assertRefusedNode(refusedNodes.get(1), LATE_NAMESPACE_NODE,
"Bad_NodeIdUnknown",
+ "the namespace URI is unknown to the server");
+
+ // the server answers again, the next attempt creates the node
+ release.countDown();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject created = singleConnection(console);
+ assertEquals(1, created.getInteger("monitoredNodes"));
+ final JsonArray stillRefused =
created.getCollection("refusedNodes");
+ assertEquals(1, stillRefused.size());
+ assertRefusedNode(stillRefused.get(0), LATE_NAMESPACE_NODE,
"Bad_NodeIdUnknown",
+ "the namespace URI is unknown to the server");
+ });
+ } finally {
+ release.countDown();
+ namespace.shutdown();
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void consumersOfAConnectionAreListedTogether() throws Exception {
+ startNodes();
+ template.sendBody("milo-server:otherItem", "Bar");
+ otherEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("other");
+ otherEndpoint.assertIsSatisfied();
+
+ final DevConsole console = resolveConsole();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject connection = singleConnection(console);
+ assertEquals(List.of("nodes", "other"),
connection.getCollection("routeIds"));
+ assertEquals(2, connection.getInteger("monitoredNodes"));
+ });
+
+ // a consumer that stops is not listed, and its node is not monitored
anymore
+ context.getRouteController().stopRoute("other");
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject connection = singleConnection(console);
+ assertEquals(List.of("nodes"),
connection.getCollection("routeIds"));
+ assertEquals(1, connection.getInteger("monitoredNodes"));
+ });
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void clientIdTellsTheConnectionsOfAServerApart() throws Exception {
+ template.sendBody("milo-server:otherItem", "Bar");
+ otherEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("other");
+ otherEndpoint.assertIsSatisfied();
+ template.sendBody("milo-server:monitoredItem", "Foo");
+ timeoutMonitoredEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("timeoutMonitored");
+ timeoutMonitoredEndpoint.assertIsSatisfied();
+
+ // both connections have the endpoint of the server, the second one
has a client id
+ final DevConsole console = resolveConsole();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonArray connections = connections(console);
+ assertEquals(2, connections.size());
+ final JsonObject other = (JsonObject) connections.get(0);
+ assertEquals("opc.tcp://foo:xxxxxx@localhost:" + getServerPort(),
other.getString("endpointUri"));
+ // omitted when none is set
+ assertFalse(other.containsKey("clientId"));
+ assertEquals(List.of("other"), other.getCollection("routeIds"));
+ assertEquals(1, other.getInteger("monitoredNodes"));
+ final JsonObject timeout = (JsonObject) connections.get(1);
+ assertEquals(other.getString("endpointUri"),
timeout.getString("endpointUri"));
+ assertEquals("timeout", timeout.getString("clientId"));
+ assertEquals(List.of("timeoutMonitored"),
timeout.getCollection("routeIds"));
+ assertEquals(1, timeout.getInteger("monitoredNodes"));
+ });
+
+ assertEquals(String.format("%n Endpoint:
opc.tcp://foo:xxxxxx@localhost:%d", getServerPort())
+ + String.format("%n Routes: other")
+ + String.format("%n Session Active: true")
+ + String.format("%n Monitored Nodes: 1")
+ + String.format("%n Refused Nodes: 0")
+ + "\n"
+ + String.format("%n Endpoint:
opc.tcp://foo:xxxxxx@localhost:%d", getServerPort())
+ + String.format("%n Client Id: timeout")
+ + String.format("%n Routes: timeoutMonitored")
+ + String.format("%n Session Active: true")
+ + String.format("%n Monitored Nodes: 1")
+ + String.format("%n Refused Nodes: 0")
+ + "\n",
+ console.call(DevConsole.MediaType.TEXT));
+ }
+
+ @Test
+ @Timeout(value = 90, unit = TimeUnit.SECONDS)
+ void lostSessionKeepsTheNodesOfItsConnection() throws Exception {
+ startNodes();
+ relay.start(relayPortHolder.getPort(), getServerPort());
+ relayedEndpoint.expectedMessageCount(2);
+ context.getRouteController().startRoute("relayed");
+ relayedEndpoint.assertIsSatisfied();
+
+ // each connection is listed with the routes that use it, in the order
of the routes
+ final DevConsole console = resolveConsole();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonArray connections = connections(console);
+ assertEquals(2, connections.size());
+ assertEquals(List.of("nodes"), ((JsonObject)
connections.get(0)).getCollection("routeIds"));
+ final JsonObject relayed = (JsonObject) connections.get(1);
+ assertEquals("opc.tcp://foo:xxxxxx@localhost:" +
relayPortHolder.getPort(), relayed.getString("endpointUri"));
+ assertEquals(List.of("relayed"),
relayed.getCollection("routeIds"));
+ assertTrue(relayed.getBoolean("sessionActive"));
+ assertEquals(1, relayed.getInteger("monitoredNodes"));
+ final JsonArray refusedNodes =
relayed.getCollection("refusedNodes");
+ assertEquals(1, refusedNodes.size());
+ assertRefusedNode(refusedNodes.get(0), MISSING_NODE,
"Bad_NodeIdUnknown", null);
+ });
+
+ // the relay goes down: the session of its connection is lost, which
keeps its nodes until it is active again
+ relay.stop();
+ await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonArray connections = connections(console);
+ assertEquals(2, connections.size());
+ assertTrue(((JsonObject)
connections.get(0)).getBoolean("sessionActive"));
+ final JsonObject relayed = (JsonObject) connections.get(1);
+ assertFalse(relayed.getBoolean("sessionActive"));
+ assertEquals(1, relayed.getInteger("monitoredNodes"));
+ final JsonArray refusedNodes =
relayed.getCollection("refusedNodes");
+ assertEquals(1, refusedNodes.size());
+ // its status is the one of the latest attempt, which the drop may
have failed
+ assertEquals(MISSING_NODE, ((JsonObject)
refusedNodes.get(0)).getString("nodeId"));
+ });
+
+ relay.start(relayPortHolder.getPort(), getServerPort());
+ await().atMost(60, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonArray connections = connections(console);
+ assertEquals(2, connections.size());
+ assertTrue(((JsonObject)
connections.get(1)).getBoolean("sessionActive"));
+ });
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void connectionWithoutSessionMonitorsNoNode() throws Exception {
+ context.getRouteController().startRoute("unreachable");
+
+ final JsonObject connection = singleConnection(resolveConsole());
+ assertEquals("opc.tcp://localhost:" + unusedPortHolder.getPort(),
connection.getString("endpointUri"));
+ assertEquals(List.of("unreachable"),
connection.getCollection("routeIds"));
+ assertFalse(connection.getBoolean("sessionActive"));
+ assertEquals(0, connection.getInteger("monitoredNodes"));
+ assertTrue(connection.<JsonArray>
getCollection("refusedNodes").isEmpty());
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void connectionBeingClosedMonitorsNoNode() throws Exception {
+ template.sendBody("milo-server:monitoredItem", "Foo");
+ timeoutMonitoredEndpoint.expectedMessageCount(1);
+ context.getRouteController().startRoute("timeoutMonitored");
+ timeoutMonitoredEndpoint.assertIsSatisfied();
+
+ // the server holds the CreateMonitoredItems call of the next consumer
until it times out, which closes the
+ // connection, and then the CloseSession call, which deletes the
monitored items
+ final CountDownLatch deleting = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ final CamelNamespace namespace = holdingNamespace(new
CountDownLatch(1), deleting, release);
+ CompletableFuture<Void> start = null;
+ try {
+ namespace.startup();
+ namespace.getOrAddItem("heldItem").update("Foo");
+ start = startRouteAsync("timeout");
+ assertTrue(deleting.await(10, TimeUnit.SECONDS));
+
+ // its nodes have been reported lost while the client waits for
the server
+ final DevConsole console = resolveConsole();
+ final JsonObject connection = singleConnection(console);
+ assertFalse(connection.getBoolean("sessionActive"));
+ assertEquals(0, connection.getInteger("monitoredNodes"));
+ assertTrue(connection.<JsonArray>
getCollection("refusedNodes").isEmpty());
+
+ // the server answers again, the connection is built up again with
both nodes
+ release.countDown();
+ start.get(30, TimeUnit.SECONDS);
+ await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
+ final JsonObject rebuilt = singleConnection(console);
+ assertEquals(List.of("timeout", "timeoutMonitored"),
rebuilt.getCollection("routeIds"));
+ assertTrue(rebuilt.getBoolean("sessionActive"));
+ assertEquals(2, rebuilt.getInteger("monitoredNodes"));
+ });
+ } finally {
+ release.countDown();
+ if (start != null) {
+ // the start ends once the client stops waiting for the
server, its failure fails the test above
+ start.handle((result, e) -> null).get(30, TimeUnit.SECONDS);
+ }
+ namespace.shutdown();
+ }
+ }
+
+ @Test
+ @Timeout(value = 60, unit = TimeUnit.SECONDS)
+ void consoleDoesNotWaitForAServiceCallInProgress() throws Exception {
+ startNodes();
+
+ // the server holds the CreateMonitoredItems call of the next
consumer, which holds the lock of the connection
+ final CountDownLatch creating = new CountDownLatch(1);
+ final CountDownLatch release = new CountDownLatch(1);
+ final CamelNamespace namespace = holdingNamespace(creating, new
CountDownLatch(1), release);
+ CompletableFuture<Void> start = null;
+ try {
+ namespace.startup();
+ namespace.getOrAddItem("heldItem").update("Foo");
+ start = startRouteAsync("held");
+ assertTrue(creating.await(10, TimeUnit.SECONDS));
+
+ final DevConsole console = resolveConsole();
+ final JsonObject connection =
assertTimeoutPreemptively(Duration.ofSeconds(5), () ->
singleConnection(console));
+ // the call was still in progress
+ assertFalse(start.isDone());
+ assertEquals(List.of("nodes"),
connection.getCollection("routeIds"));
+ assertEquals(1, connection.getInteger("monitoredNodes"));
+ assertEquals(2, connection.<JsonArray>
getCollection("refusedNodes").size());
+
+ release.countDown();
+ start.get(30, TimeUnit.SECONDS);
+ } finally {
+ release.countDown();
+ if (start != null) {
+ // the start ends once the server answers, its failure fails
the test above
+ start.handle((result, e) -> null).get(30, TimeUnit.SECONDS);
+ }
+ namespace.shutdown();
+ }
+ }
+
+ /**
+ * Starts the consumer of a node that the server has, one that it does not
have and one of a namespace that it does
+ * not have, and waits for the value of the first and the refusals of the
others
+ */
+ private void startNodes() throws Exception {
+ template.sendBody("milo-server:monitoredItem", "Foo");
+ nodesEndpoint.expectedMessageCount(3);
+ context.getRouteController().startRoute("nodes");
+ nodesEndpoint.assertIsSatisfied();
+ }
+
+ /**
+ * Starts a route off the test thread, its consumer waits for the server
while it starts
+ */
+ private CompletableFuture<Void> startRouteAsync(final String routeId) {
+ return CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().startRoute(routeId);
+ } catch (final Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+ }
+
+ /**
+ * A namespace of the server that holds the creation and the deletion of
its monitored items until it is released.
+ * The server answers a CreateMonitoredItems call once the items are
created, and a CloseSession call once the items
+ * of the session are deleted.
+ */
+ private CamelNamespace holdingNamespace(
+ final CountDownLatch creating, final CountDownLatch deleting,
final CountDownLatch release) {
+ return new CamelNamespace(HELD_NAMESPACE_URI, getServer().getServer())
{
+ @Override
+ public void onDataItemsCreated(final List<DataItem> dataItems) {
+ hold(creating, release);
+ super.onDataItemsCreated(dataItems);
+ }
+
+ @Override
+ public void onDataItemsDeleted(final List<DataItem> dataItems) {
+ hold(deleting, release);
+ super.onDataItemsDeleted(dataItems);
+ }
+ };
+ }
+
+ private static void hold(final CountDownLatch holding, final
CountDownLatch release) {
+ holding.countDown();
+ try {
+ release.await(30, TimeUnit.SECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private DevConsole resolveConsole() {
+ final DevConsole console =
PluginHelper.getDevConsoleResolver(context).resolveDevConsole("milo-client");
+ assertNotNull(console);
+ return console;
+ }
+
+ private static JsonArray connections(final DevConsole console) {
+ final JsonObject out = (JsonObject)
console.call(DevConsole.MediaType.JSON);
+ return out.getCollection("connections");
+ }
+
+ private static JsonObject singleConnection(final DevConsole console) {
+ final JsonArray connections = connections(console);
+ assertEquals(1, connections.size());
+ return (JsonObject) connections.get(0);
+ }
+
+ private static void assertRefusedNode(
+ final Object refusedNode, final String nodeId, final String
status, final String reason) {
+ final JsonObject node = (JsonObject) refusedNode;
+ assertEquals(nodeId, node.getString("nodeId"));
+ assertEquals(status, node.getString("status"));
+ if (reason == null) {
+ // omitted when the server was asked
+ assertFalse(node.containsKey("reason"));
+ } else {
+ assertEquals(reason, node.getString("reason"));
+ }
+ }
+
+ private MiloServerComponent getServer() {
+ return context.getComponent("milo-server", MiloServerComponent.class);
+ }
+}
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java
new file mode 100644
index 000000000000..b564fc117aef
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/TcpRelay.java
@@ -0,0 +1,111 @@
+/*
+ * 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.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A TCP relay to the server, whose connections a test can drop while it keeps
listening, and which can be down while
+ * the client tries to connect.
+ */
+final class TcpRelay {
+
+ private volatile ServerSocket listener;
+
+ private final List<Socket> sockets = new ArrayList<>();
+
+ void start(final int port, final int serverPort) throws IOException {
+ final ServerSocket listener = new ServerSocket(port);
+ this.listener = listener;
+ final Thread acceptor = new Thread(() -> {
+ while (!listener.isClosed()) {
+ Socket client = null;
+ try {
+ client = listener.accept();
+ final Socket server = new Socket("localhost", serverPort);
+ synchronized (sockets) {
+ sockets.add(client);
+ sockets.add(server);
+ }
+ pump(client, server);
+ pump(server, client);
+ } catch (final IOException e) {
+ // the listener was closed, or the server refused: nothing
to relay
+ if (client != null) {
+ close(client);
+ }
+ }
+ }
+ }, "relay-" + port);
+ acceptor.setDaemon(true);
+ acceptor.start();
+ }
+
+ private static void pump(final Socket from, final Socket to) {
+ final Thread thread = new Thread(() -> {
+ final byte[] buffer = new byte[8192];
+ try (InputStream in = from.getInputStream(); OutputStream out =
to.getOutputStream()) {
+ int read;
+ while ((read = in.read(buffer)) >= 0) {
+ out.write(buffer, 0, read);
+ out.flush();
+ }
+ } catch (final IOException e) {
+ // the connection was dropped
+ } finally {
+ close(from);
+ close(to);
+ }
+ }, "relay-pump");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ void dropConnections() {
+ synchronized (sockets) {
+ sockets.forEach(TcpRelay::close);
+ sockets.clear();
+ }
+ }
+
+ void stop() {
+ final ServerSocket listener = this.listener;
+ this.listener = null;
+ if (listener != null) {
+ try {
+ listener.close();
+ } catch (final IOException e) {
+ // nothing left to close
+ }
+ }
+ dropConnections();
+ }
+
+ private static void close(final Socket socket) {
+ try {
+ socket.close();
+ } catch (final IOException e) {
+ // already closed
+ }
+ }
+}
