Your own app, through Kafka

Your devices speak MQTT. Your app - a dashboard, a rules engine, a database
writer, written in Node.js, Python, Java or anything else - does not have to.
It reads and writes Kafka, and modbee carries everything between the two:

 devices ──MQTT──► modbee ──► Kafka topic "mqtt-messages" ──► your app
 devices ◄──MQTT── modbee ◄── Kafka topic "mqtt-commands" ◄── your app

This page is for building that app on one of the *-kafka setups in
docker/examples - broker-kafka
on your own computer is the quickest - which set all of this up as they ship. It assumes no Kafka experience.

Why go through Kafka rather than subscribe to MQTT directly? Kafka keeps
every message (seven days by default), so an app that was down catches up when
it starts again, a new app can read last week's data, and any number of apps
can each read everything without affecting each other or the devices. For one
small script, subscribing to MQTT is simpler, and that works too.

Kafka in one minute

Three words are all you need:

• Topic - a named list of records that Kafka keeps on disk, in order. Here
  there are two: mqtt-messages (from your devices) and mqtt-commands (to
  them). Reading a record does not remove it.
• Record - one message: a key, a value, and optional headers
  (name/value pairs). modbee turns each MQTT message into one record.
• Consumer group - the name your app reads under (groupId). Kafka
  remembers, per group, how far it has read, so a restarted app carries on
  where it stopped. Two copies of an app with the same group share the
  work; apps with different groups each get every record.

What arrives from your devices

Every MQTT message becomes one record in mqtt-messages:

The value is bytes: text for a reading like 4.25, JSON if your devices
send JSON (parse it in your app), and binary for Sparkplug B (decode it with a
Sparkplug library, such as sparkplug-payload for Node.js or Eclipse Tahu for
Python).

Order: each device's messages arrive in the order it sent them, because
they share a key, and so a partition - whichever topics they were on. That is
the order MQTT itself promises. Messages from different devices may
interleave.

Sending to your devices

Write a record to mqtt-commands:

modbee publishes it to MQTT at QoS 1, not retained, and every device
subscribed to that topic receives it. Both rules - the key as the topic, QoS
1 - are the kafka-to-mqtt mapping in the examples' modbee.yaml, and
kafka.md shows how to change them (a fixed topic, a
different QoS, retained, properties). To choose them per record instead, give
the record the headers mqtt-topic, mqtt-qos and mqtt-retain
("true"): the examples take those over the key.

The key means something different in each direction: here it is the topic to
publish on; on mqtt-messages it is the device that published. Each is its
own mapping in modbee.yaml, and each is set the way that suits it.

You will also see your own command come back on mqtt-messages - everything
published on MQTT goes there, including what apps send. It has no key and no
mqtt-clientid header, which is how you tell it apart. It never loops: modbee
does not write a message back to the Kafka topic it came from.

Where your app connects

The first three need no password, so they stay on the server, or on your own
network. Over the internet, use port 9093: set a password in .env - nothing
to build or install - and connect like this:

const kafka = new Kafka({
  kafkaJS: {
    brokers: ["mqtt.example.com:9093"],
    ssl: true,
    sasl: { mechanism: "plain", username: "app", password: process.env.KAFKA_PASSWORD },
  },
});

docker.md has the details, and the
same settings for Python and any other client.

A complete example in Node.js

examples/kafka/nodejs is a small app that reads
every MQTT message and runs one rule: when the tank level on
plant/tank-01/level goes above 8 it sends stop to plant/pump-2/cmd, and
below 2 it sends start. It uses
@confluentinc/kafka-javascript,
Confluent's official Node.js client, through its KafkaJS-style API.

Run it

With docker/examples/broker-kafka running, from the example's folder:

cd examples/kafka/nodejs
docker compose up -d --build       # joins the stack's network; Kafka is kafka:9092
docker compose logs -f

Or, with Node.js installed on the same computer:

npm install
node tank-rule.js                  # uses localhost:9094

Try it

Act as two devices, with the modbee command that is inside the stack. In one
terminal, be the pump, listening for its commands:

cd docker/examples/broker-kafka
docker compose exec modbee modbee sub -t plant/pump-2/cmd -v

In a second terminal, be the tank, publishing its level:

cd docker/examples/broker-kafka
docker compose exec modbee modbee pub -t plant/tank-01/level -m 8.6 -q 1
docker compose exec modbee modbee pub -t plant/tank-01/level -m 1.5 -q 1

