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.

This is asserted by 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:

• deduplication on an output, and dlq / retry on an input, log a warning and are skipped. The route still starts.
• weak_join on an output is a hard startup error (Unsupported publisher middleware).

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. 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, a poison payload a handler 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/connection errors, then hands a still-failing message on to be dropped (or to a following dlq). 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

Declarative JSON reshaping: field mapping, then schema-directed coercion, defaults and validation — over a single parse. Input and output.

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

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

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.

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.

deduplication

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

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. Entries expire via a MongoDB TTL index; 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 }

When MongoDB is your sink and messages carry a business key, prefer the sink’s own unique index (id_field 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

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.

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.

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. See ARCHITECTURE.md.

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.

switch

Content-based routing: picks a destination by the value of a metadata key.

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" } }

Matching is on the metadata value, not the payload — it does not read JSON fields. To route on payload content, first promote the value into metadata (for example with transform’s on_error: pass_through, which sets mqb.transform_error, or an endpoint that emits a status key such as http_status_code). A message whose key is missing or unmatched goes to default; without a default it is dropped.

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.

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. See ARCHITECTURE.md.

See also

• README.md — overview, data endpoints, request/response and CQRS patterns
• CONFIGURATION.md — full YAML examples, env vars, TLS, IDE schema validation
• ARCHITECTURE.md — internals, batching/concurrency, extension traits