Introduction

crossing streams

mq-bridge-app is a fast, single-command ETL and data-movement tool built in Rust — and, on top of the same engine, a multi-protocol bridge and traffic workbench for messaging.

It ships in three forms that share one engine and one config format: a desktop app (visual workbench), a CLI / server (headless bridge and one-line copy), and a library (embed the engine in Rust, Python, or Node.js). Design a route once, run it anywhere — no rewrite in between.

Supported integrations include Kafka, RabbitMQ (AMQP), NATS, AWS SQS, MQTT, IBM MQ (optional), HTTP, gRPC, ZeroMQ, MongoDB, Redis Streams, ClickHouse, Postgres CDC, sqlx (MySQL, MariaDB, PostgreSQL, SQLite), cloud object storage, and filesystem endpoints.

A quick taste

At its core is a zero-config copy command that moves data between databases, queues, and files in a single line of bash — no YAML, no pipeline definition, no code:

mq-bridge-app copy \
  --from 'postgres://user:pass@localhost/db?table=src&sslmode=disable' \
  --to   'file://out.jsonl?format=raw' \
  --drain

The scheme selects the endpoint and query parameters configure it, so any source→sink pair is just one URL each. And it’s quick: in benchmarks a 1,000,000-row Postgres → JSONL job sustained 266,951 rows/s at ~20 MiB peak RSS — see Performance tuning.

Philosophy

The project has one main bias: move data reliably without forcing the rest of the application to care too much about the transport. Kafka offsets, RabbitMQ nacks, HTTP responses, MongoDB polling, WebSocket frames, and file rows are all different in real life, but route code should still be able to receive a batch, process it, publish it, and commit it.

• Fast by default. Every endpoint is optimized around batch-shaped APIs, and the headless surfaces ship tuned for throughput: the copy CLI and the MCP server default to batch_size: 1024 and concurrency: 4. The low-level library/config primitive defaults to batch_size: 1, concurrency: 1 (opt in per route) so embedded routes stay predictable until you raise them — usually the first knob to reach for when throughput matters.
• Reliability is built in, not bolted on. Retries, dead-letter queues, deduplication, rate limiting, and cookie/session persistence wrap any endpoint. Ack/nack behaviour and retry/DLQ handling were designed to work with batching, including commit sequencing for cumulative-ack brokers.
• Not a framework. It is not a domain framework, an actor runtime, or a full stream processor. It cares about transport, routing, and delivery behaviour, not about prescribing your domain model.

Where to go next

• New here? Start with Installation and the Quick start.
• Want the end-to-end walkthroughs? See the Tutorials.
• Looking for a specific task? The Cookbook has short recipes.
• Need exact fields and defaults? The Reference is authoritative.
• Running it in production? See Operations, especially the Performance tuning page.
• Driving it from an AI agent? The same binary is an MCP server — the rows move without entering the model’s context.

App vs. engine

mq-bridge is the engine/library; mq-bridge-app is the application — desktop app + CLI/server + library distribution — built on that engine. This book is the user-facing home; the engine’s deep API reference lives on docs.rs. The library bindings let you embed the same engine in Rust, Python, or Node.js.

Install

mq-bridge-app ships in three forms that share one engine and one config format: the CLI / server, the desktop app (UI), and the library. Pick the install path for the form you need. To compile any of them yourself, see BUILD.md.

• CLI / server — Homebrew, cargo binstall, cargo install, Docker
• Desktop app (UI) — Homebrew cask, or a prebuilt bundle
• Library — Rust, Python, Node.js

CLI / server

The CLI (mq-bridge-app) is a single headless binary.

Homebrew (macOS, Linux) — recommended

brew install marcomq/tap/mq-bridge-app
brew upgrade mq-bridge-app     # later, to update

Prebuilt bottles cover Apple Silicon macOS and x86_64 Linux. Homebrew refreshes the tap only on brew update, so if brew upgrade doesn’t pick up a new release yet, run brew update first.

cargo binstall — prebuilt binary

Downloads the prebuilt CLI from Releases instead of compiling from source, so it installs in seconds:

cargo binstall mq-bridge-app

Prebuilt binaries are available for x86_64 Linux, Apple Silicon macOS, and x86_64 Windows. (cargo-binstall is a drop-in cargo install replacement.)

cargo install — from source

Requires a Rust toolchain and compiles all supported endpoint client libraries (except IBM MQ), so it may take a while:

cargo install mq-bridge-app

For IBM MQ, install the client library first and build with --features=ibm-mq.

Docker

The CLI is published as a multi-arch image (amd64 + arm64):

docker run --rm --name mq-bridge -p 9091:9091 ghcr.io/marcomq/mq-bridge-app:latest

To read+tail from input.log and forward its content, mount the working directory at /app and seed the config from one of the templates baked into the image at /config:

touch input.log
docker run --rm --name mq-bridge -p 9091:9091 -v "$(pwd)":/app \
  ghcr.io/marcomq/mq-bridge-app:latest --init-config=/config/file-to-http.yml

Note

The default latest image is a plain multi-arch image for amd64 and arm64. IBM MQ support is published separately as the latest-ibm-mq and ibm-mq tags on amd64 only, since there is no redistributable IBM MQ client library for arm64 yet. Start that image in emulation mode with --platform=linux/amd64, or build mq-bridge-app yourself with cargo build --release --features=ibm-mq.

Desktop app (UI)

The desktop app is a Tauri bundle of the full messaging workbench — the same UI the CLI serves in a browser, only packaged as a native app.

Homebrew cask (macOS, Apple Silicon)

brew install --cask marcomq/tap/mq-bridge

Homebrew quarantines cask apps by default and the desktop binaries are not notarized yet, so macOS blocks the app on first launch. See the note below for how to open it (via System Settings or Terminal).

Prebuilt bundles (macOS, Windows, Linux)

Bundles for every platform are attached to each release on the GitHub Releases page.

• macOS — download the .dmg / .app bundle.
• Windows — download the installer or standalone executable.
• Linux — download the bundle that suits your distribution: AppImage, .deb, .rpm, or the unpacked archive.

Note

The desktop binaries are currently not notarized, so on macOS Gatekeeper blocks the app on first launch — it’s reported as “damaged” or “cannot be opened because the developer cannot be verified”. This applies to both the Homebrew cask and a downloaded bundle. There are two ways to open it:

Option 1 — System Settings (no Terminal). Try to open the app once so the block is triggered, then go to System Settings → Privacy & Security, scroll to the Security section, and click “Open Anyway” next to the mq-bridge message. Authenticate, then launch the app again.

Option 2 — Terminal. Remove the quarantine attribute directly:

# app in /Applications (where the cask and the .dmg install it)
sudo xattr -rd com.apple.quarantine /Applications/mq-bridge.app
# app in a user-owned directory (e.g. ~/Downloads) — no sudo needed
xattr -rd com.apple.quarantine ~/Downloads/mq-bridge.app

If macOS says the app is “damaged”, the “Open Anyway” button may not appear — use the Terminal method in that case.

Library

Embed the core engine in your own code — produce or consume messages with a unified API, one config format across all three bindings:

• Rust — mq-bridge (cargo add mq-bridge)
• Python — pip install mq-bridge-py
• Node.js — npm install mq-bridge

The core of the library are the MessageConsumer and MessagePublisher traits, found in mq_bridge::traits.

Build from source

Building the CLI/server, the desktop (Tauri) app, or the Docker image from source is covered in BUILD.md. For IBM MQ specifically, see the IBM MQ Setup Guide.

Quick Start

mq-bridge-app copy moves data between two endpoints described as URIs. The scheme picks the connector; ?query=params configure it.

mq-bridge-app copy \
  --from postgres://localhost/app?table=users \
  --to 'clickhouse://localhost:8123?table=users&database=analytics'

This copies every row currently in app.users (PostgreSQL) into analytics.users (ClickHouse). No config file, no UI — just a source and a destination. Add --drain to exit once the source is empty instead of running as a continuous bridge (see Continuous vs. one-shot below).

The five sections below are complete, working commands. Each links to the full connector page for that endpoint, which lists every available option; the generated URL reference is the authoritative source for every parameter’s type, default, and description.

PostgreSQL → ClickHouse

mq-bridge-app copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to 'clickhouse://localhost:8123?table=orders&database=analytics'

Reads all rows from the orders table and bulk-inserts them into ClickHouse’s HTTP interface. Add &cursor_column=id&cursor_id=orders_sync on --from to make this resumable instead of a one-shot batch. See PostgreSQL and ClickHouse.

PostgreSQL CDC → PostgreSQL

mq-bridge-app copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to 'postgres://user:pass@otherhost/replica?table=orders&auto_create_table=true'

Streams inserts/updates/deletes from a PostgreSQL logical-replication publication into another PostgreSQL table, continuously (CDC is a change stream, so this command doesn’t drain — run it as a long-lived process). See PostgreSQL CDC.

MQTT → Kafka

mq-bridge-app copy \
  --from mqtt://broker.local:1883?topic=sensors/+/temperature \
  --to kafka://kafka.local:9092?topic=sensor-readings

Subscribes to an MQTT topic (wildcards supported) and republishes every message to a Kafka topic, continuously. See MQTT and Kafka.

RabbitMQ → HTTP

mq-bridge-app copy \
  --from rabbitmq://guest:guest@localhost:5672/%2f?queue=orders \
  --to http://internal-api.local/ingest?method=POST

Consumes messages from a RabbitMQ queue and POSTs each one to an HTTP endpoint, continuously. See RabbitMQ and HTTP.

File (CSV) → MongoDB

mq-bridge-app copy --drain \
  --from file:///data/customers.csv?format=csv \
  --to 'mongodb://localhost?database=app&collection=customers'

Reads a CSV file (first row = header) and inserts one document per row into a MongoDB collection, then exits since the source is a finite file. See File and MongoDB.

Continuous vs. one-shot

Without --drain, copy runs as a continuous bridge until Ctrl-C — the right mode for message brokers (MQTT, Kafka, RabbitMQ) and CDC sources, which never “end”. With --drain, copy exits once the source yields an empty batch — the right mode for finite sources (a file, or a full-table read from a database). --concurrency and --batch_size tune throughput on both modes.

Escape hatch: driver options and full connection strings

