# UltimateMQ

**Repository Path**: wanwuliandong/UltimateMQ

## Basic Information

- **Project Name**: UltimateMQ
- **Description**: 基于C++ 17的极致性能通用消息队列，支持AMQP/Kafka/MQTT
- **Primary Language**: C++
- **License**: Apache-2.0
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-16
- **Last Updated**: 2026-11-14

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# UltimateMQ

UltimateMQ is an experimental MQTT broker written in C++ with cursor.

## Prerequisites

- C++ compiler with C++17 support
- CMake 3.12 or higher
- OpenSSL development headers (`libssl-dev`); the broker still runs in
  plaintext when no certificate is configured

## Building

```bash
mkdir build && cd build
cmake ..
make
```

## Run

```bash
./flowmq
```

### Configuration

All configuration is read from environment variables at startup.

| Variable | Default | Description |
| --- | --- | --- |
| `MQTT_PORT` | `1883` | TCP port the broker listens on |
| `FLOWMQ_DEBUG` | unset | Emit debug traces to stderr |
| `FLOWMQ_PASSWORD_FILE` | unset | User database; enables authentication when present |
| `FLOWMQ_ACL_FILE` | unset | Per-user topic access rules; enables ACL enforcement when present |
| `FLOWMQ_STATE_FILE` | unset | Durable state file; retained messages and persistent sessions survive a restart when set |
| `FLOWMQ_MAX_PACKET_SIZE` | `1048576` (1 MiB) | Largest Remaining Length accepted from a single client, in bytes; also advertised to MQTT 5 clients |
| `FLOWMQ_TLS_CERT` | unset | Server certificate (PEM); enables TLS when set together with `FLOWMQ_TLS_KEY` |
| `FLOWMQ_TLS_KEY` | unset | Private key (PEM) matching `FLOWMQ_TLS_CERT` |
| `FLOWMQ_TLS_CA` | unset | CA bundle used to verify client certificates |
| `FLOWMQ_TLS_VERIFY_CLIENTS` | unset | When set, a client certificate signed by `FLOWMQ_TLS_CA` is required |

When neither file is set the broker stays fully open (every client may
connect and use every topic), which keeps the historical behaviour.#### Users (`FLOWMQ_PASSWORD_FILE`)

One entry per line, `#` starts a comment:

```
# plaintext
alice:secret
# pre-hashed: $sha256$<salt-hex>:<sha256(salt + password)>
bob:$sha256$a1b2c3d4:9f2c...c41d
```

Once a users file is loaded every client must send a username and password;
a missing or wrong credential is answered with CONNACK `0x86` (bad user name
or password, MQTT 5) / `0x04`-style failure for 3.1.1 and the socket is closed.

#### Access control (`FLOWMQ_ACL_FILE`)

```
# user,action,topic-filter
alice,read,sensors/#
alice,write,commands/alice
bob,read,#
*,read,public/#
```

* `read` controls subscriptions (and therefore deliveries and retained
  messages); `write` controls PUBLISH and Will messages.
* Topic filters use the normal MQTT wildcard syntax.
* The special user `*` applies to every client.
* If an ACL file is loaded, anything not explicitly granted is denied. A
  denied SUBSCRIBE returns SUBACK `0x87` (MQTT 5) or `0x80` (3.1.1); a denied
  PUBLISH returns PUBACK/PUBREC `0x87` and the message is not routed; a denied
  QoS 0 PUBLISH disconnects an MQTT 5 client and is dropped for 3.1.1.
* A session is bound to the user that created it: connecting with the same
  client id as a different user discards the old session instead of handing
  over its subscriptions and buffered messages.

#### Durable state (`FLOWMQ_STATE_FILE`)

Without this variable the broker keeps everything in memory, exactly as before.
With it, the following survive a restart:

* retained messages;
* every session created with `cleanStart=false`, together with its
  subscriptions, its next packet id and its unacknowledged QoS 1/2 messages.
  An unacknowledged message is retransmitted with `DUP=1` on reconnect.
* an inbound QoS 2 flow that is still awaiting its `PUBREL`: on reconnect (or
  after a restart) the broker resends the `PUBREC` so the client can finish the
  handshake with its `PUBREL`, instead of abandoning the flow.

The file is written by a background thread at most once per second and again on
`SIGINT`/`SIGTERM`, always atomically (a temporary file is renamed into place),
so a crash never leaves a half-written state file behind. A state file with an
unrecognised header aborts startup instead of silently discarding sessions.

```bash
FLOWMQ_STATE_FILE=/var/lib/flowmq/state.dat ./flowmq
```

QoS 0 messages are at-most-once and are not persisted. The offline queue of a
session is not stored separately: QoS>0 messages already live in its
unacknowledged set and are re-queued on reconnect, which avoids duplicates.

#### Transport security (`FLOWMQ_TLS_CERT` / `FLOWMQ_TLS_KEY`)

By default the listener speaks plaintext, so with authentication enabled the
username and password cross the network in the clear. Supplying a certificate
turns the listener into a TLS endpoint (TLS 1.2 or newer) and every byte,
credentials included, is then encrypted:

```bash
FLOWMQ_TLS_CERT=/etc/flowmq/server.crt \
FLOWMQ_TLS_KEY=/etc/flowmq/server.key \
FLOWMQ_PASSWORD_FILE=/etc/flowmq/users.txt \
./flowmq
```

* `FLOWMQ_TLS_CERT` and `FLOWMQ_TLS_KEY` (PEM) must be given together; setting
  only one aborts startup with an error naming the missing variable.
* The cipher is chosen by OpenSSL; TLS 1.3 is negotiated when the client
  supports it.
* A plaintext client connecting to a TLS listener fails during the handshake
  and never reaches the MQTT layer.
* Set `FLOWMQ_TLS_VERIFY_CLIENTS` together with `FLOWMQ_TLS_CA` to also require
  a client certificate signed by that CA (mutual TLS). A client without a valid
  certificate is dropped at the handshake.

#### Maximum packet size (`FLOWMQ_MAX_PACKET_SIZE`)

A client can declare any Remaining Length up to 256 MiB in an MQTT header. Without
a cap the broker would buffer that declaration for a client that dribbles the body
out (or never finishes), so one connection can be made to pin a large amount of
memory. `FLOWMQ_MAX_PACKET_SIZE` bounds this: as soon as a packet's Remaining
Length is read and found to exceed the limit, the connection is dropped before any
body byte is buffered.

```bash
FLOWMQ_MAX_PACKET_SIZE=8388608 ./flowmq   # allow up to 8 MiB per packet
```

* The limit defaults to 1 MiB, which comfortably covers normal telemetry.
* The limit is applied to every packet, including the first `CONNECT`.
* MQTT 5 clients are told the value through the `Maximum Packet Size` CONNACK
  property, so well-behaved clients keep their own packets within it. MQTT 3.1.1
  has no property section and is simply enforced on the wire.
* A non-numeric or zero value aborts startup with an error naming the variable.

## Contributing

Please read CONTRIBUTING.md for details on our code of conduct and the process for submitting pull requests.

## License

This project is licensed under the Apache License, Version 2.0 - see the LICENSE file for details.