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 adf38d969ea8 CAMEL-25846: camel-tahu - an Edge Node producer connects
again after its route is restarted (#28319)
adf38d969ea8 is described below
commit adf38d969ea830efe333b9971634f67b84b6023c
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 22:03:44 2026 +0530
CAMEL-25846: camel-tahu - an Edge Node producer connects again after its
route is restarted (#28319)
* CAMEL-25846: camel-tahu - an Edge Node producer connects again after its
route is restarted
TahuEdgeProducer keeps the TahuEdgeClient of each Edge Node in a static
map. When the Edge Node producer stops, it shuts its client down, but the
client stays in the map. A Tahu EdgeClient cannot run again after a
shutdown (EdgeClient.run() returns at once once stayRunning is false, and
only the constructor sets it). The producer created when the route starts
again took the shut down client from the map and "started" it: it never
connected, and every message was dropped with an ERROR log
("Failed to publish message") while the exchange completed. Device
producers kept the old client as well.
The Edge Node producer now removes its client from the map when it stops
(and shuts down the client's executor), and creates a new client when it
starts. A new client gets the metric maps of the Edge Node and of its
Devices again, and Device producers take the current client of their
Edge Node when they publish. A producer keeps its metric map for that
from its start to its stop; producers to the same endpoint share its
map, which stays while one of them runs.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25846: camel-tahu - a stopped Edge Node producer shuts down the
client of the future it removed
Review of #28319: TahuEdgeNodeProducer.doStop shut down the producer's own
client reference,
which is stale when another producer of the same Edge Node was restarted
since this producer
took its client. The future it removed belongs to the new client, so the
new client kept
running, stayed in descriptorClients and was never shut down. doStop now
takes the client with
currentClient() (the client of the Edge Node in the map).
Test:
clientIsShutDownWhenBothProducersOfTheEdgeNodeStopAfterOneWasRestarted (two
routes to
one Edge Node endpoint, one restarted, then both stopped): the second
client publishes its
NDEATH and is removed from the map. Without the change it times out waiting
for the second
NDEATH.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25846: camel-tahu - a Device producer never creates the client of
its Edge Node
Review of #28319: with the client removed from descriptorClients when the
Edge Node producer
stops, a Device producer created while the Edge Node producer is stopped
reached
computeIfAbsent with the Device endpoint, whose getDeviceIdList() throws a
NullPointerException
when the Device URI has no deviceIds (an Edge Node option), so the Device
route failed to start;
with deviceIds set, the Edge Node took a client built from the Device
endpoint's options.
Only an Edge Node producer creates the client now. A Device producer takes
the client of its
Edge Node from the map when it publishes (and adds its metric map when it
starts, as before);
when no Edge Node producer is started the exchange fails with an
IllegalStateException that
names the Device and the Edge Node, instead of publishing on a client that
was shut down.
Test:
deviceProducerCreatedWhileTheEdgeNodeIsStoppedUsesTheClientOfTheEdgeNode
(Device route
without deviceIds added while the Edge Node route is stopped: a send fails
with the
IllegalStateException, and after the Edge Node route starts the Device is
born and its data
published). Without the change the Device route fails to start with the
NullPointerException.
Docs: a sentence in tahu-edge-component.adoc (and its catalog copy) and an
upgrade guide entry.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25846: camel-tahu - an Edge Node producer creates its client only
when it starts
Review of #28319: the constructor of an Edge Node producer still called
createClient,
which clientForStart() in doStart does again (it takes the client from the
map, or creates
it when there is none), so the call in the constructor is redundant.
Removing it also means
that an Edge Node producer that is created and never started no longer
leaves a client and
its executor in the static map.
The camel-tahu unit tests pass unchanged (12, among them the 5 of
TahuEdgeProducerRestartTest).
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/tahu-edge-component.adoc | 4 +
.../src/main/docs/tahu-edge-component.adoc | 4 +
.../camel/component/tahu/TahuEdgeProducer.java | 145 ++++++++++--
.../component/tahu/handlers/TahuEdgeClient.java | 7 +
.../camel/component/tahu/FakeMqttBroker.java | 168 ++++++++++++++
.../tahu/TahuEdgeProducerRestartTest.java | 249 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
7 files changed, 569 insertions(+), 16 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-edge-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-edge-component.adoc
index 08dd2a6a34f1..e4af59a909e0 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-edge-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-edge-component.adoc
@@ -89,6 +89,10 @@ Device Endpoints inherit all MQTT Server connection
information from their assoc
Edge Node Endpoint. Setting Component- or Endpoint-level configuration values
on Device
Components or Endpoints is unnecessary and should be avoided.
+A Device Producer publishes through the client of its Edge Node Producer and
never
+creates one, so sending to a Device Endpoint while no producer of its Edge Node
+Endpoint is started fails the exchange with an `IllegalStateException`.
+
=== Edge Node and Device Endpoint Interaction
Sparkplug Edge Nodes are not required to have a Device hierarchy and physical
diff --git a/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
b/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
index 08dd2a6a34f1..e4af59a909e0 100644
--- a/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
+++ b/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
@@ -89,6 +89,10 @@ Device Endpoints inherit all MQTT Server connection
information from their assoc
Edge Node Endpoint. Setting Component- or Endpoint-level configuration values
on Device
Components or Endpoints is unnecessary and should be avoided.
+A Device Producer publishes through the client of its Edge Node Producer and
never
+creates one, so sending to a Device Endpoint while no producer of its Edge Node
+Endpoint is started fails the exchange with an `IllegalStateException`.
+
=== Edge Node and Device Endpoint Interaction
Sparkplug Edge Nodes are not required to have a Device hierarchy and physical
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeProducer.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeProducer.java
index 4412833f2f77..e5554ad60d1a 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeProducer.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeProducer.java
@@ -35,18 +35,31 @@ import org.eclipse.tahu.message.BdSeqManager;
import org.eclipse.tahu.message.model.DeviceDescriptor;
import org.eclipse.tahu.message.model.EdgeNodeDescriptor;
import org.eclipse.tahu.message.model.SparkplugBPayload;
+import org.eclipse.tahu.message.model.SparkplugBPayloadMap;
import org.eclipse.tahu.model.MqttServerDefinition;
public abstract class TahuEdgeProducer extends DefaultProducer {
protected static final ConcurrentMap<EdgeNodeDescriptor, TahuEdgeClient>
descriptorClients = new ConcurrentHashMap<>();
protected static final ConcurrentMap<EdgeNodeDescriptor, Future<?>>
descriptorFutures = new ConcurrentHashMap<>();
+ // the metric maps of the started Edge Node and Device producers, added
again to a client that replaces a stopped
+ // one (a producer adds its map when it starts and removes it when it
stops)
+ private static final ConcurrentMap<EdgeNodeDescriptor,
SparkplugBPayloadMap> descriptorMetricMaps
+ = new ConcurrentHashMap<>();
+ // the started producers that use the metric map of a descriptor (several
producers to one endpoint share its map),
+ // changed together with descriptorMetricMaps while holding its lock
+ private static final ConcurrentMap<EdgeNodeDescriptor, Integer>
descriptorMetricMapUsers = new ConcurrentHashMap<>();
private final CamelContext camelContext;
- protected final TahuEdgeClient tahuEdgeClient;
+ // the client of the Edge Node: a stopped Edge Node producer shuts its
client down, and the next start creates a
+ // new one, as a Tahu EdgeClient cannot be started again
+ protected volatile TahuEdgeClient tahuEdgeClient;
protected ExecutorService clientExecutorService;
protected final EdgeNodeDescriptor edgeNodeDescriptor;
+ // the Edge Node of the producer, whose client it uses
+ protected final EdgeNodeDescriptor clientDescriptor;
+ private final TahuEdgeEndpoint endpoint;
private TahuEdgeProducer(TahuEdgeEndpoint endpoint, EdgeNodeDescriptor
edgeNodeDescriptor) {
super(endpoint);
@@ -54,8 +67,41 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
camelContext = endpoint.getCamelContext();
this.edgeNodeDescriptor = edgeNodeDescriptor;
+ this.endpoint = endpoint;
- tahuEdgeClient = createClient(endpoint, edgeNodeDescriptor);
+ // A DeviceDescriptor looks up the client of its EdgeNodeDescriptor,
i.e. minus its deviceId.
+ clientDescriptor = edgeNodeDescriptor.isDeviceDescriptor()
+ ? ((DeviceDescriptor)
edgeNodeDescriptor).getEdgeNodeDescriptor() : edgeNodeDescriptor;
+
+ // Only an Edge Node producer creates the client of its Edge Node,
when it starts (clientForStart): the options
+ // that describe the Edge Node (deviceIds, primaryHostId, servers) are
on the Edge Node endpoint, not on a Device
+ // endpoint. A Device producer takes the client of its Edge Node when
it publishes, and adds its metric map to it
+ // when it starts.
+ }
+
+ /**
+ * The client of the Edge Node. When the Edge Node producer was stopped
and started since this producer took its
+ * client, that client is shut down and replaced by a new one (which has
the metric maps of this producer): take the
+ * new one.
+ */
+ protected TahuEdgeClient currentClient() {
+ TahuEdgeClient client = tahuEdgeClient;
+ TahuEdgeClient current = descriptorClients.get(clientDescriptor);
+ if (current != null && current != client) {
+ client = current;
+ tahuEdgeClient = client;
+ }
+ return client;
+ }
+
+ /**
+ * The client to start: the client of the Edge Node in the map, or a new
one when there is none (a stop shut it down
+ * and removed it, or no Edge Node producer started yet).
+ */
+ protected TahuEdgeClient clientForStart() {
+ TahuEdgeClient client = createClient(endpoint, edgeNodeDescriptor);
+ tahuEdgeClient = client;
+ return client;
}
@Override
@@ -83,7 +129,7 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
.ifPresent(seq ->
message.setHeader(TahuConstants.MESSAGE_SEQUENCE_NUMBER, seq));
try {
- tahuEdgeClient.publishData(edgeNodeDescriptor, payload);
+ currentClient().publishData(edgeNodeDescriptor, payload);
} catch (Throwable t) {
exchange.setException(t);
}
@@ -95,15 +141,7 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
private TahuEdgeClient createClient(TahuEdgeEndpoint endpoint,
EdgeNodeDescriptor edgeNodeDescriptor) {
- // Only create clients for Edge Nodes, not Devices. If passed a
- // DeviceDescriptor, create/lookup based on its EdgeNodeDescriptor,
i.e. minus
- // its deviceId.
- EdgeNodeDescriptor clientCreationDescriptor = edgeNodeDescriptor;
- if (edgeNodeDescriptor.isDeviceDescriptor()) {
- clientCreationDescriptor = ((DeviceDescriptor)
edgeNodeDescriptor).getEdgeNodeDescriptor();
- }
-
- TahuEdgeClient tahuEdgeClient =
descriptorClients.computeIfAbsent(clientCreationDescriptor, end -> {
+ TahuEdgeClient tahuEdgeClient =
descriptorClients.computeIfAbsent(clientDescriptor, end -> {
TahuConfiguration configuration = endpoint.getConfiguration();
@@ -131,6 +169,15 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
.clientExecutorService(clientExecutorService)
.build();
+ // a client that replaces a stopped one gets the metric maps of
the Edge Node and of its Devices again
+ descriptorMetricMaps.forEach((descriptor, metricMap) -> {
+ EdgeNodeDescriptor nodeOf = descriptor.isDeviceDescriptor()
+ ? ((DeviceDescriptor)
descriptor).getEdgeNodeDescriptor() : descriptor;
+ if (end.equals(nodeOf)) {
+ client.addDeviceMetricDataPayloadMap(descriptor,
metricMap);
+ }
+ });
+
return client;
});
@@ -142,6 +189,45 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
return tahuEdgeClient;
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ SparkplugBPayloadMap metricMap =
endpoint.getMetricDataTypePayloadMap();
+ if (metricMap != null) {
+ synchronized (descriptorMetricMaps) {
+ if (descriptorMetricMaps.get(edgeNodeDescriptor) == metricMap)
{
+ descriptorMetricMapUsers.merge(edgeNodeDescriptor, 1,
Integer::sum);
+ } else {
+ descriptorMetricMaps.put(edgeNodeDescriptor, metricMap);
+ descriptorMetricMapUsers.put(edgeNodeDescriptor, 1);
+ }
+ }
+ // the client of the Edge Node may have been created again while
this producer was stopped
+ TahuEdgeClient client = descriptorClients.get(clientDescriptor);
+ if (client != null) {
+ client.addDeviceMetricDataPayloadMap(edgeNodeDescriptor,
metricMap);
+ }
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+
+ SparkplugBPayloadMap metricMap =
endpoint.getMetricDataTypePayloadMap();
+ if (metricMap != null) {
+ synchronized (descriptorMetricMaps) {
+ // only when no other started producer (of the same endpoint)
uses the map
+ if (descriptorMetricMaps.get(edgeNodeDescriptor) == metricMap
+ && descriptorMetricMapUsers.merge(edgeNodeDescriptor,
-1, Integer::sum) <= 0) {
+ descriptorMetricMaps.remove(edgeNodeDescriptor, metricMap);
+ descriptorMetricMapUsers.remove(edgeNodeDescriptor);
+ }
+ }
+ }
+ }
+
static final class TahuEdgeNodeProducer extends TahuEdgeProducer {
private TahuEdgeNodeProducer(TahuEdgeEndpoint endpoint,
EdgeNodeDescriptor edgeNodeDescriptor) {
super(endpoint, edgeNodeDescriptor);
@@ -151,21 +237,22 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
protected void doStart() throws Exception {
super.doStart();
- descriptorFutures.computeIfAbsent(edgeNodeDescriptor, end ->
tahuEdgeClient.startup());
+ TahuEdgeClient client = clientForStart();
+ descriptorFutures.computeIfAbsent(edgeNodeDescriptor, end ->
client.startup());
}
@Override
protected void doSuspend() throws Exception {
super.doSuspend();
- tahuEdgeClient.suspend();
+ currentClient().suspend();
}
@Override
protected void doResume() throws Exception {
super.doResume();
- tahuEdgeClient.resume();
+ currentClient().resume();
}
@Override
@@ -174,7 +261,14 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
Future<?> clientFuture =
descriptorFutures.remove(edgeNodeDescriptor);
if (clientFuture != null) {
- tahuEdgeClient.shutdown();
+ // the client of the future just removed: another producer of
this Edge Node may have been restarted
+ // since this producer took its client, so its own reference
can be a client that is already shut down
+ TahuEdgeClient client = currentClient();
+ client.shutdown();
+ // a shut down client cannot connect again: the next start of
an Edge Node producer creates a new
+ // client, else it would take this one from the map and never
connect
+ descriptorClients.remove(edgeNodeDescriptor, client);
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(client.getClientExecutorService());
}
}
}
@@ -183,6 +277,25 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
private TahuEdgeDeviceProducer(TahuEdgeEndpoint endpoint,
DeviceDescriptor deviceDescriptor) {
super(endpoint, deviceDescriptor);
}
+
+ /**
+ * The client that the started Edge Node producer of this Device
created. A Device producer never creates one,
+ * and it does not publish on a client that a stopped Edge Node
producer shut down.
+ *
+ * @throws IllegalStateException when no Edge Node producer of the
Edge Node of this Device is started
+ */
+ @Override
+ protected TahuEdgeClient currentClient() {
+ TahuEdgeClient client = descriptorClients.get(clientDescriptor);
+ if (client == null) {
+ throw new IllegalStateException(
+ "Cannot publish the data of Device " +
edgeNodeDescriptor.getDescriptorString()
+ + ": no producer of its Edge
Node " + clientDescriptor.getDescriptorString()
+ + " is started");
+ }
+ tahuEdgeClient = client;
+ return client;
+ }
}
static final class Builder {
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/TahuEdgeClient.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/TahuEdgeClient.java
index 377526845eea..368ded4e919b 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/TahuEdgeClient.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/TahuEdgeClient.java
@@ -116,6 +116,13 @@ public final class TahuEdgeClient extends EdgeClient {
}
}
+ /**
+ * The executor that runs this client.
+ */
+ public ExecutorService getClientExecutorService() {
+ return clientExecutorService;
+ }
+
public SparkplugBPayloadMap addDeviceMetricDataPayloadMap(
SparkplugDescriptor metricDescriptor, SparkplugBPayloadMap
metricDataTypePayloadMap) {
return
tahuEdgeNodeMetricHandler.addDeviceMetricDataPayloadMap(metricDescriptor,
metricDataTypePayloadMap);
diff --git
a/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/FakeMqttBroker.java
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/FakeMqttBroker.java
new file mode 100644
index 000000000000..f95999082801
--- /dev/null
+++
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/FakeMqttBroker.java
@@ -0,0 +1,168 @@
+/*
+ * 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.tahu;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * A minimal in-process MQTT 3.1.1 broker for tests: it accepts every
connection and subscription, acknowledges every
+ * publish and records the topics published to it. It does not route messages
to subscribers.
+ */
+final class FakeMqttBroker implements AutoCloseable {
+
+ private final ServerSocket serverSocket;
+ private final List<Socket> sockets = new CopyOnWriteArrayList<>();
+ private final List<String> publishedTopics = new CopyOnWriteArrayList<>();
+ private final AtomicInteger connects = new AtomicInteger();
+ private volatile boolean running = true;
+
+ FakeMqttBroker() throws IOException {
+ serverSocket = new ServerSocket(0, 50,
InetAddress.getLoopbackAddress());
+ Thread acceptor = new Thread(this::accept, "fake-mqtt-acceptor");
+ acceptor.setDaemon(true);
+ acceptor.start();
+ }
+
+ int getPort() {
+ return serverSocket.getLocalPort();
+ }
+
+ List<String> getPublishedTopics() {
+ return publishedTopics;
+ }
+
+ int getConnects() {
+ return connects.get();
+ }
+
+ private void accept() {
+ while (running) {
+ try {
+ Socket socket = serverSocket.accept();
+ sockets.add(socket);
+ Thread handler = new Thread(() -> handle(socket),
"fake-mqtt-connection");
+ handler.setDaemon(true);
+ handler.start();
+ } catch (IOException e) {
+ return;
+ }
+ }
+ }
+
+ private void handle(Socket socket) {
+ try (socket) {
+ DataInputStream in = new DataInputStream(socket.getInputStream());
+ OutputStream out = socket.getOutputStream();
+ while (running) {
+ int header = in.read();
+ if (header < 0) {
+ return;
+ }
+ byte[] body = new byte[readRemainingLength(in)];
+ in.readFully(body);
+ int type = header >> 4;
+ switch (type) {
+ case 1 -> {
+ connects.incrementAndGet();
+ write(out, 0x20, 0, 0);
+ }
+ case 3 -> {
+ int qos = (header >> 1) & 3;
+ int topicLength = ((body[0] & 0xff) << 8) | (body[1] &
0xff);
+ publishedTopics.add(new String(body, 2, topicLength,
StandardCharsets.UTF_8));
+ if (qos == 1) {
+ write(out, 0x40, body[2 + topicLength], body[3 +
topicLength]);
+ } else if (qos == 2) {
+ write(out, 0x50, body[2 + topicLength], body[3 +
topicLength]);
+ }
+ }
+ case 6 -> write(out, 0x70, body[0], body[1]);
+ case 8 -> {
+ // SUBACK: the packet id and the requested QoS of
every topic filter
+ ByteArrayOutputStream granted = new
ByteArrayOutputStream();
+ int i = 2;
+ while (i < body.length) {
+ int length = ((body[i] & 0xff) << 8) | (body[i +
1] & 0xff);
+ i += 2 + length;
+ granted.write(body[i++]);
+ }
+ byte[] qos = granted.toByteArray();
+ byte[] suback = new byte[2 + qos.length];
+ suback[0] = body[0];
+ suback[1] = body[1];
+ System.arraycopy(qos, 0, suback, 2, qos.length);
+ writePacket(out, 0x90, suback);
+ }
+ case 10 -> write(out, 0xB0, body[0], body[1]);
+ case 12 -> writePacket(out, 0xD0, new byte[0]);
+ case 14 -> {
+ return;
+ }
+ default -> {
+ // nothing to answer
+ }
+ }
+ }
+ } catch (IOException e) {
+ // the client went away
+ } finally {
+ sockets.remove(socket);
+ }
+ }
+
+ private static int readRemainingLength(DataInputStream in) throws
IOException {
+ int value = 0;
+ int multiplier = 1;
+ int digit;
+ do {
+ digit = in.readUnsignedByte();
+ value += (digit & 127) * multiplier;
+ multiplier *= 128;
+ } while ((digit & 128) != 0);
+ return value;
+ }
+
+ private static void write(OutputStream out, int header, int b1, int b2)
throws IOException {
+ writePacket(out, header, new byte[] { (byte) b1, (byte) b2 });
+ }
+
+ private static synchronized void writePacket(OutputStream out, int header,
byte[] body) throws IOException {
+ out.write(header);
+ out.write(body.length);
+ out.write(body);
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ running = false;
+ serverSocket.close();
+ for (Socket socket : sockets) {
+ socket.close();
+ }
+ }
+}
diff --git
a/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerRestartTest.java
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerRestartTest.java
new file mode 100644
index 000000000000..251a67a7f184
--- /dev/null
+++
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerRestartTest.java
@@ -0,0 +1,249 @@
+/*
+ * 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.tahu;
+
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.tahu.message.model.EdgeNodeDescriptor;
+import org.eclipse.tahu.message.model.Metric.MetricBuilder;
+import org.eclipse.tahu.message.model.MetricDataType;
+import org.eclipse.tahu.message.model.SparkplugBPayload;
+import org.eclipse.tahu.message.model.SparkplugBPayloadMap;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An Edge Node producer must connect and publish again after its route was
stopped and started.
+ */
+class TahuEdgeProducerRestartTest {
+
+ @Test
+ void edgeNodePublishesAgainAfterTheRouteIsRestarted() throws Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ String uri =
"tahu-edge:RestartGroup/RestartNode?deviceIds=D1&servers=Mqtt:tcp://localhost:"
+ broker.getPort();
+ TahuEdgeEndpoint endpoint = context.getEndpoint(uri,
TahuEdgeEndpoint.class);
+ endpoint.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:node").routeId("node").to(endpoint);
+ }
+ });
+ context.start();
+
+ String ndata = "spBv1.0/RestartGroup/NDATA/RestartNode";
+ waitFor(broker, () ->
broker.getPublishedTopics().contains("spBv1.0/RestartGroup/NBIRTH/RestartNode"));
+ ProducerTemplate template = context.createProducerTemplate();
+ template.sendBody("direct:node", payload(1));
+ waitFor(broker, () -> count(broker, ndata) == 1);
+
+ context.getRouteController().stopRoute("node");
+ context.getRouteController().startRoute("node");
+
+ // without the fix the producer of the restarted route took the
client that the stop had shut down: it never
+ // connected again, and the data was dropped (logged) while the
exchange completed
+ waitFor(broker, () -> broker.getPublishedTopics().stream()
+ .filter(t ->
t.equals("spBv1.0/RestartGroup/NBIRTH/RestartNode")).count() == 2);
+ template.sendBody("direct:node", payload(2));
+ waitFor(broker, () -> count(broker, ndata) == 2);
+ assertEquals(2, broker.getConnects());
+ }
+ }
+
+ @Test
+ void deviceOfAnotherRoutePublishesAgainAfterTheEdgeNodeRouteIsRestarted()
throws Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ TahuEdgeEndpoint node = context.getEndpoint(
+
"tahu-edge:DeviceGroup/DeviceNode?deviceIds=D1&servers=Mqtt:tcp://localhost:" +
broker.getPort(),
+ TahuEdgeEndpoint.class);
+ node.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ TahuEdgeEndpoint device =
context.getEndpoint("tahu-edge:DeviceGroup/DeviceNode/D1",
TahuEdgeEndpoint.class);
+ device.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:node").routeId("node").to(node);
+ from("direct:device").routeId("device").to(device);
+ }
+ });
+ context.start();
+
+ String ddata = "spBv1.0/DeviceGroup/DDATA/DeviceNode/D1";
+ waitFor(broker, () ->
broker.getPublishedTopics().contains("spBv1.0/DeviceGroup/DBIRTH/DeviceNode/D1"));
+ ProducerTemplate template = context.createProducerTemplate();
+ template.sendBody("direct:device", payload(1));
+ waitFor(broker, () -> count(broker, ddata) == 1);
+
+ // only the route of the Edge Node is restarted: the producer of
the device keeps running
+ context.getRouteController().stopRoute("node");
+ context.getRouteController().startRoute("node");
+
+ waitFor(broker, () -> count(broker,
"spBv1.0/DeviceGroup/DBIRTH/DeviceNode/D1") == 2);
+ template.sendBody("direct:device", payload(2));
+ waitFor(broker, () -> count(broker, ddata) == 2);
+ }
+ }
+
+ @Test
+ void
deviceStillBornAfterTheEdgeNodeRestartWhenOneOfTwoProducersOfTheDeviceStopped()
throws Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ TahuEdgeEndpoint node = context.getEndpoint(
+
"tahu-edge:SharedGroup/SharedNode?deviceIds=D1&servers=Mqtt:tcp://localhost:" +
broker.getPort(),
+ TahuEdgeEndpoint.class);
+ node.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ TahuEdgeEndpoint device =
context.getEndpoint("tahu-edge:SharedGroup/SharedNode/D1",
TahuEdgeEndpoint.class);
+ device.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:node").routeId("node").to(node);
+ // two producers to the same Device endpoint share its
metric map
+ from("direct:device1").routeId("device1").to(device);
+ from("direct:device2").routeId("device2").to(device);
+ }
+ });
+ context.start();
+
+ String dbirth = "spBv1.0/SharedGroup/DBIRTH/SharedNode/D1";
+ waitFor(broker, () -> count(broker, dbirth) == 1);
+
+ // one of the two Device producers stops, the other keeps running
+ context.getRouteController().stopRoute("device2");
+ context.getRouteController().stopRoute("node");
+ context.getRouteController().startRoute("node");
+
+ // the new client of the Edge Node still has the metric map of the
Device, so it is born again
+ waitFor(broker, () -> count(broker, dbirth) == 2);
+ ProducerTemplate template = context.createProducerTemplate();
+ template.sendBody("direct:device1", payload(1));
+ waitFor(broker, () -> count(broker,
"spBv1.0/SharedGroup/DDATA/SharedNode/D1") == 1);
+ }
+ }
+
+ @Test
+ void
clientIsShutDownWhenBothProducersOfTheEdgeNodeStopAfterOneWasRestarted() throws
Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ TahuEdgeEndpoint node = context.getEndpoint(
+
"tahu-edge:TwoGroup/TwoNode?deviceIds=D1&servers=Mqtt:tcp://localhost:" +
broker.getPort(),
+ TahuEdgeEndpoint.class);
+ node.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // two producers of the same Edge Node
+ from("direct:node1").routeId("node1").to(node);
+ from("direct:node2").routeId("node2").to(node);
+ }
+ });
+ context.start();
+
+ String nbirth = "spBv1.0/TwoGroup/NBIRTH/TwoNode";
+ String ndeath = "spBv1.0/TwoGroup/NDEATH/TwoNode";
+ waitFor(broker, () -> count(broker, nbirth) == 1);
+
+ // the restarted producer shuts down the first client and starts a
second one; the other producer still
+ // refers to the first client
+ context.getRouteController().stopRoute("node1");
+ context.getRouteController().startRoute("node1");
+ waitFor(broker, () -> count(broker, nbirth) == 2);
+
+ context.getRouteController().stopRoute("node2");
+ context.getRouteController().stopRoute("node1");
+
+ // the second client is shut down (its NDEATH is published) and
removed, so that no client of this Edge
+ // Node is left running without a started producer
+ waitFor(broker, () -> count(broker, ndeath) == 2);
+ assertNull(TahuEdgeProducer.descriptorClients.get(new
EdgeNodeDescriptor("TwoGroup", "TwoNode")));
+ }
+ }
+
+ @Test
+ void
deviceProducerCreatedWhileTheEdgeNodeIsStoppedUsesTheClientOfTheEdgeNode()
throws Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ TahuEdgeEndpoint node = context.getEndpoint(
+
"tahu-edge:LateGroup/LateNode?deviceIds=D1&servers=Mqtt:tcp://localhost:" +
broker.getPort(),
+ TahuEdgeEndpoint.class);
+ node.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:node").routeId("node").to(node);
+ }
+ });
+ context.start();
+ waitFor(broker, () -> count(broker,
"spBv1.0/LateGroup/NBIRTH/LateNode") == 1);
+ context.getRouteController().stopRoute("node");
+
+ // the Device URI has none of the Edge Node options (deviceIds,
servers): without the fix its producer
+ // created the client of the Edge Node from them and failed to
start with a NullPointerException
+ TahuEdgeEndpoint device =
context.getEndpoint("tahu-edge:LateGroup/LateNode/D1", TahuEdgeEndpoint.class);
+ device.setMetricDataTypePayloadMap(new
SparkplugBPayloadMap.SparkplugBPayloadMapBuilder()
+ .addMetric(new MetricBuilder("level",
MetricDataType.Int32, 0).createMetric()).createPayload());
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:device").routeId("device").to(device);
+ }
+ });
+
+ // the Edge Node has no started producer, so the Device data
cannot be published
+ ProducerTemplate template = context.createProducerTemplate();
+ Exchange failed = template.send("direct:device", e ->
e.getMessage().setBody(payload(1)));
+ IllegalStateException e =
assertInstanceOf(IllegalStateException.class, failed.getException());
+ assertTrue(e.getMessage().contains("LateGroup/LateNode"),
e.getMessage());
+
+ // the Edge Node producer creates the client from its own
endpoint, with the metric map of the Device
+ context.getRouteController().startRoute("node");
+ waitFor(broker, () -> count(broker,
"spBv1.0/LateGroup/DBIRTH/LateNode/D1") == 1);
+ template.sendBody("direct:device", payload(2));
+ waitFor(broker, () -> count(broker,
"spBv1.0/LateGroup/DDATA/LateNode/D1") == 1);
+ }
+ }
+
+ private static void waitFor(FakeMqttBroker broker, BooleanSupplier
condition) {
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertTrue(condition.getAsBoolean(),
+ () -> "not published; connects: " + broker.getConnects() + ",
published: " + broker.getPublishedTopics()));
+ }
+
+ private static long count(FakeMqttBroker broker, String topic) {
+ return
broker.getPublishedTopics().stream().filter(topic::equals).count();
+ }
+
+ private static SparkplugBPayload payload(int value) throws Exception {
+ return new
SparkplugBPayload.SparkplugBPayloadBuilder().setTimestamp(new Date())
+ .addMetric(new MetricBuilder("level", MetricDataType.Int32,
value).createMetric()).createPayload();
+ }
+}
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 982ce7102bdc..c513494b38f6 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
@@ -2612,3 +2612,11 @@ With `crossOriginFilterOn=true`, the CORS responses now
include `Access-Control-
has an entry with a wildcard such as `https://*` or an entry `null`, the
origin is still allowed, but without credentials. Previously credentials
were allowed for every origin. A cross-origin client that sends cookies or
other credentials must have its origin
listed in `allowedOrigins`.
+
+=== camel-tahu - a Device producer uses the client of its started Edge Node
producer
+
+A Device producer (`tahu-edge:group/edgeNode/device`) no longer creates the
client of its Edge Node: only the Edge
+Node producer creates it, from the options of the Edge Node endpoint
(`deviceIds`, `primaryHostId`, `servers`). A
+Device route that starts before the Edge Node route, or while it is stopped,
therefore no longer fails with a
+`NullPointerException` when the Device URI has no `deviceIds`. Sending to a
Device endpoint while no producer of its
+Edge Node endpoint is started now fails the exchange with an
`IllegalStateException`; before, the data was dropped.