Any query parameter that isn’t a recognised config field (e.g. sslmode, replicaSet) is left on the connection URL untouched, so driver-specific options just work — including object-typed fields like tls, which can never be set from a single scalar query param and so always stays on the URL (e.g. mongodb://host/?tls=true&database=appdb passes tls=true straight through to the MongoDB driver). If you already have a complete connection string (copied from elsewhere, or one whose own options would otherwise be mis-parsed as config), skip decomposition entirely and pass it verbatim with ?url=<url-encoded string>:

mq-bridge-app copy \
  --from 'mongodb://_/?url=mongodb%3A%2F%2Fuser%3Apass%40host%2Fdb%3Ftls%3Dtrue&collection=orders' \
  --to null:

See the generated reference for each connector’s recognised field names.

The three ways to run it

mq-bridge-app is one engine with one config format, exposed three ways. Build and test a route in the UI, export the JSON/YAML, then run that config in a config-mode service or from library code. The copy CLI takes the same settings, but expressed as endpoint URIs and flags rather than a config file.

See Installation for every install method and platform.

Desktop app (UI)

The desktop app is a Tauri bundle of the full messaging workbench: manage publishers/consumers/routes, run request/response traffic (like Postman for REST), inspect message history, and import Postman/OpenAPI/AsyncAPI definitions. It is the same UI the CLI serves in a browser — only the packaging differs.

The UI is generated dynamically from the Rust configuration structures: the backend uses schemars to produce a JSON Schema for the AppConfig struct (exposed at /schema.json, also mq-bridge-app --schema), and the frontend renders a complete config form from that schema. So when a new middleware or option is added to the engine, the schema updates automatically and the UI reflects it with no frontend change.

CLI / server

The CLI (mq-bridge-app) is a headless binary that runs in three modes. They share the same engine and config format but differ in how you drive them — and only config mode serves the browser UI:

# Config mode: run a long-lived bridge from a file
mq-bridge-app --config config.yml

# Seed config.yml from a template on first run only
mq-bridge-app --config config.yml --init-config dev/config/file-to-http.yml

# Start empty, then open the UI to build your config
mq-bridge-app

In config mode the CLI also serves the browser UI (the same UI as the desktop app) on the configured port. See the CLI reference for every flag, and Configuration grammar for the config format.

The configuration-first workflow

The point of one shared config format is that you can test connections and dial in a route in the UI, export the JSON/YAML, then run that exact config unchanged — as a config-mode service or loaded from library code. A known-good route shape from the UI drops straight into production.

copy is the exception: it takes no config file, so a route you built in the UI has to be mapped by hand onto --from / --to endpoint URIs and route flags. The settings are the same, only the way you pass them differs.

Library

Beyond running standalone, the core engine is available as a library so you can produce or consume messages with a unified API — no broker-specific SDK, one config format across all three bindings:

• Rust — mq-bridge (cargo add mq-bridge)
• Python — pip install mq-bridge-py
• Node.js — npm install mq-bridge

The core of the library are the MessageConsumer and MessagePublisher traits in mq_bridge::traits. See the Language bindings API and the Embed the library tutorial.

How the UI differs from API clients

The UI overlaps with API clients like Postman, Bruno, and Insomnia, but its centre of gravity is different: it is designed around message bridging, runtime operation, and long-lived route management rather than just request composition. It adds broker pub/sub workflows, long-lived consumers/routes, bridging traffic between protocols, hex-level payload debugging, replay, local-first git-friendly config, and encrypted config — while leaving scripting and complex request workflows to the dedicated API clients. Use an API client when your main job is crafting and sharing API requests; use mq-bridge-app when you need to connect systems, move messages between protocols, inspect live traffic, and manage bridge-style runtime configuration.

The desktop / web UI

mq-bridge-app ships a visual workbench for building, testing, and running message routes. It is the same UI whether you launch the desktop app (a Tauri bundle) or let the CLI serve it in a browser — only the packaging differs. Think of it as Postman for message bridging: build and test a route, export the JSON/YAML, then run that exact config unchanged wherever you deploy.

For how the UI is built (schema-driven forms, the engine acting as its own web server), see How the app is built.

Layout

The window is split into three top-level tabs, a sidebar listing your endpoints, and a top bar:

• Publishers (↑) — endpoints you send to (build/test requests, run request/response traffic).
• Consumers (↓) — endpoints you receive from (live message log, payload inspection).
• App Config (⚙) — application-wide settings, config security, and environment variables.

The top bar shows a runtime status indicator (idle / active consumer(s)), a theme switcher (light / auto / dark), and Save. The sidebar filters endpoints and offers one-click import from Postman, OpenAPI, AsyncAPI, and existing mq-bridge configs.

Publishers

Select or add a publisher on the left, then edit it on the right. The header row holds the endpoint type (HTTP, Kafka, NATS, …), method, and URL, with a Send button to fire a request. Below it are per-endpoint tabs:

• Definition — name, endpoint type and connection settings (with Show advanced options), and a Middlewares list you can add to / remove from (metrics, retries, transforms, …).
• Body / Headers — the outgoing payload (with AUTO / TEXT / JSON / XML / HEX views) and header rows.
• History / Presets — previous sends and saved request presets.

After a Send, the response panel shows status, timing, request/response headers, and the response body (including a HEX view), with copy / json / curl shortcuts. Copy to Consumer, Clone, {} (view raw JSON), and Delete act on the selected endpoint.

Consumers

Consumers receive messages live. The header shows the connection state (Connected), a Capture messages toggle, a Keep last N limit, and Clear / Stop controls. Tabs:

• Definition — connection settings, same shape as a publisher.
• Output — the consumer’s response/output configuration.
• Messages — a live log (time + payload preview). Click a row to inspect it below: Message Headers, Message Body, and any Response Headers / Body, each with AUTO / TEXT / JSON / XML / HEX views plus copy/export.

App Config

Application-wide settings rendered directly from the AppConfig schema:

• AppConfig — default tab, log level, logger, metrics address, and UI address.
• Config Security — the storage mode (unencrypted, balanced, sensitive, durable, …) controlling how secrets and cached message history are stored. See Encryption at rest and Secrets.
• Environment Variables — key/value pairs available for interpolation in endpoint URLs and settings.

Export, Import, Reset, and {?} JSON operate on the whole configuration.

The configuration-first workflow

The point of one shared config format is that you test connections and dial in a route in the UI, export the JSON/YAML, then run that exact config unchanged — as a copy command, a config-mode service, or loaded from library code. A known-good route shape from the UI drops straight into production.

How the UI differs from API clients

The UI overlaps with API clients like Postman, Bruno, and Insomnia, but its centre of gravity is different: it is built around message bridging, runtime operation, and long-lived route management rather than one-off request composition. It adds broker pub/sub workflows, long-lived consumers/routes, protocol-to-protocol bridging, hex-level payload debugging, replay, local-first git-friendly config, and encrypted config — while leaving scripting and complex request chaining to the dedicated API clients.

Status. The UI/Tauri layer was prototyped quickly and does not mirror the mq-bridge / core/CLI standards — treat it as a working demo, not a reference implementation, and test before relying on it in production.

MCP server

data movement as a tool call

mq-bridge-app mcp runs the bridge as a Model Context Protocol server, so an agent can move data between any two supported endpoints — 15+ connectors covering databases, queues, brokers, HTTP, and files — from natural language.

Nothing is preconfigured. publish and start_route take their endpoints inline as JSON keyed by connector type, so the model picks both ends ad hoc; there is no YAML to write first and no per-connector tool to install. list_routes, route_status, route_messages, and stop_route manage what is already running.

The rows never enter the model’s context. The agent describes the job; the engine moves the bytes. Moving a 116.3 MiB dataset costs three tool calls and ~370 tokens — the same ~370 tokens whether the job is 1,000 rows or 1,000,000.

Quick start

# 1. get the binary (any install method from the Installation page)
brew install marcomq/tap/mq-bridge-app

# 2. register it with every MCP client detected on this machine
mq-bridge-app mcp install

# 3. restart the client fully — reopening a tab is not enough

Then ask the agent for the job in words:

“Drain the events Redis stream into the events table in Postgres, creating the table if needed.”

Running the server

# stdio — for local clients (Claude Code, Claude Desktop, Cursor)
mq-bridge-app mcp

# streamable HTTP — for remote/shared clients
mq-bridge-app mcp --transport http --bind 127.0.0.1:9092

No web UI is started in this mode. Logs go to stderr, because stdio transport owns stdout for the protocol itself.

Tools

Call server_info before quoting any throughput number: a debug binary reports much slower rates, and the figure would be meaningless.

start_route and route_messages are annotated not read-only — starting a route moves real data, and reading captured messages consumes the buffer — so a client should not auto-approve them the way it may auto-approve a true reader.

Endpoints

Every endpoint is a single-key JSON object naming the connector:

{"kafka":   {"url": "localhost:9092", "topic": "orders"}}
{"nats":    {"url": "nats://localhost:4222", "stream": "ORDERS", "subject": "ORDERS.new"}}
{"sqlx":    {"url": "postgres://user:pass@localhost:5432/db", "table": "events"}}
{"file":    {"path": "/tmp/out.jsonl", "format": "json"}}
{"null":    null}

The key is the connector; the object is its configuration:

Structural endpoints (fanout, switch, stream_buffer, request, reader, response, static, ref) are accepted here too and are documented under Middleware & structural endpoints.

A file/object_store source must repeat the compression/encryption its data was written with — neither is auto-detected, and a mismatch ends the route as completed having moved few or no messages rather than failing. Check the moved-message count.

Messages

A message is a payload plus optional string metadata headers and an optional message_id (a UUIDv7 is generated when omitted). A string payload is sent verbatim; any other JSON value is serialized to JSON bytes.

{
  "payload": {"id": 1, "sku": "A-100", "qty": 3},
  "metadata": {"kind": "order", "correlation_id": "abc-123"}
}

Conventional metadata keys: kind (message type, drives type-based routing), correlation_id and reply_to (request/reply). Keys starting with mqb.src. are reserved for provenance and are stripped on input.

Route options

Alongside input and output, a route accepts batch_size, concurrency, exit_on_empty (drain the source, then exit) and capture_last. Without exit_on_empty a route polls indefinitely until stop_route.

batch_size and concurrency are top-level start_route arguments, not fields inside route — the MCP server applies the tuned app defaults (1024 and 4) rather than the library’s conservative 1/1.

capture_last: N keeps the last N messages that flow through the route for route_messages to return. It is off by default: captured payloads are held in memory and handed to the model verbatim, which is exactly what “the rows never enter the context” otherwise avoids. Use it to sample a stream, not to move it.

Route status

route_status (and list_routes) report two rates, and they answer different questions:

For a route that is running now, read messages_per_second. For one that has finished — anything started with exit_on_empty — read average_messages_per_second: the instantaneous rate of a completed job is ~0 by the time any status call observes it, which says nothing about how fast it was. stop_route returns the same two fields as its parting summary.

elapsed_s and average_messages_per_second are null until the route has been sampled at least once (the sampler runs every 200 ms), so a job that finishes inside one tick reports no average.

Examples

Publish a batch to NATS JetStream:

{
  "publisher": {"nats": {"url": "nats://localhost:4222", "stream": "ORDERS", "subject": "ORDERS.new"}},
  "messages": [
    {"payload": {"id": 1, "sku": "A-100"}, "metadata": {"kind": "order"}},
    {"payload": {"id": 2, "sku": "B-200"}, "metadata": {"kind": "order"}}
  ]
}

Drain a Redis stream into Postgres, then exit:

{
  "name": "redis-to-postgres",
  "route": {
    "input":  {"redis_streams": {"url": "redis://localhost:6379", "stream": "events",
                                 "group": "g1", "read_from_start": true}},
    "output": {"sqlx": {"url": "postgres://user:pass@localhost:5432/db",
                        "table": "events", "auto_create_table": true}},
    "exit_on_empty": true
  }
}

Tail a live Kafka topic and keep the last 20 messages for inspection:

{
  "name": "orders-tail",
  "route": {
    "input":  {"kafka": {"url": "localhost:9092", "topic": "orders", "group_id": "agent"}},
    "output": {"null": null}
  },
  "capture_last": 20
}

Performance

The interface costs one round-trip, not a per-row tax: a route started through start_route moves data at the rate the copy CLI does, within run-to-run variance, and what separates them is a fixed ~55 ms of startup and completion polling.

Agent token cost is flat in the number of rows moved — three tool calls (start_route, one route_status, stop_route) totalling ~1.5 KB of JSON-RPC move 1M rows or 1,000 alike. Passing the same 116.3 MiB through a context window instead would cost ~30.5M tokens, which no context window holds at any price.

Methodology and the client used to measure it — a real MCP client over stdio, not an in-process harness — are in benches/etl/README.md, scenario 7. Engine-level tuning is on the Performance tuning page.

Registering with a client

mcp install writes the client config for you, registering the absolute path of the binary you just ran — so a target/debug build and an installed release binary each register themselves correctly.

# every client detected on this machine
mq-bridge-app mcp install

# a single client, project-scoped rather than global
mq-bridge-app mcp install --client cursor --local

# bake --report-to-ui into the registered command
mq-bridge-app mcp install --report-to-ui

Where the client ships its own CLI (Claude Code) that CLI is driven, since it stays correct across config-schema changes; otherwise the entry is merged into the client’s JSON, leaving every other registered server untouched. Installing twice is a no-op.

With --client omitted, every client detected on the machine is configured. For anything not listed above, mcp install --print-config prints the snippet:

{
  "mcpServers": {
    "mq-bridge": {
      "command": "mq-bridge-app",
      "args": ["mcp", "--transport", "stdio"]
    }
  }
}

Use an absolute path to the binary if it is not on PATH. With --transport http, point your client at http://127.0.0.1:9092/ instead.

Restart the client fully after installing — reopening a tab is not enough.

Under stdio the client spawns the server process, so a rebuilt binary is only picked up after the client restarts the server — an edit to mcp.rs alone will not change the behaviour of an already-running session.

Errors

Invalid endpoints, duplicate route names, and unknown route names come back as JSON-RPC -32602 (invalid params); a connection failure at publish time reports the underlying transport error:

invalid publisher endpoint: IO error: Connection refused (os error 61)

A partially failed batch returns a result flagged is_error with a {"status": "Partial", "sent": N, "failed": M, "errors": [...]} summary, so a partial send is not mistaken for success.

Known limitations

This originates in the upstream mq-bridge crate, not in the MCP layer.

• Finished routes are not reaped. A route started with exit_on_empty stays in list_routes after it has drained and exited; call stop_route to clear it. It is no longer indistinguishable from a running one, though: since mq-bridge 0.3.6 each entry carries "finished" plus an "outcome". While the route runs these are false / null; once its task ends, "finished": true and "outcome" is completed (drained cleanly) or failed (permanent error — the cause is in status.error).

  stop_route removes the entry from list_routes as it stops the route, so the third outcome, stopped, is not observable through these tools.

MCP Registry (server.json)

mcp install and the registry solve different problems, and neither needs the other:

So install remains the shortest path on a machine that already has the binary — it is not superseded by publishing to the registry.

server.json at the repository root is the registry entry. It publishes the server under io.github.marcomq/mq-bridge-app and offers two package types, both of which run the same mcp --transport=stdio command:

The registry only stores metadata — it never hosts the artifact — so it verifies that each package really belongs to this project by looking for an ownership marker inside the published artifact itself:

• OCI — a LABEL io.modelcontextprotocol.server.name in the Dockerfile final stage.
• Cargo — an mcp-name: line in the crate README, which crates.io renders. It must be visible markdown: crates.io strips HTML comments during rendering, so the hidden <!-- mcp-name: … --> form that works for PyPI and NuGet is silently dropped here.

Both markers must equal the name field in server.json.

Publishing happens in the publish-mcp-registry job of the release workflow. It runs after the ghcr manifest and the crates.io release exist, rewrites the versions in server.json from the release tag, authenticates with mcp-publisher login github-oidc (the io.github.* namespace is proven by the workflow’s own OIDC token, so there is no secret to rotate) and publishes. To do it by hand:

brew install mcp-publisher
mcp-publisher login github
mcp-publisher publish

Testing against local brokers

The compose files in the mq-bridge repo bring up the brokers:

cd ../mq-bridge/tests/integration/docker-compose
docker compose -f nats.yml -f postgres.yml -f redis.yml up -d

Verified end-to-end against these: batch publish to file/NATS/Redis Streams, NATS → file and Redis Streams → Postgres routes (including auto_create_table), the full route lifecycle, and the error paths above.

Architecture Overview

mq-bridge is designed as a highly extensible, protocol-agnostic message integration layer for Rust. Its architecture enables seamless bridging between diverse messaging systems, databases, and protocols, while allowing users to inject custom business logic and reliability patterns.

Core Principles

• Protocol Abstraction: All business logic operates on a unified CanonicalMessage type, decoupling your code from specific broker or database APIs.
• Extensibility: New endpoints and middleware can be added with minimal effort via trait-based factories.
• Async-First: Built on Tokio, all I/O and processing is asynchronous and concurrency-aware.
• Unopinionated: The library does not enforce a specific domain or concurrency model, focusing instead on reliable, programmable data movement.

Main Components

1. Route

A Route defines a data pipeline from one input endpoint to one output endpoint. Each route can:

• Specify concurrency and batch size
• Attach middleware for reliability, deduplication, metrics, etc.
• Attach a handler for business logic (transform, filter, respond)

2. Endpoint

Endpoints are protocol adapters for sources (consumers) and sinks (publishers). Supported types include Kafka, NATS, AMQP, MQTT, MongoDB, HTTP, SQLx, ZeroMQ, Files, AWS, IBM MQ, and the memory endpoint (in-process channels and cross-process IPC). Endpoints are created via factory functions and configured via serde (json/yml).

Beyond these protocol adapters there are structural endpoints that compose other endpoints or shape routing rather than talking to a broker: ref, fanout, switch, request, response, reader, static, stream_buffer, null and custom. All of them are documented in REFERENCE.md.

3. Middleware

Middleware wraps consumers and publishers to add cross-cutting features. There is no Middleware trait: a middleware is a decorator implementing MessageConsumer and/or MessagePublisher, which is why CustomMiddlewareFactory is defined as apply_consumer / apply_publisher. Available middleware includes:

• Retries (exponential backoff) and dead-letter queues (DLQ)
• JSON transformation (transform): mapping, type coercion, schema validation
• Deduplication (sled-based), weak joins, buffering, rate limiting
• Metrics, delays, fault injection
• Custom user middleware

The complete list, with fields, defaults and the layer-ordering rules, is in REFERENCE.md.

4. Handler

Handlers are user-defined async functions that process messages. There are two main handler types:

• CommandHandler: 1-to-1 or 1-to-0 transformation, can return a new message for publishing or as response.
• EventHandler: 1-to-N handler for event consumption. Compatible to CommandHandler, but should not return a response.
• TypeHandler: Strongly-typed handler, dispatches based on the kind metadata field and deserializes payloads.

Memory Endpoint and IPC Transport

The memory endpoint is not only an in-process channel. Its topic field (serde alias: url) doubles as a transport URL, so the same endpoint type covers both in-process queues and cross-process IPC over Unix domain sockets / Windows named pipes.

Transport URL schemes

Anything else (http://…, an empty URL, a relative unix://path) is rejected at parse time.

Named ipc:// resolution on Unix falls back in order, using the first writable location:

1. /run/mq-bridge/<name>.sock (systemd standard)
2. $XDG_RUNTIME_DIR/mq-bridge/<name>.sock
3. /tmp/mq-bridge/<name>.sock (less secure)

Because of this fallback, ipc://name can land in different places for different users or services. When both sides must agree deterministically, use an explicit path (ipc:///run/myapp/queue.sock) instead of a bare name.

Roles: consumer is the server, publisher is the client

IPC is unidirectional, publisher → consumer, and the roles are fixed:

• The consumer binds and listens on the socket/pipe (server). It removes a stale socket file before binding, creates the parent directory with mode 0700, and sets the socket to 0600. It unlinks the socket on drop.
• The publisher connects to it (client).

So the consumer process must be running before the publisher connects — otherwise the publisher fails with a connection error. The consumer serves one connection at a time; if the peer disconnects, it logs a warning and waits for a new connection rather than erroring out.

IPC requires async construction

MemoryConsumer::new, MemoryPublisher::new, and new_local are synchronous and only support memory://. Given an IPC URL they return an error (“requires async endpoint construction”). Use the async constructors:

use mq_bridge::endpoints::memory::{MemoryConsumer, MemoryPublisher};
use mq_bridge::models::MemoryConfig;

let config = MemoryConfig::new_with_url("ipc:///run/mq-bridge/orders.sock", Some(100));

// Server side (start first)
let mut consumer = MemoryConsumer::new_async(&config).await?;
// Client side, in another process
let publisher = MemoryPublisher::new_async(&config).await?;

Routes always build endpoints through the async factories, so an ipc:// URL works in YAML/JSON config with no extra code:

ipc_ingest:
  input:
    memory:
      url: "ipc:///run/mq-bridge/orders.sock"
      capacity: 256
  output:
    kafka:
      topic: "orders"
      url: "localhost:9092"

Wire format

Batches are serialized with MessagePack (Vec<CanonicalMessage>) and written as length-prefixed frames: a 4-byte big-endian length followed by the payload. Frames larger than 100 MB are rejected.

Framing is handled by tokio_util::codec::LengthDelimitedCodec in endpoints/memory/framed.rs, shared by both platforms. This matters for more than deduplication: the codec owns the partial-frame buffer, which makes reads cancel safe. Routes cancel receive_batch on shutdown via select!, and a hand-rolled read_exact pair would consume part of a frame on cancellation and desync the connection permanently.

Backpressure

A batch is written as one frame. Socket buffers are small — 8 KiB by default on macOS (net.local.stream.sendspace) — so a batch that outgrows the buffer only completes once the consumer drains it. send_batch blocking is therefore normal backpressure, not a fault.

Two consequences worth knowing:

• The consumer must actually be reading, not merely connected. A consumer that has accepted but stopped draining will stall the publisher indefinitely. After 5 seconds blocked, the publisher logs a warning naming the socket; it keeps waiting rather than dropping data.
• capacity does not create a queue here. There is no buffering between the two processes beyond the kernel socket buffer.

Acknowledgements and redelivery over IPC

enable_nack defaults to true for IPC transports (ipc://, unix://, pipe://) and false for memory://; an explicit enable_nack in config always wins.

Redelivery over IPC is consumer-local. The socket carries publisher → consumer traffic only, so a nack cannot travel back to the producer — the publisher never reads, and writing to it would strand the messages and eventually block the commit on a full socket buffer. A nacked message is therefore requeued inside the consumer and redelivered ahead of new traffic. It does not survive a consumer crash, and the publisher is never told.

If you need redelivery that survives the consumer process, use a real broker endpoint. mq-bridge deliberately does not implement a bidirectional ack protocol over IPC.

Behavioural differences vs memory://

• subscribe_mode is not supported over IPC, on either side — the EventStore/broadcast backend is in-process only.
• request_reply is not supported for IPC publishers.
• Only the publisher side may send. Calling send_batch on a consumer-side IPC transport returns an error rather than writing into a peer that never reads.
• capacity bounds the consumer-side buffer, not an internal queue — a socket has no backlog of its own. len() reports whole frames already buffered by the codec (readable without touching the socket), and endpoint status().pending adds the consumer’s own buffered and awaiting-redelivery messages.

Batching and Concurrency

Batch processing is a core concept in mq-bridge and is required for all endpoint implementations. Every consumer and publisher must implement batch receive and batch send methods (receive_batch, send_batch).

Why batch mode?

• Batching improves throughput and efficiency, especially for high-volume or high-latency backends.
• It enables the bridge to process messages concurrently and in parallel, reducing per-message overhead.

Concurrency

• Each route can be configured with a concurrency parameter, which determines how many worker tasks will process batches in parallel.
• Batch size is also configurable per route.

Helper Utilities

mq-bridge provides several helper functions to make implementing batching easier, especially if your endpoint only supports single-message operations:

• send_batch_helper: Calls send for each message in a batch and aggregates the results. Used to implement send_batch when only single-message sending is available.
• receive_batch_helper: Calls receive once and wraps the result as a batch. Used to implement receive_batch when only single-message receive is available.
• into_commit_func: Converts a batch commit function (BatchCommitFunc) into a single-message commit function (CommitFunc).
• into_batch_commit_func: Converts a single-message commit function into a batch commit function.

Sample: Using send_batch_helper

use mq_bridge::traits::send_batch_helper;
// Inside your MessagePublisher implementation:
async fn send_batch(&self, messages: Vec<CanonicalMessage>) -> Result<SentBatch, PublisherError> {
    send_batch_helper(self, messages, |pub_ref, msg| {
        Box::pin(pub_ref.send(msg))
    }).await
}

Sample: Using receive_batch_helper

// receive_batch_helper is a default method on MessageConsumer — no import needed.
// Inside your MessageConsumer implementation:
async fn receive_batch(&mut self, max_messages: usize) -> Result<ReceivedBatch, ConsumerError> {
    self.receive_batch_helper(max_messages).await
}

Sample: Commit function conversion

use mq_bridge::traits::{into_commit_func, into_batch_commit_func};
let commit: CommitFunc = into_commit_func(batch_commit_func);
let batch_commit: BatchCommitFunc = into_batch_commit_func(commit_func);

How batch receive works internally

The receive_batch method is designed to efficiently collect a batch of messages from the underlying transport. The typical pattern is:

1. Wait for the first message: The consumer awaits a message from the backend (e.g., Kafka, NATS, etc.).
2. Drain additional messages if available: After the first message is received, the consumer immediately checks if more messages are already available (without waiting). It continues to drain messages up to the batch size or until no more are available.
3. Return the batch: The batch is returned as soon as either the batch size is reached or no more messages are immediately available.

This approach minimizes latency for the first message while maximizing throughput for bursts of messages.

Pseudocode:

async fn receive_batch(&mut self, max_messages: usize) -> Result<ReceivedBatch, ConsumerError> {
	let mut messages = Vec::with_capacity(max_messages);
	// Wait for the first message
	let first = self.inner_receive().await?;
	messages.push(first);
	// Try to drain more messages without waiting
	while messages.len() < max_messages {
		match self.try_receive_now()? {
			Some(msg) => messages.push(msg),
			None => break,
		}
	}
	Ok(ReceivedBatch { messages, commit: ... })
}

Example: Receiving and committing a batch

let batch = consumer.receive_batch(100).await?;
// Process each message in the batch...
batch.commit(vec![MessageDisposition::Ack; batch.messages.len()]).await?;

Example: Sending a batch

let messages = vec![msg1, msg2, msg3];
publisher.send_batch(messages).await?;

See the README and tests for more advanced batching and concurrency patterns.

Getting Started

Below are minimal examples for the three main usage patterns in mq-bridge. For more, see the README and tests.

1. Typed Handler (Event-driven, Type-safe)

Use for strongly-typed, event-driven communication. Register Rust types per message kind; the bridge deserializes payloads automatically. Supports request-response where the protocol allows. Multiple types can be handled by a single route.

use mq_bridge::{msg, Handled, Route, publisher::Publisher, models::Endpoint};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug, Clone)]
struct OrderPlaced {
    order_id: u64,
    amount: f64,
}

let input = Endpoint::new_memory("in", 10);
let output = Endpoint::null();
let route = Route::new(input, output)
    .add_handler("order_placed", |msg: OrderPlaced| async move {
        println!("Order #{}: ${}", msg.order_id, msg.amount);
        Ok(Handled::Ack)
    });
route.deploy("typed_handler_example").await.unwrap(); // or use route.run()

let input_publisher = Publisher::new(route.input.clone()).await.unwrap();
let event = OrderPlaced { order_id: 42, amount: 19.99 };
input_publisher.send(msg!(&event, "order_placed")).await.unwrap();
// ...

2. Compute Handler (Generic)

Use a generic handler to process, transform, or filter messages against the raw CanonicalMessage. One handler per-route. Suitable for pipelines, ETL, or side-effect processing.

use mq_bridge::{CanonicalMessage, Handled, Route, models::Endpoint};

let input = Endpoint::new_memory("in", 10);
let output = Endpoint::new_memory("out", 10);
let handler = |mut msg: CanonicalMessage| {
    msg.set_payload_str(format!("processed: {}", msg.get_payload_str()));
    async move { Ok(Handled::Publish(msg)) }
};
let route = Route::new(input, output).with_handler(handler);
route.deploy("compute_handler_example").await.unwrap();
// ...

3. Direct Endpoint Usage (Manual Control)

Use send / send_batch and receive / receive_batch directly on endpoints. Gives full manual control over batching, commit, concurrency, and sequencing. Useful for advanced scenarios or integration with external async runtimes.

use mq_bridge::endpoints::memory::{MemoryConsumer, MemoryPublisher};
use mq_bridge::{CanonicalMessage, traits::MessageDisposition};

let publisher = MemoryPublisher::new_local("my_topic", 100);
let mut consumer = MemoryConsumer::new_local("my_topic", 100);
let msg = CanonicalMessage::new(b"hello world".to_vec(), None);
publisher.send(msg).await.unwrap();
let received = consumer.receive().await.unwrap();
// Process the message...
// Acknowledge (required for most endpoints):
(received.commit)(MessageDisposition::Ack).await.unwrap();

// Batch variant:
let batch = consumer.receive_batch(10).await.unwrap();
batch.commit(vec![MessageDisposition::Ack; batch.messages.len()]).await.unwrap();

Extending mq-bridge

• Custom Endpoints: Implement the CustomEndpointFactory trait and register your type.
• Custom Middleware: Implement the CustomMiddlewareFactory trait.
• Typed Handlers: Use TypeHandler to add new message types and logic.

Configuration

• All routes, endpoints, and middleware are defined via YAML, JSON, or environment variables.
• See CONFIGURATION.md for a full reference and examples.

Example: Route Lifecycle

1. Define a route either as json or code
2. Create endpoints and apply middleware
3. Attach a handler (optional)
4. Deploy or run the route (spawns async workers)
5. Inject or receive messages
6. Route processes, transforms, and delivers messages according to config and handler logic

More Information

• See the README for usage patterns and code examples.
• See CONFIGURATION.md for configuration details.
• See the source code for trait definitions and extension points.

How the app is built

mq-bridge-app is a demonstration of the mq-bridge engine used on itself: the application serves its own management UI through the engine rather than through a separate web framework. If you want to understand the engine’s request-reply model in practice, this is a worked example. For the engine internals themselves, see Learn the architecture.

Backend: mq-bridge as a web server

Instead of using a traditional web framework like Actix or Axum directly for the management API, the application routes HTTP through the engine:

1. HTTP input — an http input endpoint listens on the configured UI port and converts incoming HTTP requests into CanonicalMessages.
2. WebUiHandler — a custom Handler processes these messages as a router: serving static files (HTML, JS) or handling API requests (e.g. /config, /schema.json).
3. Response output — the handler returns a response message, sent to a response output endpoint, completing the HTTP request-response cycle.

This showcases the library’s ability to handle request-reply patterns and act as a lightweight web server.

Frontend: vanilla-schema-forms

The Web UI is generated dynamically from the Rust configuration structures — no hand-written form code:

1. Schema generation — the backend uses schemars to generate a JSON Schema for the AppConfig struct at runtime, exposed via /schema.json (also available on the CLI: mq-bridge-app --schema dev/config/schema.json).
2. Dynamic form — the frontend uses vanilla-schema-forms to render a complete configuration form from that schema alone.
3. No UI code changes — when a new feature or config option is added to the Rust code (e.g. a new middleware), the schema updates automatically and the UI reflects it without any frontend changes.

Project status

This project is in active development. It originally served as the primary reference implementation and testbed for the mq-bridge engine.

The UI/Tauri layer was largely prototyped quickly and does not mirror the mq-bridge / mq-bridge-app core/CLI standards — treat it as a working demo, not a reference implementation, and test before relying on it in production.

Core concepts

Four terms recur throughout this book.

Route

A route is a named data pipeline that defines a flow from one input to one output. It carries optional middlewares on either side, a handler, and the tuning knobs batch_size, concurrency, and commit_concurrency_limit. In config, the top-level keys are route names:

orders_bridge:                 # <- route name
  input:  { kafka: { topic: "orders", url: "localhost:9092" } }
  output: { nats:  { subject: "orders.processed", url: "nats://localhost:4222" } }

Endpoint

An endpoint is a source or sink for messages, written as a single-key object naming the connector. Transport endpoints talk to a broker or store (kafka, nats, sqlx, mongodb, file, memory, …); structural endpoints (fanout, switch, response, ref, null, …) compose other endpoints or shape routing. The same endpoint type can act as consumer or publisher depending on which side of the route it is on. See the full catalog in Endpoints (connectors).

Middleware

Middleware intercepts and processes messages around an endpoint — retries, dead-letter queues, transformation, deduplication, rate limiting, encryption, metrics. It attaches as a middlewares: list on the input, output, or both. Ordering matters: on an output the last entry is the outermost layer, so dlq goes last. See Middleware.

Handler

A handler is a programmatic component for business logic, used when you embed the library (config-only routes forward messages without one). There are three kinds:

• CommandHandler — a 1-to-1 or 1-to-0 transformation. It takes a message and can return a new message to publish or return as a response.
• EventHandler — a terminal 1-to-N handler for event consumption. It reads messages without removing them for other event handlers, and should not return a response.
• TypeHandler — a strongly-typed handler. It dispatches on the kind metadata field and deserializes payloads into concrete Rust types, so handlers don’t repeat parsing code.

let typed_handler = TypeHandler::new()
    .add("create_user", |cmd: CreateUser| async move {
        // handle create_user; Ok(()) maps to Handled::Ack
    })
    .add("delete_user", |cmd: DeleteUser| async move {
        // handle delete_user
    });
let route = Route::new(input, output).with_handler(typed_handler);

Handlers return a Handled value: Handled::Ack (consume), Handled::Publish(msg) (forward msg down the publisher chain), and so on. Message selection for TypeHandler is driven by the kind metadata field, which msg!(&value, "kind") sets. See the Embed the library tutorial and Language bindings API.

CanonicalMessage

Every handler works with a CanonicalMessage — the unified message type that decouples your code from any specific transport. It carries a payload (bytes), string metadata headers, and a message_id (a UUIDv7 when omitted). Conventional metadata keys: kind (message type / typed routing), correlation_id and reply_to (request/reply). Keys starting with mqb.src. are reserved for provenance and stripped on input.

Delivery model

Endpoints default to a consumer (queue) pattern; subscriber (pub/sub) behaviour is opt-in per backend. Delivery is at-least-once where a durable position or lease exists — so for ETL, pair it with an idempotent write at the sink for effective exactly-once. Ack/nack and retry/DLQ handling are designed to work with batching; see Learn the architecture.

Build from source

Developer build instructions for mq-bridge-app — the CLI/server, the desktop (Tauri) app, and the Docker image. For prebuilt installs (cargo binstall, release bundles, Docker Hub images) see the main README.

Prerequisites

• Rust toolchain (latest stable version recommended)
• Access to the message brokers you want to connect (e.g. Kafka, NATS, RabbitMQ)

CLI / server

1. Clone the repository:

   git clone https://github.com/marcomq/mq-bridge-app
   cd mq-bridge-app
   

2. Build and run the CLI (empty):

   cargo run --release
   

3. Build and run with a config:

   cargo run --release -- --config dev/config/file-to-http.yml
   

Create a config.yml in the project root or set environment variables — or start without one and use the UI to define config.yml.

For IBM MQ, install the client library first and build with --features=ibm-mq. See the IBM MQ Setup Guide.

Desktop app (Tauri)

The desktop UI is a Svelte frontend driven by a Tauri backend:

npm install
npm run dev

To build the UI bundle and serve it from the Rust backend:

npm run build:ui
cargo run --release

Docker image (no local Rust required)

Requires Docker and Docker Compose:

docker-compose up --build

This builds and starts the bridge CLI application.

Postgres CDC → JSONL

Stream every insert/update/delete from a PostgreSQL table into a newline-delimited JSON file, continuously. This is change-data-capture (CDC): instead of reading a table snapshot, mq-bridge follows the write-ahead log via logical replication, so new changes keep flowing until you stop it.

Prerequisites

• PostgreSQL with logical replication enabled (wal_level = logical).

• A publication on the source table:

  CREATE PUBLICATION mqb_pub FOR TABLE orders;
  

  publication must already exist; slot_name is created automatically if missing (it is a permanent, resumable replication slot).

The one-liner

mq-bridge-app copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to file:///data/orders.jsonl?format=json

--from uses the postgres-cdc:// scheme (alias pgcdc://); the URL underneath is a plain Postgres URL. --to writes each change as one JSON object per line (format=json); the path comes from the URI path itself, not a query param.

Each change arrives as a CanonicalMessage whose payload is the flat row and whose postgres.operation metadata marks the operation (insert/update/delete) — the same convention as MongoDB CDC, so typed handlers work identically across both.

Equivalent config file

The same thing as a route in mq-bridge.yaml, for the config-driven run forms:

orders_cdc:
  input:
    postgres_cdc:
      url: "postgres://user:pass@localhost:5432/app"
      publication: "mqb_pub"     # CREATE PUBLICATION mqb_pub FOR TABLE orders;
      slot_name: "mqb_slot"      # created if missing (permanent slot, resumable)
  output:
    file:
      path: "/data/orders.jsonl"
      format: "json"

Notes

• Resumability. The named replication slot persists progress on the server, so a restart resumes from the last confirmed LSN rather than re-copying from the start.
• This runs forever. CDC is a continuous stream; the process stays up tailing new changes. For a one-shot bounded copy of an existing table, use a plain postgres://…?table=… source instead (see Quick start).
• Performance. For sustained CDC/ETL throughput and how batching interacts with file sinks, see Performance tuning.

See also

• PostgreSQL CDC connector — URL format and examples.
• postgres-cdc parameter reference — every recognised field.
• File connector — formats and options for the sink.

HTTP webhook → MongoDB

Accept incoming HTTP webhooks on a local port and persist each request body as a document in MongoDB, with automatic retries on transient write failures. This is a config-driven route rather than a one-liner, because it wires middleware into the pipeline.

The route

mq-bridge.yaml:

webhook_to_mongo:
  input:
    http:
      url: "127.0.0.1:8080"
      # Force the normal route pipeline instead of the inline HTTP response fast path.
      inline_response_fast_path: false
    middlewares:
      - retry:
          max_attempts: 3
          initial_interval_ms: 500
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "app_db"
      collection: "webhooks"
      format: "json"   # a bit slower, but stores readable documents

Run it in config mode:

mq-bridge-app --config mq-bridge.yaml

Then POST to it:

curl -X POST http://localhost:8080 \
  -H 'content-type: application/json' \
  -d '{"event":"signup","user":"alice"}'

The document lands in app_db.webhooks.

What each piece does

• http input listens on 127.0.0.1:8080, i.e. localhost only. The endpoint is unauthenticated, so bind it to 0.0.0.0 only behind an authenticating proxy or a network you control. Setting inline_response_fast_path: false forces requests through the full route pipeline (input → middleware → output) instead of the fast inline-response path, so the retry middleware and the MongoDB write both run before the HTTP response is sent.
• retry middleware re-attempts a failed write up to max_attempts times, starting at initial_interval_ms and backing off. See the Retries & backoff recipe for the full knob set and its interaction with a dead-letter queue.

Note

Delivery here is at-least-once: a write that actually succeeded but whose acknowledgement was lost gets retried and inserts a second document. For an idempotent sink, set id_field on the mongodb output to a stable business key so the value becomes the document _id: a retry of an already-written message is then rejected by the unique _id instead of appending a new document — see Upserts.

• mongodb output with format: "json" stores each payload as a readable JSON document rather than the default full-message serialization.

Variations

• Want a synchronous reply to the caller? Attach a handler and use a response output — see the Request / reply tutorial.
• High request rates? The http input scales with connection concurrency; see Performance tuning.

See also

• Configuration grammar — this is Route 2 of the annotated example.
• MongoDB connector and parameter reference.
• HTTP connector and parameter reference.

Cross-process IPC bridge

The memory endpoint is more than an in-process channel: its url (alias topic) is a transport URL, so the same endpoint can bridge separate processes on one machine over a Unix socket or Windows named pipe — no broker required.

Transport URLs

The url field selects the transport:

The consumer side binds/listens and must be started before the publisher connects. IPC does not support subscribe_mode or request_reply. enable_nack defaults to true, but redelivery is consumer-local: a nacked message is retried inside the consumer and is lost if the consumer process dies.

Example: ingest process → Kafka

One process reads from a Unix socket and forwards to Kafka. Start this consumer first so the socket exists before any producer connects:

ipc_ingest:
  input:
    memory:
      url: "ipc://orders"
      capacity: 256
  output:
    kafka:
      topic: "orders"
      url: "localhost:9092"

mq-bridge-app --config ipc_ingest.yaml

The named form is used rather than an explicit path so no privileged directory has to exist up front: /run/mq-bridge/ needs root, and ipc://orders falls back to $XDG_RUNTIME_DIR/mq-bridge or /tmp/mq-bridge. Give an absolute ipc:///abs/path.sock only once you have created and permissioned that directory.

Any other process on the same host can now publish into ipc://orders — via its own mq-bridge route, or programmatically through a Publisher built on a memory endpoint (see the embedding tutorial) — and the messages flow through to Kafka.

When to use it

• Decouple a producer and a sink into separate processes without standing up a broker — e.g. an app writes to the socket, a long-lived bridge process owns the Kafka/DB connections and batching.
• Language boundary. A Python or Node process publishes into the socket; a Rust bridge process does the heavy I/O.

capacity bounds the in-flight channel; tune it alongside the guidance in Performance tuning.

See also

• Configuration grammar — this is Route 9 of the annotated example, with the full URL grammar in comments.
• Endpoints (concepts).

Request / reply

mq-bridge supports request-response patterns for interactive services such as web APIs. A client sends a request and waits for the matching response, while the bridge keeps the correlation details away from your handler.

The response output is the most direct option and the safest one under concurrency: the response stays in the same execution context as the request, so concurrent requests do not share a reply queue or race on correlation IDs.

How it works

1. An input endpoint that supports request-reply (like http) receives a request.
2. The message passes through the route’s processing chain — this is where you attach a handler to process the request and generate a response payload.
3. The final message is sent to the output.
4. If the output is response: {}, the bridge sends the message back to the original input source, which delivers it as the reply (e.g. the HTTP response body).

Not every backend can do request-reply. It is only supported/tested for endpoints that natively support or emulate it. SQLx, files, AWS, IBM MQ, and Sled do not support request-reply. Check the backend table in the engine README.

Example: MongoDB request/response

A sender writes a request document to MongoDB and waits for a reply. The bridge reads the document, runs the handler, and writes the result back to the reply collection:

mongo_responder:
  input:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "app_db"
      collection: "requests"
  output:
    # 'response' sends the processed message back to the reply collection
    # (whatever reply_to the sender set).
    response: {}

Attaching the handler (Rust)

The response output only echoes what the pipeline produced — you supply the logic by attaching a handler to the route:

use mq_bridge::models::Handled;
use mq_bridge::CanonicalMessage;

let handler = |mut msg: CanonicalMessage| async move {
    let request_body = String::from_utf8_lossy(&msg.payload);
    let response_body = format!("Handled response for: {}", request_body);
    msg.payload = response_body.into();
    Ok(Handled::Publish(msg))
};
// load the route from YAML, then attach `handler` to its output and run it.

See Embed the library for the full load-and-run scaffolding, and Core concepts for the handler model (CommandHandler / EventHandler / TypeHandler).

See also

• request / response structural endpoints — the reference definitions.
• Engine README, Patterns: Request-Response — the authoritative write-up.

Embed the library (Rust / Python / Node)

The core engine is a Rust library, and the same engine ships as native bindings for Python and Node.js. The Tokio runtime, broker I/O, routing, and batching all stay in Rust; the binding is a thin layer for handlers and configuration. Behaviour and reliability match the Rust engine regardless of which language calls in.

The config-first workflow

The natural way to embed the library is to design the route as config, then load that exact config from your code. Design and test a route in the desktop UI (or hand-write the YAML), export the JSON/YAML, and load it — the running route behaves identically to the --config CLI form.

Constructor names are kept aligned across languages (Python snake_case, Node camelCase):

The name argument is optional: pass it to select one entry from a routes:/publishers: document, or omit it to treat the config as a single bare route/endpoint body.

Rust

The Rust crate exposes the full engine. The main types:

• Route::new(input, output) — a pipeline from one endpoint to one endpoint, with .with_batch_size(n), .with_handler(h), .add_handler(kind, f), .deploy(name) / .run().
• Endpoint — protocol adapters (Endpoint::new_memory, Endpoint::null, …) plus the serde-configured variants.
• Publisher::new(endpoint) — publish into a route’s input or any endpoint.
• Handlers: CommandHandler (1-to-1 / 1-to-0), EventHandler (terminal 1-to-N), and TypeHandler (dispatches on the kind metadata field, deserializing payloads).
• CanonicalMessage — the unified message type all handlers work with; msg!(&value, "kind") builds one with a kind.

The MessageConsumer and MessagePublisher traits (in mq_bridge::traits) are the core abstractions. See the Rust docs on docs.rs for the full API.

See also

• Language bindings API — the reference for this material.
• Core concepts and Learn the architecture — the handler and message model.
• The three ways to run it — library vs CLI-server vs desktop.

Upserts & insert-if-absent

For ETL, at-least-once delivery plus an idempotent write at the sink gives you effective exactly-once: a replayed or retried record must not create a duplicate row. The most robust place to enforce this is the sink’s own unique constraint — it’s already shared across every writer, so no extra state store is needed.

MongoDB — id_field

Point id_field at a top-level payload field and its value becomes the document _id. Re-inserting the same business key then hits the unique _id index and is treated as an idempotent success (the duplicate is skipped, not errored):

orders_to_mongo:
  input:  { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "shop"
      collection: "orders"
      format: json
      id_field: "order_id"   # payload {"order_id": "A-1", ...} → _id = "A-1"

The field’s JSON type is preserved (a number stays a BSON integer). The payload must contain the field, otherwise the message is dead-lettered rather than written with a random _id. Use id_field on sink collections only — a business-key _id is incompatible with the consumer/subscriber competing-consumer modes, which require a UUID _id.

SQL (sqlx) — ON CONFLICT / ON DUPLICATE KEY

insert_query is user-supplied, so you write the dialect’s upsert directly. This requires a pre-existing UNIQUE/PRIMARY KEY on the key column, and is incompatible with bulk_copy (COPY cannot express ON CONFLICT) — so you trade peak throughput for deduplication.

-- PostgreSQL — insert if absent (drop duplicates):
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING

-- PostgreSQL — upsert (last write wins):
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body})
  ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body

