Middleware & Structural Endpoint Reference

Complete listing of every middleware and every structural endpoint mq-bridge ships.

Structural endpoints are the ones that do not talk to a broker or store: they compose other endpoints, shape routing, or terminate a request. Data endpoints (kafka, nats, mqtt, sqlx, …) are covered in README.md and CONFIGURATION.md.

• Middleware
• Structural endpoints

Middleware

Middleware attaches to an endpoint via a middlewares: list, on the input, the output, or both:

my_route:
  input:
    middlewares:
      - deduplication: { sled_path: "/var/lib/mqb/dedup", ttl_seconds: 3600 }
    kafka: { topic: "orders", url: "localhost:9092" }
  output:
    middlewares:
      - retry: { max_attempts: 5 }
      - dlq: { endpoint: { file: { path: "failed.jsonl" } } }
    nats: { subject: "orders.processed", url: "nats://localhost:4222" }

Ordering — read this before combining middleware

Output (publisher) middlewares wrap in list order, so the last entry is the outermost layer and sees the failures of the ones before it. Put dlq last.

Input (consumer) middlewares are applied in reverse, so the first entry is outermost and runs first on an incoming message.

Consequence: a route that reads back what another route wrote needs the reversed list. Writing with [compression, encryption] produces compress(encrypt(payload)); a reader given that same list would try to decrypt first and fail. The reading route must say [encryption, compression]. The lists mirror — they are not copied.

A route handler sits outside every output middleware, so it runs once per message and the middlewares act on what it returned. In particular retry re-attempts only the publish, never the handler — a handler with a side effect fires once however many times the sink is retried. The trade: a dlq cannot capture a handler failure, only a send failure; a handler error propagates to the route and is reported there.

This is asserted by route::tests::test_retryable_handler_error_is_not_retried_by_output_middleware (the handler runs once and retry does not re-run it), route::tests::test_dlq_and_retry_batch_integration, middleware::transform::tests::test_rejected_message_reaches_the_dlq_through_the_config_wiring, and reference_docs_test::publisher_middleware_wraps_last_entry_outermost, and is documented on apply_middlewares_to_publisher in src/middleware/mod.rs.

# Correct: transform rejects -> retry gives up -> dlq captures.
middlewares:
  - transform: { schema_file: "user.json" }
  - retry: { max_attempts: 3 }
  - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }

What exists

Two kinds of compression. The compression middleware compresses each message payload on any transport and decompresses it on the far side. Separately, the batch compression field on the file and object_store endpoints (none / gzip / lz4 / zstd, same compression feature) compresses whole write batches so the file stays decodable with zcat / lz4 -d. Use the field for CLI-readable data at rest, the middleware for over-the-wire payloads. Don’t stack either compression with the encryption middleware on the same route — ciphertext does not compress; for compressed-and-encrypted data at rest use the endpoints’ own compression/encryption fields (compress-then-encrypt per batch).

Putting a middleware on the wrong side behaves in two different ways, so check the table above rather than assuming:

• dlq / retry on an input log a warning and are skipped. The route still starts.
• deduplication, weak_join and id on an output are hard startup errors. Deduplication cannot work on the publish side, and silently starting an un-deduplicated route is worse than refusing to start.

A middleware whose feature is not compiled in (deduplication without dedup, metrics without metrics) is likewise a startup error, not a silent no-op.

retry

Retries failed sends with exponential backoff. Output only.

- retry: { max_attempts: 5, initial_interval_ms: 200, max_interval_ms: 10000, multiplier: 2.0 }

Only Retryable and connection errors are retried; NonRetryable failures pass straight through. Once attempts are exhausted the error is marked so a following dlq treats it as permanent. Pair the two.

dlq

Sends permanently-failed messages to a separate endpoint instead of failing the batch. Output only.

- dlq:
    endpoint:
      file: { path: "dead-letters.jsonl" }

Captures NonRetryable failures and Retryable ones whose retries are exhausted. Connection errors are not dead-lettered — they propagate so the route can reconnect. Nor are handler failures: the handler runs outside the middlewares (see Ordering), so a dlq only ever sees what failed on the way to the sink. The DLQ endpoint is a full endpoint, so it can itself have middleware. If the DLQ send fails with a connection error that error propagates rather than silently dropping the message.

