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:

mqb 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 338,066 rows/s at ~40 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 library/config primitive defaults to batch_size: 512, concurrency: 1: batches fill opportunistically — a route takes whatever is already queued rather than waiting — so throughput comes for free while parallelism stays a deliberate per-route choice.
• 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 is a single headless binary, installed under two names: mq-bridge-app and the short mqb, which is what the docs use and what you’ll normally type. mqb is a small launcher that hands straight over to the real binary, so the two are interchangeable — existing scripts and MCP registrations keep working untouched.

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 --ui --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

mqb copy SOURCE TARGET moves data between two endpoints described as URIs. The scheme picks the connector; ?query=params configure it. The existing --from SOURCE --to TARGET form is equivalent and remains supported.

mqb copy \
  'postgres://localhost/app?table=users' \
  '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 common copy controls are deliberately small:

mqb copy SOURCE TARGET [--filter EXPR] [--resume] [--drain]

--filter evaluates a readable expression against each top-level JSON payload, for example amount > 100 or status == "paid". A false result intentionally drops and acknowledges the message; malformed JSON and invalid expressions are errors, while a field that is absent or not a scalar counts as no match and is warned about once. It is an in-process filter and is not translated into a database query. Filtering into cloud object storage also changes how the objects are named — see Filtering.

--resume asks the source to use its native durable position and fails before the route starts when that is not safe. The generated state identity includes the credential-redacted source, destination, and filter, so changing pipeline semantics starts a new checkpoint while rotating a password does not.

The examples 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

mqb 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. For a resumable non-destructive scan, add &cursor_column=id on --from and pass --resume; the CLI supplies the stable cursor id and the SQL source stores the checkpoint in its own database. An explicit cursor_id or checkpoint_store in the URI still takes precedence. See PostgreSQL and ClickHouse.

Filtered, resumable Kafka copy

mqb copy \
  'kafka://localhost:9092?topic=orders' \
  'postgres://localhost/app?table=orders' \
  --filter 'status == "paid"' \
  --resume

The generated Kafka consumer group is stable for this source, destination, and filter. Kafka offsets advance only after the destination succeeds, or after a message is intentionally filtered out.

PostgreSQL CDC → PostgreSQL

mqb 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

mqb 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

mqb 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

mqb 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>:

mqb 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 mqb --schema <path>), 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 (mqb) 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
mqb --config config.yml

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

# Start empty, then open the UI to build your config
mqb --ui

In config mode the CLI can also serve the browser UI (the same UI as the desktop app) on the configured port. It is never opened implicitly: set ui_addr in the config, pass --ui, or answer the prompt — an unattended run without --ui stays headless. See the CLI reference for the full rule and every flag, and Configuration grammar for the config format.

What a config file has to contain

Everything a config file defines is started at boot, in both headless and UI runs. A route can be held back with enabled: false; the UI starts it on demand.

The full form is a routes: map alongside the application settings, which is what the UI writes out. Routes may also be written at the top level, so a file that is nothing but a route needs neither the wrapper nor a name — these three are the same bridge:

# 1. the full form
routes:
  proxy:
    input:  { http: { url: "0.0.0.0:8443" } }
    output: { http: { url: "https://upstream.internal/" } }

# 2. named at the top level, as in the Configuration grammar
proxy:
  input:  { http: { url: "0.0.0.0:8443" } }
  output: { http: { url: "https://upstream.internal/" } }

# 3. a single unnamed route — it runs as `route`
input:  { http: { url: "0.0.0.0:8443" } }
output: { http: { url: "https://upstream.internal/" } }

Application settings (ui_addr, metrics_addr, log_level, plugins, …) stay at the top level in every form. In form 3 every other top-level key belongs to the route, so route options such as batch_size and exit_on_empty go there — and a misspelled one is reported by name rather than ignored.

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

mqb 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 ~381 tokens — the same ~381 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
mqb 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)
mqb mcp

# streamable HTTP — for remote/shared clients
mqb 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 may be written inside route or as top-level start_route arguments. They resolve most-specific-first:

1. the top-level argument, if given;
2. otherwise the same key inside route, if given;
3. otherwise the app default — 1024 and 4, rather than the library’s 512/1.

Only a value you never wrote is replaced, so an explicit 1 is honoured either way. start_route echoes the resolved batch_size/concurrency in its result.

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.

Middleware

Either endpoint may carry a middlewares array, so delivery behaviour is added without changing the route:

{
  "input":  {"kafka": {"url": "localhost:9092", "topic": "orders", "group_id": "agent"}},
  "output": {
    "sqlx": {"url": "postgres://user:pass@localhost:5432/db", "table": "orders"},
    "middlewares": [
      {"retry": {"max_attempts": 5}},
      {"dlq": {"endpoint": {"file": {"path": "/tmp/failed.jsonl"}}}}
    ]
  }
}

Available: retry, dlq, deduplication, limiter, transform, compression, encryption, buffer, delay, weak_join, cookie_jar, metrics. Each is documented with its options in the Cookbook and Middleware reference.

Waiting for a job to finish

For a job started with exit_on_empty, call wait_route rather than polling route_status in a loop:

{"name": "redis-to-postgres", "timeout_ms": 300000}

It returns once the route’s task ends, reporting finished, outcome, messages, elapsed_s and average_messages_per_second. On timeout it returns finished: false and leaves the route running, so the agent can wait again or stop it. The default timeout is 60 s and the ceiling is 5 h; the server polls internally every 50 ms, which the client never pays for.

This is what keeps agent token cost flat: one call covers a job of any duration, where a polling loop costs a call per tick.

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 ~30-55 ms of startup and completion polling. In the latest run the MCP path measured marginally above the CLI on the same dataset — that is variance on a sub-second job, not the tool call being faster than the command.

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
mqb mcp install

