davsclaus opened a new pull request, #29015: URL: https://github.com/apache/camel/pull/29015
Fixes [CAMEL-26181](https://issues.apache.org/jira/browse/CAMEL-26181). ## 1. tahu-host gave two exchanges for each Sparkplug message **What was wrong.** Tahu's `TahuPayloadHandler` calls the callback of the message type (`onNodeBirthArrived`, `onNodeDataArrived`, `onNodeDeath`, `onDeviceBirthArrived`, `onDeviceDataArrived`, `onDeviceDeath`) and then `onMessage` for every message. `MultiTahuHostApplicationEventHandler` passed the message to the consumer from both. So the route got every message twice. An `NDATA` came three times, because Tahu 1.0.21 calls `onNodeDataArrived` a second time after the metrics (where `onNodeDataComplete` was probably meant). **What changed.** The message is passed to the consumer from `onMessage` only. The type callbacks do nothing now. **Related bug in the same consumer, fixed here.** In `TahuHostConsumer.onMessageConsumer`, the metric headers were set with `setHeaders(...)`. That replaced all headers. So for a message with metrics, `CamelTahuMessageType`, `CamelTahuEdgeNodeDescriptor`, `CamelTahuMessageTimestamp`, `CamelTahuMessageSequenceNumber` and `CamelTahuMessageUUID` were lost. Only a message without metrics (such as a `DDEATH`) kept them. The metrics are now added one by one with `setHeader`. This also removes the `IllegalStateException` that `Collectors.toMap` threw when a payload had the same metric name twice (the last one now wins). **Per-metric exchanges.** They are unchanged: `onBirthMetric`, `onDataMetric` and `onStale` still send one exchange per metric. Such an exchange has `CamelTahuEdgeNodeDescriptor` and its metric header, but no `CamelTahuMessageType`. This is now documented. The YAML example in the doc filters on `CamelTahuMessageType == 'DDATA'`, so it logs each temperature once instead of twice. ## 2. tahu-edge could never set the payload body from the message body **What was wrong.** `TahuEdgeProducer.process` built the payload from the message only for a null body. It did this by setting the message itself as the body, and the converter skips a body that is the message. A text or bytes body went to `getMandatoryBody(SparkplugBPayload.class)`, which has no converter for it, so it failed with an `InvalidPayloadException`. Also, `TahuEdgePayloadConverter` looked up a `byte[]` to `byte[]` converter in the registry, and there is none. So a `byte[]` body was dropped even if it reached the converter. **What changed.** - A `SparkplugBPayload` body, or a body that has a converter to `SparkplugBPayload`, is published as is (as before). - For any other body, the producer converts the message itself to `SparkplugBPayload`. The metrics come from the `CamelTahuMetric.<name>` headers, and the body, converted to bytes, is the payload body. A null body gives a payload without a body (as before). - If the body does not convert to bytes, the exchange fails with an `InvalidPayloadException` and nothing is published. Silently dropping the body would lose data. - The converter takes a `byte[]` body as is. - The producer no longer sets the message as its own body and then resets it. ## Tests - `TahuHostConsumerMessageTest` (new) sends `NBIRTH`, `NDATA`, `DBIRTH`, `DDATA`, `DDEATH` and `NDEATH` to a real `tahu-host` consumer through the in-process `FakeMqttBroker`. It checks that each message gives exactly one exchange, in order, with its headers. Before the fix it got 2 message exchanges (headers lost). With only the header fix applied it got 13 (6 messages: twice each, `NDATA` three times). - `TahuEdgeProducerBodyTest` (new) publishes through the `FakeMqttBroker` and decodes the payload of the `NDATA`. It covers a text body, a `byte[]` body, no body, a `SparkplugBPayload` body, and a POJO body that fails. Before the fix the text and bytes tests failed. - `FakeMqttBroker` now records the published payloads and the subscriptions. It can send a message to the subscribers, and it writes the MQTT remaining length correctly for packets over 127 bytes. - `mvn clean install -Psourcecheck` in `camel-tahu`: 18 unit tests pass. The integration tests with the Sparkplug TCK container pass too (7 run, 2 manual ones skipped). In one earlier local run, `TahuEdgeProducerIT` failed 3 TCK profiles on `payloads-timestamp-in-UTC`, including session establishment, which sends no data. It passed on a re-run with this change, and on main, so it looks flaky. ## Docs and generated files - `tahu-edge-component.adoc`: new "Message Body" section. - `tahu-host-component.adoc`: one exchange per message, the per-metric exchanges, and the example filter. - `@BodyTypes` of `tahu-edge`: `producerIn` now lists text and bytes. It no longer says that text and bytes fail. Regenerated `tahu-edge.json` and copied it and both docs to `camel-catalog`. No yaml-dsl-validator test refers to tahu. - Upgrade guide 4.24: new `camel-tahu` section next to the existing one. ## Not changed - The per-metric exchanges stay. Removing them would be a larger behaviour change (see the open question). - `TahuHostConsumer` keeps its host application in a static map keyed by host id. Its callbacks are bound to the first consumer, and its server definitions come from the first endpoint. A second CamelContext in the same JVM with the same host id reuses the stopped application. This PR leaves that alone; the new test uses a host id of its own. ## Open question Should the per-metric exchanges of `tahu-host` stay? The message exchange already has every metric as a header, so each metric reaches the route twice. The per-metric exchanges are the only exchanges for metrics that turn stale when a node or device dies. My recommendation: keep them for now, since this PR documents them. Later, maybe add an option to turn them off, or send the stale metrics as part of the death message exchange. _Claude Code on behalf of Claus Ibsen (davsclaus)_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