Without a dlq middleware, a message that fails permanently — a data/type error the sink rejects — is logged at error level and dropped, and the route keeps processing the rest of the batch. dlq is the only retention mechanism: retry alone does not retain a permanently-failed message nor prevent it from being dropped — it only re-attempts Retryable errors (a connection error is passed straight through for the route to reconnect on, not retried), then hands a still-failing message on to be dropped (or to a following dlq). This is why a sink that fails with a connection error never reaches its dlq, whether it is a route’s sole output or one leg of a fanout. This tolerate-and-continue policy keeps one bad message from halting the whole stream, but it means a systematic failure (e.g. every row hitting a column-type mismatch) drains the input while committing nothing and still ends completed. Add a dlq to capture the failures for inspection/replay, or watch the route’s logs — a burst of Dropping message … due to non-retryable error is the signal. Note that transient errors are handled separately: several endpoints retry connection/timeout errors internally, and the retry middleware adds backoff on top, so only genuinely permanent errors reach this drop path.

transform

JSON reshaping with field mapping, Zen Expression, and schema processing, in that order.

schema and schema_file are mutually exclusive. A mapping rule is a bare path string or { path, default, required }.

schema must be a JSON object, not a string containing one. A flat key=value middleware syntax (such as the |transform?schema=… form in a connection URI) can only pass strings, so it cannot express schema or any mapping rule beyond a bare path — use schema_file, or move the route into a config file.

- transform:
    mapping:
      firstName: "$.first_name"
      id: "$.user_id"
      "address.city": { path: "$.city", default: "unknown" }
    schema_file: "schemas/user.json"

For calculated output, use expression (available with the zen Cargo feature):

- transform:
    mapping:
      first: "$.first_name"
      last: "$.last_name"
    expression: >-
      { fullName: first + ' ' + last, source: meta.source }

Paths accept $.field, $.a.b, and $.items[0]; the $. prefix is optional. Dots in the output key nest the result. An absent optional source field is omitted rather than emitted as null.

Schema keywords honoured: type, properties, required, default, items, nullable (also "type": ["string","null"]), enum, contentMediaType, contentSchema. Everything else is ignored, so an existing fuller schema can be used as-is. Coercions are limited to the lossless ones: string → integer, string → number, string → boolean (true/false/1/0), number → string.

Empty strings

CSV and many SQL exports spell “no value” as an empty string. coerce_empty_as_null: true reads every "" the schema visits as null, which is then handled like any other null — a nullable field keeps it, a default replaces it:

- transform:
    coerce_empty_as_null: true
    schema:
      type: object
      properties:
        note: { type: string, nullable: true }
        tier: { type: string, default: standard }

note: "" arrives as null and tier: "" as "standard". A field that is neither nullable nor defaulted is rejected, naming the coercion. Only fields the schema declares are affected; " " is not empty.

Embedded JSON

A field carrying a JSON document as a string is decoded by contentMediaType, following JSON Schema 2020-12:

- transform:
    schema:
      type: object
      properties:
        payload:
          type: string
          contentMediaType: application/json
          contentSchema:
            type: object
            properties:
              qty: { type: integer }

The string is replaced by the parsed document, and contentSchema — if given — is applied to it with the same coercion, defaults and validation as anywhere else, so the inner qty: "7" arrives as 7. Without contentSchema the value is parsed but not validated. A root-level schema of this shape decodes a double-encoded message body.

This is not a coercion, and coerce: true never performs it: widening "42" to 42 is lossless, whereas evaluating a string as a document is a parse that can succeed on input never meant as JSON. It is opt-in per field, as the JSON Schema spec requires. Note that the spec treats contentSchema as annotation-only; applying it is the opt-in behaviour it carves out.

Media types ending in +json (and text/json) are decoded too; parameters like ; charset=utf-8 are ignored. A media type we cannot decode, or one paired with a contentEncoding, leaves the string untouched rather than failing. A string that does not parse fails with kind content: transform failed at $.payload [content]: contentMediaType is JSON but the string does not parse: ....

