mq-bridge-app

mq-bridge

ストリームをつなぐ · ライブラリ

ブローカー、データベース、ファイル、HTTP 間でデータを確実に移動する Rust エンジンを、Rust、Python、Node.js のサービスに組み込めます。

🖥️ mq-bridge-app · 📊 ETL ベンチマーク · 🏗️ アーキテクチャ · ⚙️ 設定 · 📚 ミドルウェアとエンドポイントのリファレンス

mq-bridge とは？

mq-bridge は、Kafka、NATS、RabbitMQ、MQTT、MongoDB、PostgreSQL/MySQL/SQLite、Postgres CDC、ClickHouse、Redis Streams、HTTP、WebSocket、gRPC、ZeroMQ、AWS SQS/SNS、IBM MQ、クラウドオブジェクトストレージ、ファイル、メモリ内チャネルなどのシステム間でメッセージとデータを移動する、非同期 Rust ライブラリです。すべてのエンドポイントは、同じバッチ形式の receive_batch / send_batch トレイトを介するため、任意のソースから任意のシンクにデータを送れます。

ルートは、1 つの入力と 1 つの出力を接続します。その間に設定だけで、データの変換、フィルタリング、ファンアウト、再試行、レート制限、重複排除、リクエストへの応答を行えます。設定だけでは足りない場合は、Rust、Python、Node.js でハンドラーを追加できます。バッチ処理、ack/nack、コミット順序、ブローカーとの I/O は Tokio ベースのコアが担うため、コードでは通常の CanonicalMessage を扱うだけで、Kafka オフセットや AMQP の nack を処理する必要はありません。

これは独自のサービスに組み込むライブラリであり、運用するデーモンやプラットフォームではありません。コードを書かずに使いたい場合は、mq-bridge-app を使うと、同じエンジンと設定を CLI、サーバー、デスクトップアプリ、MCP サーバーとして実行できます。

ユースケース

課題	実行可能なサンプル付きガイド
RAG またはセマンティック検索インデックスが Postgres のデータソースと同期しなくなる	Postgres とベクトルインデックスを同期する：CDC → 埋め込み → Qdrant
サービスでデータベースの更新とイベントの公開をアトミックに行う必要がある	Postgres と Kafka によるトランザクショナルアウトボックス
Kafka、NATS、RabbitMQ のコンシューマーのテストに、稼働中のブローカーが必要	ブローカーなしでメッセージハンドラーをテストする

クイックスタート

お使いの言語のライブラリをインストールします。

言語	パッケージ	インストール
Rust	mq-bridge	cargo add mq-bridge --features kafka,nats,yaml
Python	mq-bridge (PyPI)	pip install mq-bridge
Node.js	mq-bridge (npm)	npm install mq-bridge
C / C++	mq_bridge.h	cargo build --release -p mq-bridge-c（ソースからビルド）

routes.yaml でルートを定義します。ここでは、Kafka から NATS へのルートにリトライを設定します。

kafka_to_nats:
  input:
    kafka: { url: "localhost:9092", topic: "orders", group_id: "bridge" }
  output:
    middlewares:
      - retry: { max_attempts: 5 }
    nats: { url: "nats://localhost:4222", subject: "orders.processed" }

実行します。

from mq_bridge import Route
Route.from_file("routes.yaml", "kafka_to_nats").run()  # blocks; use start() to keep going

import { Route } from "mq-bridge";
Route.fromFile("routes.yaml", "kafka_to_nats").start();

mq_bridge::deploy_file("routes.yaml").await?; // every route in the file, in the background

ビジネスロジックを途中に挟む場合は、ハンドラー（with_handler / withHandler）を追加します。コードをまったく書きたくない場合は？ mq-bridge-app を使うと、コマンドラインまたはデスクトップ UI から同じエンジンを実行できます。

mqb copy 'kafka://localhost:9092?topic=orders' 'nats://localhost:4222?subject=orders.processed'

機能の概要

各行の機能は、Rust クレート、Python および Node.js パッケージ、コード不要のアプリのいずれからも利用できます。Cargo フィーチャーが必要なのは、Rust クレートを自分でビルドする場合のみです。リンク先のブックでは、ライブラリとアプリの両方について説明しています。

