Repository navigation
Replies: 2 comments
1. Motivation & Operational ContextMQTT is the industry-standard messaging protocol for IoT devices, microcontrollers, and edge gateways. Providing an MQTT entry point enables existing MQTT clients to publish to and consume from Iggy. To maintain architectural clarity, two distinct integration paths are defined:
The standalone gateway introduces an independently deployed process that must be configured, secured, monitored, scaled, and made highly available. This operational cost is justified specifically for brokerless edge deployments. 2. Architecture & Process BoundariesThe MQTT gateway operates as an independent, standalone process under Process Boundary Guarantees
Proposed Crate Structure3. Protocol Scope & Compatibility BaselineBaseline Specification
Control Packet Scope (11 of 14 MQTT 3.1.1 Control Packets)
Explicit Exclusions & Error BehaviorThe following features remain explicitly out of scope:
Note on Error Codes: Unsupported features return MQTT-specific reason codes or disconnect behavior according to spec, rather than generic "unsupported version" errors [29]. 4. Authentication, Authorization & Validation PipelineWhen an MQTT client connects, the gateway processes the request through a multi-stage pipeline.
5. Topic Mapping & Partitioning StrategyTopic Mapping AbstractionMQTT topic names are mapped to Iggy Streams and Topics through an explicit mapping layer within the gateway:
Partitioning & OrderingThe gateway selects the target Iggy partition before forwarding an MQTT
Execution FlowAll topic mapping and partition-selection logic is performed inside the gateway process. Iggy Core remains unaware of MQTT-specific topic semantics and receives a standard SDK request targeting the resolved Iggy Stream, Topic, and partition. 6. Inbound Publish & Outbound Delivery Flow (ACK Disambiguation)A crucial architectural distinction is drawn between Inbound Publish Acknowledgements and Outbound Delivery Acknowledgements:
7. Shared-Reader Fan-Out & Acknowledgement RulesShared Reader Model
Contiguous Safe Offset Advancement (
|
0 replies
MQTT Gateway for Apache Iggy
TL;DR
1. Where this fitsThere are two ways to connect MQTT devices to Iggy:
flowchart LR
D["MQTT devices<br/>sensors · edge gateways<br/>ports 1883 / 8883 (TLS)"] -->|"PUBLISH / SUBSCRIBE"| G
subgraph G["Gateway core — standalone process, gateways/mqtt/"]
C1["Bounded wire codec"]
C2["Session + ACL check"]
C3["ToMaR mapping engine"]
C4["Shared readers + fan-out"]
C5["ACK ledger · bounded queues"]
end
G --> B["IggyBridge<br/>official Rust SDK<br/>send · poll · commit"]
B --> I[("Iggy Server<br/>streams · topics<br/>partitions · offsets")]
G -.-> H["Health: GET :8080/health · :8080/ready"]
Two state rules:
Layout: one flat crate at 2. First connectionNothing reaches Iggy until a connection passes three gates: sequenceDiagram
participant Dev as Device
participant GW as Gateway
participant Iggy as Iggy (via SDK)
Dev->>GW: 1 · TCP / TLS connect → 1883 / 8883
Dev->>GW: 2 · CONNECT (ClientID, credentials, CleanSession)
Note over GW: 3 · Validate frame + version + flags<br/>(size cap 8092 B · UTF-8 · reject QoS 2 flags)
GW->>Iggy: 4 · Authenticate principal
GW-->>Dev: 5 · CONNACK — session created, capabilities advertised
Note over Dev,Iggy: 6 · From here, every PUBLISH / SUBSCRIBE is ACL-checked<br/>against its mapped Iggy stream/topic before any SDK call
The CONNACK is a capability firewall: the gateway advertises only what it implements 3. Scope: 11 of 14 packetsMQTT 3.1.1 defines 14 packet types. This gateway supports 11:
Excluded features, and what a client observes on the wire:
Why so small? Every broker feature is stored state: retained messages are a store, 4. Produce mapping (MQTT → Iggy)
4.1 Definitions
| Resolution | The result of mapping: `(stream, iggy topic, partition key \ | none) | Envelope | The Iggy message as stored: payload unchanged, plus headers 4.2 Compile time: what the gateway checks before it serves anyoneAt startup the gateway compiles the whole topic-map file and refuses to start if any
4.3 The produce algorithmOn
flowchart TD
A["PUBLISH (topic T, QoS q, pid)"] --> B{"P1 · frame valid?<br/>size · UTF-8 · no wildcards · QoS ≠ 2"}
B -->|"no"| X["Protocol reject<br/>close connection · count"]
B -->|"yes"| C["P2 · Match(T)<br/>first rule in file order"]
C --> D["P3 · Resolve<br/>captures → stream · topic · key<br/>(no match → configured fallback)"]
D --> E{"P4 · target exists<br/>and ACL allows D?"}
E -->|"no"| Y["No append · NO PUBACK<br/>device retries · ops alerted"]
E -->|"yes"| F["P5 · partition = hash(key) mod N<br/>no key → balanced"]
F --> G["P6 · stamp envelope<br/>mqtt.topic = T · mqtt.qos · ClientID+PacketID"]
G --> H["P7 · append via SDK"]
H --> I{"append confirmed?"}
I -->|"yes, q = 1"| J["PUBACK(pid)"]
I -->|"yes, q = 0"| K["done — no ack exists"]
I -->|"no"| Y
4.4 One packet, step by stepDevice publishes
Three more packets against the same config:
4.5 The aggregation, in one pictureflowchart LR
T1["building/north/room/42/temp"] --> M
T2["building/north/room/42/humidity"] --> M
T3["building/south/room/7/temp"] --> M
M["Match + Resolve<br/>first match · captures → templates"] --> O["ONE Iggy topic: building_north / telemetry<br/>partition by room_$2<br/>each message keeps its full original<br/>topic in the mqtt.topic header"]
Thousands of device topics collapse into a handful of Iggy topics. **The collapse is 4.6 The only regex an operator needs
Rules of thumb:
4.7 The config, annotated# topic-map.toml — one complete document (check C4)
default_stream = "sensors" # fallback resolution, part 1
default_topic = "messages_default" # fallback resolution, part 2 (name is Strimzi's)
[[rule]] # rule 1 — evaluated first (Match = first match in file order)
mqtt_topic = 'building/(\w+)/room/(\d+)/.*' # $1 = building word, $2 = room number
stream = 'building_$1' # one stream per building (bounded set)
topic = 'telemetry' # constant: all metrics aggregate here
partition_key = 'room_$2' # stable identity ⇒ ordering per room
[[rule]] # rule 2
mqtt_topic = 'sensors/([^/]+)/data' # $1 = any single level, dashes allowed
stream = 'sensors'
topic = 'sensor_data'
partition_key = 'sensor_$1'Use TOML single-quoted (literal) strings for patterns, so regex backslashes need no 4.8 Outcomes: every publish ends in exactly one visible state
There is no sixth outcome and no silent path — that totality is what "coverage" means for 4.9 Edge cases, decided
4.10 Invariants (testable)
5. Fetch mapping (Iggy → MQTT subscribers)
5.1 Definitions
5.2 Resolve: how a subscription finds its partitionKey insight: a subscription topic is a **concrete string of exactly the same form a Consequences, stated explicitly:
5.3 Subscribe-time algorithmOn
On 5.4 Delivery-time algorithm (Route)The shared reader for a triple polls a batch with
flowchart LR
LOG[("Iggy partition<br/>M101 · M102 · M103 …")] -->|"poll once<br/>shared reader"| R["Fan-out router<br/>reads mqtt.topic header<br/>registry lookup → delivery set"]
R -->|"1 · personalised copy<br/>enqueue — non-blocking"| QA["Client A queue · dashboard<br/>queued bytes ≈ 0 KB"]
R -->|"1 · personalised copy<br/>enqueue — non-blocking"| QB["Client B queue · phone<br/>queued bytes ≈ 240 KB"]
R -->|"1 · personalised copy<br/>enqueue — non-blocking"| QC["Client C queue · stuck sensor<br/>queued bytes growing …"]
QA -->|"2 · writer drains<br/>as fast as A receives"| WA["Connection writer A"] --> CA["Client A"]
QB -->|"2 · writer drains<br/>slower than fan-out adds"| WB["Connection writer B"] --> CB["Client B"]
QC -->|"2 · writer cannot drain<br/>C is stuck / not reading"| WC["Connection writer C"] --> CC["Client C ✗"]
QC --> Q{"3 · queued_bytes ><br/>max_outbound_queue_bytes_per_client<br/>(e.g. 1 MB)?"}
Q -->|"QoS 0 copy"| DROP["Drop the new frame<br/>QoS 0 promised nothing"]
Q -->|"QoS 1 copy"| PAUSE["Pause delivery to C<br/>still not draining<br/>within the timeout → EVICT C<br/>purge queue · strike from ledger"]
R -.->|"QoS 1 copies also create<br/>a ledger entry"| LED[("ACK ledger")]
CA -.->|"PUBACK"| LED
CB -.->|"PUBACK"| LED
classDef stuck fill:#ffe3e3,stroke:#d64545,color:#7a1020;
class QC,CC stuck;
5.5 Worked exampleRegistry state and one polled batch from triple Timing, pinned down: if the phone subscribes to 5.6 Edge cases, decided
5.7 Invariants (testable)
flowchart LR
P[("One partition triple<br/>M101 · M102 · M103…")] -->|"poll once<br/>auto_commit = false"| R["Shared reader (I1)"]
R --> F["Route: hash-lookup header<br/>in registry → delivery set (I2)"]
F --> QA["Client A queue<br/>bounded · per-client packet IDs (I5)"]
F --> QB["Client B queue<br/>bounded"]
F --> QC["Client C queue<br/>bounded"]
5.8 The round trip, end to endsequenceDiagram
participant S as Sensor
participant GW as Gateway
participant Log as Iggy log
participant D as Dashboard
S->>GW: PUBLISH building/north/room/42/temp
GW->>Log: append to building_north/telemetry (key room_42)<br/>headers: mqtt.topic = building/north/room/42/temp
GW-->>S: PUBACK (only after commit)
D->>GW: SUBSCRIBE building/north/room/42/temp
Note over GW: S1–S5: validate → Resolve → registry upsert → reader ensured → SUBACK
Log-->>GW: reader polls the message
Note over GW: R1–R5: header lookup → set {dashboard} → personalised copy → queue
GW->>D: PUBLISH topic = building/north/room/42/temp (fresh packet ID)
D-->>GW: PUBACK → ledger (R6) → offset may advance (§6)
One line: *aggregation on the way in is lossy; reconstruction on the way out is exact — 6. How the two sides bind: POLL ≠ ACK6.1 Inbound rule
6.2 Outbound rulePolling a message is not delivering it. The gateway calls
6.3 A stall, and its resolutionOne silent client (C, above) freezes the partition's offset. The design answers in three
6.4 Four rules that keep the ledger sound
6.5 Slow-consumer policy, per QoS
7. Limits and healthAll limits live in one
Plus a global outbound-queue byte cap, so per-client limits can't multiply into an 8. Where each decision comes fromInspired, not invented — unless §9 says otherwise:
Deliberately not copied: the Kafka gateway's ACL gap (auth treated as a view, not 9. What is genuinely newTwo decisions have no precedent in any surveyed system. They are new because the
12. Build planFollowing the Kafka gateway's route (Discussion → foundation issue → batch-filed
Phasing inside the design: Phase 1 inbound publish path; Phase 2 subscription |
0 replies
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Uh oh!
There was an error while loading. Please reload this page.
The discussion is branching out from the #4181 around the architectural and design of the gateway/mqtt. The design details and implementation details will be added here
All reactions