Failures are always non-retryable and name the field, e.g. transform failed at $.items[1].qty [coercion]: cannot coerce string "oops" to integer. On an output endpoint the message is failed so a following dlq captures it; on an input endpoint it is dropped from the batch and acknowledged, keeping invalid data out of the route. on_error: pass_through instead forwards the original payload with the reason in the mqb.transform_error metadata key, which a switch can route on.

Schemas and paths compile once at startup; schema_file is read a single time. A transform with neither stage configured leaves the payload untouched without parsing it.

id

Renders a template into the mqb.id metadata key, giving a message a business identity that survives a re-read. Input only.

The value is a bare interpolation template string (see ${namespace:selector}).

- id: "${payload:order_id}"

Most sources mint a fresh message_id on every read, so it identifies the delivery, not the record — see DELIVERY.md for which ones do carry a stable id. mqb.id fills that gap: derived from the message itself, it is the same on every re-read. Unlike message_id (a u128) it keeps the key as a string, so a sink can use it verbatim, and unlike mqb.src.* it is not stripped on publish — an identity describes the record, not the hop, so it propagates downstream.

Order matters, and not the way it reads. Consumer middlewares wrap in reverse, so the entry closest to the end of the list touches an incoming message first. Anything that consumes mqb.id must therefore be listed before the id that produces it:

- deduplication: { store: "sled:///var/lib/mq-bridge/dedup", ttl_seconds: 3600, key: "${metadata:mqb.id}" }
- id: "${payload:order_id}"

Reversing those two leaves mqb.id unset when deduplication reads it, which falls back to message_id with only a warning. Pinned by middleware::id::tests::the_last_listed_consumer_middleware_runs_first.

A partial identity is no identity. The key is set only when every selector in the template resolves; if any one is missing the message passes through with mqb.id unset (warned once per route, then at debug). This matters for multi-part templates: "${payload:tenant}-${payload:order_id}" would otherwise render "acme-" for every message missing order_id and hand them all the same identity — which, used as a deduplication key, drops all but the first.

Malformed templates fail at startup, and so does a template with no ${...} token at all, since a constant would give every message one identity.

filter

Keeps only the messages for which an expression is true; the rest are dropped. Input and output. Requires the filter feature (pulls the zen-expression engine), which is part of middleware/full but not of portable.

The value is a bare expression string:

- filter: "amount > 100"

Put it on the input whenever you can. A filter on the input drops the message before the rest of the pipeline touches it, and acknowledges it at the source; on the output the message has already paid for the whole route.

When filtering splits a full input batch, the consumer reads additional full source batches until it refills the requested batch size. A naturally short source batch remains a flush boundary, so live routes do not wait indefinitely merely to fill a batch. This lets sinks such as MongoDB continue using bulk writes after filtering without adding input buffer middleware.

What an expression can read:

• Payload fields by bare name, including nested paths — amount, order.status. The payload must be a JSON object; anything else produces a per-message error and fails the batch. Indexed paths such as items[0].qty are unsupported: an expression that uses one is rejected at startup rather than silently dropping every message.
• Metadata under the reserved meta. prefix — meta.http_status_code, meta.kind. Metadata is always text, so a numeric comparison needs an explicit cast: number(meta.http_status_code) >= 400.

If an expression names no payload field at all, the payload is never parsed — a metadata-only filter costs no JSON decode.

- filter: "order.status == \"open\" and number(meta.retry_count) < 3"

&& and || are rewritten to and / or for you, so both spellings work.

A field that is absent is supplied to the expression as null, so an or branch or negation can still match. A null or non-scalar field (an array or object where the expression expects a scalar) logs a warning. A payload that is not a JSON object, or an expression that does not evaluate to a boolean, is an error and fails the batch; those are configuration mistakes, and dropping every message would hide them.

To send the non-matching messages somewhere instead of discarding them, use switch’s when mode rather than a filter.

