This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26181-tahu-duplicate-and-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit ae8fd6bb3c70e9671cbccdee858d0fbb5967cf51 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 23:57:31 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 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 10f5a47c1b89..a7432337c2f7 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 @@ -4238,6 +4238,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`,