機能	方法（エンドポイント／ミドルウェア · Cargo 機能）	ドキュメント
メッセージブローカー：Kafka、NATS（JetStream）、RabbitMQ（AMQP）、MQTT、Redis Streams、AWS SQS/SNS、IBM MQ、ZeroMQ	kafka、nats、amqp、mqtt、redis_streams、aws、ibmmq、zeromq エンドポイント · 機能 kafka、nats、amqp、mqtt、redis-streams、aws、ibm-mq、zeromq	コネクター
変更データキャプチャ（CDC）：Postgres または MongoDB の変更を Kafka、ファイル、データウェアハウスなどにストリーミング	postgres_cdc（論理レプリケーション、pgoutput） · postgres-cdc；capture_all / capture_new（変更ストリーム）を指定した mongodb · mongodb	Postgres CDC チュートリアル · MongoDB
SQL データベース：ソース（カーソルポーリング）またはシンク（挿入／アップサート）としての PostgreSQL、MySQL、MariaDB、SQLite	sqlx エンドポイント · sqlx	PostgreSQL／MySQL／SQLite · アップサート
ソースまたはシンクとしての MongoDB	mongodb エンドポイント · mongodb	MongoDB
ClickHouse の一括挿入と再開可能な読み取り	clickhouse エンドポイント · clickhouse	ClickHouse
検索エンジンおよびその他の一括ドキュメント API（Meilisearch、Typesense、Elasticsearch、Qdrant、PostgREST／Supabase）	http_bulk エンドポイント · http-bulk	HTTP 一括処理
OpenAI 互換 API（OpenAI、Ollama、vLLM など）を介した LLM と埋め込み：各メッセージの分類、要約、または埋め込み生成	openai_chat、openai_embeddings エンドポイント · http-bulk	OpenAI 互換 API · LLM クックブック
オブジェクトストレージ：S3（および R2 などの S3 互換ストレージ）、GCS、Azure Blob、またはローカルディレクトリを Parquet、JSONL、CSV 形式で使用	format: parquet | raw | json | csv を指定した object_store エンドポイント · object-store、parquet	オブジェクトストレージ
ファイル：CSV と JSONL の読み取りと書き込み	file エンドポイント · 組み込み	ファイル
分析／データウェアハウス：Snowflake、BigQuery、Databricks／Spark、Athena／Trino、S3/GCS/Azure 上の Parquet 経由の DuckDB。これらはネイティブコネクターではありません。mq-bridge が Parquet ファイルを書き込み、データウェアハウスがそれを読み込むかクエリします	format: parquet を指定した object_store · parquet	Snowflake · BigQuery · Databricks／Spark · Athena／Trino · DuckDB
サーバーまたはクライアントとしての HTTP、gRPC、WebSocket（リクエスト／リプライを含む）	http、grpc、websocket エンドポイント · http、grpc、websocket	HTTP · gRPC · リクエスト／リプライ
スキーマ検証と型変換（JSON Schema のサブセット、フィールドマッピング）	transform ミドルウェア · 組み込み	変換とスキーママッピング
信頼性：再試行、デッドレターキュー、重複排除、レート制限	retry、dlq、deduplication、limiter ミドルウェア · 重複排除用の dedup	ミドルウェアリファレンス · 配信保証
ブローカーを使わないテスト：Docker なしで、インメモリチャネルに対してルートまたはハンドラーを実行	memory エンドポイント · 組み込み	ライブラリの組み込み
コード不要の ETL	mq-bridge-app：mqb copy <from> <to>、YAML ルート、デスクトップ UI、MCP サーバー	クイックスタート：mqb copy
プラグインによる対応システムの拡張：Apache Pulsar、Meilisearch、Redpanda Connect の入力、出力、プロセッサー	ネイティブプラグイン · plugin	プラグインエンドポイント · Connect プラグイン：入力、出力、プロセッサー
Connect プラグインには、Redpanda Community License のコードを含まない Redpanda Connect コンポーネントのみがリンクされています（バンドルされているコードは Apache-2.0 および MIT です）。そのため、Kafka、AWS、Snowflake などの一部の上流コンポーネントは含まれていません。詳しくは含まれていないものをご覧ください。