-- MySQL / MariaDB:
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body})
  ON DUPLICATE KEY UPDATE body = VALUES(body)

-- SQLite:
INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING

A plain INSERT without a conflict clause instead fails the row as a non-retryable error (captured by a dlq, or logged and dropped without one). ${payload:field} binds a typed value from the JSON payload; ${metadata:key} binds a metadata string.

ClickHouse — ReplacingMergeTree

ClickHouse has no unique constraints; dedup is a table-engine property. Create the target as ReplacingMergeTree(version) keyed by your business key via ORDER BY, using a monotonic column as the version (an ingest timestamp, or postgres.lsn from a CDC source):

CREATE TABLE orders (id UInt64, body String, version UInt64)
ENGINE = ReplacingMergeTree(version) ORDER BY id;
-- mq-bridge just inserts rows; duplicates for the same id collapse on merge.
SELECT * FROM orders FINAL;

Read with FINAL (or argMax) to see the deduplicated result. ClickHouse also natively dedupes identical re-inserted blocks (default one-hour window) — treat that only as retry-safety, and rely on ReplacingMergeTree for logical dedup.

Branch on insert vs. duplicate (MongoDB report_outcome)

To act on whether a record was newly inserted or already existed, set report_outcome: true; the Mongo publisher tags the returned message with mongodb.outcome = inserted or existed. Wrap it in a request endpoint to forward that tagged message into a switch that routes on the outcome:

orders_upsert_branch:
  input: { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    request:
      to:
        mongodb:
          url: "mongodb://localhost:27017"
          database: "shop"
          collection: "orders"
          format: json
          id_field: "order_id"     # deterministic _id → insert-if-absent
          report_outcome: true     # → mongodb.outcome = inserted | existed
      forward_to:
        switch:
          metadata_key: "mongodb.outcome"
          cases:
            inserted: { ref: "enrich_new_order" }
            existed:  { ref: "handle_duplicate" }

report_outcome is sink-only and pairs with id_field. See also Deduplication for the middleware-based complement, and the Postgres CDC tutorial for CDC-specific idempotency (postgres.lsn as the version to drop stale replays).

Dead-letter queues

By default, a message that fails permanently is logged at error level and dropped — the route tolerates it and keeps going. A dlq middleware is the only mechanism that retains those failures for inspection or replay.

Basic DLQ

Add dlq to the output middleware list. The DLQ endpoint is a full endpoint, so failed messages can land anywhere — a file, another queue, a database:

output:
  middlewares:
    - dlq:
        endpoint:
          file: { path: "dead-letters.jsonl" }
  kafka: { topic: "orders", url: "localhost:9092" }

From the copy CLI, dlq’s endpoint is a URL-encoded endpoint URI:

--to 'kafka://broker:9092?topic=orders|dlq?endpoint=file%3A%2F%2F%2Ftmp%2Ffailed.jsonl'

Pair it with retry — and mind the order

retry alone does not retain a failed message; once its attempts are exhausted it hands the still-failing message on to be dropped (or to a following dlq). Put retry before dlq so the dlq is the outermost layer and captures what retry gives up on:

output:
  middlewares:
    - retry: { max_attempts: 3 }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }   # last = outermost