# a single client, project-scoped rather than global
mqb mcp install --client cursor --local

# bake --report-to-ui into the registered command
mqb 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.

Ordering

Two independent guarantees, each decided once per route by the endpoint itself:

• Commits follow MessageConsumer::commit_requires_order() (default true). Cumulative-ack sources such as Kafka funnel their commits through one sequencer; individually-acking sources commit concurrently, bounded by commit_concurrency_limit.
• Publishing follows MessagePublisher::requires_ordered_publish() (default false). Above concurrency: 1 workers call send_batch in parallel, so whole batches can reach the sink out of source order — rows keep their order within a batch. The file sink declares true and the route sequences the send_batch calls; batch prep and commits stay parallel, so the ordered path measures within noise of an unordered one.

Only file declares it, for two reasons that any other candidate has to clear:

• The cost must be low. The file sink already holds a write lock across the whole batch, so sequencing changes who writes, not how many write at once. Sequencing a sink that ends in a network round trip instead costs the whole concurrency factor — one batch in flight instead of N.
• The guarantee must be reachable. object_store does not declare it: read order there is object-key order, and the uuidv7 key randomises everything below the millisecond, so ordering the writes would buy nothing. Ordered cloud export needs a source-sequenced key first.

Broker sinks keep it off. NATS (both Core and JetStream) publishes a batch through send_batch_helper, which keeps up to SEND_BATCH_CONCURRENCY publishes in flight, so wire order inside one batch is already unordered — only the results are re-sorted. Kafka and AMQP preserve order within a batch (they submit the whole batch, then await confirms), but for them the order-critical part is the cheap submit loop, not the round trip, so a useful fix would have to release the sequence after submit rather than after send_batch. For per-key Kafka ordering today, use concurrency: 1.

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

See EXTENDING.md for the full guide, with worked examples in all three languages.

• Custom Endpoints: Implement the CustomEndpointFactory trait and register it with extensions::register_endpoint_factory. Any endpoint key mq-bridge does not recognise is looked up in that registry, so a registered pulsar factory makes input: { pulsar: {...} } work with no core change. Endpoints for transports we do not want in this repository’s dependency tree live in their own crates.
• Custom Middleware: Implement the CustomMiddlewareFactory trait and register it with extensions::register_middleware_factory; use it as custom: { name, config } in an endpoint’s middlewares list.
• From Python / Node: register_endpoint / register_middleware (Python) and registerEndpoint / registerMiddleware (Node) take a host-language object instead of a Rust type, for endpoints that only have a Python/JS SDK.
• 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: mqb --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
   cd mq-bridge
   

2. Build and run the CLI (empty):

   cargo run --release -p mq-bridge-app
   

3. Build and run with a config:

   cargo run --release -p mq-bridge-app -- --config apps/mq-bridge-app/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 --prefix apps/mq-bridge-app install
npm --prefix apps/mq-bridge-app run dev

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

npm --prefix apps/mq-bridge-app run build:ui
cargo run --release -p mq-bridge-app

Docker image (no local Rust required)

Requires Docker and Docker Compose:

docker compose -f apps/mq-bridge-app/docker-compose.yml 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

mqb 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:

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

mqb --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.
• Writing endpoints & middleware — plugging your own endpoint or middleware into the engine from Rust, Python, or Node.
• 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 competing-consumer mode, which requires 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, the Postgres CDC tutorial for CDC-specific idempotency. When the same key can change more than once in a transaction, enable source_metadata: true and order by both mqb.src.postgres_lsn and mqb.src.postgres_ordinal; an LSN-only version cannot order those changes. The linked deduplication guide shows the complete predicate. Also see Delivery guarantees for which source/sink pairings add up to effectively-once.

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 status and logs

Without a dlq, a systematic failure (e.g. every row hitting a column-type mismatch) drains the input while delivering nothing to the sink and still ends completed. Route status retains the drop count and last rejection cause, and the logs contain a burst of Dropping message … due to non-retryable error. Do not use the outcome alone as a delivery check. 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:

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

Picking a store

store selects the backend by URL scheme, and the scheme decides whether deduplication is process-local or shared across every instance of the route:

The collection/table defaults to mqb_dedup_<route>. Point a shared store at the deployment your sink already uses rather than standing up extra infrastructure. sled_path is the legacy spelling of a local sled store, equivalent to store: "sled://<path>".

# Shared across every instance of this route.
- deduplication: { store: "mongodb://localhost:27017/etl", ttl_seconds: 3600 }

A sled store is per-process, not cluster-wide, so for multi-writer pipelines use either a shared store above or the sink constraint (below). Even with a shared store, the sink’s own unique constraint remains the more robust choice when the sink has one — it is already the authority, with no second write.

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 that includes the table, key, operation, commit LSN, and intra-transaction ordinal. Replayed changes therefore deduplicate through the deduplication middleware while distinct changes in one transaction remain distinct — but only within that middleware’s reach: a replay arriving more than ttl_seconds after the original is no longer remembered, and a local sled store only remembers what this route instance saw. The sink’s own constraint (id_field / ON CONFLICT) is the durable guarantee; the middleware only saves the sink the write.

An LSN-only sink predicate is not enough when one transaction changes the same key more than once: those changes share a commit LSN, so the first accepted row can block a later row. Enable source_metadata: true on the postgres_cdc source and order sink versions by the pair (mqb.src.postgres_lsn, mqb.src.postgres_ordinal) instead. Persist both metadata values and compare the pair lexicographically in the upsert predicate:

Both columns must be typed and NOT NULL. lsn as text sorts 0/9… above 0/16…, which is backwards, and a NULL on either side makes the whole WHERE predicate NULL, so the update is skipped and the row silently stops advancing:

ALTER TABLE orders
  ADD COLUMN lsn pg_lsn NOT NULL DEFAULT '0/0',
  ADD COLUMN ordinal bigint NOT NULL DEFAULT 0;