With an object_store sink on name_by: auto, the route switches to write_time names. A source-range name covers one contiguous run of source positions, so a batch with holes punched in it would be written as one object per surviving run — a filter keeping 80% of rows turns one upload into roughly a hundred. The route logs one line at startup saying it made the switch. The same applies to every other middleware that removes messages from a batch (deduplication, weak_join, transform with on_error: reject) and to a switch in when mode with no default. Set name_by: source_position explicitly to keep replay-safe names and accept the fragmentation.

deduplication

Drops messages whose key was already seen within the TTL. Input only. Requires the dedup feature (pulls sled).

key is an interpolation template (see ${namespace:selector}), typically "${payload:order_id}". Without it the key is the message_id, which most sources regenerate on every read — so re-reading the same source deduplicates nothing and only in-flight redeliveries are suppressed. Set key to a business key whenever you need dedup to survive a re-read.

store selects the backend by URL scheme:

• sled:///path (or a bare path) — a local sled database; per-process, not cluster-wide.
• mongodb://host/db[/collection] — a shared collection, so multiple instances of a route deduplicate against one another. Requires the mongodb feature. Expiry is judged on read, so a ttl_seconds boundary is honoured exactly; the TTL index only reclaims space afterwards (MongoDB’s sweep can lag by up to a minute). The collection defaults to mqb_dedup_<route>. Point it at the same deployment your sink already uses to avoid running extra infrastructure.
• postgres|mysql|mariadb|sqlite://…[/table] — a shared SQL table (dedup_key PK, expire_at), so multiple instances deduplicate against one another. Requires the sqlx feature. SQL has no native TTL, so expired rows are swept periodically; the table defaults to mqb_dedup_<route>.

sled_path is the legacy spelling of a local sled store and is equivalent to store: "sled://<path>".

- deduplication: { store: "sled:///var/lib/mq-bridge/dedup", ttl_seconds: 3600 }

- deduplication: { store: "mongodb://localhost:27017/etl", ttl_seconds: 3600 }

- deduplication: { store: "postgres://user:pass@localhost/etl", ttl_seconds: 3600 }

- deduplication: { store: "sled:///var/lib/mq-bridge/dedup", ttl_seconds: 3600, key: "${payload:order_id}" }

When MongoDB is your sink and messages carry a business key, prefer the sink’s own unique index (id_field, which also accepts templates, on the mongodb output) over this middleware — the target collection then is the deduplication authority, with no second write. See the idempotency notes in README.

weak_join

Correlates messages by a metadata key and emits them as one joined message. Input only.

# Count mode: wait for any 3 messages sharing a correlation_id, emit a JSON array.
- weak_join: { group_by: "correlation_id", expected_count: 3, timeout_ms: 5000 }

# Branch mode: wait for named branches, emit a branch-keyed JSON object.
- weak_join:
    group_by: "correlation_id"
    expected_count: 2
    timeout_ms: 5000
    branch_by: "source"
    required: ["inventory", "pricing"]
    on_timeout: discard

group_by reads message metadata only — never the payload. A message that lacks the key falls into a shared "default" group, so a mistyped key or a source that never sets it joins unrelated messages instead of failing. If the value lives in the payload, lift it into metadata first (a transform mapping, or the source’s own metadata options).

Setting branch_by switches to branch mode, where required overrides expected_count. On timeout an incomplete group is either emitted partially (fire) or dropped (discard). Messages are acknowledged on receipt, so a crash before the group completes loses the buffered members.

buffer

Accumulates single sends and forwards them as one batch. Input and output.

- buffer: { max_messages: 500, max_delay_ms: 20 }

Flushes when either bound is hit. Useful in front of an endpoint whose per-call overhead dominates. Adds up to max_delay_ms of latency.

With route concurrency greater than 1, buffering preserves order inside each batch but does not guarantee source order across concurrent destination writes. Use concurrency: 1 when destination order matters; route validation emits a warning for this combination.

limiter

Paces throughput to a target rate. Input and output.

- limiter: { messages_per_second: 250 }

Best-effort pacing that accounts for batch size, not just call count.

delay

Sleeps a fixed duration before each receive or send. Input and output.

- delay: { delay_ms: 100 }

Mainly for testing and for crude pacing of a downstream system; prefer limiter for real rate control.