AI エージェント向け：llms.txtは、ドキュメントを厳選してまとめたインデックスです。

ベンチマーク

ゼロコードのmq-bridge-appを使った、条件をそろえたETL比較です。数値、ベースライン、コマンドはすべてETLベンチマークハーネスに記載しています。

シナリオ	mq-bridge	比較対象
CSV → JSONL、混合型の行1M件（約116 MiB）	3,134,796行/秒、RAM約29 MiB	Meltano（tap-csv → target-jsonl）：9,771行/秒、RAM約444 MiB — mq-bridgeは約321倍高速 DuckDB、全コア使用：2,109,704行/秒 — mq-bridgeは約1.5倍高速、メモリ使用量は約18分の1
Kafka → ファイル、行1M件、変換なし	878,105行/秒、RAM約160 MiB	Sea Streamer、mimallocビルド：492,465行/秒、RAM約894 MiB — mq-bridgeは約1.8倍高速（デフォルトアロケーターのビルドとの比較でも約1.8倍）

CSVの行では、バイト単位で同一のフィクスチャを使った、既存の独立公開済みのCSV → JSONL、100万行のワークロードを再利用しています。混合型の列7つ、シード値42、121,981,421バイト、SHA-256 a84894e0…0c45b221（定義の全文）。変換なしでプロセス全体の実時間を計測し、Meltanoも同じマシンとファイルで実行しています。

すべての数値はmq-bridge 0.4.20のものです。DuckDBは変換処理自体のスループット上限を示すものであり、ETLツールではありません。

外部ベンチマーク： Rustライブラリ（mq-bridge）とPythonバインディング（mq-bridge-py）は、サードパーティのhttp-arena.comランキングに掲載されています。このランキングでは、HTTPフレームワークを毎秒リクエスト数で比較します（リアルタイムで更新されるため、順位は時間とともに変動します）。計測対象はmq-bridgeのHTTP（およびWebSocket）配信経路です。これは多数あるエンドポイントのうちの1つであり、ブローカーやETLのスループットではありません。これらについてはベンチマークで扱っています。ローカル比較用ハーネスについてはPython分析メモをご覧ください。

mq-bridge を選ぶ理由

どの言語から呼び出しても、1 つのエンジンで次のことができます。

• 20 以上のネイティブトランスポート、プラグイン経由でさらに 100 以上、API は 1 つ：上記のすべてのコネクターに加え、dir_spool（クラッシュセーフなディレクトリ FIFO キュー）も、共通の receive_batch ／ send_batch 形式で利用できます。
• Redpanda Connect の広範な対応範囲：Connect プラグインにより、Redpanda Connect の本番環境で実績のあるコンポーネントを利用できます。エンドポイントとして 56 個の入力と 68 個の出力に加え、86 個のプロセッサーを備え、その多くは任意のエンドポイントでミドルウェアとしても動作します。一方、ルーティング、バッチ処理、リトライ、DLQ、重複排除は mq-bridge が担います。
• 変更データキャプチャ：Postgres（論理レプリケーション／pgoutput）および MongoDB（変更ストリーム）から行単位の変更を、操作マーカー付きのフラットな行としてストリーミングします。
• 再起動に強い配信：累積 ACK を使用するブローカー向けに、コミット順序制御を備えたバッチ対応の ack/nack を提供します。統合テストでは、Postgres CDC の再起動安全性テストを含め、ブローカーの処理中に再起動してもデータ損失がないことを確認しています。
• フレームワークではなく、信頼性を高めるミドルウェア：リトライ、デッドレターキュー、重複排除、レート制限、Cookie ／セッションの永続化を、任意のエンドポイントに適用できます。
• 1 つのエンジンで複数言語に対応：同じ Rust コアを、ネイティブの Python および Node.js バインディングとして提供します。ルーティング、バッチ処理、ブローカーとの I/O は Rust 内で行われます。
• あらゆる箇所で TLS に対応し、設定形式は 1 つ：CA バンドル、mTLS 用クライアント証明書／キー、セキュリティチェックのスキップをまとめた単一の TlsConfig ブロックを、複数のトランスポートで再利用できます。
• セルフホスト可能で、デーモン不要：オプションの UI で設定を生成し、コードに貼り付けて、プロセス内で実行します。ホスト型のコントロールプレーンも、別個のスケジューラーも必要ありません。