On a table that already has these columns as text or as nullable, backfill before switching the predicate on — UPDATE orders SET lsn = '0/0' WHERE lsn IS NULL (same for ordinal), then ALTER COLUMN … TYPE pg_lsn USING lsn::pg_lsn and SET NOT NULL. A '0/0' floor means the first change event for each existing row wins, which is what a re-snapshot should do.

INSERT INTO orders (id, body, lsn, ordinal)
VALUES (${payload:id}, ${payload:body}, ${metadata:mqb.src.postgres_lsn}::pg_lsn, ${metadata:mqb.src.postgres_ordinal}::bigint)
ON CONFLICT (id) DO UPDATE
SET body = EXCLUDED.body, lsn = EXCLUDED.lsn, ordinal = EXCLUDED.ordinal
WHERE (EXCLUDED.lsn, EXCLUDED.ordinal) > (orders.lsn, orders.ordinal)

The row comparison decides on lsn and only consults ordinal when the two LSNs are equal.

See the Postgres CDC → JSONL tutorial for the full CDC idempotency picture, and Delivery guarantees for what identity each source provides and which sinks absorb a duplicate write.

Checkpoints & resumable copies

A checkpoint is a durable record of how far a source has been read, so a restart continues where the last run stopped instead of re-copying from the beginning. It is what turns a one-shot copy into a repeatable incremental sync you can put on a timer.

Two settings control it, on the source endpoint:

Positions are namespaced as <source>:<cursor_id>, so several routes may share one store without colliding. checkpoint_store may embed credentials and is treated as a secret.

Which sources checkpoint

An SQLx source with no cursor_column is a destructive work queue, not a resumable read — progress is the deletion of claimed rows, so there is nothing to checkpoint. See Endpoints.

Picking a store

checkpoint_store selects the backend by URL scheme; a value with no scheme is a plain table/collection name in the source datastore.

Notes:

• ClickHouse requires an external store. It cannot cheaply upsert cursor rows, so a source-datastore checkpoint is rejected; cursor_id without a checkpoint_store silently disables resume (with a warning).
• Object-store sources must point checkpoint_store at a different bucket or prefix than they read — a cursor object written under the source prefix would be listed and re-read as data. The source rejects an overlapping location.
• A file store is written atomically (temp file + rename) and concurrent writers in one process are serialized, so several routes may share one file.
• Cloud object-store checkpoints need the object-store feature compiled in.

Using it with copy

cursor_id, cursor_column, and checkpoint_store are ordinary endpoint config fields, so on the CLI they are just query parameters on --from:

# Incremental table → table sync. Re-run it as often as you like: each run copies
# only rows whose `id` is greater than the last successfully written row.
mqb copy \
  --from 'postgres://user:pass@localhost/app?table=orders&cursor_column=id&cursor_id=orders_sync' \
  --to   'clickhouse://localhost:8123?table=orders&database=analytics' \
  --drain

# Read-only source: keep the cursor next to the job instead of in the source DB.
mqb copy \
  --from 'mysql://ro_user:pass@reporting/app?table=events&cursor_column=event_id&cursor_id=events_export&checkpoint_store=file%3A%2F%2F%2Fvar%2Flib%2Fmqb%2Fcursors.json' \
  --to   'file:///data/events.jsonl' \
  --drain

# MongoDB bulk read that survives a restart mid-copy.
mqb copy \
  --from 'mongodb://localhost:27017/app?collection=orders&consume=capture_all&cursor_id=orders_dump' \
  --to   'file:///data/orders.jsonl'