cookie_jar

Persists HTTP cookies and arbitrary session values across messages. Input and output.

- cookie_jar:
    shared_scope: "login-session"
    capture_metadata_keys: ["x-csrf-token"]
    export_metadata_prefix: "session."

Reads set-cookie from responses and injects cookie into later requests. With shared_scope, instances using the same name share one store across endpoints and routes in the process — that is how a login route and a data route reuse one session.

encryption

Encrypts each message payload into a self-describing AEAD envelope on the output side and decrypts it on the input side. Metadata and routing keys stay in the clear. Input and output. Requires the encryption feature.

- encryption: { key: "${env:MQB_ENC_KEY}" }

The envelope records the cipher and key_id, so key rotation works by sealing with a new key_id/key while listing the old key under decrypt_keys on the consuming side. Each payload is authenticated independently: any bit-level tampering, a torn frame, or a missing/wrong key is a hard consumer error, not a silent drop. The AEAD binds only the payload (empty associated data): metadata and routing keys are not authenticated against the ciphertext, since they are not guaranteed to survive transport round-trips (many endpoints regenerate the message_id or drop kind). A sealed payload can therefore be replayed under different metadata; use the deduplication middleware or a sink uniqueness constraint if that matters. Note that this authenticates each payload, not the file as a whole — like any append-structured file, an at-rest file that loses whole trailing frames (truncation at a frame boundary) reads back as a shorter stream with no error, so rely on the consumer’s checkpoint/cursor for completeness rather than on the encryption layer.

Do not combine this middleware with a sink’s batch compression on the same route: ciphertext does not compress. For compressed and encrypted data at rest, use the file / object_store endpoints’ own fields instead, which apply compress-then-encrypt per batch:

output:
  file:
    path: "data.enc"
    format: raw
    compression: lz4          # none | gzip | lz4 | zstd  (`compression` feature)
    encryption: { key: "${env:MQB_ENC_KEY}" }

Both endpoints accept the same compression and encryption fields (object_store derives its default object extension from them, e.g. .jsonl.gz / .jsonl.lz4, and adds a trailing .enc when encryption is on since the object is ciphertext, not a directly decompressible .gz). An encrypted file is written as length-prefixed sealed frames (one per batch) and is only readable through a matching consumer; a compressed-only file stays a standard .gz/.lz4 stream. File compression/encryption supports only the default consume mode. csv works too: the header row is written into the first member, so the decoded stream is a normal CSV file.

A file source must declare the same compression/encryption the data was written with. A mismatch (wrong key, wrong codec, or a missing field) is a permanent decode failure: the route ends failed with the error in its status, rather than completing as if the file were empty. Reading a compressed file with no compression set is likewise rejected up front by sniffing the leading magic bytes, so raw compressed bytes are never emitted as messages.

f64 precision. Numbers move through payloads as JSON. serde_json’s default parser shifts ~1 ULP on ~19% of 17-significant-digit doubles, so a postgres → file → postgres hop of a double precision column can change the last bit. Build with the float-roundtrip feature for bit-exact float parsing across every endpoint (it trades a little parse speed for it).

compression

Compresses each message payload on the output side and decompresses it on the input side. Metadata and routing keys are untouched. Input and output. Requires the compression feature.

- compression: { algorithm: zstd }

Each payload is compressed independently into a single self-contained member, so this works over any transport, not just files. algorithm: none is a passthrough. A truncated or corrupt frame is a permanent consumer error (the poison message is not re-read indefinitely), as is a payload that exceeds max_decompressed_bytes. Put the same algorithm on both the input and output side of a route.

Unlike the file / object_store batch compression field — which keeps whole write batches decodable with zcat / lz4 -d — this middleware frames per message and is only readable through a matching consumer. Do not combine it with the encryption middleware (ciphertext does not compress); for compressed-and-encrypted data at rest, use the endpoints’ own compression/encryption fields instead.

metrics

Emits throughput, latency and error metrics for the endpoint. Input and output. Requires the metrics feature. Takes no options; its presence enables collection.

- metrics: {}

Input and output are labelled separately, so attaching it to both sides is meaningful.