The pump's terminal prints plant/pump-2/cmd stop, then
plant/pump-2/cmd start, and the app's log (docker compose logs -f in the
example's folder) shows each step:

connected to Kafka at kafka:9092 - waiting for MQTT messages
<- MQTT plant/tank-01/level: 8.6  (QoS 1, from modbee-pub-178)
-> MQTT plant/pump-2/cmd: stop
<- MQTT plant/pump-2/cmd: stop  (QoS 1, from a Kafka app)
<- MQTT plant/tank-01/level: 1.5  (QoS 1, from modbee-pub-199)
-> MQTT plant/pump-2/cmd: start
<- MQTT plant/pump-2/cmd: start  (QoS 1, from a Kafka app)

The parts that matter

Connecting, and reading under a group:

const { Kafka, logLevel } = require("@confluentinc/kafka-javascript").KafkaJS;

const kafka = new Kafka({ kafkaJS: { brokers: ["localhost:9094"], logLevel: logLevel.WARN } });
const consumer = kafka.consumer({ kafkaJS: { groupId: "tank-rule", fromBeginning: true } });
const producer = kafka.producer();

fromBeginning: true means the first time this group runs, it starts with
everything Kafka still holds (up to seven days); after that it carries on from
where it stopped. Use false to start with new messages only.

Reading an MQTT message:

await consumer.subscribe({ topics: ["mqtt-messages"] });
await consumer.run({
  eachMessage: async ({ message }) => {
    const mqttTopic = message.headers["mqtt-topic"]?.toString();   // plant/tank-01/level
    const payload = message.value?.toString() ?? "";                // 4.25
    const device = message.key?.toString();                         // tank-01, or none from an app
  },
});

Sending one:

await producer.send({
  topic: "mqtt-commands",
  messages: [{ key: "plant/pump-2/cmd", value: "stop" }],
});

The same in Python

With confluent-kafka
(pip install confluent-kafka):

from confluent_kafka import Consumer, Producer

consumer = Consumer({
    "bootstrap.servers": "localhost:9094",
    "group.id": "my-python-app",
    "auto.offset.reset": "earliest",   # a new group starts with what is already there
})
producer = Producer({"bootstrap.servers": "localhost:9094"})
consumer.subscribe(["mqtt-messages"])

while True:
    record = consumer.poll(1.0)
    if record is None or record.error():
        continue
    headers = {name: value.decode() for name, value in (record.headers() or [])}
    mqtt_topic = headers.get("mqtt-topic")   # the MQTT topic
    payload = record.value() or b""         # the MQTT payload, as bytes
    device = record.key()                    # the publisher's client id; None from an app
    print(mqtt_topic, payload.decode(errors="replace"), headers.get("mqtt-qos"))

    # An empty payload clears a retained message; it is not a reading.
    if mqtt_topic == "plant/tank-01/level" and payload.strip() and float(payload) > 8:
        producer.produce("mqtt-commands", key="plant/pump-2/cmd", value="stop")
        producer.flush()

Any other language works the same way with any Kafka client: connect to
the address above, read mqtt-messages under a group id of your choosing
(the MQTT topic is the mqtt-topic header), and write records keyed by MQTT
topic to mqtt-commands.

Things that catch people out

• The two addresses are not interchangeable. From your computer, outside
  Docker, it is localhost:9094 - port 9092 is not open outside Docker, so
  localhost:9092 is refused. From a container, it is kafka:9092 -
  localhost inside a container is the container itself. Kafka tells each
  client which address to use from then on, and each of the two is only right
  for one side.
• An app in another compose file cannot find kafka until it joins the
  modbee network - the three lines at the bottom of
  the example's compose.yml. Start the
  stack first, so the network exists.
• A record may arrive twice after your app or Kafka restarts: Kafka
  guarantees each record at least once, not exactly once. Make what your app
  does safe to repeat - setting a value is; adding to a total is not.
• A new Kafka topic for commands must exist before anything is written to
  it, or the first record is missed. The examples create mqtt-commands;
  if you add another, add it to the kafka-topics service in compose.yml
  (kafka.md).
• Which MQTT topics reach Kafka, and which Kafka topics reach MQTT, is
  configured in modbee.yaml. As shipped, every MQTT message goes to
  mqtt-messages. To send only plant/#, or to split topics across several
  Kafka topics, edit the mappings - kafka.md has every option -
  and run docker compose restart modbee.
• Many Node.js examples online use the kafkajs package. It works, but it
  is no longer actively developed; Confluent's client used here runs the same
  style of code.