A checkpoint_store URL inside a URI must be percent-encoded (:// → %3A%2F%2F), since it is a query-parameter value. In a YAML config it is written plainly:

orders_sync:
  input:
    postgres:
      url: "postgres://user:pass@localhost/app"
      table: orders
      cursor_column: id
      cursor_id: orders_sync
      checkpoint_store: "file:///var/lib/mqb/cursors.json"
  output:
    clickhouse: { url: "http://localhost:8123", table: orders, database: analytics }

--drain and checkpoints

--drain exits when the source yields an empty batch — i.e. when the checkpoint has caught up with the table. That is exactly the shape you want for a cron/systemd-timer job: each invocation drains the backlog since last time and exits 0. Without --drain, the same command runs forever, polling for new rows every polling_interval_ms (100 ms by default) — or backing off exponentially up to max_polling_interval_ms while drained, if you set it.

MongoDB capture_all follows a change stream once its initial read is done, so a --drain run ends when that stream goes quiet rather than at a known end of data. capture_new only ever emits changes made after it starts and is continuous by nature. For a one-shot read with a real end, use consume: snapshot — non-destructive, no replica set, and not resumable (it rejects cursor_id).

An object-store source ends a drain run when it reaches the end of the objects it listed, which can be before the prefix is exhausted — the checkpoint makes this safe rather than lossy: each run resumes at the last fully-acked object key, so repeated runs advance until one reports zero messages. Loop the job until it moves nothing if you need a single pass to cover everything.

Delivery semantics

Checkpoints are at-least-once, never at-most-once:

• The position advances only after the sink acknowledges, and only across the contiguous run of acks from the front of the batch. The first nack stops the advance.
• On a partial failure the in-memory read cursor rolls back to the committed boundary, so nacked rows are re-read on the next poll rather than skipped until a restart.
• A crash between “rows written” and “checkpoint flushed” replays that batch. Make the sink idempotent — see Upserts & insert-if-absent and Deduplication.
• If persisting the cursor fails, the route logs a warning and keeps running; rows may be reprocessed on restart.

Gotchas

• cursor_column must be monotonic and non-decreasing for new rows (WHERE col > $last ORDER BY col ASC). An autoincrement id or an append-only timestamp works; a mutable updated_at does not give you deletes, and a column that can go backwards loses rows.
• Cursor polling captures appends only. Updates and deletes to already-copied rows are not observed. For those, use CDC (postgres_cdc, MongoDB capture_*).
• Equal-value groups must fit in a batch. If more rows share one cursor_column value than batch_size, the reader refuses to advance rather than skipping the remainder. It reports cursor_column '…' has a group of equal values larger than batch_size and retries that poll indefinitely rather than exiting, so a --drain job hangs instead of failing — watch for the repeating log line. Raise batch_size above the largest group, or pick a more unique column.
• The cursor column must be integer or text. Other types the SQL Any driver can’t decode fail permanently; expose the column as BIGINT/TEXT through a view.
• cursor_column and delete_after_read are mutually exclusive — one is non-destructive, the other consumes.
• Changing cursor_id or checkpoint_store starts over. The position is keyed by both; a new key means a full re-copy. Reuse the same pair to continue a sync, and give unrelated jobs distinct cursor_ids.
• capture_all’s initial snapshot is not incrementally checkpointed. What gets persisted is the change-stream resume token, written once streaming begins; a run interrupted during the snapshot re-snapshots from the beginning. Size the sink’s idempotency accordingly.
• ClickHouse and SQL Server: ClickHouse is polling-only and needs an external store; cursor_column mode is not supported on Microsoft SQL Server at all.

See also

• Endpoints (concepts) — read modes and CDC
• PostgreSQL parameters · MongoDB · ClickHouse · Postgres CDC
• CLI commands — copy flags and URI grammar

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 one destination per message. Output only. It has two modes and uses exactly one of them: value lookup on a metadata key, or when predicates over the payload. Naming both is a startup error.

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.

From the CLI, the same endpoint is a URI: switch:?metadata_key=country_code&case.US=<uri>&default=<uri> — see Structural endpoints in the URI.

Route on the payload (when)

when takes an ordered list of predicates and sends the message to the first one that matches. The expression language is the one --filter uses: payload fields by bare name including nested paths, metadata under meta., and and / or.

output:
  switch:
    when:
      - if: "amount > 10000"
        to: { kafka: { topic: "large_orders", url: "kafka:9092" } }
      - if: "order.status == 'refunded'"
        to: { nats: { subject: "refunds", url: "nats://localhost:4222" } }
    default: { file: { path: "/var/data/orders.jsonl" } }

A predicate parses the payload, while value lookup is a HashMap get on metadata — which is why the modes cannot be mixed in one endpoint, and why a metadata key you already have is the cheaper branch. A message matching no predicate goes to default, and without a default it is dropped, exactly as in value-lookup mode.

The CLI spells this as when=<expression> / to=<uri> pairs: switch:?when=amount > 10000&to=<uri>&default=<uri>.

Promote a payload value into metadata

Value-lookup mode matches on metadata, not payload fields. When you want that mode — an exact-match table rather than predicates — promote the value into metadata first. 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" }

From the CLI

copy can build a fan-out too — fanout:?to=<uri>&mirror=<uri>, where a mirror branch’s response and failures are discarded. See Structural endpoints in the URI.

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).

A sealed payload is binary, so a file/object_store sink that stores it must use format=normal — format=json/text render the payload as a JSON value and it cannot be read back (the reader reports “unsupported encryption envelope version 91”). Stacking it with compression works, but the reading route must list the two middlewares in the reverse order, since middlewares apply in list order on both ends.

From the copy CLI it is an inline middleware; keep the key in ${env:VAR} so it never reaches the process list or shell history ({/} percent-encode to %7B/%7D):

MQB_ENC_KEY=… mqb copy \
  --from 'file:///tmp/orders.jsonl' \
  --to   'nats://localhost:4222?stream=secure&subject=secure.orders|encryption?key=$%7Benv:MQB_ENC_KEY%7D'

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

There are two kinds of compression, both behind the compression feature:

• the batch compression field on the file / object_store endpoints — compresses whole write batches, so the output stays readable with zcat / lz4 -d;
• the compression middleware — compresses each message payload, so it works over any transport (Kafka, NATS, HTTP, …), not just files.

Endpoint batch compression

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

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.

The compression middleware

To compress payloads over the wire — a Kafka topic, a NATS subject, an HTTP body — attach the compression middleware instead. It compresses each message payload on the output side and decompresses it on the input side; metadata and routing keys are untouched. It works on input and output:

orders_bridge:
  input:
    middlewares:
      - compression: { algorithm: zstd }
    kafka: { topic: "orders", url: "localhost:9092" }
  output:
    middlewares:
      - compression: { algorithm: zstd }
    nats: { subject: "orders.out", url: "nats://localhost:4222" }

Put the same algorithm on both sides of a route. Each payload is framed independently, so unlike the endpoint field this is only readable through a matching consumer — a truncated or corrupt frame is a permanent consumer error rather than an endlessly re-read poison message.

From the copy CLI it is an inline middleware like any other:

mqb copy \
  --from 'nats://localhost:4222?stream=orders&subject=orders.in|compression?algorithm=zstd' \
  --to   'file:///tmp/orders.jsonl'

A compressed payload is binary, so a file/object_store sink that stores it must use format=normal. With format=json or text the payload is rendered as a JSON value and cannot be read back. Verified round trip:

mqb copy --drain \
  --from 'file:///tmp/in.jsonl?format=json' \
  --to   'file:///tmp/packed.bin?format=normal|compression?algorithm=zstd'