random_panic

Deliberate fault injection for testing recovery paths. Input and output.

- random_panic: { mode: disconnect, trigger_on_message: 500 }

disconnect and timeout produce retryable errors; json_format_error produces a non-retryable one — useful for exercising a dlq. Keep enabled: false in committed configs rather than deleting the block.

On the input side, json_format_error/nack never call the real consumer at all — they substitute a synthetic message (or error) on every triggered receive. Leaving trigger_on_message unset means every poll is faulted, so the real source is never read and exit_on_empty/--drain never sees the empty batch it waits for — the route runs forever, manufacturing synthetic messages. Always set trigger_on_message to a specific count when testing a drain-mode route with input-side fault injection. And since dlq/retry on an input are no-ops (see above), pair an input-side fault with a real assertion on the consumer’s recovery, not a dlq.

The middleware block alone is not enough: fault injection is gated per route by allow_fault_injection, which defaults to false. Copying only the snippet above leaves the middleware inert (the route logs that it is disabled). A complete, working configuration:

flaky_test_route:
  allow_fault_injection: true
  input:
    memory: { topic: "in" }
    middlewares:
      - random_panic: { mode: disconnect, trigger_on_message: 500 }
  output:
    memory: { topic: "out" }

allow_fault_injection: true is intended for test configurations only. Do not enable it — or the random_panic middleware — in production configs.

custom (middleware)

Delegates to a factory you registered programmatically.

- custom:
    name: "my_enricher"
    config: { lookup_url: "http://enrich.internal" }

Implement CustomMiddlewareFactory (apply_consumer and/or apply_publisher, each defaulting to pass-through) and register it before starting routes. It can also be written in Python (register_middleware, hooks on_receive / on_send) or JavaScript (registerMiddleware, hooks onReceive / onSend). See EXTENDING.md for the full guide.

Structural endpoints

These appear wherever an endpoint is expected — as a route input/output, or nested inside another structural endpoint.

They live under src/endpoints/structural/, and each of the variants above carries "format": "structural_endpoint" in the generated JSON schema (mq-bridge.schema.json), so external tooling can tell them apart from the transport endpoints.

ref

Reuses an endpoint registered under a name, instead of repeating its configuration.

The name is a registry key, not a topic name. Register it from Rust before starting the routes:

use mq_bridge::models::Endpoint;
use mq_bridge::route::register_endpoint;

register_endpoint("common_queue", Endpoint::new_memory("shared_memory_topic", 100));

enrich:
  input: { ref: "common_queue" }
  output: { nats: { subject: "enriched", url: "nats://localhost:4222" } }

A route can also publish its own output under a name with Route::register_output_endpoint(Some("name")), which is how one route’s output becomes another’s input.

The value is a bare string. Resolution looks in the endpoint registry first, then in registered publishers. Middleware on the ref itself is applied outside the referenced endpoint’s own middleware. Circular references are detected and rejected at startup, and nesting depth is bounded.

fanout

Publishes each message to every listed endpoint. Output only.

output:
  fanout:
    - kafka: { topic: "audit", url: "localhost:9092" }
    - file: { path: "audit.jsonl" }
    - nats: { subject: "audit", url: "nats://localhost:4222" }

The value is a plain list of endpoints, each of which may have its own middleware and may itself be structural. All branches receive the same message.

A fan-out can also reply. If one of the branches produces a response — a response or static leg, or a request whose forward_to replies — that response is returned to the caller, so a request/reply input can fan its message out and still answer. Branches that must not answer need a forward_to that does not reply ({} is the null endpoint, which discards).

# Mirror every call to staging, but answer the caller from production only.
proxy:
  input: { http: { url: "0.0.0.0:8443", path: "test" } }
  output:
    fanout:
      - request:
          to: { http: { url: "http://127.0.0.1:1444/" } }
          forward_to: {}                 # discard staging's response
      - request:
          to: { http: { url: "http://127.0.0.1:1445/" } }
          forward_to: { response: {} }   # only this one replies to the caller

A caller has one reply channel, so only one branch may answer a given message: if several do, the first in list order wins and the others are dropped. That is a configuration mistake which would otherwise repeat on every message, so the route warns once and logs later drops at debug level.

