Route IoT sensor data from MQTT to separate Server-Sent Events (SSE) streams based on the message subject, using the config-driven rules resolver.
An MQTT broker receives IoT sensor data on sensors/#. A web dashboard needs separate SSE streams for different data types so that each panel subscribes only to its relevant feed:
sensors/temperature/...) goes to SSE endpoint /events/temperaturesensors/humidity/...) goes to SSE endpoint /events/humidity/events/otherGoBridge inspects each incoming message’s subject and routes it to exactly one SSE sender using the rules resolver with prefix matching. Browser clients connect to the SSE endpoints and receive a filtered, real-time event stream.
flowchart LR
subgraph MQTT Broker
T["sensors/#"]
end
subgraph GoBridge
R[Receiver\nsensor-in]
Route[Route\nsse-dispatch]
Res["Resolver\ntype: rules\nsubject prefix matching"]
end
subgraph "HTTP Server (SSE)"
SSE1["/events/temperature"]
SSE2["/events/humidity"]
SSE3["/events/other"]
end
subgraph Browser Clients
B1["Dashboard\nTemperature Panel"]
B2["Dashboard\nHumidity Panel"]
B3["Dashboard\nOther Panel"]
end
T -->|subscribe| R
R --> Route
Route --> Res
Res -->|"subject: sensors/temperature/*"| SSE1
Res -->|"subject: sensors/humidity/*"| SSE2
Res -->|"no rule match (default)"| SSE3
SSE1 -->|EventSource| B1
SSE2 -->|EventSource| B2
SSE3 -->|EventSource| B3
style Route fill:#f96,stroke:#333
style Res fill:#ff9,stroke:#333
style GoBridge fill:#eef,stroke:#333
bridge:
id: sensor-sse-router
sessions:
- id: mqtt-conn
transport: mqtt
# direct_hold relies on the broker redelivering what a crashed process never
# acknowledged; only a persistent (or exclusive) session does that.
session_mode: persistent
options:
session:
broker_url: tcp://mqtt.iot.local:1883
client_id: sensor-sse-router-01
clean_start: false
session_expiry_interval: 3600
keep_alive: 30
# The builder sizes ingress memory from max_payload_bytes,
# receive_maximum and the route's max_in_flight before it opens
# anything; 500 in flight at the default 256 KiB payload cap needs
# ~460 MiB, above the 256 MiB default budget. State the budget.
ingress_memory_budget_bytes: 536870912 # 512 MiB
stores:
# A persistent session keeps an exact record of the filters it installed on
# the broker (ADR 0003); seed the baseline once, before the first start:
# gobridge -config bridge.yaml -seed-managed-subscriptions mqtt-conn
managed_subscriptions:
type: sqlite
options:
path: /var/lib/gobridge/state/managed-subscriptions.db
# Where a message the route gives up on is kept.
dlq:
type: sqlite
options:
path: /var/lib/gobridge/state/dlq.db
receivers:
- id: sensor-in
session_id: mqtt-conn
topics:
- topic: "sensors/#"
qos: 1
senders:
- id: sse-temperature
transport: http
options:
mode: sse
path: /events/temperature
heartbeat_interval: 15s
max_clients: 500
- id: sse-humidity
transport: http
options:
mode: sse
path: /events/humidity
heartbeat_interval: 15s
max_clients: 500
- id: sse-other
transport: http
options:
mode: sse
path: /events/other
heartbeat_interval: 15s
max_clients: 500
bindings:
- id: to-temperature
sender_id: sse-temperature
# Naming the session on the binding is what makes the bridge manage it:
# connect, subscribe, reconcile. A session nobody manages never subscribes.
session_id: mqtt-conn
address: temperature
- id: to-humidity
sender_id: sse-humidity
address: humidity
- id: to-other
sender_id: sse-other
address: other
routes:
- id: sse-dispatch
receiver_id: sensor-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-temperature, to-humidity, to-other]
resolver:
type: rules
default_binding: to-other
rules:
- binding_id: to-temperature
match:
- field: subject
operator: prefix
value: "sensors/temperature/"
- binding_id: to-humidity
match:
- field: subject
operator: prefix
value: "sensors/humidity/"
policy:
max_in_flight: 500
# Exactly one replica consumes this subscription; a second copy of this
# process would double-deliver. See Scenario 8 for fenced ownership.
allow_unfenced: true
Each SSE sender registers an HTTP endpoint that browser clients connect to via the EventSource API. The mode: sse option is currently the only sender mode supported by the HTTP transport.
| Option | Purpose |
|---|---|
mode |
Must be sse. Selects Server-Sent Events output. |
path |
HTTP GET path where clients connect (e.g., /events/temperature). |
heartbeat_interval |
How often to send SSE comment heartbeats to keep connections alive. Default: 30s. |
max_clients |
Maximum concurrent SSE connections per sender. Default: 10000. |
All three SSE senders share the same HTTP transport factory and internal http.ServeMux. The factory’s Handler() method returns a single http.Handler that dispatches to the correct SSE sender based on the request path.
The resolver block replaces the need for programmatic MatchFunc registration. The rules are evaluated in order (first-match-wins):
subject starts with sensors/temperature/, select binding to-temperature.subject starts with sensors/humidity/, select binding to-humidity.to-other (the default_binding).A message on topic sensors/temperature/room-3 matches rule 1 and is broadcast to all clients connected to /events/temperature. A message on topic sensors/pressure/lab-1 matches no rule and falls through to the default binding to-other.
SSE is an ephemeral transport. Connected clients receive events in real-time; disconnected clients miss events. There is no durable queue or outbox backing the SSE stream.
This makes direct_hold the correct delivery mode:
The HTTP transport advertises the CapHTTPEndpoint capability. Routes using HTTP sources or SSE senders operate correctly with direct_hold because the transport does not require visibility extension or deferred acknowledgment.
The address field in each binding is stored as metadata on the dispatch plan. For SSE senders the address is informational only – the sender broadcasts to all connected clients regardless of the address value. The actual SSE endpoint path is configured in the sender’s path option.
The max_in_flight: 500 policy limits concurrent messages flowing through the route. If MQTT delivers faster than the SSE broadcast can complete, backpressure propagates to the MQTT receiver. Each SSE client has a 256-event internal buffer; when a client’s buffer is full, events are dropped for that client with a warning log.
sequenceDiagram
participant MQTT as MQTT Broker
participant R as Receiver (sensor-in)
participant Route as RouteRunner
participant Res as RuleResolver
participant SSE as SSE Sender (sse-temperature)
participant B as Browser Client
Note over B,SSE: Client connects: GET /events/temperature
B->>SSE: EventSource connect
SSE-->>B: HTTP 200, Content-Type: text/event-stream
MQTT->>R: sensors/temperature/room-3 (QoS 1)
R->>Route: Envelope{Subject: "sensors/temperature/room-3"}
Route->>Res: Resolve(envelope)
Note over Res: Rule 1: subject prefix "sensors/temperature/" -- MATCH
Res-->>Route: DispatchPlan{BindingID: "to-temperature"}
Route->>SSE: Send(envelope)
SSE->>SSE: JSON marshal, format SSE frame
SSE->>B: id: msg-001\nevent: message\ndata: {...}
SSE-->>Route: OK
Route->>R: ACK (PUBACK)
package main
import (
"context"
"log/slog"
"net/http"
"github.com/mariotoffia/gobridge/bridge"
cfgparser "github.com/mariotoffia/gobridge/config/parser"
"github.com/mariotoffia/gobridge/ports"
adaptershttp "github.com/mariotoffia/gobridge/adapters/http/transport"
"github.com/mariotoffia/gobridge/adapters/mqtt/transport/paho"
)
func main() {
logger := slog.Default()
// Build the plugin registry and register each linked adapter's config
// decoder. ParseFile requires a non-nil registry.
reg := ports.NewRegistry()
_ = paho.Register(reg)
_ = adaptershttp.Register(reg)
cfg, _ := cfgparser.ParseFile("bridge.yaml", cfgparser.FormatAuto, reg)
httpFactory := adaptershttp.NewFactory(adaptershttp.WithFactoryLogger(logger))
rt, _ := bridge.NewBuilder(cfg, bridge.WithLogger(logger)).
RegisterTransportFactory("mqtt", paho.NewFactory(logger)).
RegisterTransportFactory("http", httpFactory).
Build(context.Background())
// Mount SSE endpoints on an HTTP server
go func() {
mux := http.NewServeMux()
mux.Handle("/events/", httpFactory.Handler())
_ = http.ListenAndServe(":8080", mux)
}()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
rt.Start(ctx)
// ... wait for signal ...
rt.Stop(ctx)
}
The HTTP transport factory accumulates all SSE sender paths. Mounting httpFactory.Handler() on a standard http.ServeMux exposes /events/temperature, /events/humidity, and /events/other as SSE endpoints.
Use regex instead of prefix for more complex topic patterns:
resolver:
type: rules
default_binding: to-other
rules:
- binding_id: to-temperature
match:
- field: subject
operator: regex
value: "^sensors/temperature/(room|lab)-\\d+$"
- binding_id: to-humidity
match:
- field: subject
operator: regex
value: "^sensors/humidity/[a-z]+-\\d+$"
This restricts matching to specific topic name formats, rejecting malformed topics.
Route based on a type field inside the JSON payload instead of the MQTT topic:
resolver:
type: rules
default_binding: to-other
rules:
- binding_id: to-temperature
match:
- field: $.type
operator: eq
value: "temperature"
- binding_id: to-humidity
match:
- field: $.type
operator: eq
value: "humidity"
This is useful when multiple sensor types share the same MQTT topic but differentiate via a payload field like {"type": "temperature", "value": 22.5}.
Protect SSE streams with a per-sender API key:
senders:
- id: sse-temperature
transport: http
options:
mode: sse
path: /events/temperature
api_key: "dashboard-key-min-16ch"
Clients must include the key via X-API-Key header or Authorization: Bearer token. Connections without a valid key receive HTTP 401. The key is compared using SHA-256 constant-time hashing to prevent timing and length-based information leaks.
Add a fourth binding to archive all sensor data to SQS while still streaming to SSE:
senders:
- id: sqs-archive
transport: sqs
options:
queue_url: https://sqs.us-west-1.amazonaws.com/123456789/sensor-archive
region: us-west-1
bindings:
- id: to-archive
sender_id: sqs-archive
address: sensor-archive
routes:
- id: sse-dispatch
receiver_id: sensor-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-temperature, to-humidity, to-other, to-archive]
resolver:
type: all
Changing the resolver to type: all converts this into a fan-out route where every binding receives every message. To combine fan-out with filtering, use separate routes or the processor-based filter approach from Scenario 4.
fan_out dispatch, which sends to all bindings.Send call succeeds with no recipients. No error is returned.heartbeat_interval sends SSE comments (: heartbeat\n\n) to prevent proxies and load balancers from closing idle connections.binding_id values in resolver rules reference bindings listed in the route’s bindings array. Invalid references cause a startup error, not a runtime surprise.NewSession returns (nil, nil). There is no sessions entry for HTTP senders.