mqb copy --drain \
  --from 'file:///tmp/packed.bin?format=normal|compression?algorithm=zstd' \
  --to   'file:///tmp/out.jsonl?format=json'

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 — consume: capture_all (the default) reads existing documents first and then keeps capturing (no gap, at-least-once); consume: capture_new watches an existing collection for changes from now on. Both emit insert/update/replace/delete tagged with mongodb.operation and checkpoint under cursor_id. Change streams need a replica set; without one, use consume: snapshot for a one-shot read.
• 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. consumer buys exclusivity via four round trips per batch and deletes what it reads, so it is for work queues, not for reading a collection; where you need a single-reader bulk read or ETL pass, the default capture_all is roughly 5x faster (500k docs: 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.

Both capture_* modes read the oplog and therefore require a replica set — a single-node one is enough — and refuse to start without one. On a standalone mongod, use snapshot for a one-shot non-destructive read. snapshot delivers what exists when the run starts and is not a tail: it pages by _id, which is assigned client-side and so does not follow commit order, so it rejects cursor_id rather than resuming above a stored _id and silently skipping whatever a concurrent writer commits below that mark. Incremental reads need commit order — i.e. the oplog, i.e. a replica set.

Removed in 0.4.0: consume: subscriber. It polled seq > last_seq and advanced its watermark to the highest seq it had seen, so a batch whose seq block was reserved first but committed second was skipped permanently — silent loss, with no error and no gap in the delivery count. Use capture_new for ephemeral fan-out (it now reads arbitrary collections, not only documents written by the bridge’s own publisher). The deprecated change_stream: true boolean resolves to capture_new.

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).

Delivery guarantees

What mq-bridge promises about duplicates and loss, what it needs from you, and which source/sink combinations give you which guarantee.

Short version. mq-bridge is at-least-once. A message is acked only after the output chain reports success, so nothing is lost on a crash — but a crash between the write and the ack replays the message. Combine at-least-once delivery with an idempotent write at the sink and you get effective exactly-once: the record lands once no matter how many times it is delivered. Everything below is about how to arrange that.

Enabling effective exactly-once

There is no global exactly_once switch. The guarantee follows from ordinary endpoint configuration: provide a replay-stable identity when the source does not already have one, then configure an idempotent sink write. At startup, mq-bridge inspects the route and reports the inferred guarantee as effectively-once or at-least-once; it does not silently change how the sink writes data.

For a source without stable identity, derive one once and consume it at the sink:

input:
  middlewares:
    - id: "${payload:order_id}"       # writes metadata mqb.id
  file: { path: "orders.jsonl" }

output:
  sqlx:
    url: "sqlite://orders.db"
    table: orders
    insert_query: >
      INSERT INTO orders (id, body)
      VALUES (${metadata:mqb.id}, ${payload:body})
      ON CONFLICT (id) DO NOTHING

The target column must actually have a PRIMARY KEY or UNIQUE constraint. DO NOTHING gives insert-once semantics; an appropriate DO UPDATE clause gives convergent upsert semantics.

What exactly-once actually requires

It is four separate properties, and they fail independently:

1. Deterministic identity — a key for the record that is the same on every replay. Without this, nothing downstream can recognise a duplicate.
2. An idempotent (or transactional) write — the sink must absorb a repeat of the same key.
3. A durable source position tied to the write — so recovery resumes at the right place.
4. Fencing — a restarted or duplicated instance must not race the old one.

mq-bridge gives you 1 and 2 across most of the matrix, 3 for two sources, and does not implement 4. See What is not provided.

Sources: what identity you get

message_id is a u128. When the source can derive it from the record it carries, it is stable across replay and usable as a deduplication key. When it cannot, it defaults to a fresh fast_uuid_v7 value per read — which identifies this delivery, not that record, and is therefore useless for dedup across a restart.

Two consequences worth internalising:

• A file or object_store source cannot deduplicate on message_id. Re-reading the same file produces entirely new ids. Derive a business key instead — see Giving a source an identity.
• mq-bridge writes mq_bridge.message_id on the sink side for Kafka, NATS and Redis Streams, so a mq-bridge → broker → mq-bridge hop preserves identity end to end even when the broker itself has no id concept.

Giving a source an identity

For the sources in the bottom rows, the id middleware derives one from the message itself and stores it in the mqb.id metadata key:

input:
  middlewares:
    - id: "${payload:order_id}"
  file: { path: "orders.jsonl" }

Re-reading the same record now yields the same mqb.id, even though each read mints a fresh message_id. Three properties make it the right carrier:

• It is a string, so it holds the business key in its original form. message_id is a u128 and could only hold a hash of it — which a sink cannot then use as a readable _id.
• It propagates. mqb.id sits deliberately outside the mqb.src.* namespace, so publishers do not strip it: an identity describes the record, not the hop it arrived on. Kafka already forwards every non-mqb.src.* metadata entry as a header, so it survives that hop today.
• It is opt-in. No id entry means no wrapper and no cost.

Once set, read it anywhere metadata is available — ${metadata:mqb.id} in a deduplication key, a Kafka partition_key, or a handler.

Watch the ordering. Consumer middlewares wrap in reverse, so the entry closest to the end of the list touches an incoming message first. Anything reading mqb.id must be listed before the id that produces it:

input:
  middlewares:
    - deduplication: { store: "sled:///var/lib/mqb/dedup", ttl_seconds: 3600, key: "${metadata:mqb.id}" }
    - id: "${payload:order_id}"
  file: { path: "orders.jsonl" }

Reversed, mqb.id is still unset when deduplication reads it, and dedup silently falls back to message_id.

Identity is not a version. mqb.id answers “which record is this”, not “which revision”. Do not point a CDC route’s dedup key at it: every update to a row shares one business key, so they would collapse into one. That is why postgres_cdc builds its message_id from schema.table + key + **lsn** — dedup on a change stream needs identity and version, and should keep defaulting to message_id. Use mqb.id for sink keying and correlation.

id is input-only; on an output it is a startup error rather than a silent no-op. The key is set only when every selector in the template resolves — a partial render such as "acme-" would hand one identity to every message missing the field, so it is dropped instead.

Sinks: how a duplicate write is absorbed

Picking a combination