コードを書きたくない場合は？ mq-bridge-app を使うと、同じエンジンを YAML または環境変数だけで設定できる、スタンドアロンのゼロコード ETL サービスとして実行できます。ルートに対するメッセージの作成、送信、確認に使える Postman スタイルの UIを備え、Postman コレクションや AsyncAPI ドキュメントをインポートして、ルートとエンドポイントのひな型を作成できます。アプリのインストール、CLI、UI、レシピ集、チューニングに関する情報はすべて、📖 ブックにあります。

mq-bridge を使う場面

• ハイブリッドメッセージング：システムごとに異なるプロトコル（例：MQTT と Kafka）を使用していても、組み合わせごとにカスタムアダプターを作成せずに接続できます。
• バッチ処理が中心のパイプライン：メッセージごとの ack/nack の判断を維持しながら、メッセージをバッチで移動してスループットを向上させます。
• インフラストラクチャの抽象化：CanonicalMessage を基準にビジネスロジックを記述し、後から基盤のトランスポートを切り替えられます。
• 耐障害性の高いパイプライン：リトライ、DLQ、重複排除、リミッター、Cookie ／セッションの動作を、エンドポイント全体に一貫して適用できます。
• データベース連携：データベースとメッセージブローカーを組み合わせて、たとえばメッセージを SQL ／ MongoDB に取り込んだり、アウトボックスの行をブローカーに転送したりできます。
• サイドカー／ゲートウェイ：別のサービスの横でブリッジを実行し、メッセージがコアアプリケーションに届く前に取り込み、フィルタリング、ルーティングを行います。

mq-bridge を使わない場面

• ステートフルなストリーム処理：ウィンドウ処理、結合、時間軸に沿った複雑な集計には、専用のストリーム処理エンジンのほうが適しています。
• ドメイン集約の管理：ドメイン集約のライフサイクル、バージョン管理、再生（イベントソーシング）を管理するフレームワークが必要な場合は、専用のライブラリを使用してください。mq-bridge が扱うのは バス であり、エンティティ ではありません。
• プロトコル固有の高度な機能：mq-bridge が意図的に提供するのは、共通部分です。具体的には、パブリッシュ／コンシューム、可能な場合のパブ／サブやリクエスト／リプライ、バッチ処理、ミドルウェア、ack/nack の処理です。アプリケーションがブローカー固有の高度な機能に依存する場合は、そのブローカーのネイティブクライアントを直接使用するほうが適していることがあります。

ドキュメント

📖 mq-bridge book では、ライブラリとアプリについて説明しています。コネクター、ミドルウェア、チューニングでは、どちらも同じ設定を使用します。

したいこと	参照先
ライブラリを組み込み、ハンドラーを作成し、コードから発行する	クイックスタート：ライブラリ · ライブラリの組み込み · Bindings API · docs.rs
コードを書かずにデータを移動する	クイックスタート：mqb copy · デスクトップ UI · CLI · MCP サーバー
コネクターを設定する	コネクター。それぞれに YAML と URL の例があります
各バックエンドがサポートする機能（サブスクライバーモード、リクエスト／リプライ、nack、CDC とポーリングの違い）を確認する	概要と機能 · 機能一覧（確認応答、コミット順序、リプライ、TLS）
リトライ、DLQ、重複排除、変換、ルーティングを追加する	クックブック · ミドルウェアと構造化エンドポイント
配信保証と冪等性のあるシンクを理解する	配信保証
リクエスト／リプライまたは CQRS 形式のフローを構築する	リクエスト／リプライ · ライブラリの組み込み：CQRS
スループットを調整する	パフォーマンスチューニング
独自のエンドポイント、ミドルウェア、プラグインを作成する	EXTENDING.md · PLUGINS.md
内部構造を理解する	ARCHITECTURE.md · CONFIGURATION.md