On an output, publisher middlewares wrap in list order, so the last entry is outermost. See the ordering rule.

What gets dead-lettered

• ✅ NonRetryable failures (data/type errors the sink rejects, poison payloads).
• ✅ Retryable failures whose retries are exhausted.
• ❌ Connection errors are not dead-lettered — they propagate so the route can reconnect. (If the DLQ send itself fails with a connection error, that error propagates rather than silently dropping the message.)

Catching transform rejections

A transform failure on an output is non-retryable and flows to a following dlq, so invalid records are captured with the reason:

output:
  middlewares:
    - transform: { schema_file: "schemas/user.json" }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }
  kafka: { topic: "users", url: "localhost:9092" }

No DLQ? Watch the logs

Without a dlq, a systematic failure (e.g. every row hitting a column-type mismatch) drains the input while committing nothing and still ends completed. The signal is a burst of Dropping message … due to non-retryable error in the logs. See Troubleshooting.

Deduplication

Two complementary ways to keep replayed or retried records from creating duplicates: the deduplication middleware (filters before the sink), and the sink’s own unique constraint (the robust choice for multi-writer ETL, covered in Upserts & insert-if-absent).

The deduplication middleware

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

input:
  middlewares:
    - deduplication: { sled_path: "/var/lib/mq-bridge/dedup", ttl_seconds: 3600 }
  kafka: { topic: "orders", url: "localhost:9092" }

State is a local sled database, so deduplication is per-process, not cluster-wide. For multi-writer pipelines prefer the sink constraint (below); use the middleware as a complement, not a replacement.

Sink-side dedup (the robust path)

The most robust place to dedup is the sink’s own unique constraint — it’s already shared across every writer:

• MongoDB — id_field maps a business key to the unique _id; a duplicate is an idempotent skip.
• SQL — ON CONFLICT (key) DO NOTHING / ON DUPLICATE KEY UPDATE.
• ClickHouse — ReplacingMergeTree(version) collapses duplicates by sort key at merge time.

Full examples in Upserts & insert-if-absent.

Deduplicating CDC replays

A postgres_cdc change event’s message_id is a stable hash of schema.table + key + lsn, so a replayed change deduplicates through the deduplication middleware, and the sink’s own constraint (id_field / ON CONFLICT) makes the write idempotent. Use postgres.lsn as the version to drop stale replays:

INSERT INTO orders (id, body, lsn) VALUES (${payload:id}, ${payload:body}, ${metadata:postgres.lsn})
  ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body, lsn = EXCLUDED.lsn
  WHERE EXCLUDED.lsn > orders.lsn

Known edge: if the same primary key changes twice within one transaction, both events share that transaction’s commit LSN and therefore the same message_id — the middleware treats the second as a duplicate and drops it. The sink still converges to the final row, but the intermediate revision is not delivered. If you need every intra-txn revision, don’t rely on the message_id/middleware path for those rows.

See the Postgres CDC → JSONL tutorial for the full CDC idempotency picture.

Retries & backoff

The retry middleware retries failed sends with exponential backoff. Output only.

output:
  middlewares:
    - retry: { max_attempts: 5, initial_interval_ms: 200, max_interval_ms: 10000, multiplier: 2.0 }
  kafka: { topic: "orders", url: "localhost:9092" }

From the copy CLI:

Attach the chain to --to, since retry is output-only:

--to 'kafka://localhost:9092?topic=orders|retry?max_attempts=5&initial_interval_ms=200'

What gets retried

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.

Always pair retry with dlq

retry does not retain a message it gives up on — it hands the still-failing message on to be dropped, or to a following dlq. Put dlq after retry so it’s the outermost layer and captures the exhausted failures:

output:
  middlewares:
    - retry: { max_attempts: 3 }
    - dlq: { endpoint: { file: { path: "rejected.jsonl" } } }

Don’t over-tune

Several endpoints already retry connection/timeout errors internally, and the retry middleware adds backoff on top — so you rarely need a large max_attempts. Cap max_interval_ms to match your latency budget so retries never stall a route indefinitely. See Performance tuning → Retry & backoff.

Transform & schema mapping

The transform middleware reshapes JSON payloads declaratively, so renaming fields and fixing types doesn’t need a custom handler. It runs two optional stages over a single parse: mapping (rename, move, nest) then schema (coerce, apply defaults, validate). Input and output.

Rename + type-fix a CSV feed

csv_to_kafka:
  input:
    file: { path: "users.csv", format: csv }
  output:
    middlewares:
      - transform:
          mapping:
            firstName: "$.first_name"
            lastName: "$.last_name"
            id: "$.user_id"
            "address.city": { path: "$.city", default: "unknown" }
          schema_file: "schemas/user.json"
      - dlq:
          endpoint: { file: { path: "rejected.jsonl" } }
    kafka: { topic: "users", url: "localhost:9092" }

With the CSV producing {"first_name":"John","last_name":"Smith","user_id":"42"}, the mapping yields {"firstName":"John","lastName":"Smith","id":"42","address":{"city":"unknown"}} ($.city is absent, so address.city falls back to its default) and the schema then coerces id to the integer 42.

• Paths accept $.field, $.a.b, $.items[0] (the $. prefix is optional). Dots in the output key nest the result. An absent optional source field is omitted, not emitted as null.
• Coercions are the lossless ones only: string → integer, string → number, string → boolean, number → string.

Decode an embedded JSON string

A field carrying a JSON document as a string is decoded via contentMediaType (JSON Schema 2020-12) — opt-in per field, never done by plain coercion:

- 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 with the same coercion/defaults/validation — so an inner qty: "7" arrives as 7.

Handling failures

Failures are non-retryable and name the field, e.g. transform failed at $.items[1].qty [coercion]: cannot coerce string "oops" to integer.

