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 91c23da3054b CAMEL-26181: camel-tahu - tahu-host gives one exchange
for each Sparkplug message, and tahu-edge publishes a text or bytes body as the
payload body (#29015)
91c23da3054b is described below
commit 91c23da3054b9c4c0ebb2bf6dbca1ffe4bab78f0
Author: Claus Ibsen <[email protected]>
AuthorDate: Tue Nov 3 07:41:56 2026 +0100
CAMEL-26181: camel-tahu - tahu-host gives one exchange for each Sparkplug
message, and tahu-edge publishes a text or bytes body as the payload body
(#29015)
…plug message, and tahu-edge publishes a text or bytes body as the payload
body
Tahu calls onMessage right after the callback of each message type
(onNodeBirthArrived,
onNodeDataArrived, ...), and the event handler gave the message to the
consumer from both,
so the route got each message twice (an NDATA three times, as Tahu calls
onNodeDataArrived
again after its metrics). The message is now given from onMessage only. The
metric headers
also replaced the other headers of a message with metrics (setHeaders);
they are now added.
The tahu-edge producer only built the payload from the message for a null
body, and a byte[]
body has no byte[] to byte[] converter in the registry, so a text or bytes
body never became
the payload body. A body that is not a SparkplugBPayload (and has no
converter to it) is now
the body of a payload made from the message; a body that does not convert
to bytes fails.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../apache/camel/catalog/components/tahu-edge.json | 2 +-
.../camel/catalog/docs/tahu-edge-component.adoc | 8 ++
.../camel/catalog/docs/tahu-host-component.adoc | 15 +-
.../org/apache/camel/component/tahu/tahu-edge.json | 2 +-
.../src/main/docs/tahu-edge-component.adoc | 8 ++
.../src/main/docs/tahu-host-component.adoc | 15 +-
.../camel/component/tahu/TahuEdgeEndpoint.java | 9 +-
.../component/tahu/TahuEdgePayloadConverter.java | 5 +-
.../camel/component/tahu/TahuEdgeProducer.java | 24 ++--
.../camel/component/tahu/TahuHostConsumer.java | 11 +-
.../MultiTahuHostApplicationEventHandler.java | 18 ++-
.../camel/component/tahu/FakeMqttBroker.java | 56 +++++++-
.../component/tahu/TahuEdgeProducerBodyTest.java | 156 +++++++++++++++++++++
.../tahu/TahuHostConsumerMessageTest.java | 107 ++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 ++
15 files changed, 404 insertions(+), 44 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tahu-edge.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tahu-edge.json
index ec3d1734e85b..b8239a6da329 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tahu-edge.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/tahu-edge.json
@@ -46,7 +46,7 @@
"CamelTahuMessageSequenceNumber": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The sequence number of a Sparkplug
message", "constantName":
"org.apache.camel.component.tahu.TahuConstants#MESSAGE_SEQUENCE_NUMBER" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "null" ], "description": "A Tahu
SparkplugBPayload with the metrics to publish, or no body: the metrics are then
made from the CamelTahuMetric.<name> headers. Text, bytes or another body
fails, as it does not convert to a SparkplugBPayload" },
+ "producerIn": { "carriers": [ "pojo", "text", "bytes", "null" ],
"description": "A Tahu SparkplugBPayload with the metrics to publish. For any
other body the metrics are made from the CamelTahuMetric.<name> headers, and
the body, converted to bytes, is the body of the payload (no body gives a
payload without body). A body that does not convert to bytes fails" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
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 e4af59a909e0..7a336f8f9026 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
@@ -111,6 +111,14 @@ using the `metricDataTypePayloadMap` Endpoint property
setter method.
These requirements allow Sparkplug 3.0.0 compliant behavior.
+=== Message Body
+
+A message body that is a Tahu `SparkplugBPayload` is published as it is. For
any other body the producer makes the
+payload from the message: each `CamelTahuMetric.<name>` header is a metric (a
header whose value is a Tahu `Metric` is
+added as it is), and the body, converted to bytes, is the body of the payload.
A text or a byte array body is therefore
+the body of the published payload, and a message without a body gives a
payload without body. A body that does not
+convert to bytes fails the exchange with an `InvalidPayloadException`.
+
=== Example
The following routes publish a temperature for device `D1` of Edge Node
`EdgeNode` in group `Basic`:
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-host-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-host-component.adoc
index d90e901c6a67..e4cfcc9bb781 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-host-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/tahu-host-component.adoc
@@ -35,9 +35,16 @@ include::partial$component-endpoint-headers.adoc[]
== Usage
-A Host Application consumer receives the Sparkplug messages of the Edge Nodes
and Devices. Each metric of a message is
-a header named `CamelTahuMetric.` followed by the name of the metric, and its
value is a Tahu `Metric`.
-The following route logs the temperature of the device in the
xref:tahu-edge-component.adoc[Tahu Edge] example:
+A Host Application consumer receives the Sparkplug messages of the Edge Nodes
and Devices, and gives one exchange for
+each message, with the `CamelTahuMessageType` header. Each metric of a message
is a header named `CamelTahuMetric.`
+followed by the name of the metric, and its value is a Tahu `Metric`.
+
+In addition, the consumer gives an exchange for each metric: each metric of a
birth or data message, and each metric
+that turns stale when its Edge Node or Device dies. Such an exchange has the
`CamelTahuEdgeNodeDescriptor` header and
+the header of its metric, but no `CamelTahuMessageType` header.
+
+The following route logs the temperature of the device in the
xref:tahu-edge-component.adoc[Tahu Edge] example, once
+for each data message:
[source,yaml]
----
@@ -51,7 +58,7 @@ The following route logs the temperature of the device in the
xref:tahu-edge-com
- filter:
expression:
simple:
- expression: "${header[CamelTahuMetric.temperature]} != null"
+ expression: "${header.CamelTahuMessageType} == 'DDATA' &&
${header[CamelTahuMetric.temperature]} != null"
steps:
- log:
message: "Temperature:
${header[CamelTahuMetric.temperature].value}"
diff --git
a/components/camel-tahu/src/generated/resources/META-INF/org/apache/camel/component/tahu/tahu-edge.json
b/components/camel-tahu/src/generated/resources/META-INF/org/apache/camel/component/tahu/tahu-edge.json
index ec3d1734e85b..b8239a6da329 100644
---
a/components/camel-tahu/src/generated/resources/META-INF/org/apache/camel/component/tahu/tahu-edge.json
+++
b/components/camel-tahu/src/generated/resources/META-INF/org/apache/camel/component/tahu/tahu-edge.json
@@ -46,7 +46,7 @@
"CamelTahuMessageSequenceNumber": { "index": 4, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Long", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The sequence number of a Sparkplug
message", "constantName":
"org.apache.camel.component.tahu.TahuConstants#MESSAGE_SEQUENCE_NUMBER" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "pojo", "null" ], "description": "A Tahu
SparkplugBPayload with the metrics to publish, or no body: the metrics are then
made from the CamelTahuMetric.<name> headers. Text, bytes or another body
fails, as it does not convert to a SparkplugBPayload" },
+ "producerIn": { "carriers": [ "pojo", "text", "bytes", "null" ],
"description": "A Tahu SparkplugBPayload with the metrics to publish. For any
other body the metrics are made from the CamelTahuMetric.<name> headers, and
the body, converted to bytes, is the body of the payload (no body gives a
payload without body). A body that does not convert to bytes fails" },
"producerOut": { "carriers": [ "unchanged" ] }
},
"properties": {
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 e4af59a909e0..7a336f8f9026 100644
--- a/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
+++ b/components/camel-tahu/src/main/docs/tahu-edge-component.adoc
@@ -111,6 +111,14 @@ using the `metricDataTypePayloadMap` Endpoint property
setter method.
These requirements allow Sparkplug 3.0.0 compliant behavior.
+=== Message Body
+
+A message body that is a Tahu `SparkplugBPayload` is published as it is. For
any other body the producer makes the
+payload from the message: each `CamelTahuMetric.<name>` header is a metric (a
header whose value is a Tahu `Metric` is
+added as it is), and the body, converted to bytes, is the body of the payload.
A text or a byte array body is therefore
+the body of the published payload, and a message without a body gives a
payload without body. A body that does not
+convert to bytes fails the exchange with an `InvalidPayloadException`.
+
=== Example
The following routes publish a temperature for device `D1` of Edge Node
`EdgeNode` in group `Basic`:
diff --git a/components/camel-tahu/src/main/docs/tahu-host-component.adoc
b/components/camel-tahu/src/main/docs/tahu-host-component.adoc
index d90e901c6a67..e4cfcc9bb781 100644
--- a/components/camel-tahu/src/main/docs/tahu-host-component.adoc
+++ b/components/camel-tahu/src/main/docs/tahu-host-component.adoc
@@ -35,9 +35,16 @@ include::partial$component-endpoint-headers.adoc[]
== Usage
-A Host Application consumer receives the Sparkplug messages of the Edge Nodes
and Devices. Each metric of a message is
-a header named `CamelTahuMetric.` followed by the name of the metric, and its
value is a Tahu `Metric`.
-The following route logs the temperature of the device in the
xref:tahu-edge-component.adoc[Tahu Edge] example:
+A Host Application consumer receives the Sparkplug messages of the Edge Nodes
and Devices, and gives one exchange for
+each message, with the `CamelTahuMessageType` header. Each metric of a message
is a header named `CamelTahuMetric.`
+followed by the name of the metric, and its value is a Tahu `Metric`.
+
+In addition, the consumer gives an exchange for each metric: each metric of a
birth or data message, and each metric
+that turns stale when its Edge Node or Device dies. Such an exchange has the
`CamelTahuEdgeNodeDescriptor` header and
+the header of its metric, but no `CamelTahuMessageType` header.
+
+The following route logs the temperature of the device in the
xref:tahu-edge-component.adoc[Tahu Edge] example, once
+for each data message:
[source,yaml]
----
@@ -51,7 +58,7 @@ The following route logs the temperature of the device in the
xref:tahu-edge-com
- filter:
expression:
simple:
- expression: "${header[CamelTahuMetric.temperature]} != null"
+ expression: "${header.CamelTahuMessageType} == 'DDATA' &&
${header[CamelTahuMetric.temperature]} != null"
steps:
- log:
message: "Temperature:
${header[CamelTahuMetric.temperature].value}"
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeEndpoint.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeEndpoint.java
index e7fa9bacdfef..4edf52fe59d3 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeEndpoint.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgeEndpoint.java
@@ -50,10 +50,11 @@ import org.eclipse.tahu.message.model.SparkplugBPayloadMap;
producerOnly = true,
category = { Category.MESSAGING, Category.IOT,
Category.MONITORING },
headersClass = TahuConstants.class)
-@BodyTypes(producerIn = @BodyType(value = { Carrier.POJO, Carrier.NULL },
- description = "A Tahu SparkplugBPayload with
the metrics to publish, or no body: the"
- + " metrics are then made from
the CamelTahuMetric.<name> headers. Text, bytes"
- + " or another body fails, as
it does not convert to a SparkplugBPayload"),
+@BodyTypes(producerIn = @BodyType(value = { Carrier.POJO, Carrier.TEXT,
Carrier.BYTES, Carrier.NULL },
+ description = "A Tahu SparkplugBPayload with
the metrics to publish. For any other body"
+ + " the metrics are made from
the CamelTahuMetric.<name> headers, and the"
+ + " body, converted to bytes,
is the body of the payload (no body gives a"
+ + " payload without body). A
body that does not convert to bytes fails"),
producerOut = @BodyType(Carrier.UNCHANGED))
public class TahuEdgeEndpoint extends TahuDefaultEndpoint implements
HeaderFilterStrategyAware {
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgePayloadConverter.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgePayloadConverter.java
index 2d5099cb8a0f..d8413ab8f35a 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgePayloadConverter.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuEdgePayloadConverter.java
@@ -58,7 +58,10 @@ public class TahuEdgePayloadConverter implements
HeaderFilterStrategyAware {
dataPayloadBuilder.setTimestamp(payloadDate);
Object body = messageCopy.getBody();
- if (body != null && body != message) {
+ if (body instanceof byte[] bytes) {
+ // the registry has no converter from byte[] to byte[]
+ dataPayloadBuilder.setBody(bytes);
+ } else if (body != null && body != message) {
TypeConverterRegistry registry =
exchange.getContext().getTypeConverterRegistry();
TypeConverter converter = registry.lookup(byte[].class,
body.getClass());
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 e5554ad60d1a..88d568d8e1f8 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
@@ -27,6 +27,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.InvalidPayloadException;
import org.apache.camel.Message;
+import org.apache.camel.TypeConverter;
import org.apache.camel.component.tahu.handlers.CamelBdSeqManager;
import org.apache.camel.component.tahu.handlers.TahuEdgeClient;
import org.apache.camel.support.DefaultProducer;
@@ -105,17 +106,22 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
}
@Override
- public void process(Exchange exchange) throws InvalidPayloadException {
+ public void process(Exchange exchange) throws Exception {
Message message = exchange.getMessage();
- boolean messageHasNullBody = message.getBody() == null;
-
- if (messageHasNullBody) {
- message.setBody(message, Message.class);
+ TypeConverter typeConverter = exchange.getContext().getTypeConverter();
+ Object body = message.getBody();
+ // a SparkplugBPayload body (or a body with a converter to it) is
published as is
+ SparkplugBPayload payload = body != null ?
typeConverter.convertTo(SparkplugBPayload.class, exchange, body) : null;
+ if (payload == null) {
+ // else the payload is made from the message: the metrics from its
headers, and the payload body from its
+ // body converted to bytes
+ payload =
typeConverter.mandatoryConvertTo(SparkplugBPayload.class, exchange, message);
+ if (body != null && payload.getBody() == null) {
+ throw new InvalidPayloadException(exchange, byte[].class);
+ }
}
- SparkplugBPayload payload =
message.getMandatoryBody(SparkplugBPayload.class);
-
String messageType = (edgeNodeDescriptor.isDeviceDescriptor()) ?
"DDATA" : "NDATA";
message.setHeader(TahuConstants.MESSAGE_TYPE, messageType);
@@ -133,10 +139,6 @@ public abstract class TahuEdgeProducer extends
DefaultProducer {
} catch (Throwable t) {
exchange.setException(t);
}
-
- if (messageHasNullBody) {
- message.setBody(null);
- }
}
private TahuEdgeClient createClient(TahuEdgeEndpoint endpoint,
EdgeNodeDescriptor edgeNodeDescriptor) {
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuHostConsumer.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuHostConsumer.java
index 015a7b305338..74c64d041cf0 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuHostConsumer.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/TahuHostConsumer.java
@@ -17,11 +17,9 @@
package org.apache.camel.component.tahu;
import java.util.List;
-import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
@@ -132,12 +130,9 @@ public class TahuHostConsumer extends DefaultConsumer {
camelMessage.setBody(payload.getBody(), byte[].class);
}
- Map<String, Object> payloadMetrics =
payload.getMetrics().stream()
- .map(m -> new Object[] {
TahuConstants.METRIC_HEADER_PREFIX + m.getName(), m })
- .collect(Collectors.toMap(arr -> (String) arr[0], arr
-> arr[1]));
-
- if (!payloadMetrics.isEmpty()) {
- camelMessage.setHeaders(payloadMetrics);
+ // each metric is a header, next to the headers of the message
+ for (Metric metric : payload.getMetrics()) {
+ camelMessage.setHeader(TahuConstants.METRIC_HEADER_PREFIX
+ metric.getName(), metric);
}
getProcessor().process(exchange);
diff --git
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/MultiTahuHostApplicationEventHandler.java
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/MultiTahuHostApplicationEventHandler.java
index d991e10bb64a..83f1cbc4e9b4 100644
---
a/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/MultiTahuHostApplicationEventHandler.java
+++
b/components/camel-tahu/src/main/java/org/apache/camel/component/tahu/handlers/MultiTahuHostApplicationEventHandler.java
@@ -39,7 +39,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onNodeBirthArrived(MqttServerName serverName,
EdgeNodeDescriptor edgeNodeDescriptor, Message message) {
- onMessageConsumer.accept(edgeNodeDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -48,7 +48,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onNodeDataArrived(MqttServerName serverName,
EdgeNodeDescriptor edgeNodeDescriptor, Message message) {
- onMessageConsumer.accept(edgeNodeDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -57,7 +57,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onNodeDeath(MqttServerName serverName, EdgeNodeDescriptor
edgeNodeDescriptor, Message message) {
- onMessageConsumer.accept(edgeNodeDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -66,7 +66,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onDeviceBirthArrived(MqttServerName serverName,
DeviceDescriptor deviceDescriptor, Message message) {
- onMessageConsumer.accept(deviceDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -75,7 +75,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onDeviceDataArrived(MqttServerName serverName,
DeviceDescriptor deviceDescriptor, Message message) {
- onMessageConsumer.accept(deviceDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -84,7 +84,7 @@ public class MultiTahuHostApplicationEventHandler implements
MultiHostApplicatio
@Override
public void onDeviceDeath(MqttServerName serverName, DeviceDescriptor
deviceDescriptor, Message message) {
- onMessageConsumer.accept(deviceDescriptor, message);
+ // the message is given to the consumer by onMessage
}
@Override
@@ -106,6 +106,12 @@ public class MultiTahuHostApplicationEventHandler
implements MultiHostApplicatio
acceptMetric(sparkplugDescriptor, metric);
}
+ /**
+ * Tahu calls onMessage for every NBIRTH, NDATA, NDEATH, DBIRTH, DDATA and
DDEATH message, right after the callback
+ * of its type (onNodeBirthArrived, onNodeDataArrived, ...): the message
is given to the consumer here only, else
+ * the route would get the same message twice (or three times for an
NDATA, as Tahu calls onNodeDataArrived again
+ * after its metrics).
+ */
@Override
public void onMessage(MqttServerName serverName, SparkplugDescriptor
sparkplugDescriptor, Message message) {
if (sparkplugDescriptor.isDeviceDescriptor()) {
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
index f95999082801..74a521f2e144 100644
---
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
@@ -24,19 +24,24 @@ import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
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.
+ * publish and records the topics and payloads published to it. It does not
route messages to subscribers: a test sends
+ * a message to the subscribed clients with {@link #publish(String, byte[])}.
*/
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 List<Published> published = new CopyOnWriteArrayList<>();
+ private final List<String> subscriptions = new CopyOnWriteArrayList<>();
+ private final List<OutputStream> subscribers = new
CopyOnWriteArrayList<>();
private final AtomicInteger connects = new AtomicInteger();
private volatile boolean running = true;
@@ -55,10 +60,36 @@ final class FakeMqttBroker implements AutoCloseable {
return publishedTopics;
}
+ List<Published> getPublished() {
+ return published;
+ }
+
+ List<String> getSubscriptions() {
+ return subscriptions;
+ }
+
int getConnects() {
return connects.get();
}
+ /**
+ * Sends a message with QoS 0 to every client that subscribed to a topic.
+ */
+ void publish(String topic, byte[] payload) throws IOException {
+ byte[] topicBytes = topic.getBytes(StandardCharsets.UTF_8);
+ byte[] body = new byte[2 + topicBytes.length + payload.length];
+ body[0] = (byte) (topicBytes.length >> 8);
+ body[1] = (byte) topicBytes.length;
+ System.arraycopy(topicBytes, 0, body, 2, topicBytes.length);
+ System.arraycopy(payload, 0, body, 2 + topicBytes.length,
payload.length);
+ for (OutputStream out : subscribers) {
+ writePacket(out, 0x30, body);
+ }
+ }
+
+ record Published(String topic, byte[] payload) {
+ }
+
private void accept() {
while (running) {
try {
@@ -74,9 +105,10 @@ final class FakeMqttBroker implements AutoCloseable {
}
private void handle(Socket socket) {
+ OutputStream out = null;
try (socket) {
DataInputStream in = new DataInputStream(socket.getInputStream());
- OutputStream out = socket.getOutputStream();
+ out = socket.getOutputStream();
while (running) {
int header = in.read();
if (header < 0) {
@@ -93,7 +125,10 @@ final class FakeMqttBroker implements AutoCloseable {
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));
+ String topic = new String(body, 2, topicLength,
StandardCharsets.UTF_8);
+ publishedTopics.add(topic);
+ int payloadStart = 2 + topicLength + (qos > 0 ? 2 : 0);
+ published.add(new Published(topic,
Arrays.copyOfRange(body, payloadStart, body.length)));
if (qos == 1) {
write(out, 0x40, body[2 + topicLength], body[3 +
topicLength]);
} else if (qos == 2) {
@@ -107,6 +142,7 @@ final class FakeMqttBroker implements AutoCloseable {
int i = 2;
while (i < body.length) {
int length = ((body[i] & 0xff) << 8) | (body[i +
1] & 0xff);
+ subscriptions.add(new String(body, i + 2, length,
StandardCharsets.UTF_8));
i += 2 + length;
granted.write(body[i++]);
}
@@ -116,6 +152,9 @@ final class FakeMqttBroker implements AutoCloseable {
suback[1] = body[1];
System.arraycopy(qos, 0, suback, 2, qos.length);
writePacket(out, 0x90, suback);
+ if (!subscribers.contains(out)) {
+ subscribers.add(out);
+ }
}
case 10 -> write(out, 0xB0, body[0], body[1]);
case 12 -> writePacket(out, 0xD0, new byte[0]);
@@ -131,6 +170,9 @@ final class FakeMqttBroker implements AutoCloseable {
// the client went away
} finally {
sockets.remove(socket);
+ if (out != null) {
+ subscribers.remove(out);
+ }
}
}
@@ -152,7 +194,13 @@ final class FakeMqttBroker implements AutoCloseable {
private static synchronized void writePacket(OutputStream out, int header,
byte[] body) throws IOException {
out.write(header);
- out.write(body.length);
+ // the remaining length: 7 bits per byte, the high bit set when
another byte follows
+ int length = body.length;
+ do {
+ int digit = length % 128;
+ length /= 128;
+ out.write(length > 0 ? digit | 128 : digit);
+ } while (length > 0);
out.write(body);
out.flush();
}
diff --git
a/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerBodyTest.java
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerBodyTest.java
new file mode 100644
index 000000000000..e580666b7b3a
--- /dev/null
+++
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuEdgeProducerBodyTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.tahu.message.SparkplugBPayloadDecoder;
+import org.eclipse.tahu.message.model.Metric;
+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.eclipse.tahu.model.MetricDataTypeMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * The message body of an Edge Node producer: a SparkplugBPayload is published
as is, any other body is the body of a
+ * payload whose metrics are the metric headers.
+ */
+class TahuEdgeProducerBodyTest {
+
+ private static final String NDATA = "spBv1.0/BodyGroup/NDATA/BodyNode";
+
+ private FakeMqttBroker broker;
+ private CamelContext context;
+ private ProducerTemplate template;
+
+ @BeforeEach
+ void startEdgeNode() throws Exception {
+ broker = new FakeMqttBroker();
+ context = new DefaultCamelContext();
+ TahuEdgeEndpoint endpoint = context.getEndpoint(
+
"tahu-edge:BodyGroup/BodyNode?deviceIds=D1&servers=Mqtt:tcp://localhost:" +
broker.getPort(),
+ 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").to(endpoint);
+ }
+ });
+ context.start();
+ template = context.createProducerTemplate();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() ->
broker.getPublishedTopics().contains("spBv1.0/BodyGroup/NBIRTH/BodyNode"));
+ }
+
+ @AfterEach
+ void stopEdgeNode() throws Exception {
+ context.close();
+ broker.close();
+ }
+
+ @Test
+ void textBodyIsThePayloadBody() throws Exception {
+ template.sendBodyAndHeader("direct:node", "Hello Sparkplug",
TahuConstants.METRIC_HEADER_PREFIX + "level", 7);
+
+ // without the fix a text body failed with an InvalidPayloadException,
as it does not convert to a
+ // SparkplugBPayload
+ SparkplugBPayload payload = publishedData();
+ assertArrayEquals("Hello Sparkplug".getBytes(StandardCharsets.UTF_8),
payload.getBody());
+ assertLevel(payload, 7);
+ }
+
+ @Test
+ void bytesBodyIsThePayloadBody() throws Exception {
+ byte[] body = { 1, 2, 3 };
+ template.sendBodyAndHeader("direct:node", body,
TahuConstants.METRIC_HEADER_PREFIX + "level", 8);
+
+ SparkplugBPayload payload = publishedData();
+ assertArrayEquals(body, payload.getBody());
+ assertLevel(payload, 8);
+ }
+
+ @Test
+ void noBodyGivesAPayloadWithoutBody() throws Exception {
+ template.sendBodyAndHeader("direct:node", null,
TahuConstants.METRIC_HEADER_PREFIX + "level", 9);
+
+ SparkplugBPayload payload = publishedData();
+ assertNull(payload.getBody());
+ assertLevel(payload, 9);
+ }
+
+ @Test
+ void sparkplugPayloadBodyIsPublishedAsIs() throws Exception {
+ SparkplugBPayload body = new
SparkplugBPayload.SparkplugBPayloadBuilder().setTimestamp(new Date())
+ .setBody(new byte[] { 4, 5 })
+ .addMetric(new MetricBuilder("level", MetricDataType.Int32,
10).createMetric()).createPayload();
+ // the header is not used, as the body is the payload
+ template.sendBodyAndHeader("direct:node", body,
TahuConstants.METRIC_HEADER_PREFIX + "level", 11);
+
+ SparkplugBPayload payload = publishedData();
+ assertArrayEquals(new byte[] { 4, 5 }, payload.getBody());
+ assertLevel(payload, 10);
+ }
+
+ @Test
+ void bodyThatDoesNotConvertToBytesFails() {
+ Exchange exchange = template.send("direct:node", e ->
e.getMessage().setBody(new Reading(12)));
+
+ assertInstanceOf(InvalidPayloadException.class,
exchange.getException());
+ assertFalse(broker.getPublishedTopics().contains(NDATA), () ->
"published: " + broker.getPublishedTopics());
+ }
+
+ private SparkplugBPayload publishedData() throws Exception {
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> broker.getPublished().stream().anyMatch(p ->
NDATA.equals(p.topic())));
+ byte[] bytes = broker.getPublished().stream().filter(p ->
NDATA.equals(p.topic())).findFirst().orElseThrow()
+ .payload();
+ // the data types of the metrics are only in the birth message
+ MetricDataTypeMap dataTypes = new MetricDataTypeMap();
+ dataTypes.addMetricDataType("level", MetricDataType.Int32);
+ return new SparkplugBPayloadDecoder().buildFromByteArray(bytes,
dataTypes);
+ }
+
+ private static void assertLevel(SparkplugBPayload payload, int level) {
+ Metric metric = payload.getMetrics().stream().filter(m ->
"level".equals(m.getName())).findFirst()
+ .orElseThrow(() -> new AssertionError("no level metric: " +
payload.getMetrics()));
+ assertEquals(level, ((Number) metric.getValue()).intValue());
+ }
+
+ record Reading(int value) {
+ }
+}
diff --git
a/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuHostConsumerMessageTest.java
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuHostConsumerMessageTest.java
new file mode 100644
index 000000000000..474ce4cf8aad
--- /dev/null
+++
b/components/camel-tahu/src/test/java/org/apache/camel/component/tahu/TahuHostConsumerMessageTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.tahu.message.SparkplugBPayloadEncoder;
+import org.eclipse.tahu.message.model.Metric;
+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.SparkplugMeta;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * A Host Application consumer gives one exchange for each Sparkplug message.
+ */
+class TahuHostConsumerMessageTest {
+
+ private static final String NODE = "spBv1.0/MessageGroup/%s/MessageNode";
+ private static final String DEVICE =
"spBv1.0/MessageGroup/%s/MessageNode/D1";
+
+ @Test
+ void oneExchangeForEachMessage() throws Exception {
+ try (FakeMqttBroker broker = new FakeMqttBroker(); CamelContext
context = new DefaultCamelContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a host id of its own, as a consumer keeps its host
application in a static map
+ from("tahu-host:MessageHost" + broker.getPort()
+ +
"?clientId=MessageHostClient&servers=Fake:tcp://localhost:" + broker.getPort())
+ // the exchanges of the messages, not those of
each metric
+
.filter(header(TahuConstants.MESSAGE_TYPE).isNotNull())
+ .to("mock:messages");
+ }
+ });
+ context.start();
+ MockEndpoint messages = context.getEndpoint("mock:messages",
MockEndpoint.class);
+ messages.expectedMessageCount(6);
+ // wait a while after the last message for duplicates
+ messages.setAssertPeriod(1000);
+
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
broker.getSubscriptions().contains("spBv1.0/#"));
+ broker.publish(NODE.formatted("NBIRTH"), payload(0,
metric(SparkplugMeta.SPARKPLUG_BD_SEQUENCE_NUMBER_KEY,
+ MetricDataType.Int64, 0L), metric("temperature",
MetricDataType.Double, 20.5)));
+ broker.publish(NODE.formatted("NDATA"), payload(1,
metric("temperature", MetricDataType.Double, 21.5)));
+ broker.publish(DEVICE.formatted("DBIRTH"), payload(2,
metric("level", MetricDataType.Int32, 1)));
+ broker.publish(DEVICE.formatted("DDATA"), payload(3,
metric("level", MetricDataType.Int32, 2)));
+ broker.publish(DEVICE.formatted("DDEATH"), payload(4));
+ broker.publish(NODE.formatted("NDEATH"), payload(null,
metric(SparkplugMeta.SPARKPLUG_BD_SEQUENCE_NUMBER_KEY,
+ MetricDataType.Int64, 0L)));
+
+ // without the fix each message came twice (an NDATA three times):
from the callback of its type
+ // (onNodeBirthArrived, ...) and from onMessage
+ MockEndpoint.assertIsSatisfied(context, 20, TimeUnit.SECONDS);
+ assertEquals(List.of("NBIRTH", "NDATA", "DBIRTH", "DDATA",
"DDEATH", "NDEATH"),
+ messages.getReceivedExchanges().stream()
+ .map(e ->
e.getMessage().getHeader(TahuConstants.MESSAGE_TYPE, String.class)).toList());
+ // without the fix the metric headers replaced the other headers
of a message that has metrics
+ Exchange ddata = messages.getReceivedExchanges().get(3);
+ assertEquals("MessageGroup/MessageNode/D1",
ddata.getMessage().getHeader(TahuConstants.EDGE_NODE_DESCRIPTOR));
+ assertEquals(3L,
ddata.getMessage().getHeader(TahuConstants.MESSAGE_SEQUENCE_NUMBER));
+ assertEquals(2,
ddata.getMessage().getHeader(TahuConstants.METRIC_HEADER_PREFIX + "level",
Metric.class)
+ .getValue());
+ }
+ }
+
+ private static Metric metric(String name, MetricDataType type, Object
value) throws Exception {
+ return new MetricBuilder(name, type, value).createMetric();
+ }
+
+ private static byte[] payload(Integer seq, Metric... metrics) throws
Exception {
+ SparkplugBPayload.SparkplugBPayloadBuilder builder
+ = new
SparkplugBPayload.SparkplugBPayloadBuilder().setTimestamp(new Date());
+ if (seq != null) {
+ builder.setSeq(seq.longValue());
+ }
+ for (Metric metric : metrics) {
+ builder.addMetric(metric);
+ }
+ return new
SparkplugBPayloadEncoder().getBytes(builder.createPayload(), false);
+ }
+}
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 e004aadc1355..6ed514d6f90f 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
@@ -4288,6 +4288,18 @@ Device route that starts before the Edge Node route, or
while it is stopped, the
`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.
+=== camel-tahu - one exchange for each Sparkplug message, and the payload body
of the Edge Node producer
+
+The `tahu-host` consumer gave two exchanges for each Sparkplug message (three
for an `NDATA`). It now gives one. The
+exchange of a message with metrics now keeps the `CamelTahuMessageType`,
`CamelTahuEdgeNodeDescriptor`,
+`CamelTahuMessageTimestamp`, `CamelTahuMessageSequenceNumber` and
`CamelTahuMessageUUID` headers next to the
+`CamelTahuMetric.<name>` headers; before, the metric headers replaced them.
The exchange of each metric is unchanged:
+it has no `CamelTahuMessageType` header.
+
+The `tahu-edge` producer publishes a message body that is not a
`SparkplugBPayload`, such as a text or a byte array, as
+the body of the payload, whose metrics are made from the
`CamelTahuMetric.<name>` headers. Before, such a body failed
+with an `InvalidPayloadException`. A body that does not convert to a byte
array still fails.
+
=== camel-aws - producers without an operation, and headers of the wrong type
The producers of `aws2-ec2`, `aws2-ecs`, `aws2-eks`, `aws2-mq`, `aws2-msk`,
`aws2-rekognition`, `aws2-step-functions`,