A branch that fails nacks the whole fan-out: every branch is delivered at-least-once, so the answering branch’s response is discarded and the caller gets a 500 rather than an answer that hides a lost message. That is stricter than nginx’s mirror, which ignores failed mirror subrequests entirely.

The mirror pattern above is unaffected, because a request branch with a non-replying forward_to absorbs its own failure — that is where nginx’s “ignore the mirror” semantics live, opted into per branch. For a plain branch (- kafka: { … } directly in the list) the equivalent is a dlq on that branch: the failure is parked, the branch acks, and the answering branch still replies.

proxy_with_parked_mirror:
  input: { http: { url: "0.0.0.0:8443", path: "test" } }
  output:
    fanout:
      - kafka: { topic: "audit", url: "localhost:9092" }
        middlewares:
          - dlq: { endpoint: { file: { path: "audit-failures.jsonl" } } }
      - request:
          to: { http: { url: "http://127.0.0.1:1445/" } }
          forward_to: { response: {} }

switch

Content-based routing: picks one destination per message. Two modes, and a switch uses exactly one of them — naming both, or neither, is a startup error.

Value lookup matches a metadata value exactly. It is a HashMap get and never reads the payload, so prefer it when the routing key already is metadata.

output:
  switch:
    metadata_key: "http_status_code"
    cases:
      "200": { nats: { subject: "ok", url: "nats://localhost:4222" } }
      "404": { file: { path: "not-found.jsonl" } }
    default: { file: { path: "other.jsonl" } }

Predicate mode routes on an expression, so it can branch on payload content directly. Cases are evaluated in order and the first match wins, which is what makes overlapping thresholds safe to write:

output:
  switch:
    when:
      - if: "amount > 100"
        to: { kafka: { topic: "large-orders", url: "localhost:9092" } }
      - if: "amount <= 100"
        to: { nats: { subject: "small-orders", url: "nats://localhost:4222" } }
    default: { file: { path: "unrouted.jsonl" } }

if takes the same expression language as the filter middleware — payload fields by bare name (amount, order.status), metadata under meta. and always as text (number(meta.http_status_code) >= 400), and/or or &&/||. Predicate mode therefore needs the filter feature; a when list in a build without it is a startup error, not a silent fallback. A payload the expression cannot read fails the send rather than dropping the message silently. As with filter, indexed payload paths such as items[0].qty are unsupported and are rejected at startup.

In either mode, a message that matches nothing goes to default; without a default it is dropped with a warning. Value lookup is the cheaper mode and stays the right choice when the key is already in metadata — for payload-derived keys you can either promote the value into metadata first (for example with transform’s on_error: pass_through, which sets mqb.transform_error) or just use when.

request

Sends each message to a request-capable endpoint and forwards the response somewhere else, turning a request/reply exchange into a one-way flow.

output:
  request:
    to: { http: { url: "https://api.internal/score" } }
    forward_to: { ibmmq: { queue: "RESULTS", url: "mq(1414)", queue_manager: "QM1", channel: "APP.SVRCONN" } }

to must support request/reply: http, or a nats/mongodb/memory endpoint with request_reply: true. On error or timeout the original message is forwarded instead of a response, so nothing is lost — distinguish the two downstream with a switch on a status key such as http_status_code.

For batch input, requests still run individually because each needs its own reply. They run concurrently unless to requires ordered publishing, in which case they are issued one at a time in source order. Their responses and error fallbacks are restored to input order and passed to forward_to in one send_batch call. Batch-capable sinks such as MongoDB can therefore use their native bulk write for the forwarding leg.

Whatever forward_to returns is passed back up. A plain sink acks, forward_to: {} (the null endpoint, also spelled null) discards, and forward_to: { response: {} } replies to the origin of the current request — which is how a fanout branch answers the caller.

The error fallback never becomes that reply: when forward_to would answer the caller, a failed request surfaces the error instead of echoing the original back as a success. The route then nacks (HTTP 500), and a retry or dlq middleware on the endpoint sees the failure as usual. Forwarding-to-a-sink still acks, so the switch pattern above is unchanged.