ステータス

mq-bridge は 2025 年に作成されました。信頼性に関する動作は、サポート対象の各エンドポイントで、そのエンドポイントがサポートするキューモードとサブスクライバーモードのそれぞれについて、自動統合テストおよびパフォーマンステストスイートで検証されています。

• すべてのエンドポイントで、ブローカーの処理中の再起動によるデータ損失はありませんでした。MQTT の発行確認は、カオステストで処理中の損失がゼロになるまで強化されました。
• Postgres CDC には再起動安全性テストがあります。未確認の処理中バッチは、データベースの再起動後に損失や欠落なく再配信されます。

armature では Rust ライブラリとして、omniload では Python ライブラリとして使用されています。

安定性と Rust のバージョン

1.0 以降、Rust API はセマンティックバージョニングに従います。対象はクレートのルートと、モジュール models、traits、errors、outcomes、extensions、plugin（SDK）、type_handler、publisher、route、checkpoint です。実装モジュール endpoints、middleware、support、test_utils はマイナーリリースで変更される場合があります。また、再エクスポートされたドライバークレート（mq_bridge::sqlx、mq_bridge::rdkafka、...）のメジャーバージョンも変更される場合があります。

マイナーリリースでは、設定フィールドや設定またはエラーの列挙型のバリアントが追加されたり、必要な Rust の最小バージョンが引き上げられたりする場合があります。..Default::default() を使って設定構造体を記述し、これらの列挙型に対する match に _ アームを用意しているコードは、このようなリリース間で変更なくコンパイルできます。

Rust の最小バージョンは 1.89 です。aws、sqlx、postgres-cdc の各機能、およびそれらすべてを含む full には 1.94.1 が必要です。zeromq-omq には 1.93 が必要です。

テストの実行

このプロジェクトには、統合テストとパフォーマンステストが含まれています。ほとんどのバックエンドテストには Docker が必要です。

サポートされているすべてのバックエンドのパフォーマンスベンチマークを実行するには：

cargo test --test integration_test --release -- --ignored --nocapture --test-threads=1

criterion ベンチマークを実行するには：

cargo bench --features "full"

criterion の数値は実行するマシンによって異なります。比較可能なスループットの数値を得るには、上記の統合パフォーマンステストを使用するか、ETL ベンチマークを参照してください。

コントリビューション

コントリビューションを歓迎します。セットアップに関する注意事項、コードスタイル、プルリクエストのガイドラインについては、CONTRIBUTING.mdを参照してください。

AI に関する免責事項

このライブラリは、AI の支援を多く受けて作成されました。コア部分は私自身のコードとして始まりましたが、Gemini、CodeRabbit、Claude、Codex を活用して多くのエンドポイントやドキュメントを拡充しました。エンドポイントのトレイトが安定した後は、これによってトランスポートの追加が大幅に速くなりました。

生成されたコードは一見もっともらしく見えても、重要な詳細が抜けていることがあります。そのため、すべてのコミットを手作業で確認し、出力を整理してリファクタリングするとともに、統合テストスイートを使ってリグレッションを検出しています。**現在のコードは、自分で書いた場合と同じくらい信頼しています。**機能が多いため、未解決の課題が残っている可能性があります。現在はテストとドキュメントに注力しています。

ライセンス

mq-bridge は、任意で、MIT ライセンスまたは Apache License, Version 2.0のいずれかのデュアルライセンスで提供されます。

バイナリ配布物には、適用されるサードパーティの利用条件が THIRD_PARTY_LICENSES.txt に記載されています。現在のメンテナンスプロセスについては、 docs/THIRD_PARTY_LICENSES.mdに記載されています。

コントリビューション

特に明記しない限り、Apache-2.0 ライセンスで定義される、mq-bridge への収録を目的としてあなたが意図的に提出したコントリビューションは、追加の条件を伴わず、上記と同じデュアルライセンスでライセンスされます。