Any source → a database sink. The easy case. You do not need a replayable source position at all — the sink’s unique constraint is already shared across every writer and is the authority. Give it a deterministic key (id_field, or a ON CONFLICT column) and you are done.

Any source → files or object storage. A filesystem has no unique constraint, so this route needs a replayable source position and therefore works only from kafka or postgres_cdc. See Files & object storage.

A source → a broker sink (Kafka, NATS, MQTT, …). The sink cannot deduplicate. Either filter before it with the deduplication middleware, or accept at-least-once and make the downstream consumer idempotent.

A route with a handler. Sink-side idempotency happens after the handler runs. If that matters, see Handlers.

Deduplication & idempotent writes

For ETL, at-least-once delivery plus an idempotent write 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 database’s own unique constraint — it is already shared across every writer, so no extra state store is needed. Both database sinks lean on this instead of an application-side cache.

MongoDB — id_field

Point id_field at a top-level payload field and its value becomes the document _id. Alternatively, use the template form id_field: "${metadata:mqb.id}" to consume an identity produced by the input id middleware. 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 equivalent configuration using the shared identity carrier is:

input:
  middlewares:
    - id: "${payload:order_id}"
output:
  mongodb:
    # ... connection and collection ...
    id_field: "${metadata:mqb.id}"

A plain id_field preserves the payload value’s BSON type; the template form renders a string and accepts only replay-stable payload or metadata tokens.

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

SQL (sqlx) — ON CONFLICT / ON DUPLICATE KEY

The 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_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING"

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

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

# SQLite:
insert_query: "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: a configured dlq captures it, and without one it is logged and dropped. Add the conflict clause when replays are expected. ${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 (e.g. an ingest timestamp, or postgres.lsn from a CDC source). ClickHouse collapses rows with the same sort key at merge time, keeping the highest version; read with FINAL (or argMax) to see the deduplicated result:

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;

ReplacingMergeTree deduplicates by business key at merge time. Separately, ClickHouse also dedupes identical re-inserted blocks natively: a retried send_batch that resends the same block is dropped server-side (default one-hour window) — insert_deduplication_token lets you make that explicit, but mq-bridge does not set one, so rely on ReplacingMergeTree for logical dedup and treat block-level dedup only as retry-safety.

Postgres CDC — deterministic id + postgres.key

The postgres_cdc source resumes from the slot’s durable confirmed_flush_lsn. In-band standby feedback is asynchronous and is not flushed when the stream stops, so the last acks are made durable by the consumer’s Drop, which stops the stream and advances the slot synchronously. Re-delivery is therefore avoided only on a restart that actually runs that teardown — a host process that exits without dropping the route (or one on a current-thread Tokio runtime, where the blocking advance is skipped) replays everything since the last asynchronous feedback tick. Set checkpoint_store to a file:// path for a second, per-ack durable position that survives teardown regardless. Treat the source as at-least-once and make the sink idempotent.

Each change event carries the full row (so the primary key is in the payload), postgres.lsn (a monotonic version), postgres.operation/schema/table, and — when the table has a primary key / replica identity — postgres.key (the key value). The event’s message_id is a stable hash of schema.table + key + lsn, so a replayed change deduplicates through the deduplication middleware, and Mongo id_field or a SQL ON CONFLICT on the key column make the sink write idempotent. Use postgres.lsn as the version to drop stale replays (... DO UPDATE ... WHERE excluded.lsn > orders.lsn).

Known edge: if the same primary key is changed twice within a single transaction, both events share that transaction’s commit LSN, so they produce the same message_id. The deduplication middleware then treats the second as a duplicate and drops it. The sink still converges to the final row state, but the intermediate change is not delivered — if you need every intra-txn revision, do not rely on the message_id/middleware path for those rows.

The deduplication middleware

The middleware is a complement, not a replacement: it filters duplicates before the sink, and it is consumer-only (configuring it on an output is a startup error). Prefer the sink constraint for multi-writer ETL; reach for the middleware when the sink has no constraint to lean on, or when you need to keep duplicates away from a handler.

Two things decide whether it actually works:

• key. The default keys on message_id, which most sources mint fresh per read (see the source table) — the route then looks configured and dedupes nothing. Set key to a business key, either directly ("${payload:order_id}") or via "${metadata:mqb.id}" when an id middleware already derived one. An unresolvable template also falls back to message_id with only a warning, so a typo fails silently.
• store. sled is single-instance. Point it at a shared MongoDB or SQL deployment for anything scaled out.

input:
  middlewares:
    - deduplication:
        store: "mongodb://localhost:27017/shop"
        ttl_seconds: 86400
        key: "${payload:order_id}"
  kafka: { topic: "orders", url: "localhost:9092" }

middlewares sits beside the endpoint type, not nested inside it. Requires the dedup feature.

MongoDB — branch on insert vs. duplicate (report_outcome)

Sometimes you need to act on whether a record was newly inserted or already existed — enrich only fresh rows, or reply with the existing entry for duplicates. Set report_outcome: true and the Mongo publisher returns the message tagged with metadata mongodb.outcome = inserted (fresh write) or existed (dup-key on the unique _id). Wrap it in a request endpoint to forward that tagged message into a switch that routes on mongodb.outcome:

orders_upsert_branch:
  input: { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    request:                       # calls `to`, forwards its response to `forward_to`
      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" }   # e.g. build entry X, reply
            existed:  { ref: "handle_duplicate" }   # e.g. reply with parts of X
          default:   { file: { path: "orders-unrouted.jsonl" } }

report_outcome is sink-only and pairs with id_field; without a deterministic _id there is no duplicate to detect. Do not also set request_reply: true — that switches the publisher to the reply-polling path, which never reports an outcome and times out with nothing answering. Left unwrapped by request, the tagged message is returned as the route’s response as usual.

Keep the default arm: when the Mongo send errors, request forwards the original message, which carries no mongodb.outcome, and a switch with no default drops unmatched messages.

The outcome is only truthful on the first attempt. If the process dies after the insert committed but before the branch’s downstream send committed, the replay hits a duplicate key, Mongo answers existed, and a genuinely-new record takes the duplicate branch. Write the existed branch as “may or may not have been handled — check and repair”, not “definitely already done”.

Files & object storage — name_by

Check first whether you need any of this. If your records already carry a business key — an id field in the payload, or mqb.id from the id middleware — then a key-addressed sink (Mongo id_field, SQL ON CONFLICT) already gives you effective exactly-once, and the order objects happen to land in does not matter. Positional naming below is for the case where the sink itself has to recognise a replay, or where a downstream reader depends on replay order. Turning it on when you do not need it only adds restrictions.

A filesystem has no unique constraint, so the file and object_store sinks get replay safety from the name they write under — which works only as far as the source repeats its positions: a re-read that numbers the same records differently produces different names, and different names are not a replay to anything downstream. name_by picks the scheme:

Under source_position the sink groups each batch into runs of consecutive source positions and writes one immutable part per run, named for the range it covers:

kafka_to_s3:
  input:
    kafka: { topic: "orders", url: "localhost:9092", source_metadata: true }
  output:
    object_store:
      url: "s3://my-bucket/orders"
      name_by: source_position   # → part-orders-<partition>-<start>-<end>.jsonl (zero-padded)

auto applies to object_store only. Either scheme leaves that sink writing whole immutable objects — write_time one per flushed batch, source_position one per contiguous run within it — so only the name and the grouping change, and deriving them is safe. The file sink is different: source_position turns path from a file into a directory of part files, which is a change of structure rather than of name, so a file sink stays on write_time until you ask for parts explicitly.

All of this belongs to the source_position path. Before its first write such a sink lists the prefix once and parses the ranges out of the part names already there, so a replay skips them without re-encoding or re-uploading. Filtering is per record, not per batch, so batch boundaries are free to differ across restarts — that is what makes it work without a checkpoint protocol, and it is why the listing cannot wait for a name to collide: a replay on different boundaries names different objects and collides with nothing. A repeat that does land on the same name is caught anyway, since the PUT uses PutMode::Create and treats AlreadyExists as success. A write_time sink never lists — its names are unique per write, so there is nothing to recognise; on the source_position path a fresh prefix pays one empty listing per publisher. A listing that fails is logged rather than failing the batch, leaving same-name-only deduplication until it succeeds: a transient failure is retried on the next batch, a denied ListObjects is not retried. Local files stage to a temp name, fsync, then rename (atomic within a filesystem); object stores have no atomic rename, so a single PUT under the final name is the commit.

This needs a replayable source position. Today that means:

A route whose output resolves to source_position turns source_metadata on for its input automatically; set it explicitly only when you want the mqb.src.* keys for something else. An input from the table above is what auto looks for; anything else keeps the sink on write_time. An explicit name_by: source_position over an input that has no position is rejected when the route starts, because there is then no way to honour what was asked for. NATS and AMQP also accept source_metadata and emit provenance keys, but a subject or routing key is not a replayable offset, so they cannot drive positional naming.

Under write_time, an object_store sink names objects in write order, and key order is the only order a bucket has. At concurrency: 1 that is source order; above it the name is minted inside the worker pool, so it is arrival order, and replaying a change stream through the bucket can reorder updates to the same key. For an input that carries a replay position, auto already avoids this. For one that does not — NATS, MQTT, HTTP, gRPC, ZeroMQ, IBM MQ, Redis Streams — there is no positional name to fall back to, and concurrency: 1 is the only remedy. A concurrent route into a write_time object-store sink logs a warning at startup either way: over an input with a position it points at name_by: source_position, over one without it at concurrency: 1.

Deprecated: idempotency: true|false is the old spelling of name_by: source_position|write_time. It is still read, but an explicit name_by wins over it.

Numbers inside a part name are zero-padded so that ASCII sort equals numeric sort — a bucket is listed lexicographically and that listing order is the replay order. A bucket written by a version before this padding contains unpadded names; the two forms do not sort correctly against each other, so do not mix them in one prefix.

MongoDB capture_all reads the collection, then streams changes. Both phases are numbered into the key so the snapshot always sorts ahead of the changes; a change can never replay before the document it modifies. The snapshot does not resume across runs — it re-scans from the start — but the numbering is deterministic, so a restart reproduces the same names and the covered-range recovery skips them. The exception is inserts or deletes landing in the collection between the two runs: those shift the numbering, and some documents are then written twice.

The sqlx cursor source uses the cursor_column value itself as the position — the same shape Kafka Connect’s S3 sink gives a Kafka offset. That only works for a unique, non-negative integer column: a text cursor pages fine but has no contiguous numeric order, and a repeated value would resolve two rows to one position and drop one of them. The reader rejects both at read time rather than naming records wrongly. Consecutive ids coalesce into one part file; gaps split it.

The file source numbers records by their index in the file, not their byte offset (byte offsets are not consecutive, so every record would become its own part).

That index repeats across runs only in consume mode, which always reads from byte 0 — and repeating is the point: a re-read produces the same names, so the sink recognises them as already covered and writes nothing. subscribe starts at the current end of the file and group_subscribe resumes at a stored byte offset, so their index restarts at 0 over records an earlier run already numbered. Those two modes therefore carry a run epoch in the object name.

The epoch keeps names distinct and keeps runs in order (a later run reads later records, so its objects sort after the earlier run’s). What it does not give you is deduplication across a restart: records re-read after a crash are written again under new names. That is ordinary at-least-once, and it is the honest guarantee for a source with no durable per-record position — these modes are allowed, not rejected, because that guarantee is fine for plenty of pipelines.

For the file sink, name_by: source_position changes what path means: it is the directory that receives the part files, not the file that is appended to. The sink creates it on startup, so pointing it at an existing regular file fails there with “Failed to create part-file sink directory”.

compression and encryption work as usual.

Current limits, all of which the sink rejects or logs rather than silently mishandling:

• No csv (each part would need its own header row).
• date_partition is ignored — parts are written flat under the prefix, since the name already carries the range. Logged at startup.
• One part file per contiguous run per batch. There is no size-based rolling yet, so a large backfill produces many small files (~1000 per 1M rows at batch_size: 1024). For postgres_cdc this is per-transaction, so a high-commit-rate stream produces one small file per commit. Rolling would require buffering across batches the route has already acked, which is not safe today.

Handlers

A handler is not a stage between the consumer and the publisher — it is a publisher wrapper. The order on the output side is:

handler  →  output middlewares  →  sink

So every sink-side idempotency mechanism on this page runs after the handler has already executed. For a pure ETL route that is fine. For a handler with side effects — enriching, calling an API, emitting an event — it means duplicates reach your code.

To keep duplicates away from a handler, put deduplication on the route’s input. It filters them out of the batch before the route ever calls the publisher chain, and the duplicates are acked straight back at the source. Set key to a business key, as above.

This is still at-least-once, deliberately. The middleware reserves a key on receive and only promotes it to “processed” when the message is acked. That in-flight reservation expires after five seconds, precisely so that a crash between reserve and commit frees the key for redelivery rather than losing the message. A crash after the handler ran but before the commit will re-run the handler.

To make a handler effectively-once, its own effect has to be idempotent:

• If the handler writes to a store with a unique key, you already have it — the write is absorbed on replay. The handler may run twice; it cannot land twice.
• If the handler makes a non-idempotent external call (charge a card, send mail), mq-bridge has no primitive for this today. You need an idempotency-key cache around the effect itself.

What is not provided

• Exactly-once across systems. Not achievable without a transaction spanning the source’s offset commit and the sink’s write. Nothing here claims it.
• Kafka transactional EOS. enable.idempotence is on, but transactional.id and send_offsets_to_transaction are not used, so Kafka→Kafka is not exactly-once.
• Fencing on checkpoint stores. The file/s3/mongodb/postgres checkpoint stores have no lease, owner id or epoch. Kafka consumer groups and Postgres replication slots fence structurally; the checkpoint stores do not, so a zombie instance after a partial partition can re-emit.
• A transactional outbox. When source position and sink write live in the same database, they could be committed together — mq-bridge has no way to express that today.

See also

• REFERENCE.md — every middleware and structural endpoint, with fields and defaults.
• ARCHITECTURE.md — internal design and extension points.

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 mqb 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:

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

Write with auto-created table (destination):

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

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

Consume with a durable consumer group, continuous:

mqb 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:

mqb 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:

mqb 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:

mqb 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):