• On an output, the message is failed so a following dlq captures it.
• On an input, it’s dropped from the batch and acknowledged, keeping invalid data out.
• 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.

Full option list, the supported schema subset, and the exact coercion rules are in the middleware reference → transform.

Content-based routing (switch)

The switch structural endpoint picks a destination by the value of a metadata key. Output only.

output:
  switch:
    metadata_key: "country_code"
    cases:
      US: { kafka: { topic: "us_orders", url: "kafka-us:9092" } }
      EU: { nats:  { subject: "eu_orders", url: "nats-eu:4222" } }
    default: { file: { path: "/var/data/unroutable_orders.log" } }

A message whose key is missing or unmatched goes to default; without a default it is dropped.

Route on the payload, not just metadata

switch matches on metadata, not payload fields. To route on payload content, first promote the value into metadata. Two common ways:

• An endpoint that already emits a status key — e.g. http_status_code from an HTTP request, or mongodb.outcome from a Mongo upsert (see Upserts).
• transform with on_error: pass_through, which sets mqb.transform_error on failed records so you can shunt them aside:

output:
  middlewares:
    - transform: { schema_file: "schemas/order.json", on_error: pass_through }
  switch:
    metadata_key: "mqb.transform_error"
    cases: {}                                  # (no exact-match cases)
    default: { kafka: { topic: "orders", url: "localhost:9092" } }

Split HTTP responses by status

Pair switch with request, which forwards a response (or, on error/timeout, the original message) tagged with a status key:

output:
  request:
    to: { http: { url: "https://api.internal/score" } }
    forward_to:
      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" } }

See also Fan-out to send to all destinations instead of picking one.

Fan-out

The fanout structural endpoint publishes each message to every listed endpoint. Output only. The value is a plain list — each branch may have its own middleware and may itself be structural.

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

All branches receive the same message. Use it to mirror a stream to an archive while it also flows to its primary sink, to tee traffic to an audit log, or to emulate a static subscriber set (one branch per subscriber).

Fan-out vs. switch

• fanout → every destination gets the message.
• switch → one destination, chosen by a metadata key.

Give one branch its own reliability

Because each branch is a full endpoint, you can wrap just the fragile one with retry/dlq while the others stay plain:

output:
  fanout:
    - kafka: { topic: "orders", url: "localhost:9092" }
    - middlewares:
        - retry: { max_attempts: 5 }
        - dlq: { endpoint: { file: { path: "audit-failed.jsonl" } } }
      http: { url: "https://audit.internal/ingest", method: "POST" }

Weak join / correlation

The weak_join middleware correlates messages by a metadata key and emits them as one joined message. Input only. It has two modes.

Count mode

Wait for any N messages sharing a correlation key, then emit them as a JSON array:

input:
  middlewares:
    - weak_join: { group_by: "correlation_id", expected_count: 3, timeout_ms: 5000 }
  kafka: { topic: "fragments", url: "localhost:9092" }

Branch mode

Set branch_by to wait for named branches (e.g. one message from inventory and one from pricing), then emit a branch-keyed JSON object. required overrides expected_count:

- weak_join:
    group_by: "correlation_id"
    expected_count: 2
    timeout_ms: 5000
    branch_by: "source"
    required: ["inventory", "pricing"]
    on_timeout: discard

From the copy CLI, object/array fields take a JSON literal:

--from '...|weak-join?group_by=cid&expected_count=2&timeout_ms=1000&required=["inventory","pricing"]'

On timeout

An incomplete group is either emitted partially (on_timeout: fire, the default) or dropped (on_timeout: discard).

Durability caveat. Messages are acknowledged on receipt, so a crash before the group completes loses the buffered members — weak_join correlates in memory and is not a durable join. For correlation that must survive a restart, land the fragments in a store and join there.

Full field list in the middleware reference → weak_join.

Encryption at rest

Two ways to encrypt data, depending on whether you want it encrypted in transit between endpoints or at rest in a file / object.

The encryption middleware — payloads in transit

The encryption middleware seals 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. Requires the encryption feature.

output:
  middlewares:
    - encryption: { key: "${env:MQB_ENC_KEY}" }
  nats: { subject: "secure.orders", url: "nats://localhost:4222" }

The key is a base64-encoded 32-byte key; ${env:VAR} reads it from the environment (see Secrets & interpolation). Cipher defaults to xchacha20poly1305 (aes256gcm also available).

Key rotation: seal with a new key_id/key while listing the old key under decrypt_keys on the consuming side. Each payload is authenticated independently — tampering, a torn frame, or a wrong/missing key is a hard consumer error, not a silent drop.

The AEAD binds only the payload (empty associated data), so a sealed payload can be replayed under different metadata. Use deduplication or a sink uniqueness constraint if that matters.

File / object encryption at rest

To store data compressed and encrypted at rest, use the file / object_store endpoints’ own compression + encryption fields — they apply compress-then-encrypt per batch, which is what you want (ciphertext does not compress):

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

Do not stack the encryption middleware on top of a sink’s batch compression on the same route — ciphertext won’t compress. Use the endpoint fields above instead.

An encrypted file is written as length-prefixed sealed frames (one per batch) and is only readable through a matching consumer. object_store derives its object extension from these fields (e.g. .jsonl.gz / .jsonl.lz4, plus a trailing .enc when encryption is on).

Reading it back

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. See Troubleshooting and the Compression recipe.

Secrets & interpolation

Keep credentials out of committed config, and template message bodies from request data. There are two related interpolation systems: config-value env references (resolved when the config loads) and ${namespace:selector} message templating (resolved per message, in static bodies and a few endpoint fields).

Config values from the environment

Reference environment variables anywhere in JSON/YAML/UI values with ${ENV_VARIABLE_NAME:-default_if_not_found}:

orders_out:
  output:
    kafka:
      topic: "orders"
      url: "${KAFKA_URL:-localhost:9092}"
      # a password sourced from the environment, never committed
      sasl_password: "${KAFKA_PASSWORD}"

For local development, drop a .env file in the working directory — it is loaded automatically. In containers/Kubernetes, set the vars in the environment (and override any config field with MQB__{ROUTE}__{PATH}; see Configuration grammar). This is the recommended way to keep secrets out of source — see the deploying security checklist.

Encryption keys from the environment

The encryption middleware and the file endpoints read their key with the ${env:VAR} form:

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

Message templating with ${namespace:selector}

The static endpoint’s body is a template compiled once at startup. Tokens use the ${namespace:selector} form:

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

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

SQL / query token mapping

The sqlx insert_query supports the same request-binding tokens so a typed value goes into the statement: ${payload:field} binds a typed JSON value, ${metadata:key} binds a metadata string. See Upserts & insert-if-absent.

Compression

The file and object_store endpoints can compress each batch on write with the compression field (requires the compression feature):

output:
  file:
    path: "data.jsonl"
    format: json
    compression: lz4        # none | gzip | lz4 | zstd

object_store derives its default object extension from the codec (e.g. .jsonl.gz / .jsonl.lz4).

Reading it back

A file source must declare the same compression the data was written with:

input:
  file:
    path: "data.jsonl"
    format: json
    compression: lz4

A mismatch — wrong codec, or a missing field — is a permanent decode failure: the route ends failed with the error in its status rather than silently 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.

File compression 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.

Compression and encryption

Do not stack the encryption middleware on top of a sink’s batch compression on the same route — ciphertext does not compress. For compressed and encrypted data at rest, use the endpoints’ own fields, which apply compress-then-encrypt per batch:

output:
  file:
    path: "data.enc"
    format: raw
    compression: lz4
    encryption: { key: "${env:MQB_ENC_KEY}" }

An encrypted file is written as length-prefixed sealed frames (one per batch) and is only readable through a matching consumer; object_store adds a trailing .enc since the object is ciphertext, not a directly decompressible .gz.

See also

• Encryption at rest — the encryption side and key handling.
• Middleware & structural endpoints — the authoritative compression/encryption field docs.
• Performance tuning → Compression & encryption cost.

Endpoints (connectors)

An endpoint is a source (consumer) or sink (publisher) for messages. Every endpoint — whatever the underlying protocol — is driven behind the same receive_batch / send_batch shape, so any source can feed any sink. In config, an endpoint is a single-key object naming the connector:

input:  { kafka:   { url: "localhost:9092", topic: "orders" } }
output: { mongodb: { url: "mongodb://localhost:27017", database: "app", collection: "orders" } }

This page catalogs the transport endpoints and their behaviour. For per-connector query parameters used by copy (name, type, default, required, description), see the generated URL parameter reference and the hand-written connector pages. Structural endpoints (ref, fanout, switch, response, …) are documented separately in Structural endpoints.

Supported transports

Kafka, NATS, AMQP (RabbitMQ), MQTT, MongoDB, Postgres CDC (logical replication), PostgreSQL / MySQL / SQLite / MariaDB (SQLx), ClickHouse, HTTP, WebSocket, gRPC, ZeroMQ, Redis Streams, AWS SQS/SNS, cloud object storage (S3 / GCS / Azure), IBM MQ, files, and in-memory channels (memory, in-process and cross-process IPC).

Consumer vs. subscriber, and nack support

Endpoints default to a Consumer pattern (a queue: messages are distributed among workers). To get Subscriber (pub/sub) behaviour, per-backend config is required.

• Request-Reply — Native uses protocol-level correlation (HTTP connection, NATS reply subject). Emulated publishes a new message to a reply destination (the reply_to metadata field) carrying a correlation_id.
• Nack — Yes is explicit negative acknowledgement triggering redelivery. Eventual means redelivery depends on timeout or a connection drop. Simulated is handled in-memory. Consumer-local means the nacked message is retried inside the consumer process and is lost if that process dies — the producer is never notified.

Database sources: change capture vs. polling

Databases have no native pub/sub, so a database source is read one of two ways:

• Change Data Capture (CDC) tails the database’s own change log — inserts and updates/deletes — and resumes from a durable log position.
  • PostgreSQL — the postgres_cdc endpoint streams a logical-replication slot (pgoutput). It emits flat JSON rows tagged with postgres.operation metadata (insert/update/delete/truncate). Acking a batch confirms the LSN back to the server; a nack or interrupted run does not advance the confirmed LSN, so replication resumes from the last acknowledged position — at-least-once. Enable with the postgres-cdc feature; requires wal_level = logical and a publication. See the Postgres CDC → JSONL tutorial.
  • MongoDB — set consume: capture_new to watch an existing collection for changes from now on, or consume: capture_all to read existing documents first and then keep capturing (no gap, at-least-once). It emits insert/update/replace/delete tagged with mongodb.operation and checkpoints under cursor_id. Change streams need a replica set.
• Cursor polling pages an existing table by a monotonic cursor_column (WHERE col > $last ORDER BY col ASC), persisting the last read value under cursor_id. Captures appends only — updates and deletes are not observed. Available on SQLx (PostgreSQL / MySQL / MariaDB / SQLite) and ClickHouse; the poll interval backs off exponentially between polling_interval_ms and max_polling_interval_ms. SQLite and ClickHouse are polling-only.

SQLx read mode is chosen by config, not by driver. With no cursor_column, an SQLx source is a competing-consumers work queue: it atomically claims rows via a locked_until lease and deletes them on ack. This requires the table to carry the queue schema — an id, a payload, and a locked_until column — on every driver; it is not a plain full-table read. To read an arbitrary table non-destructively (the ETL path), set cursor_column to switch to cursor polling. A source table missing locked_until fails fast with a permanent error rather than reconnecting forever.

MongoDB consume modes

These are not interchangeable. The default consumer buys exclusivity via four round trips per batch; where you only need a single-reader bulk read or ETL pass, use capture_all — it is roughly 5x faster (500k docs, standalone MongoDB: consumer → null 23,667 rows/s vs. capture_all → jsonl 120,308 rows/s). Note concurrency does not speed up a MongoDB source; batches are fetched serially.

Cloud object storage (S3 / GCS / Azure)

