# Cameleer at the edge > Cameleer runs integration routes on gateways and small devices: a single native binary of > about 10 MB of memory that ingests MQTT, syslog and HL7 locally, filters and aggregates on > site, buffers when the uplink is down, and forwards to Kafka or HTTP in the cloud. Edge gateways sit between machines and the cloud. They need to speak local protocols, keep working offline, and fit next to other software on small hardware. Apache Camel does all of this on a JVM; Cameleer runs the same routes without one. This cookbook shows five patterns, each a complete route file in [`examples/edge/`](../../examples/edge/) that runs end to end in the test suite (`crates/cameleer-main/tests/docs_guides.rs`) against the embedded MQTT and Kafka brokers and HTTP mocks. | pattern | file | what it shows | |---|---|---| | [Sensor rollups](#filter-and-aggregate-mqtt-sensor-readings) | `mqtt-sensor-rollup.camel.yaml` | MQTT in, validation, per-sensor aggregation, Kafka out | | [Store and forward](#store-and-forward-when-the-uplink-is-down) | `store-and-forward.camel.yaml` | retries, a local file buffer and replay when the uplink is down | | [Syslog to Kafka](#syslog-to-kafka) | `syslog-to-kafka.camel.yaml` | protocol bridging with severity filtering | | [HL7 over MLLP](#hl7-v2-over-mllp-to-an-http-api) | `hl7-mllp-to-http.camel.yaml` | healthcare messages with acknowledgements after delivery | | [Gateway presence](#gateway-presence-with-retained-messages-and-a-last-will) | `gateway-heartbeat.camel.yaml` | retained status and MQTT last will | ## Filter and aggregate MQTT sensor readings Sensors publish a reading every few seconds; the cloud needs a summary. The route subscribes to `sensors/+/temperature`, takes the sensor id from the topic, drops readings that are not numbers or are outside the sensor's range, and aggregates five readings per sensor (or what arrived within a minute) into one rollup with count, min, max and average. Only rollups go to Kafka, keyed by sensor, so a sensor's rollups stay in order on one partition. ```yaml - route: id: sensor-rollup from: uri: "paho:sensors/+/temperature?brokerUrl={{env:MQTT_BROKER:tcp://localhost:1883}}&qos=1&clientId={{env:GATEWAY_ID:gateway-1}}-rollup" steps: - convertBodyTo: type: java.lang.String - setHeader: name: sensor simple: "${header.CamelMqttTopic.split('/')[1]}" - choice: when: - simple: "${body} !regex '-?[0-9]+(\\.[0-9]+)?'" steps: - log: "dropped malformed reading '${body}' from ${header.sensor}" - stop: {} - convertBodyTo: type: java.lang.Double - choice: when: - simple: "${body} < -40 || ${body} > 125" steps: - log: "dropped out-of-range reading ${body} from ${header.sensor}" - stop: {} - aggregate: correlationExpression: header: sensor aggregationStrategy: "#class:org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy" completionSize: 5 completionInterval: 60000 steps: - setBody: jq: '{gateway: "{{env:GATEWAY_ID:gateway-1}}", sensor: header("sensor"), count: length, min: min, max: max, avg: (add / length)}' - marshal: json: {} - setHeader: name: kafka.KEY simple: "${header.sensor}" - to: "kafka:sensor-rollups?brokers={{env:KAFKA_BROKERS:localhost:9092}}" ``` The test publishes `20`, `garbage`, `21.5`, `999`, `22`, `23.5` and `18` for sensor `t1` and one reading for `t2`. Kafka receives one rollup for `t1` with count 5, min 18, max 23.5 and average 21; `garbage` and `999` are dropped, and `t2` waits for its minute. MQTT payloads arrive as bytes, so the route converts them to text first. The `jq` step computes the statistics on the list the `GroupedBodyAggregationStrategy` collected. To keep a rollup per minute regardless of the number of readings, drop `completionSize` and keep `completionInterval`. Aggregation state is in memory: rollups in progress are lost on a restart, which is usually fine for telemetry; use `forceCompletionOnStop: true` to flush them on a clean shutdown. ## Store and forward when the uplink is down Gateways lose their uplink: cellular dead zones, maintenance windows, cloud outages. This route keeps every message: 1. Telemetry from MQTT goes into a `seda` queue, so the MQTT consumer is never blocked by a slow uplink. 2. The uplink route posts each message to the cloud. A dead letter channel retries three times with exponential back-off, then writes the original message to a buffer folder. 3. A replay route polls the buffer folder every 30 seconds, posts each file, and moves it to `.sent` when the cloud accepts it. While the uplink is still down, the post fails and the file stays for the next poll. ```yaml - route: id: telemetry-in from: uri: "paho:telemetry/#?brokerUrl={{env:MQTT_BROKER:tcp://localhost:1883}}&qos=1&clientId={{env:GATEWAY_ID:gateway-1}}-uplink" steps: - convertBodyTo: type: java.lang.String - to: "seda:uplink?size=10000&blockWhenFull=true" - route: id: uplink errorHandler: deadLetterChannel: deadLetterUri: "file:{{env:BUFFER_DIR:buffer}}?fileName=${date:now:yyyyMMddHHmmssSSS}-${exchangeId}.msg" useOriginalMessage: true redeliveryPolicy: maximumRedeliveries: 3 redeliveryDelay: "{{env:RETRY_DELAY_MS:1000}}" useExponentialBackOff: true from: uri: "seda:uplink?size=10000" steps: - setHeader: name: X-Gateway constant: "{{env:GATEWAY_ID:gateway-1}}" - enrich: uri: "{{env:UPLINK_URL:http://localhost:9000}}/telemetry?httpMethod=POST" - route: id: replay-buffer from: uri: "file:{{env:BUFFER_DIR:buffer}}?include=.*\\.msg&sortBy=file:name&move=.sent&delay={{env:REPLAY_DELAY_MS:30000}}" steps: - convertBodyTo: type: java.lang.String - setHeader: name: X-Gateway constant: "{{env:GATEWAY_ID:gateway-1}}" - setHeader: name: X-Replayed constant: "true" - enrich: uri: "{{env:UPLINK_URL:http://localhost:9000}}/telemetry?httpMethod=POST" ``` The test starts with the cloud mock answering `503`: the message is tried four times and lands in the buffer. When the mock recovers, the replay posts it (with `X-Replayed: true`) and moves the file to `.sent`, and new messages go straight through. Why `enrich` and not `to` for the uplink call: `enrich` sends a copy, so every retry posts the original message. With `to`, Cameleer's `http` producer puts the error response into the message before it fails, and a retry would post that instead (Apache Camel leaves the message unchanged). Things to tune for production: - Mount the buffer on persistent storage (`-v cameleer-buffer:/buffer`); the container root can stay read-only. - `seda` is in memory: messages queued there when the process stops are lost. For zero loss across restarts, write to the buffer folder first and let the replay route be the only sender, at the cost of disk writes for every message. - Order is kept per buffer folder (`sortBy=file:name`), not between live and replayed messages. Put a timestamp in the payload if the cloud must reorder. ## Protocol bridging Bridges translate between what the site speaks and what the cloud accepts. The two routes above already bridge MQTT to Kafka and MQTT to HTTP. Two more common ones follow. ### Syslog to Kafka Routers, switches, PLCs and Linux hosts all speak syslog. The `syslog` component receives RFC 5424 and RFC 3164 messages over UDP or TCP (and TLS), parses them into a map, and sets headers such as `CamelSyslogSeverity`, `CamelSyslogFacility` and `CamelSyslogHostname`. This route keeps warnings and worse and sends them to Kafka as JSON. ```yaml - route: id: syslog-alerts from: uri: "syslog:udp://{{env:SYSLOG_HOST:0.0.0.0}}:{{env:SYSLOG_PORT:5514}}" steps: - filter: simple: "${header.CamelSyslogSeverity} in 'EMERG,ALERT,CRIT,ERR,WARNING'" steps: - setHeader: name: kafka.KEY simple: "${header.CamelSyslogHostname}" - marshal: json: {} - to: "kafka:syslog-alerts?brokers={{env:KAFKA_BROKERS:localhost:9092}}" ``` The test sends a `local4.notice` message (dropped) and an `auth.crit` message (forwarded to the `syslog-alerts` topic, keyed by `router-1`). ### HL7 v2 over MLLP to an HTTP API Hospital systems exchange HL7 v2 over MLLP. The `mllp` component frames and acknowledges messages and sets `CamelMllp*` headers (message type, event, control id, sending application); the `hl7terser` language reads fields by path (`PID-3-1` is the patient id). The acknowledgement goes back after the route finishes, so the sender gets `AA` only once the EHR accepted the message, and `AE` when it still failed after five retries, so it keeps the message and resends. ```yaml - route: id: hl7-to-ehr errorHandler: defaultErrorHandler: redeliveryPolicy: maximumRedeliveries: 5 redeliveryDelay: "{{env:RETRY_DELAY_MS:2000}}" from: uri: "mllp://{{env:MLLP_HOST:0.0.0.0}}:{{env:MLLP_PORT:2575}}" steps: - filter: simple: "${header.CamelMllpEventType} == 'ADT'" steps: - setHeader: name: X-Patient-Id hl7terser: "PID-3-1" - setHeader: name: X-Message-Control-Id simple: "${header.CamelMllpMessageControlId}" - setHeader: name: Content-Type constant: application/hl7-v2 - enrich: uri: "{{env:EHR_URL:http://localhost:9000}}/hl7/v2?httpMethod=POST" ``` The test sends an ADT message (posted, `AA`), an ORU message (acknowledged, not posted) and an ADT message the EHR rejects (six posts with the original message, then `AE`). For TLS between the systems, add `sslContextParameters=#ssl` to the `mllp` URI. ## Gateway presence with retained messages and a last will Dashboards need to know which gateways are alive without polling them. The heartbeat route publishes a retained `online` status every 30 seconds. Its MQTT connection registers a last will of `offline`, retained too: if the gateway loses power or network, the broker publishes it. Anyone subscribing to `gateways/+/status` gets the current state of every gateway at once. ```yaml - route: id: gateway-heartbeat from: uri: "timer:heartbeat?period={{env:HEARTBEAT_MS:30000}}" steps: - setBody: constant: online - to: "paho:gateways/{{env:GATEWAY_ID:gateway-1}}/status?brokerUrl={{env:MQTT_BROKER:tcp://localhost:1883}}&clientId={{env:GATEWAY_ID:gateway-1}}-status&qos=1&retained=true&willTopic=gateways/{{env:GATEWAY_ID:gateway-1}}/status&willPayload=offline&willQos=1&willRetained=true" ``` The test checks that the broker holds the retained `online` status and that the gateway is connected under its own client id. The broker sends the will only on an abnormal disconnect; a clean stop of the route does not mark the gateway offline, so publish `offline` from an `onCompletion` or a shutdown route if you need that too. ## Small footprint deployment The Docker image `pmsudhi/cameleer` is one native binary on a distroless base, for `linux/amd64` and `linux/arm64`, about 20 MB to download (12.5 MB built with `FEATURES=core`). A running route uses about 10 MB of memory (8 MB at startup in the [benchmarks](../BENCHMARKS.md)) and starts its routes in milliseconds. Nothing is written to the filesystem unless a route writes files, so the container runs with a read-only root: ```sh docker run -d --name edge-routes --restart unless-stopped \ --read-only --memory 64m --cpus 0.5 \ --network host \ -e MQTT_BROKER=tcp://localhost:1883 -e KAFKA_BROKERS -e GATEWAY_ID=plant-7 \ -v "$PWD/routes:/routes:ro" \ pmsudhi/cameleer run /routes/mqtt-sensor-rollup.camel.yaml ``` - **arm64 boards:** the same image runs on 64-bit Raspberry Pi OS (Pi 3, 4 and 5), NVIDIA Jetson, AWS Graviton and other arm64 Linux devices; Docker picks the right platform. There is no published 32-bit ARM image. - **No container runtime:** `cargo install cameleer-main` builds the `cameleer` binary on the device (or cross-compile it: Cameleer has no dependencies outside the Rust standard library), then run it under systemd, for example `cameleer run /etc/cameleer/rollup.camel.yaml`. - **Several route files:** pass several files to `run` (or none, to load every route file in the working directory); they share one process. - **Health:** add `--prop camel.server.enabled=true --prop camel.server.healthCheckEnabled=true` for `/q/health/live` and `/q/health/ready` on port 8080. See the [deployment guide](../DEPLOYMENT.md). ## Offline-first patterns - **Decide locally.** Filters, thresholds and aggregation run on the gateway, so alarms fire without the cloud. A `choice` that writes to a local actuator topic (`paho:...`) keeps working when the uplink is gone. - **Buffer, then forward.** A dead letter channel to `file:` plus a replay route is a durable outbox in a few lines of YAML. Use `seda` with `blockWhenFull=true` in front of slow calls. - **Make delivery idempotent.** Replays and MQTT QoS 1 can deliver a message twice. Give each message an id at the edge and use an `idempotentConsumer` at the receiving side. - **Report presence with the broker.** Retained status messages and a last will tell the cloud what is offline without a polling loop. - **Thin out telemetry.** `sample` and `throttle` keep the uplink bill under control; aggregate before sending. ## Common questions ### Can Cameleer run on a Raspberry Pi? Yes, on a 64-bit OS: the `pmsudhi/cameleer` image is published for `linux/arm64`, and a route uses about 10 MB of memory. 32-bit ARM images are not published. ### How do I bridge MQTT to Kafka? Use a route from `paho:topic/#?brokerUrl=tcp://...` to `kafka:topic?brokers=...`, converting the payload with `convertBodyTo` and setting `kafka.KEY` from the MQTT topic if you need per-device ordering. The [sensor rollup example](#filter-and-aggregate-mqtt-sensor-readings) shows it with filtering and aggregation. ### What happens to messages when the internet connection drops? Without precautions, a failed call fails the exchange. With a dead letter channel to a local `file:` folder and a replay route, as in [store and forward](#store-and-forward-when-the-uplink-is-down), messages wait on disk and are sent when the uplink returns. ### Does Cameleer support MQTT 5 and TLS? Yes. `paho` speaks MQTT 3.1.1 and `paho-mqtt5` MQTT 5.0, both with QoS 0, 1 and 2, retained messages, wills, persistent sessions, automatic reconnect and `ssl://` broker URLs. WebSocket transports and shared subscriptions are not supported. ### Which industrial and healthcare protocols are supported? MQTT, syslog, HL7 v2 over MLLP, TCP and UDP (`netty`), HTTP, WebSocket, FTP and SFTP, Kafka, AMQP and STOMP. There are no OPC UA, Modbus or PLC4X components; reach those through an MQTT or HTTP gateway. ## Related - [EIPs](eips.md): [aggregate](eips/split-aggregate.md#aggregator), [sample and throttle](eips/flow-control.md) - [Error handling](error-handling.md), [deployment guide](../DEPLOYMENT.md), [benchmarks](../BENCHMARKS.md) - Components: [paho](../components/paho.md), [kafka](../components/kafka.md), [syslog](../components/syslog.md), [mllp](../components/mllp.md), [file](../components/file.md) ## Found a problem? If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: [open an issue](https://github.com/pmsudhi/cameleer/issues/new) that describes what you saw, then fork the repository, fix it, add or update a test, run `cargo test --workspace`, and open a pull request. [CONTRIBUTING.md](https://github.com/pmsudhi/cameleer/blob/main/CONTRIBUTING.md) explains the ground rules.