Route messages to different MQTT topics or bindings based on message headers and content. This is one of GoBridge’s most powerful features for content-based routing.
A factory automation system receives events from an SQS queue. Each event has a factory_id header identifying which factory it belongs to. GoBridge must publish each event to the correct MQTT topic per factory: factory/{factory_id}/events.
flowchart LR
subgraph AWS
Q["SQS Queue\nfactory-events"]
end
subgraph GoBridge
R[Receiver\nsqs-in]
Route[Route\ndispatch]
Resolve["Address Template\nfactory/{factory_id}/events"]
end
subgraph MQTT Broker
T1["factory/A/events"]
T2["factory/B/events"]
T3["factory/C/events"]
end
Q --> R
R --> Route
Route --> Resolve
Resolve --> T1
Resolve --> T2
Resolve --> T3
style Route fill:#f96,stroke:#333
style Resolve fill:#ff9,stroke:#333
Binding addresses can contain {placeholder} tokens. At runtime, GoBridge replaces each placeholder with the corresponding value from the envelope’s headers:
Template: factory/{factory_id}/orders/{device_id}
Headers: factory_id = "A", device_id = "42"
Result: factory/A/orders/42
The rendered address becomes DispatchPlan.Address. The runtime carries it across the egress path on ports.OutboundMessage.Address, and the MQTT sender uses it as the publish topic. Envelope.Subject is not mutated — it remains the logical event subject and travels alongside the message as the gobridge.subject MQTT user property.
Safety guarantees:
+/#, no empty segments)bridge:
id: factory-router
sessions:
- id: mqtt-conn
transport: mqtt
options:
session:
broker_url: tcp://mqtt.factory.local:1883
client_id: factory-router-01
receivers:
- id: sqs-in
transport: sqs
options:
queue_url: https://sqs.us-west-1.amazonaws.com/123456789/factory-events
region: us-west-1
senders:
- id: mqtt-out
session_id: mqtt-conn
options:
sender:
qos: 1
bindings:
- id: to-factory-topic
sender_id: mqtt-out
# Naming the session on the binding is what makes the bridge manage it:
# a session nobody manages never connects, and every publish fails.
session_id: mqtt-conn
address: "factory/{factory_id}/events"
stores:
# The default policy dead-letters permanent failures and expired messages;
# a route that says so needs a store to write them to.
dlq:
type: sqlite
options:
path: /var/lib/gobridge/dlq.db
routes:
- id: dispatch
receiver_id: sqs-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-factory-topic]
Envelopefactory_id: "A")factory/{factory_id}/events becomes factory/A/eventsports.OutboundMessage{Envelope: <copy>, Address: "factory/A/events"}. Dispatch headers are merged onto the envelope copy; the source Envelope.Subject is left untouched.factory/A/events (taken from OutboundMessage.Address). The logical subject is propagated as the gobridge.subject MQTT user property so downstream consumers can reconstruct it.Note: The default_topic on the sender is a fallback used only when OutboundMessage.Address is empty. The publish topic is never read from Envelope.Subject.
sequenceDiagram
participant R as Receiver
participant Route as RouteRunner
participant Res as BindingResolver
participant S as MQTT Sender
R->>Route: Envelope (Subject="<logical>", headers: factory_id=A)
Route->>Res: Resolve(envelope)
Res->>Res: RenderAddress("factory/{factory_id}/events", headers)
Res-->>Route: DispatchPlan{Address: "factory/A/events"}
Route->>Route: build OutboundMessage{Envelope: copy, Address: "factory/A/events"}
Route->>S: Send(ctx, OutboundMessage)
S->>S: topic = OutboundMessage.Address ("factory/A/events")
S->>S: user property gobridge.subject = Envelope.Subject
S-->>Route: ACK
Route incoming MQTT messages to different output topics based on headers:
bridge:
id: topic-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://localhost:1883
client_id: topic-router-01
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: mqtt-in
session_id: mqtt-conn
topics:
- topic: "events/#"
qos: 1
senders:
- id: mqtt-out
session_id: mqtt-conn
options:
sender:
qos: 1
bindings:
- id: to-region-topic
sender_id: mqtt-out
# 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: "processed/{region}/{event_type}"
routes:
- id: route-by-region
receiver_id: mqtt-in
delivery_mode: direct_hold
dispatch_mode: single
bindings: [to-region-topic]
policy:
# Exactly one replica consumes this subscription; a second copy of this
# process would double-deliver. See Scenario 8 for fenced ownership.
allow_unfenced: true
A message on events/temperature with headers region: "eu-west", event_type: "temperature" is published to processed/eu-west/temperature.
For more complex routing where different bindings (different senders/queues) are selected based on headers, use MatchByHeader programmatically:
import "github.com/mariotoffia/gobridge/runtime"
// Define bindings for different factories
bindings := []domain.DestinationBinding{
{ID: "bind-factory-a", SenderID: "mqtt-a", Address: "factory/a/orders"},
{ID: "bind-factory-b", SenderID: "mqtt-b", Address: "factory/b/orders"},
}
// Select binding based on "factory" header value
resolver := runtime.NewBindingResolver(bindings,
runtime.MatchByHeader("factory", map[string]string{
"A": "bind-factory-a",
"B": "bind-factory-b",
}),
)
This selects entirely different senders (potentially different MQTT sessions or different queues) based on header values.
| Function | Description | Use Case |
|---|---|---|
MatchByID(id) |
Always selects a specific binding | Static single-target routes |
MatchAll |
Selects every binding | Fan-out to all targets |
MatchByHeader(key, map) |
Maps header value to binding ID | Route by tenant, factory, region |
Custom MatchFunc implementations can inspect any part of the envelope:
// Custom: route by payload content
customMatch := func(env *domain.Envelope, b domain.DestinationBinding) bool {
// Parse payload and make routing decisions
var event map[string]any
json.Unmarshal(env.Payload(), &event)
priority, _ := event["priority"].(string)
if priority == "high" && b.ID == "high-priority-binding" {
return true
}
return b.ID == "default-binding"
}
Combine fan_out dispatch with address templates to send to multiple dynamically-resolved destinations:
bindings:
- id: to-factory-orders
sender_id: mqtt-out
address: "factory/{factory_id}/orders"
- id: to-factory-audit
sender_id: mqtt-out
address: "audit/{factory_id}/events"
routes:
- id: dispatch
receiver_id: sqs-in
delivery_mode: shared_outbox
dispatch_mode: fan_out
bindings: [to-factory-orders, to-factory-audit]
stores:
outbox: { type: memory, options: { acknowledge_volatile: true } }
Each message is sent to both bindings, each with its own rendered address. A message with factory_id: "A" produces:
factory/A/ordersaudit/A/eventsNote: Fan-out requires shared_outbox delivery mode.
How OutboundMessage.Address (resolved from the binding template) is used depends on the transport. In every case, Envelope.Subject is the logical event subject and is carried over the wire either in a native subject field or in the gobridge.subject user-property/header — never as the destination.
| Transport | Where OutboundMessage.Address goes |
Where Envelope.Subject rides |
|---|---|---|
| MQTT | Publish topic. Overrides default_topic; validated for wildcards. |
MQTT user property gobridge.subject |
| AMQP 0-9-1 | Routing key – takes precedence; the sender routing_key is the fallback when Address is empty. |
AMQP header gobridge.subject |
| AMQP 1.0 | Validated against the configured sender link address (mismatch fails fast). Per-address dynamic links are deferred. | Message.Properties.Subject |
| SQS | Reserved for future dynamic queue selection. The queue URL/name remains static on the sender today. | Subject message attribute |
| Azure SB | Reserved for future dynamic entity selection. The queue/topic remains static on the sender today. | Message.Subject |
| HTTP/SSE | Path is static on the sender. | JSON subject field |
For MQTT and AMQP 0-9-1, dynamic addressing is fully effective — each message can land on a different topic / routing key. For SQS, Azure SB, and HTTP, the destination is fixed per sender for now; OutboundMessage.Address is preserved as transport-neutral metadata.
In addition to address templates and programmatic MatchFunc, GoBridge supports a config-driven resolver that selects bindings based on envelope content – headers, subject, or JSON payload fields. This eliminates the need for Go code in most content-based routing scenarios.
flowchart LR
subgraph GoBridge
R[Receiver]
Proc["Processor Chain"]
Resolver["Resolver\n(rules / header_map)"]
SR["SenderRegistry"]
end
subgraph Targets
SA["Sender A\n(bind-a)"]
SB["Sender B\n(bind-b)"]
SD["Sender Default\n(bind-default)"]
end
R --> Proc
Proc --> Resolver
Resolver -->|"binding_id"| SR
SR --> SA
SR --> SB
SR --> SD
style Resolver fill:#ff9,stroke:#333
style SR fill:#9cf,stroke:#333
The resolver evaluates rules against each incoming envelope and returns a binding ID. The SenderRegistry maps binding IDs to senders, allowing a single route to dispatch to entirely different transports or connections per message.
Route messages to different Azure Service Bus queues based on the x-tenant header:
senders:
- id: sb-acme
transport: servicebus
options:
connection:
connection_string: "${ACME_SB_CONN}"
sender:
queue_name: acme-events
- id: sb-globex
transport: servicebus
options:
connection:
connection_string: "${GLOBEX_SB_CONN}"
sender:
queue_name: globex-events
- id: sb-default
transport: servicebus
options:
connection:
connection_string: "${DEFAULT_SB_CONN}"
sender:
queue_name: unrouted-events
bindings:
- id: bind-acme
sender_id: sb-acme
address: acme-events
- id: bind-globex
sender_id: sb-globex
address: globex-events
- id: bind-default
sender_id: sb-default
address: unrouted-events
routes:
- id: tenant-router
receiver_id: http-in
delivery_mode: direct_hold
bindings: [bind-acme, bind-globex, bind-default]
resolver:
type: rules
default_binding: bind-default
rules:
- binding_id: bind-acme
match:
- field: header.x-tenant
operator: eq
value: acme
- binding_id: bind-globex
match:
- field: header.x-tenant
operator: eq
value: globex
Rules are evaluated top-to-bottom. The first rule whose conditions all match (AND logic) wins. If no rule matches, default_binding is used. If neither matches, the message is rejected.
Route based on a field inside the JSON payload using $. path syntax:
routes:
- id: priority-router
receiver_id: sqs-in
bindings: [bind-high-priority, bind-normal]
resolver:
type: rules
default_binding: bind-normal
rules:
- binding_id: bind-high-priority
match:
- field: $.priority
operator: eq
value: high
- field: $.metadata.region
operator: in
value: ["us-west-1", "eu-west-1"]
The payload is parsed lazily on first $. access and cached for all subsequent condition evaluations within the same message.
For simple header-value-to-binding mappings, header_map is more concise than rules:
routes:
- id: factory-router
receiver_id: sqs-in
bindings: [bind-factory-a, bind-factory-b]
resolver:
type: header_map
header_key: factory_id
header_map:
A: bind-factory-a
B: bind-factory-b
The filter processor’s ActionRoute sets a x-bridge.route-override header during the processor chain. The runtime now consumes this header in both direct_hold and shared_outbox delivery modes:
HeaderRouteOverride to a binding IDThis enables processors to override routing decisions dynamically, complementing the config-driven resolver.
SQS queue selection is static – You cannot dynamically choose which SQS queue to send to based on message content. The queue URL is fixed in sender config. To route to different queues, use separate senders with different bindings and a resolver to select between them.
Address templates only resolve from headers – The {placeholder} syntax in binding addresses resolves from envelope headers only, not from JSON payload fields. To use payload values in address templates, first extract them into headers with a transform processor (see Payload-Based Routing via Transform). However, the resolver’s $. conditions can select different bindings based on payload content, and each binding has its own static address – so many payload-routing scenarios do not need address templates at all.
Extract payload fields into headers, then use address templates:
// Step 1: Transform extracts payload field into a header
transformProc, _ := transform.New(transform.Config{
Name: "extract-region",
Mappings: []transform.FieldMapping{
{Source: "$.metadata.region", Target: "header.region"},
},
})
// Step 2: Address template uses the extracted header
// binding address: "events/{region}/processed"
routes:
- id: region-route
receiver_id: mqtt-in
processors: [extract-region]
bindings: [to-region-topic]
bindings:
- id: to-region-topic
sender_id: mqtt-out
address: "events/{region}/processed"
This two-step pattern lets you route based on any JSON payload field.