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 3f11cf9f9991 CAMEL-25835: camel-milo - the consumer reports a lost
connection and refreshes its values after it (#28128)
3f11cf9f9991 is described below
commit 3f11cf9f999187f498afcdc595074a23a8dcd284
Author: Serdar Gökay <[email protected]>
AuthorDate: Tue Oct 20 22:01:37 2026 +0300
CAMEL-25835: camel-milo - the consumer reports a lost connection and
refreshes its values after it (#28128)
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../camel/catalog/docs/milo-client-component.adoc | 25 ++
.../src/main/docs/milo-client-component.adoc | 25 ++
.../milo/client/internal/SubscriptionManager.java | 221 ++++++++++++++-
.../component/milo/AbstractMiloServerTest.java | 16 ++
.../component/milo/CommunicationLossTest.java | 304 +++++++++++++++++++++
.../component/milo/MonitorItemLifecycleTest.java | 25 +-
.../milo/MonitorItemMultiConnectionsTest.java | 14 +-
.../milo/ServerCertificateValidationTest.java | 11 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 22 ++
9 files changed, 635 insertions(+), 28 deletions(-)
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 fec24e18aa2b..3ba0ced7902e 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
@@ -209,6 +209,31 @@ The default `omitNullValues=true` also omits a null value
with a Good status tha
does not learn that the node recovered. Set `omitNullValues=false` for a node
whose value can be null, such as a
`String`.
+=== Communication loss and recovery
+
+When the client loses communication with the server, every node that a
consumer monitors delivers an exchange with
+the status `Bad_NoCommunication` and a null value, stamped with the time the
client recognised the loss. Communication
+is lost when the session with the server stops being active, for example
because the network connection closed or
+the server stopped answering, and when Camel rebuilds the connection. With the
default settings, a connection that
+stops answering without closing is recognised after 10 to 15 seconds. A
consumer that starts while the server cannot
+be reached delivers the status for its nodes when it starts. A node delivers
the status once per loss; a node that the
+server refused to monitor keeps the status of that refusal instead, and
nothing is
+delivered when a consumer stops or the Camel context shuts down.
+
+Once communication is restored, every node delivers its current value, also
when the value did not change. In the
+existing session, Camel asks the server to send the current values again with
the `ResendData` method; when the
+server does not provide it, Camel creates the subscription again, and the
values the server queued during the loss
+are not delivered. Values can arrive more than once after a loss, and a value
received before Camel rebuilt the
+connection can arrive after the first value of the new connection. A route
that counts values, computes differences
+or detects changes must tolerate this. A value that did not change during the
loss keeps its earlier timestamps.
+
+Routes run while communication is lost. An OPC UA request that a route makes
while the client reconnects, for example
+a write, waits for the next reconnect attempt and is sent if that attempt
succeeds, so it can take effect after
+communication is restored. A request that fails rebuilds the connection.
+
+Camel keeps no last value. A route that needs one keeps the last good value of
each node itself. To recognise the
+status, check the status code of the `DataValue` body, for example with
`${body.statusCode.bad}`.
+
=== Subscriptions that stop publishing
The consumers of a connection share one subscription on the server. When the
server stops publishing for it (no
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 fec24e18aa2b..3ba0ced7902e 100644
--- a/components/camel-milo/src/main/docs/milo-client-component.adoc
+++ b/components/camel-milo/src/main/docs/milo-client-component.adoc
@@ -209,6 +209,31 @@ The default `omitNullValues=true` also omits a null value
with a Good status tha
does not learn that the node recovered. Set `omitNullValues=false` for a node
whose value can be null, such as a
`String`.
+=== Communication loss and recovery
+
+When the client loses communication with the server, every node that a
consumer monitors delivers an exchange with
+the status `Bad_NoCommunication` and a null value, stamped with the time the
client recognised the loss. Communication
+is lost when the session with the server stops being active, for example
because the network connection closed or
+the server stopped answering, and when Camel rebuilds the connection. With the
default settings, a connection that
+stops answering without closing is recognised after 10 to 15 seconds. A
consumer that starts while the server cannot
+be reached delivers the status for its nodes when it starts. A node delivers
the status once per loss; a node that the
+server refused to monitor keeps the status of that refusal instead, and
nothing is
+delivered when a consumer stops or the Camel context shuts down.
+
+Once communication is restored, every node delivers its current value, also
when the value did not change. In the
+existing session, Camel asks the server to send the current values again with
the `ResendData` method; when the
+server does not provide it, Camel creates the subscription again, and the
values the server queued during the loss
+are not delivered. Values can arrive more than once after a loss, and a value
received before Camel rebuilt the
+connection can arrive after the first value of the new connection. A route
that counts values, computes differences
+or detects changes must tolerate this. A value that did not change during the
loss keeps its earlier timestamps.
+
+Routes run while communication is lost. An OPC UA request that a route makes
while the client reconnects, for example
+a write, waits for the next reconnect attempt and is sent if that attempt
succeeds, so it can take effect after
+communication is restored. A request that fails rebuilds the connection.
+
+Camel keeps no last value. A route that needs one keeps the last good value of
each node itself. To recognise the
+status, check the status code of the `DataValue` body, for example with
`${body.statusCode.bad}`.
+
=== Subscriptions that stop publishing
The consumers of a connection share one subscription on the server. When the
server stops publishing for it (no
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 fe9f12ff406f..88aae7485114 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
@@ -67,6 +67,7 @@ import org.eclipse.milo.opcua.stack.core.UaException;
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.DateTime;
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;
@@ -115,8 +116,19 @@ public class SubscriptionManager {
StatusCodes.Bad_SecurityChecksFailed,
StatusCodes.Bad_TcpSecureChannelUnknown,
StatusCodes.Bad_RequestTypeInvalid);
+ /**
+ * The status of a monitored node while communication with the server is
lost: the source of the value is configured
+ * but cannot be reached, and no last value is kept (OPC 10000-8 Table 61,
OPC 10000-14 Table 80).
+ */
+ private static final StatusCode NO_COMMUNICATION = new
StatusCode(StatusCodes.Bad_NoCommunication);
+
private final AtomicLong clientHandleCounter = new AtomicLong();
+ /**
+ * The generation of the last connection built, zero before the first.
+ */
+ private final AtomicLong generationCounter = new AtomicLong();
+
/**
* Listens to the subscription of a connection and to the state of its
session. It is registered on the client
* before the client connects, so that it sees every change of the session
state.
@@ -124,16 +136,63 @@ public class SubscriptionManager {
private final class SubscriptionListenerImpl
implements OpcUaSubscription.SubscriptionListener,
SessionActivityListener {
- private volatile boolean sessionActive;
+ /**
+ * The generation of the connection, which orders its values and loss
reports against those of the connection
+ * that replaces it.
+ */
+ private final long generation;
+
+ /**
+ * Whether the session is active. The client connects before the
connection is used, and Milo dispatches the
+ * activity callbacks in the order of the transitions.
+ */
+ private volatile boolean sessionActive = true;
+
+ /**
+ * The subscription of the connection, once it is created.
+ */
+ private volatile OpcUaSubscription subscription;
+
+ /**
+ * Set once the connection has been disposed: the session the client
closes then is no loss of communication.
+ */
+ private volatile boolean closed;
+
+ /**
+ * The session that the client lost last, while it has not been
activated again.
+ */
+ private volatile NodeId lostSessionId;
+
+ SubscriptionListenerImpl(final long generation) {
+ this.generation = generation;
+ }
@Override
public void onSessionActive(final UaSession session) {
this.sessionActive = true;
+ final NodeId lostSessionId = this.lostSessionId;
+ this.lostSessionId = null;
+ final OpcUaSubscription subscription = this.subscription;
+ if (lostSessionId != null && subscription != null && !this.closed)
{
+ if (lostSessionId.equals(session.getSessionId())) {
+ // the same session again: the server sends only changes,
so ask it for every value
+ resendData(subscription, this);
+ } else {
+ LOG.debug("Session {} replaced by {}, the server sends the
current values", lostSessionId,
+ session.getSessionId());
+ }
+ }
}
@Override
public void onSessionInactive(final UaSession session) {
this.sessionActive = false;
+ final OpcUaSubscription subscription = this.subscription;
+ if (subscription != null && !this.closed) {
+ this.lostSessionId = session.getSessionId();
+ LOG.info("Session {} lost, reporting the monitored nodes",
session.getSessionId());
+ reportLoss(subscription, this.generation,
SubscriptionManager.this.subscriptions);
+ }
}
/**
@@ -184,6 +243,22 @@ public class SubscriptionManager {
private MonitorFilterConfiguration monitorFilterConfiguration;
private final Integer queueSize;
+ /**
+ * The generation of the connection that delivered the last value,
guarded by this
+ */
+ private long generation;
+
+ /**
+ * Whether the last delivery to the consumer was a loss report,
guarded by this
+ */
+ private boolean reported;
+
+ /**
+ * Whether the server refused to monitor the node: its status is that
refusal, whatever the communication, until
+ * its item is created
+ */
+ private volatile boolean refused;
+
Subscription(ExpandedNodeId nodeId, final Double samplingInterval,
Integer queueSize,
final Consumer<DataValue> valueConsumer,
final MonitorFilterConfiguration
monitorFilterConfiguration) {
@@ -206,8 +281,38 @@ public class SubscriptionManager {
return this.queueSize;
}
- public Consumer<DataValue> getValueConsumer() {
- return this.valueConsumer;
+ /**
+ * Delivers a value of the node to the consumer, which ends a loss
report.
+ *
+ * @param generation the generation of the connection that received
the value
+ */
+ synchronized void deliver(final long generation, final DataValue
value) {
+ this.generation = Math.max(this.generation, generation);
+ this.reported = false;
+ this.valueConsumer.accept(value);
+ }
+
+ /**
+ * Delivers the status with which the server refused to monitor the
node.
+ */
+ void refuse(final long generation, final DataValue status) {
+ this.refused = true;
+ deliver(generation, status);
+ }
+
+ /**
+ * Delivers a loss report to the consumer, unless the node is reported
or refused already, or a later connection
+ * has delivered a value. The report of a replaced connection is
queued behind its last values, which may race
+ * the first values of its replacement.
+ *
+ * @param generation the generation of the connection whose
communication was lost
+ */
+ synchronized void report(final long generation, final DataValue value)
{
+ if (this.reported || this.refused || this.generation > generation)
{
+ return;
+ }
+ this.reported = true;
+ this.valueConsumer.accept(value);
}
public MonitoringFilter createMonitoringFilter() {
@@ -223,23 +328,27 @@ 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)
{
+ private record PendingValue(Subscription subscription, long generation,
DataValue value) {
void deliver() {
- consumer.accept(value);
+ subscription.refuse(generation, value);
}
}
private class Connected {
private OpcUaClient client;
private final OpcUaSubscription manager;
+ private final SubscriptionListenerImpl listener;
+ private final long generation;
private final Map<UInteger, OpcUaMonitoredItem> goodSubscriptions =
new HashMap<>();
private final Map<String, UShort> namespaceCache = new
ConcurrentHashMap<>();
- Connected(final OpcUaClient client, final OpcUaSubscription manager) {
+ Connected(final OpcUaClient client, final OpcUaSubscription manager,
final SubscriptionListenerImpl listener) {
this.client = client;
this.manager = manager;
+ this.listener = listener;
+ this.generation = listener.generation;
}
/**
@@ -275,7 +384,7 @@ public class SubscriptionManager {
final StatusCode status = new
StatusCode(StatusCodes.Bad_NodeIdUnknown);
LOG.warn("Monitored item for node {} could not be created,
the namespace URI is unknown to the server: {}",
s.getNodeId().toParseableString(), status);
- pendingValues.add(new PendingValue(s.getValueConsumer(),
new DataValue(status)));
+ pendingValues.add(new PendingValue(s, this.generation, new
DataValue(status)));
} else {
final ReadValueId itemId = new ReadValueId(node,
AttributeId.Value.uid(), null, QualifiedName.NULL_VALUE);
final OpcUaMonitoredItem item = new
OpcUaMonitoredItem(itemId, MonitoringMode.Reporting);
@@ -290,7 +399,7 @@ public class SubscriptionManager {
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));
+ item.setDataValueListener((monitoredItem, dataValue) ->
s.deliver(this.generation, dataValue));
items.add(item);
// Keep track of which subscription this item belongs to
itemToClientHandle.put(item, entry.getKey());
@@ -315,6 +424,7 @@ public class SubscriptionManager {
final UInteger clientHandle = itemToClientHandle.get(item);
if (result.isGood()) {
this.goodSubscriptions.put(clientHandle, item);
+ subscriptions.get(clientHandle).refused = false;
} else if (result.serviceResult().isBad()) {
failedItems.add(item);
rejectedServiceResult = result.serviceResult();
@@ -325,7 +435,7 @@ public class SubscriptionManager {
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)));
+ pendingValues.add(new PendingValue(s, this.generation,
new DataValue(status)));
}
}
@@ -426,6 +536,8 @@ public class SubscriptionManager {
}
public void dispose() {
+ // the session closed by the disconnect is no loss of communication
+ this.listener.closed = true;
if (this.client != null) {
try {
this.client.disconnect();
@@ -774,9 +886,14 @@ public class SubscriptionManager {
private Connected connected;
private volatile Throwable lastConnectFailure;
- private boolean disposed;
+ private volatile boolean disposed;
private Future<?> reconnectJob;
- private final Map<UInteger, Subscription> subscriptions = new HashMap<>();
+
+ /**
+ * The registered nodes by client handle, written under {@code lock}. A
loss report reads them without the lock,
+ * which a registration holds while the server creates its items.
+ */
+ private final Map<UInteger, Subscription> subscriptions = new
ConcurrentHashMap<>();
public SubscriptionManager(final MiloClientConfiguration configuration,
final ScheduledExecutorService reconnectExecutor,
final Executor executor, final long
reconnectTimeout) {
@@ -927,10 +1044,76 @@ public class SubscriptionManager {
return !this.disposed && this.connected != null &&
this.connected.manager == subscription;
}
+ /**
+ * Asks the server to send the current value of every monitored item of
the subscription again, after the session
+ * was lost and activated again. The server sends only changes through a
reactivated session, so a node whose value
+ * did not change would keep its loss report. If the server does not
resend, the subscription is deleted and created
+ * again in the session, and its new items deliver their first values.
+ */
+ private void resendData(final OpcUaSubscription subscription, final
SubscriptionListenerImpl listener) {
+ final UInteger subscriptionId =
subscription.getSubscriptionId().orElse(null);
+ if (subscriptionId == null) {
+ LOG.debug("Subscription has no ID, it is being recovered");
+ return;
+ }
+ final CallMethodRequest request = new CallMethodRequest(
+ NodeIds.Server, NodeIds.Server_ResendData, new Variant[] { new
Variant(subscriptionId) });
+
subscription.getClient().callAsync(List.of(request)).whenCompleteAsync((response,
e) -> {
+ final StatusCode status;
+ if (e != null) {
+ status = UaException.extractStatusCode(e).orElse(new
StatusCode(StatusCodes.Bad_UnexpectedError));
+ } else if (response.getResults() == null ||
response.getResults().length == 0) {
+ status = new StatusCode(StatusCodes.Bad_UnexpectedError);
+ } else {
+ status = response.getResults()[0].getStatusCode();
+ }
+ if (status.isGood()) {
+ LOG.info("Session activated again, subscription {} resends the
values of its nodes", subscriptionId);
+ } else if (listener.isSessionLost(status)) {
+ LOG.info("Session lost while subscription {} was asked to
resend its values: {}", subscriptionId,
+ status);
+ } else {
+ LOG.warn("Server does not resend the values of subscription {}
({}), recreating it", subscriptionId,
+ status);
+ recoverSubscription(subscription, listener);
+ }
+ }, clientExecutor(subscription));
+ }
+
+ /**
+ * Queues the loss report of the given nodes on the delivery queue of the
subscription, behind the values it already
+ * holds. The queue keeps running after the client has disconnected, so a
replaced connection delivers the report
+ * after its last values.
+ *
+ * @param generation the generation of the connection whose communication
was lost
+ */
+ private void reportLoss(
+ final OpcUaSubscription subscription, final long generation, final
Map<UInteger, Subscription> nodes) {
+ final DateTime time = DateTime.now();
+ subscription.getDeliveryQueue().execute(() -> deliverLoss(nodes,
generation, time));
+ }
+
+ /**
+ * Delivers the loss report to the given nodes that are still registered.
+ */
+ private void deliverLoss(final Map<UInteger, Subscription> nodes, final
long generation, final DateTime time) {
+ final DataValue value = new DataValue(Variant.NULL_VALUE,
NO_COMMUNICATION, time, time);
+ for (final Map.Entry<UInteger, Subscription> node : nodes.entrySet()) {
+ if (this.disposed) {
+ return;
+ }
+ if (this.subscriptions.get(node.getKey()) == node.getValue()) {
+ node.getValue().report(generation, value);
+ }
+ }
+ }
+
private void handleConnectionFailure(final Throwable e) {
lock.lock();
try {
if (this.connected != null) {
+ // the nodes lose their communication with the connection,
after the values it still delivers
+ reportLoss(this.connected.manager, this.connected.generation,
this.subscriptions);
this.connected.dispose();
this.connected = null;
}
@@ -1101,7 +1284,7 @@ public class SubscriptionManager {
}
});
- final SubscriptionListenerImpl listener = new
SubscriptionListenerImpl();
+ final SubscriptionListenerImpl listener = new
SubscriptionListenerImpl(this.generationCounter.incrementAndGet());
client.addSessionActivityListener(listener);
try {
@@ -1114,8 +1297,9 @@ public class SubscriptionManager {
subscription.setPublishingInterval(this.configuration.getRequestedPublishingInterval());
subscription.setSubscriptionListener(listener);
subscription.create();
+ listener.subscription = subscription;
- return new Connected(client, subscription);
+ return new Connected(client, subscription, listener);
// final OpcUaSubscription manager =
client.getSubscriptionManager()
//
.createSubscription(this.configuration.getRequestedPublishingInterval()).get();
@@ -1309,6 +1493,17 @@ public class SubscriptionManager {
try {
this.subscriptions.putAll(subscriptions);
+ if (this.connected == null) {
+ // no communication with the server: the nodes are reported
off the thread that starts the consumer,
+ // with a generation below every connection's, so the report
is dropped once any connection has
+ // delivered their values, also one that was being built while
they were registered
+ final DateTime time = DateTime.now();
+ this.executor.execute(() -> deliverLoss(subscriptions, 0L,
time));
+ } else if (!this.connected.listener.sessionActive) {
+ // the session is lost; the nodes are reported before the
values of their items, on the same queue
+ reportLoss(this.connected.manager, this.connected.generation,
subscriptions);
+ }
+
whenConnected(connected ->
pendingValues.addAll(connected.putSubscriptions(subscriptions)));
} finally {
lock.unlock();
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/AbstractMiloServerTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/AbstractMiloServerTest.java
index 6c255aec3331..27e9f60d4195 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/AbstractMiloServerTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/AbstractMiloServerTest.java
@@ -28,6 +28,7 @@ import
org.apache.camel.component.milo.server.MiloServerComponent;
import org.apache.camel.component.mock.AssertionClause;
import org.apache.camel.test.AvailablePortFinder;
import org.apache.camel.test.junit6.CamelTestSupport;
+import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -38,6 +39,7 @@ import org.slf4j.LoggerFactory;
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.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
public abstract class AbstractMiloServerTest extends CamelTestSupport {
@@ -106,6 +108,20 @@ public abstract class AbstractMiloServerTest extends
CamelTestSupport {
};
}
+ /**
+ * The report of a node whose communication with the server is lost: a
null value with {@code Bad_NoCommunication},
+ * stamped with the time the loss was recognised.
+ */
+ public static Consumer<DataValue> assertNoCommunication() {
+ return value -> {
+ assertNotNull(value);
+ assertEquals(StatusCodes.Bad_NoCommunication,
value.getStatusCode().getValue());
+ assertNull(value.getValue().getValue());
+ assertNotNull(value.getSourceTime());
+ assertEquals(value.getSourceTime(), value.getServerTime());
+ };
+ }
+
@Override
protected CamelContext createCamelContext() throws Exception {
final CamelContext context = super.createCamelContext();
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
new file mode 100644
index 000000000000..958faad13dd4
--- /dev/null
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/CommunicationLossTest.java
@@ -0,0 +1,304 @@
+/*
+ * 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;
+
+import org.apache.camel.EndpointInject;
+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.apache.camel.test.AvailablePortFinder;
+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.NodeId;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * A consumer reports its nodes with {@code Bad_NoCommunication} when
communication with the server is lost, and
+ * delivers their current values again once it is restored. The client reaches
the server through a TCP relay of the
+ * test, which drops the connection: the client then activates its session
again, and the server resends the values.
+ */
+public class CommunicationLossTest extends AbstractMiloServerTest {
+
+ private static final String DIRECT_START_1 = "direct:start1";
+
+ private static final String MILO_SERVER_ITEM_1 = "milo-server:myitem1";
+
+ private static final String MILO_CLIENT_BASE =
"milo-client:opc.tcp://foo:bar@localhost:@@relay@@"
+ +
"?allowedSecurityPolicies=None&overrideHost=true&overridePort=true"
+ +
"&samplingInterval=50&requestedPublishingInterval=100";
+
+ private static final String MILO_CLIENT_ITEM_1
+ = MILO_CLIENT_BASE + "&node=" +
NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1");
+
+ private static final String MOCK_ITEM_1 = "mock:item1";
+
+ private static final String MOCK_LATE = "mock:late";
+
+ private static final String MOCK_UNKNOWN = "mock:unknown";
+
+ private static final String MILO_CLIENT_UNKNOWN
+ = MILO_CLIENT_BASE + "&node=" +
NodeIds.nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "unknownItem");
+
+ @EndpointInject(MOCK_ITEM_1)
+ protected MockEndpoint item1Endpoint;
+
+ @EndpointInject(MOCK_LATE)
+ protected MockEndpoint lateEndpoint;
+
+ @EndpointInject(MOCK_UNKNOWN)
+ protected MockEndpoint unknownEndpoint;
+
+ @Produce(DIRECT_START_1)
+ protected ProducerTemplate producer1;
+
+ @RegisterExtension
+ AvailablePortFinder.Port relayPort = AvailablePortFinder.find();
+
+ @RegisterExtension
+ AvailablePortFinder.Port lateRelayPort = AvailablePortFinder.find();
+
+ private final Relay relay = new Relay();
+
+ private final Relay lateRelay = new Relay();
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(DIRECT_START_1).to(MILO_SERVER_ITEM_1);
+
+ // the consumers deliver the current values while they start,
so the tests start these routes
+ from(resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@",
Integer.toString(relayPort.getPort())))
+ .routeId("item1").autoStartup(false).to(MOCK_ITEM_1);
+
+ // a node the server refuses, on the same connection
+ from(resolve(MILO_CLIENT_UNKNOWN).replace("@@relay@@",
Integer.toString(relayPort.getPort())))
+
.routeId("unknown").autoStartup(false).to(MOCK_UNKNOWN);
+
+ // its own relay, so its own connection, which the test starts
while the relay is down
+ from(resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@",
Integer.toString(lateRelayPort.getPort())))
+ .routeId("late").autoStartup(false).to(MOCK_LATE);
+ }
+ };
+ }
+
+ @Override
+ protected void doPostSetup() throws Exception {
+ relay.start(relayPort.getPort(), getServerPort());
+ }
+
+ @Override
+ protected void doPostTearDown() throws Exception {
+ // after the context, so that its clients disconnect through the relays
+ relay.stop();
+ lateRelay.stop();
+ }
+
+ @Test
+ void lostSessionReportsTheNodeAndResendsItsValue() throws Exception {
+ producer1.sendBody("Foo");
+
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Foo"));
+ context.getRouteController().startRoute("item1");
+ item1Endpoint.assertIsSatisfied();
+
+ // the refused node gets its refusal once, and keeps it through the
losses below
+ unknownEndpoint.expectedMessageCount(1);
+ testBody(unknownEndpoint.message(0),
+ value -> assertEquals(StatusCodes.Bad_NodeIdUnknown,
value.getStatusCode().getValue()));
+ context.getRouteController().startRoute("unknown");
+ unknownEndpoint.assertIsSatisfied();
+ unknownEndpoint.reset();
+ unknownEndpoint.expectedMessageCount(0);
+ unknownEndpoint.setSleepForEmptyTest(500);
+
+ final MiloClientEndpoint endpoint = context.getEndpoint(
+ resolve(MILO_CLIENT_ITEM_1).replace("@@relay@@",
Integer.toString(relayPort.getPort())),
+ MiloClientEndpoint.class);
+ final MiloClientConnection connection = endpoint.createConnection();
+ try {
+ final OpcUaClient client = connection.getOpcUaClient();
+ final NodeId sessionId = client.getSession().getSessionId();
+
+ // the connection drops: the node is reported, then the
reactivated session resends its unchanged value.
+ // The test server answers the Publish requests it still queues
for the dropped channel first, one per
+ // keep-alive interval, so the resent value reaches the client
through its republish of the missed
+ // notification, 10 to 20 seconds later
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(2);
+ item1Endpoint.setResultWaitTime(30_000);
+ item1Endpoint.setAssertPeriod(2_000);
+ testBody(item1Endpoint.message(0), assertNoCommunication());
+ testBody(item1Endpoint.message(1), assertGoodValue("Foo"));
+ relay.dropConnections();
+ item1Endpoint.assertIsSatisfied();
+
+ // through the same client and session: the connection was not
rebuilt, the session not replaced
+ assertSame(client, connection.getOpcUaClient());
+ assertEquals(sessionId, client.getSession().getSessionId());
+
+ // changes arrive as before
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(1);
+ testBody(item1Endpoint.message(0), assertGoodValue("Bar"));
+ producer1.sendBody("Bar");
+ item1Endpoint.assertIsSatisfied();
+
+ // a second loss reports the node once more
+ item1Endpoint.reset();
+ item1Endpoint.expectedMessageCount(2);
+ item1Endpoint.setResultWaitTime(30_000);
+ item1Endpoint.setAssertPeriod(2_000);
+ testBody(item1Endpoint.message(0), assertNoCommunication());
+ testBody(item1Endpoint.message(1), assertGoodValue("Bar"));
+ relay.dropConnections();
+ item1Endpoint.assertIsSatisfied();
+
+ assertSame(client, connection.getOpcUaClient());
+ assertEquals(sessionId, client.getSession().getSessionId());
+ unknownEndpoint.assertIsSatisfied();
+ } finally {
+ endpoint.releaseConnection(connection);
+ }
+ }
+
+ @Test
+ void consumerStartedWithoutTheServerReportsTheNodeThenReceivesItsValue()
throws Exception {
+ producer1.sendBody("Foo");
+
+ // the relay is down: the consumer cannot connect and reports its node
while it starts
+ lateEndpoint.expectedMessageCount(1);
+ lateEndpoint.setAssertPeriod(5_000);
+ testBody(lateEndpoint.message(0), assertNoCommunication());
+ context.getRouteController().startRoute("late");
+ lateEndpoint.assertIsSatisfied();
+
+ // the server becomes reachable: the next connection attempt delivers
the current value
+ lateEndpoint.reset();
+ lateEndpoint.expectedMessageCount(1);
+ lateEndpoint.setResultWaitTime(60_000);
+ lateEndpoint.setAssertPeriod(2_000);
+ testBody(lateEndpoint.message(0), assertGoodValue("Foo"));
+ lateRelay.start(lateRelayPort.getPort(), getServerPort());
+ lateEndpoint.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/MonitorItemLifecycleTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemLifecycleTest.java
index f6797ff65d19..a770a93c20d8 100644
---
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
@@ -267,12 +267,19 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
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());
+ // the connection is rebuilt: the new consumer reports its nodes
without communication, unless the first values
+ // of the new connection get there first, and ends with the current
values, not a creation status
+ items12Endpoint.expectedMinimumMessageCount(2);
+ items12Endpoint.allMessages().predicate(exchange -> {
+ final StatusCode status =
exchange.getMessage().getBody(DataValue.class).getStatusCode();
+ return status.isGood() || status.getValue() ==
StatusCodes.Bad_NoCommunication;
+ });
context.getRouteController().startRoute("items12");
items12Endpoint.assertIsSatisfied();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertGoodValue("Foo").accept(lastExchange(items12Endpoint,
NODE_ITEM_1).getMessage().getBody(DataValue.class));
+ assertGoodValue("Bar").accept(lastExchange(items12Endpoint,
NODE_ITEM_2).getMessage().getBody(DataValue.class));
+ });
producer1.sendBody("Baz");
await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
@@ -468,6 +475,16 @@ public class MonitorItemLifecycleTest extends
AbstractMiloServerTest {
return exchanges.get(exchanges.size() - 1);
}
+ private static Exchange lastExchange(final MockEndpoint endpoint, final
String nodeId) {
+ final List<Exchange> exchanges = endpoint.getReceivedExchanges();
+ for (int i = exchanges.size() - 1; i >= 0; i--) {
+ if
(nodeId.equals(exchanges.get(i).getMessage().getHeader(MiloConstants.HEADER_NODE_ID,
String.class))) {
+ return exchanges.get(i);
+ }
+ }
+ throw new AssertionError("No exchange of node " + nodeId);
+ }
+
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/MonitorItemMultiConnectionsTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemMultiConnectionsTest.java
index 152582e3bb00..890c73bab02f 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemMultiConnectionsTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MonitorItemMultiConnectionsTest.java
@@ -103,13 +103,13 @@ public class MonitorItemMultiConnectionsTest extends
AbstractMiloServerTest {
this.test1Endpoint.setExpectedCount(1);
this.test1Endpoint.setSleepForEmptyTest(5_000);
- // item 2
- this.test2Endpoint.setExpectedCount(0);
- this.test2Endpoint.setSleepForEmptyTest(5_000);
-
- // item 3
- this.test3Endpoint.setExpectedCount(0);
- this.test3Endpoint.setSleepForEmptyTest(5_000);
+ // items 2 and 3: their clients cannot connect, so their consumers
report the node without communication
+ this.test2Endpoint.setExpectedCount(1);
+ testBody(this.test2Endpoint.message(0), assertNoCommunication());
+ this.test2Endpoint.setAssertPeriod(5_000);
+ this.test3Endpoint.setExpectedCount(1);
+ testBody(this.test3Endpoint.message(0), assertNoCommunication());
+ this.test3Endpoint.setAssertPeriod(5_000);
// set server value, then start the consumers
this.producer1.sendBody("Foo");
diff --git
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java
index 12352d91f9c6..ff2abef17cbd 100644
---
a/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java
+++
b/components/camel-milo/src/test/java/org/apache/camel/component/milo/ServerCertificateValidationTest.java
@@ -143,10 +143,13 @@ class ServerCertificateValidationTest extends
AbstractMiloServerTest {
this.trustAllEndpoint.setExpectedCount(1);
this.certificateValidatorEndpoint.setExpectedCount(1);
- this.emptyTrustListEndpoint.setExpectedCount(0);
- this.emptyTrustListEndpoint.setSleepForEmptyTest(5_000);
- this.notConfiguredEndpoint.setExpectedCount(0);
- this.notConfiguredEndpoint.setSleepForEmptyTest(5_000);
+ // the clients that cannot connect report their node without
communication, and nothing else
+ this.emptyTrustListEndpoint.setExpectedCount(1);
+ testBody(this.emptyTrustListEndpoint.message(0),
assertNoCommunication());
+ this.emptyTrustListEndpoint.setAssertPeriod(5_000);
+ this.notConfiguredEndpoint.setExpectedCount(1);
+ testBody(this.notConfiguredEndpoint.message(0),
assertNoCommunication());
+ this.notConfiguredEndpoint.setAssertPeriod(5_000);
// the value is written before the consumers start, so it is the first
notification of each client that
// connects (and not the initial Bad value of the item)
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 833976bfad7e..6804586084fc 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
@@ -1930,6 +1930,28 @@ the server does not have, with the status code
`Bad_NodeIdUnknown`, and the othe
affected. Previously an awaited write or read completed without an error and
with a `null` body, and a browse failed
with a `NullPointerException`.
+=== camel-milo - the consumer reports a lost connection and refreshes its
values after it
+
+The OPC UA client consumer delivers `Bad_NoCommunication` with a null value
for every monitored node when
+communication with the server is lost, and the current value of every node
once communication is restored. Before,
+the consumer delivered nothing during an outage, and values that did not
change were not delivered again after it.
+
+A consumer that starts while the server cannot be reached, or whose client
cannot connect at all, for example because
+of wrong credentials or an untrusted server certificate, delivers the status
for its nodes too; the cause is in the
+log. A connection that Camel rebuilds after a failed request delivers it for
every node of the connection. Each loss and
+each recovery delivers one exchange per node.
+
+A route that uses only the value of the `DataValue` body receives a null for
these exchanges, which the type converters
+turn into `0` or `false`; a route that stores values checks the status code of
the body first, for example with
+`${body.statusCode.bad}`. A value can now arrive again after an outage, so a
route that counts values, computes
+differences or detects changes must tolerate repeats. The consumer's route
also runs during an outage: an OPC UA write
+that it makes while the client reconnects waits for the next reconnect attempt
and can take effect after communication
+is restored, and a request that fails rebuilds the connection.
+
+To keep the previous behaviour, filter the exchanges with a bad status out
right after the consumer, for example with
+`filter(simple("${body.statusCode.bad} == false"))`. This also drops the other
bad statuses, such as the refusal of a
+node, and the current values delivered after an outage still arrive.
+
=== camel-protobuf - a bytes field in a Map
When a protobuf message is converted to a `Map` (for example
`getBody(Map.class)` after unmarshalling), the value of