mqb 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:

mqb 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:

mqb 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:

mqb 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:

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

Consume with a durable group into ClickHouse, continuous:

mqb 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:

mqb 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:

mqb 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:

mqb 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):

mqb 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:

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

Only accept a specific path:

mqb 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:

mqb 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. A client consumer can also call any service described by a compiled protobuf descriptor, without generated Rust code — see Calling an arbitrary service.

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:

mqb 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:

mqb 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'

Calling an arbitrary service

Point the endpoint at a compiled FileDescriptorSet and name the service, the method and the request as JSON. Responses are decoded dynamically and emitted using protobuf’s canonical JSON representation:

input:
  grpc:
    url: https://grpc.example.com:443
    descriptor_set_path: proto/events.bin
    service_name: events.EventService
    method_name: Tail
    server_streaming: true
    request:
      topic: audit

Generate the descriptor with imports included:

protoc --descriptor_set_out=proto/events.bin --include_imports -I proto proto/events.proto

Unary and server-streaming methods are supported; client-streaming is rejected. A descriptor describes a wire format but not an acknowledgement protocol, so a dynamic source has no ACK operation and its delivery semantics are the remote API’s own. Where route-level ACK/NACK and at-least-once matter, use the built-in mqbridge.Bridge protocol (the default) — its ACK/NACK are real RPCs, and unacknowledged messages are retained and redelivered to the same consumer_id while the server process lives.

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:

mqb 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 — replica set required):

mqb 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 or deletes its documents. This mode reads the oplog, so it requires a replica set (a single-node one is enough) and refuses to start without one.

Read a standalone mongod (no replica set), one-shot:

mqb copy --drain \
  --from 'mongodb://localhost?database=app&collection=customers&consume=snapshot' \
  --to file:///data/customers.jsonl

snapshot pages the collection by _id and ends the route on drain. It is non-destructive and needs no replica set, but it is not a tail and not resumable — it delivers what exists when the run starts, and rejects cursor_id.

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

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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:

mqb 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

Wire format changed in 0.4.0. format used to default to json; it is now raw_framed, which is binary-safe and still carries headers. A 0.4 peer and a 0.3 peer no longer understand each other on the same socket unless one of them uses format=json in the URL query. REQ/REP replies are the exception to format entirely: a REP peer always answers with a JSON array of canonical messages, and a REQ publisher always decodes one.

Both backends cover the whole socket set, REQ/REP included. backend: omq is the faster path when the zeromq-omq feature is compiled in; naming omq or zmq explicitly makes that backend a hard requirement rather than a preference. On omq, backpressure is applied by the socket’s high-water mark and internal_buffer_size is ignored.

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:

mqb 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:

mqb 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:

mqb 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):

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

Export a table to JSONL, one-shot:

mqb 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:

mqb 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 mqb 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://
• Cloud Object Storage — schemes: s3://, s3a://, gs://, gcs://, az://, azure://, abfs://, abfss://
• 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.