response

Replies to the origin of the current request. Output only, and the recommended way to build request/reply routes.

http_echo:
  input: { http: { url: "0.0.0.0:8080" } }
  output: { response: {} }

Takes no options. Requires an input that carries a reply channel (http, websocket, grpc, or a request/reply nats/mongodb/memory). With an http or websocket input and no middleware, response (and static) enables an inline fast path that skips the normal route pipeline. See README.md.

reader

An output endpoint that ignores the incoming payload and instead reads one message from the wrapped consumer, returning it as the response. The inbound message is purely a trigger.

# HTTP GET pulls the next message off a Kafka topic.
poll_api:
  input: { http: { url: "0.0.0.0:8080", method: "GET" } }
  output:
    reader:
      kafka: { topic: "queue", url: "localhost:9092" }

The value is a single nested endpoint, which must be valid as a consumer. The message read is acknowledged immediately, before the caller has necessarily received it — so a crash in between loses it. Use it for polling APIs, not for guaranteed delivery.

static

A fixed, pre-rendered message. Usable as an output (a constant reply) or an input (a constant source).

Accepts either a bare string or the full map form:

output: { static: "OK" }                       # shorthand, body JSON-encoded

output:
  static:
    body: '{"status":"ok"}'
    raw: true                                  # send verbatim, do not JSON-encode
    metadata: { content-type: "application/json" }

raw: true sends body byte-for-byte; the default JSON-encodes it as a string. Like response, a static output enables the HTTP inline fast path.

Placeholders

body is a template compiled once at startup; rendering a message never re-parses it. Tokens use the ${namespace:selector} form:

payload/metadata/message read the request, so they are the useful ones on an output (e.g. an error reply that echoes the request); on an input (load-test source) only gen/env produce values. When the body’s content-type metadata is a JSON type, interpolated request values are JSON-escaped by default so external data cannot break the structure — append | raw to a token to splice it verbatim. To emit a literal, un-interpolated ${…}, write $${…} (a bare $$ is left as-is); any ${…} with an unknown namespace is also left untouched.

output:
  static:
    body: '{"error":"not found","id":"${message:id}","at":"${gen:now}"}'
    raw: true
    metadata: { content-type: "application/json" }

stream_buffer

An in-memory stream partitioned by correlation ID, used to carry streaming request/response bodies between routes.

output:
  stream_buffer: { topic: "responses" }        # publisher: no correlation_id

input:
  stream_buffer: { topic: "responses", correlation_id: "req-123" }   # consumer

A consumer without correlation_id is a startup error; a publisher with one logs a warning and ignores it. Primarily wired up via HttpConfig::stream_response_to.

null

Discards every message. Output only. This is the default output when a route omits one.

drain:
  input: { kafka: { topic: "noisy", url: "localhost:9092" } }
  output: null          # a bare YAML null

Spelling trap: it is a bare YAML null (or ~, or the explicit null: null). null: {} does not parse. Omitting output: entirely gives the same result.

Useful for consume-and-handle routes where a handler does the work and there is nothing to forward, and for benchmarking an input in isolation.

custom (endpoint)

Delegates to a factory you registered programmatically.

output:
  custom:
    name: "my_sink"
    config: { target: "internal://thing" }

Implement CustomEndpointFactory and register it before starting routes. Once registered, the name also works as a bare endpoint key — input: { my_sink: {...} } — since any unrecognised key is looked up in the custom-endpoint registry. Use the explicit custom: form above if you validate configs against mq-bridge.schema.json, which cannot know your key. Endpoints can also be written in Python (register_endpoint) or JavaScript (registerEndpoint). See EXTENDING.md for the full guide.

See also

• README.md — overview, data endpoints, request/response and CQRS patterns
• CONFIGURATION.md — full YAML examples, env vars, TLS, IDE schema validation
• DELIVERY.md — delivery guarantees, per-source identity, per-sink idempotency
• ARCHITECTURE.md — internals, batching/concurrency, extension traits
• EXTENDING.md — writing your own endpoint or middleware, in Rust, Python or Node