The object_store endpoint (alias s3) reads and writes Amazon S3, Google Cloud Storage, Azure Blob, Cloudflare R2, and anything else the object_store crate speaks. Enable with the object-store feature. Credentials and backend options come from the environment (AWS_ACCESS_KEY_ID, AWS_REGION, GOOGLE_SERVICE_ACCOUNT, AZURE_STORAGE_ACCOUNT, …); the URL scheme picks the backend (s3://, gs://, az://).

• As a sink, each flushed batch is encoded with the file formats (normal JSONL, json, text, raw) and written as one immutable object at <prefix>/[YYYY/MM/DD/]<uuidv7>.<ext>. Objects are write-once.
• As a source, objects under the prefix are listed in key order, fetched whole, split on the delimiter, and emitted. Progress is a durable cursor holding the last fully-acked object key: set cursor_id and an external checkpoint_store (file://, s3://, postgres://, mongodb://) so a restart resumes. Objects are never deleted or rewritten.

archive_to_s3:
  input:  { memory: { topic: "events" } }
  output: { object_store: { url: "s3://my-bucket/events", format: normal } }

replay_from_s3:
  input:
    object_store:
      url: "s3://my-bucket/events"
      cursor_id: "replayer-1"
      checkpoint_store: "file:///var/lib/mqb/s3-cursor.json"
  output: { nats: { subject: "events.replay", url: "nats://localhost:4222" } }

Point checkpoint_store at a different bucket/prefix than the source reads; a cursor object under the source prefix would be listed and re-read as data.

File and object formats

Both file and object_store share the encodings: normal/json/text write the {message_id, payload, metadata} wrapper (the message id survives the round trip); raw writes payloads verbatim (bare documents, no wrapper). csv is supported as a source (and on file sinks). The file endpoints also carry their own compression and encryption fields — see the Compression and Encryption at rest recipes.

Memory endpoint and IPC

The memory endpoint is both an in-process channel and a cross-process IPC transport over Unix domain sockets / Windows named pipes. Its topic field (alias url) doubles as a transport URL. See the Cross-process IPC bridge tutorial and Learn the architecture for the URL schemes, roles (consumer is server, publisher is client), wire format and backpressure.

IBM MQ

IBM MQ is included in the full feature set via the ibm-mq feature, which loads the IBM MQ client library at runtime via dlopen — no IBM SDK is needed to build. The redistributable client only has to be present at runtime, and only if you actually use an IBM MQ endpoint (it is loaded lazily on first connect). The loader finds the client via the platform default name, MQ_INSTALLATION_PATH (e.g. /opt/mqm), or an explicit MQB_IBM_MQ_LIB path. To link statically at build time, use ibm-mq-static (requires the IBM MQ SDK). IBM MQ has its own TLS shape (IbmTlsConfig): tls.cert_file (alias key_repository) is a CMS key repository path, not a PEM file. See the IBM MQ setup guide.

Connection sharing

Publishers that target the same server reuse one underlying transport client by default, consolidating TCP connections, background threads, and batching. Sharing applies to Kafka, NATS, MongoDB, SQLx, HTTP, and gRPC; the client is keyed by connection-level settings (URL, auth, TLS, client options), never by topic/subject/collection. Set shared: false on a publisher to give it a dedicated connection — see Configuration grammar and Performance tuning.

Idempotent writes

For ETL, at-least-once delivery plus an idempotent write gives effective exactly-once. The most robust place to enforce this is the sink’s own unique constraint. See the Upserts & insert-if-absent and Deduplication recipes for the per-sink mechanisms (id_field on MongoDB, ON CONFLICT/ON DUPLICATE KEY on SQL, ReplacingMergeTree on ClickHouse, deterministic message_id + postgres.key on Postgres CDC).

Connectors

Hand-written per-connector docs: purpose, URL format, and practical examples. For the full, auto-generated list of every recognised query parameter (type, default, required, description), see the matching page under ../reference/.

• PostgreSQL / MySQL / MariaDB / SQLite (including PostgreSQL CDC)
• ClickHouse
• MQTT
• Kafka
• RabbitMQ (AMQP)
• NATS
• Redis Streams
• HTTP
• WebSocket
• gRPC
• MongoDB
• AWS SQS / SNS
• ZeroMQ
• IBM MQ
• File (CSV / JSON / JSONL)

Every connector recognised by mq-bridge-app copy now has a hand-written page. For the full, auto-generated parameter tables see the matching page under ../reference/.

PostgreSQL / MySQL / MariaDB / SQLite

The sqlx connector reads and writes rows in a relational database table. The same connector backs all four schemes — the URL scheme just selects the driver. Postgres additionally has a dedicated CDC connector for streaming change data instead of table reads.

URL format

postgres://[user:pass@]host[:port]/database?table=<name>

The scheme (postgres/postgresql, mysql, mariadb, sqlite) and everything up to the query string is passed straight through as the driver connection string; table (and any other recognised field below) is pulled out of the query string into config. For SQLite, host/database are replaced by a file path, e.g. sqlite:///var/data/app.db?table=orders.

Examples

Full-table read (source), one-shot:

mq-bridge-app copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to null:

Write with auto-created table (destination):

mq-bridge-app copy --drain \
  --from file:///data/orders.csv?format=csv \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Resumable incremental read, keyed by a monotonic column, continuous:

mq-bridge-app copy \
  --from 'postgres://user:pass@localhost/app?table=orders&cursor_column=id&cursor_id=orders_export' \
  --to kafka://kafka.local:9092?topic=orders

Each restart resumes from the last id seen (persisted via cursor_id) instead of re-copying from the start.

Custom multi-column insert, MySQL:

mq-bridge-app copy --drain \
  --from postgres://user:pass@src/app?table=orders \
  --to 'mysql://user:pass@dst/app?table=orders&insert_query=INSERT+INTO+orders+%28id%2C+sku%2C+qty%29+VALUES+%28%24%7Bpayload%3Aid%7D%2C+%24%7Bpayload%3Asku%7D%2C+%24%7Bpayload%3Aqty%7D%29'

(insert_query is shown URL-encoded above — the SQL is INSERT INTO orders (id, sku, qty) VALUES (${payload:id}, ${payload:sku}, ${payload:qty}).)

Key options

Any other query parameter (e.g. sslmode=disable) is left on the connection URL untouched and passed to the driver as-is.

Full field list, types, and defaults: reference/postgres.md.

PostgreSQL CDC

A separate connector for streaming logical-replication changes (insert/ update/delete) instead of reading a table snapshot. Uses postgres-cdc:// (alias pgcdc://) to select the endpoint kind; the connection URL underneath it is a plain Postgres URL.

postgres-cdc://[user:pass@]host[:port]/database?publication=<name>&slot_name=<name>

Stream changes from a publication into Kafka, continuous:

mq-bridge-app copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to kafka://kafka.local:9092?topic=app-changes

Replicate a table into another PostgreSQL instance, continuous:

mq-bridge-app copy \
  --from 'postgres-cdc://user:pass@localhost/app?publication=mqb_pub&slot_name=mqb_slot' \
  --to 'postgres://user:pass@otherhost/replica?table=orders&auto_create_table=true'

publication must already exist on the source (CREATE PUBLICATION mqb_pub FOR TABLE orders;); slot_name is created automatically if missing.

Full field list: reference/postgres-cdc.md.

ClickHouse

Reads from or bulk-inserts into a ClickHouse table over ClickHouse’s HTTP interface (port 8123 by default; 8443 for HTTPS).

URL format

clickhouse://host[:port]?table=<name>[&database=<name>]

clickhouse:// is rewritten to http:// (and clickhouses:// to https://) before being handed to the ClickHouse client — the scheme only selects the endpoint kind on the CLI.

Examples

Bulk insert from a full-table Postgres read, one-shot:

mq-bridge-app copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to 'clickhouse://localhost:8123?table=orders&database=analytics'

Async insert for high-throughput streaming writes, continuous:

mq-bridge-app copy \
  --from kafka://kafka.local:9092?topic=events \
  --to 'clickhouse://user:pass@ch.local:8123?table=events&database=analytics&async_insert=true'

Resumable, non-destructive read of an existing table into Kafka:

mq-bridge-app copy \
  --from 'clickhouse://localhost:8123?table=events&database=analytics&cursor_column=id&cursor_id=events_export' \
  --to kafka://kafka.local:9092?topic=events

Each restart resumes from the last id seen instead of re-reading from the start. (Per-column mapping via columns is a map field, so it can’t be set from a query param — use a YAML route config for that.)

Key options

Full field list: reference/clickhouse.md.

MQTT

Publishes to or subscribes from an MQTT broker (v5 by default, v3 supported).

URL format

mqtt://[user:pass@]host[:port]?topic=<topic>

mqtt:// is rewritten to tcp:// (and mqtts:// to ssl://) before being handed to the MQTT client — the scheme only selects the endpoint kind on the CLI. MQTT topic wildcards (+, #) are supported on the source side.

Examples

Subscribe to a wildcard topic and forward to Kafka, continuous:

mq-bridge-app copy \
  --from mqtt://broker.local:1883?topic=sensors/+/temperature \
  --to kafka://kafka.local:9092?topic=sensor-readings

Publish a file’s lines to a topic, one-shot:

mq-bridge-app copy --drain \
  --from file:///data/events.jsonl?format=json \
  --to mqtts://user:pass@broker.local:8883?topic=events

Fixed client ID and QoS 2 for exactly-once delivery semantics:

mq-bridge-app copy \
  --from 'mqtt://broker.local:1883?topic=alerts&client_id=mqb-alerts-01&qos=2' \
  --to null:

Key options

Full field list: reference/mqtt.md.

Kafka

Produces to or consumes from a Kafka topic via librdkafka.

URL format

kafka://broker[:port]?topic=<topic>

kafka:// is stripped before being handed to librdkafka’s bootstrap.servers, which expects a bare host:port list, not a URI — the scheme only selects the endpoint kind on the CLI. For a multi-broker cluster, use the ?url= escape hatch (see Quick Start): kafka://_/?url=broker1:9092,broker2:9092&topic=orders.

Examples

Forward an MQTT stream into a Kafka topic, continuous:

mq-bridge-app copy \
  --from mqtt://broker.local:1883?topic=sensors/+/temperature \
  --to kafka://kafka.local:9092?topic=sensor-readings

Consume with a durable consumer group, continuous:

mq-bridge-app copy \
  --from 'kafka://kafka.local:9092?topic=orders&group_id=mqb-orders-sync' \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Without group_id, the consumer runs in ephemeral subscriber mode (unique group ID, starts from the latest offset).

SASL-authenticated broker:

mq-bridge-app copy \
  --from postgres://user:pass@localhost/app?table=orders \
  --to 'kafka://kafka.local:9093?topic=orders&username=svc&password=secret'

TLS (an object-typed field, not a scalar) can’t be set via a query param; use the ?url= escape hatch to pass a librdkafka connection string with TLS options embedded.

Key options

Full field list: reference/kafka.md.

RabbitMQ (AMQP)

Publishes to or consumes from a RabbitMQ queue over the AMQP 0-9-1 protocol.

URL format

rabbitmq://[user:pass@]host[:port]/<vhost>?queue=<name>

amqp:///amqps:// (the native AMQP scheme) and rabbitmq:/// rabbitmqs:// (an alias) are both accepted; rabbitmq(s) is rewritten to amqp(s) before being handed to the driver. The default vhost / must be percent-encoded as %2f in the URL path, per the AMQP URI spec.

Examples

Consume a queue and forward each message to an HTTP endpoint, continuous:

mq-bridge-app copy \
  --from rabbitmq://guest:guest@localhost:5672/%2f?queue=orders \
  --to http://internal-api.local/ingest?method=POST

Publish to an exchange instead of a default-exchange queue:

mq-bridge-app copy --drain \
  --from file:///data/events.jsonl?format=json \
  --to 'amqp://guest:guest@localhost:5672/%2f?exchange=events&queue=events'

Fan-out subscriber mode (ephemeral queue bound to the exchange):

mq-bridge-app copy \
  --from 'amqp://guest:guest@localhost:5672/%2f?exchange=events&subscribe_mode=true' \
  --to kafka://kafka.local:9092?topic=events

Key options

Full field list: reference/rabbitmq.md.

NATS

Publishes to or consumes from NATS subjects. Uses JetStream by default (durable, acked); set no_jetstream=true for fire-and-forget Core NATS.

URL format

nats://host[:port]?subject=<subject>[&stream=<name>]

For a multi-server cluster, pass a comma-separated list via the ?url= escape hatch: nats://_/?url=nats://n1:4222,nats://n2:4222&subject=orders. Consumers require stream (the JetStream stream name), even with no_jetstream=true (validated but unused there). If the bridge auto-creates a stream it is scoped to {stream}.>, so prefix your subject accordingly.

Examples

Load a JSONL file into a JetStream subject, one-shot:

mq-bridge-app copy --drain \
  --from file:///data/orders.jsonl?format=json \
  --to 'nats://localhost:4222?subject=orders&stream=ORDERS'

Consume a durable JetStream stream into Postgres, continuous:

mq-bridge-app copy \
  --from 'nats://localhost:4222?subject=orders&stream=ORDERS' \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Core NATS request/reply (no JetStream), continuous:

mq-bridge-app copy \
  --from http://0.0.0.0:8080/rpc \
  --to 'nats://localhost:4222?subject=rpc.echo&no_jetstream=true&request_reply=true'

Key options

Full field list: reference/nats.md.

Redis Streams

Publishes to or consumes from a Redis stream (XADD / XREADGROUP). Consumers use a consumer group by default for durable, acked delivery; subscriber_mode=true reads ephemerally via XREAD.

URL format

redis://[user:pass@]host[:port]?stream=<key>

redis:// (plain) and rediss:// (TLS) are both accepted; redis_streams:// is an explicit alias. If stream is omitted it defaults to the route name.

Examples

Load a CSV file into a stream, one-shot:

mq-bridge-app copy --drain \
  --from file:///data/events.csv?format=csv \
  --to redis://localhost:6379?stream=events

Consume with a durable group into ClickHouse, continuous:

mq-bridge-app copy \
  --from 'redis://localhost:6379?stream=events&group=analytics&consumer_name=w1' \
  --to 'clickhouse://localhost:8123?table=events&database=analytics'

Unclaimed entries pending longer than redelivery_timeout_ms (default 60s) are re-delivered via XAUTOCLAIM.

Ephemeral tail of new messages (no group, no acks), continuous:

mq-bridge-app copy \
  --from 'redis://localhost:6379?stream=events&subscriber_mode=true' \
  --to kafka://kafka.local:9092?topic=events

Key options

Full field list: reference/redis.md.

HTTP

As a publisher, sends each message as an HTTP request to a target URL. As a consumer, runs an embedded HTTP server and turns incoming requests into messages.

URL format

http://host[:port][/path]?method=<verb>

http:///https:// pass through unchanged — they’re already the native scheme the driver expects (the target path, if any, is just part of the URL, not a separate query param).

Examples

Consume a RabbitMQ queue and POST each message to an API, continuous:

mq-bridge-app copy \
  --from rabbitmq://guest:guest@localhost:5672/%2f?queue=orders \
  --to http://internal-api.local/ingest?method=POST

Run an HTTP listener as the source (webhook receiver), continuous:

mq-bridge-app copy \
  --from http://0.0.0.0:8080?method=POST \
  --to kafka://kafka.local:9092?topic=webhooks

Non-blocking publisher (don’t wait for the downstream response):

mq-bridge-app copy --drain \
  --from file:///data/events.jsonl?format=json \
  --to 'https://api.example.com/ingest?method=POST&request_timeout_ms=5000'

Key options

Full field list: reference/http.md.

WebSocket

Acts as a WebSocket server (consumer — listens for connections) or client (publisher — connects to a target URL). As a consumer it receives frames; as a publisher it sends them.

URL format

# Consumer (listen):   ws://<bind-address>
# Publisher (connect):  ws://host[:port]/path

ws:// (plain) and wss:// (TLS) are both accepted. For a consumer the URL is the listen address (e.g. ws://0.0.0.0:9000); for a publisher it is the target server URL.

Examples

Listen for WebSocket frames and forward them to Kafka, continuous:

mq-bridge-app copy \
  --from ws://0.0.0.0:9000 \
  --to kafka://kafka.local:9092?topic=ws-events

Only accept a specific path:

mq-bridge-app copy \
  --from ws://0.0.0.0:9000?path=/ingest \
  --to file:///data/ws.jsonl?format=json

Push a stream to a remote WebSocket server, continuous:

mq-bridge-app copy \
  --from redis://localhost:6379?stream=events \
  --to wss://feed.example.com/socket

Key options

Full field list: reference/websocket.md.

gRPC

Sends or receives messages over gRPC. As a client (default) it connects to a remote server and issues Publish / PublishBatch RPCs; as a server (server_mode=true) it starts an embedded tonic server that accepts those RPCs.

URL format

grpc://host[:port]?topic=<topic>

grpc:// (plain) and grpcs:// (TLS) are both accepted. In client mode the URL is the remote server (e.g. grpc://localhost:50051); in server mode it is the bind address (e.g. grpc://0.0.0.0:50051).

Examples

Forward a Kafka topic to a remote gRPC service, continuous:

mq-bridge-app copy \
  --from kafka://kafka.local:9092?topic=orders \
  --to grpc://orders-svc.local:50051?topic=orders

Run an embedded gRPC server that ingests into Postgres, continuous:

mq-bridge-app copy \
  --from 'grpc://0.0.0.0:50051?server_mode=true&topic=orders' \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Key options

Full field list: reference/grpc.md.

MongoDB

Reads from or writes to a MongoDB collection. database and collection are separate query params — MongoDB URIs don’t encode the database in the path the way this connector reads them (unlike PostgreSQL).

URL format

mongodb://[user:pass@]host[:port]?database=<db>&collection=<name>

Examples

Load a CSV file into a collection, one-shot:

mq-bridge-app copy --drain \
  --from file:///data/customers.csv?format=csv \
  --to 'mongodb://localhost?database=app&collection=customers'

Non-destructive read of an existing collection (default source behavior):

mq-bridge-app copy --drain \
  --from 'mongodb://localhost?database=app&collection=customers' \
  --to 'postgres://user:pass@localhost/app?table=customers&auto_create_table=true'

By default (no consume given), a MongoDB source reads existing documents then watches for changes (capture_all) — pointing at a collection never claims/deletes its documents, unlike the library’s own consumer default.

Watch for new documents only (change stream), continuous:

mq-bridge-app copy \
  --from 'mongodb://localhost?database=app&collection=orders&consume=capture_new' \
  --to kafka://kafka.local:9092?topic=orders

Key options

Full field list: reference/mongodb.md.

AWS SQS / SNS

Consumes from an SQS queue or publishes to an SQS queue / SNS topic. Credentials come from the standard AWS provider chain (environment, profile, instance role) unless supplied explicitly as query params.

URL format

aws://?queue_url=<sqs-url>&region=<region>

aws:// and aws-sqs:// are both accepted. The host is unused — the queue is selected by queue_url (required for consumers, and for publishers unless topic_arn is set for SNS). Point endpoint_url at LocalStack for local testing.

Examples

Drain an SQS queue into a file, one-shot:

mq-bridge-app copy --drain \
  --from 'aws://?queue_url=https://sqs.us-east-1.amazonaws.com/1234/orders&region=us-east-1' \
  --to file:///data/orders.jsonl?format=json

Publish a Postgres table to an SNS topic, one-shot:

mq-bridge-app copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to 'aws://?topic_arn=arn:aws:sns:us-east-1:1234:orders&region=us-east-1'

LocalStack, continuous:

mq-bridge-app copy \
  --from 'aws://?queue_url=http://localhost:4566/000000000000/orders&region=us-east-1&endpoint_url=http://localhost:4566&access_key=test&secret_key=test' \
  --to null:

Warning

access_key / secret_key in the URL are shown here only because LocalStack’s test/test are throwaway values. Never put real credentials in a connector URL — they leak into shell history and process listings. Against real AWS, omit both and let the standard provider chain (env vars, ~/.aws/credentials, IAM role) supply them; see Secrets.

Key options

Full field list: reference/aws.md.

ZeroMQ

Sends or receives messages over ZeroMQ sockets (PUSH/PULL, PUB/SUB, REQ/REP). The socket_type selects the pattern and bind decides whether the endpoint binds or connects.

URL format

zeromq://<transport>?socket_type=<type>[&bind=true]

zeromq:// and zmq:// are both accepted. The URL is a ZeroMQ transport address such as tcp://127.0.0.1:5555. Choose bind=true on exactly one side of a socket pair; the other connects.

Examples

Pull from a PUSH producer and write to a file, continuous:

mq-bridge-app copy \
  --from 'zeromq://tcp://127.0.0.1:5555?socket_type=pull&bind=true' \
  --to file:///data/events.jsonl?format=json

Publish a Kafka topic to a PUB socket, continuous:

mq-bridge-app copy \
  --from kafka://kafka.local:9092?topic=events \
  --to 'zeromq://tcp://0.0.0.0:5556?socket_type=pub&bind=true'

Subscribe to a topic on a remote PUB socket, continuous:

mq-bridge-app copy \
  --from 'zeromq://tcp://feed.local:5556?socket_type=sub&topic=orders' \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Key options

Full field list: reference/zeromq.md.

IBM MQ

Puts to or gets from an IBM MQ queue (point-to-point) or topic (publish/subscribe). Requires the IBM MQ client libraries and the optional IBM MQ build — see IBM MQ setup for prerequisites.

URL format

ibmmq://host(port)?queue_manager=<QM>&channel=<CHANNEL>&queue=<QUEUE>

ibmmq:// and ibm-mq:// are both accepted. The host is given in IBM MQ’s host(port) form, and a comma-separated list provides failover (host1(1414),host2(1414)). queue_manager and channel (the SVRCONN channel) are always required; supply queue or topic.

Examples

Drain a queue into a file, one-shot:

mq-bridge-app copy --drain \
  --from 'ibmmq://mq.local(1414)?queue_manager=QM1&channel=DEV.APP.SVRCONN&queue=DEV.QUEUE.1&username=app&password=<password>' \
  --to file:///data/mq.jsonl?format=json

Credentials on the command line land in shell history and process listings. Keep them out of the URL by interpolating an environment variable instead — see Secrets.

Bridge a queue into Kafka, continuous:

mq-bridge-app copy \
  --from 'ibmmq://mq.local(1414)?queue_manager=QM1&channel=DEV.APP.SVRCONN&queue=DEV.QUEUE.1' \
  --to kafka://kafka.local:9092?topic=mq-events

Subscribe to a topic (pub/sub), continuous:

mq-bridge-app copy \
  --from 'ibmmq://mq.local(1414)?queue_manager=QM1&channel=DEV.APP.SVRCONN&topic=orders/new' \
  --to 'postgres://user:pass@localhost/app?table=orders&auto_create_table=true'

Key options

Full field list: reference/ibmmq.md. Setup and prerequisites: IBM MQ setup.

File (CSV / JSON / JSONL)

Reads from or writes to a local file. Useful as a one-shot source/sink for migrating data in or out of the other connectors.

URL format

file:///absolute/path/to/file?format=<normal|json|text|raw|csv>

The path comes from the URI path itself (file:///...), not a query param. format defaults to normal (the full message serialized as JSON).

Examples

Load a CSV file into MongoDB, one-shot (first row = header):

mq-bridge-app copy --drain \
  --from file:///data/customers.csv?format=csv \
  --to 'mongodb://localhost?database=app&collection=customers'

Export a table to JSONL, one-shot:

mq-bridge-app copy --drain \
  --from postgres://user:pass@localhost/app?table=orders \
  --to file:///data/orders.jsonl?format=json

Tail a file as it grows (broadcast/subscribe mode), continuous:

mq-bridge-app copy \
  --from file:///var/log/app/events.log?mode=subscribe \
  --to kafka://kafka.local:9092?topic=app-events

Key options

Full field list: reference/file.md.

URL Parameter Reference

Auto-generated from the same JSON Schemas mq-bridge-app copy uses to parse --from/--to query parameters (see crates/cli/src/main.rs::endpoint_from_uri). For workflow examples, see the connector pages.

• PostgreSQL / MySQL / MariaDB / SQLite — schemes: postgres://, postgresql://, mysql://, mariadb://, sqlite://
• PostgreSQL CDC (logical replication) — schemes: postgres-cdc://, pgcdc://
• ClickHouse — schemes: clickhouse://, clickhouses://
• Kafka — schemes: kafka://
• MQTT — schemes: mqtt://, mqtts://
• RabbitMQ (AMQP) — schemes: amqp://, amqps://, rabbitmq://, rabbitmqs://
• HTTP — schemes: http://, https://
• WebSocket — schemes: ws://, wss://
• MongoDB — schemes: mongodb://
• File (CSV / JSON / JSONL) — schemes: file://
• NATS — schemes: nats://
• Redis Streams — schemes: redis://, rediss://, redis_streams://
• gRPC — schemes: grpc://, grpcs://
• AWS SQS / SNS — schemes: aws://, aws-sqs://
• ZeroMQ — schemes: zeromq://, zmq://
• IBM MQ — schemes: ibmmq://, ibm-mq://

PostgreSQL / MySQL / MariaDB / SQLite

Schemes: postgres://, postgresql://, mysql://, mariadb://, sqlite://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

PostgreSQL CDC (logical replication)

Schemes: postgres-cdc://, pgcdc://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

publication_tables

[]

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

MongoDB

Schemes: mongodb://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

Kafka

Schemes: kafka://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

NATS

Schemes: nats://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

RabbitMQ (AMQP)

Schemes: amqp://, amqps://, rabbitmq://, rabbitmqs://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

MQTT

Schemes: mqtt://, mqtts://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

Redis Streams

Schemes: redis://, rediss://, redis_streams://

Query parameters recognised as config fields for this connector. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

ClickHouse

Schemes: clickhouse://, clickhouses://

Query parameters recognised as config fields for this connector. The object-typed columns is set with a JSON literal, e.g. ?columns={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

HTTP

Schemes: http://, https://

Query parameters recognised as config fields for this connector. The object-typed custom_headers is set with a JSON literal, e.g. ?custom_headers={...}. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

WebSocket

Schemes: ws://, wss://

Query parameters recognised as config fields for this connector. Any other ?key=value pair is passed through unchanged as a driver option on the connection URL.

gRPC

Schemes: grpc://, grpcs://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

Struct-typed fields

tls

TLS configuration for secure connections.

Configures Transport Layer Security (TLS/SSL) for encrypted communication. Supports both client certificate (mutual TLS) and server certificate validation.

Examples

use mq_bridge::models::TlsConfig;

let tls = TlsConfig {
    required: true,
    ca_file: Some("/path/to/ca.pem".to_string()),
    cert_file: Some("/path/to/cert.pem".to_string()),
    key_file: Some("/path/to/key.pem".to_string()),
    ..Default::default()
};

ZeroMQ

Schemes: zeromq://, zmq://

Query parameters recognised as config fields for this connector. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

IBM MQ

Schemes: ibmmq://, ibm-mq://

Query parameters recognised as config fields for this connector. The object-typed tls is set with a JSON literal, e.g. ?tls={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

Struct-typed fields

tls

TLS configuration for the IBM MQ native client.

The IBM MQ client doesn’t consume PEM files, so this uses MQ-native field names rather than the generic [TlsConfig] used by the other endpoints.

AWS SQS / SNS

Schemes: aws://, aws-sqs://

Query parameters recognised as config fields for this connector. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

File (CSV / JSON / JSONL)

Schemes: file://

Query parameters recognised as config fields for this connector. The object-typed encryption is set with a JSON literal, e.g. ?encryption={...}. Unrecognised parameters are not forwarded as driver options, so any other ?key=value pair is rejected rather than silently ignored.

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.