From 055cabe2ee72c2ec38bf779c8c4417a3b7e783c7 Mon Sep 17 00:00:00 2001 From: amrit <252372762+amr8t@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:44:48 -0700 Subject: [PATCH] feat(connectors): add RabbitMQ sink --- .github/workflows/_build_rust_artifacts.yml | 2 +- .github/workflows/edge-release.yml | 1 + Cargo.lock | 705 ++++++++++++++++-- Cargo.toml | 2 + core/connectors/README.md | 1 + .../connectors/rabbitmq_sink.toml | 38 + core/connectors/sinks/README.md | 1 + .../connectors/sinks/rabbitmq_sink/Cargo.toml | 45 ++ core/connectors/sinks/rabbitmq_sink/README.md | 18 + .../sinks/rabbitmq_sink/config.toml | 38 + .../connectors/sinks/rabbitmq_sink/src/lib.rs | 388 ++++++++++ core/integration/Cargo.toml | 1 + .../tests/connectors/fixtures/mod.rs | 5 + .../connectors/fixtures/rabbitmq/container.rs | 302 ++++++++ .../tests/connectors/fixtures/rabbitmq/mod.rs | 25 + .../connectors/fixtures/rabbitmq/sink.rs | 211 ++++++ core/integration/tests/connectors/mod.rs | 1 + .../tests/connectors/rabbitmq/mod.rs | 18 + .../connectors/rabbitmq/rabbitmq_sink.rs | 299 ++++++++ .../tests/connectors/rabbitmq/sink.toml | 20 + scripts/bump-version.sh | 2 +- 21 files changed, 2066 insertions(+), 57 deletions(-) create mode 100644 core/connectors/runtime/example_config/connectors/rabbitmq_sink.toml create mode 100644 core/connectors/sinks/rabbitmq_sink/Cargo.toml create mode 100644 core/connectors/sinks/rabbitmq_sink/README.md create mode 100644 core/connectors/sinks/rabbitmq_sink/config.toml create mode 100644 core/connectors/sinks/rabbitmq_sink/src/lib.rs create mode 100644 core/integration/tests/connectors/fixtures/rabbitmq/container.rs create mode 100644 core/integration/tests/connectors/fixtures/rabbitmq/mod.rs create mode 100644 core/integration/tests/connectors/fixtures/rabbitmq/sink.rs create mode 100644 core/integration/tests/connectors/rabbitmq/mod.rs create mode 100644 core/integration/tests/connectors/rabbitmq/rabbitmq_sink.rs create mode 100644 core/integration/tests/connectors/rabbitmq/sink.toml diff --git a/.github/workflows/_build_rust_artifacts.yml b/.github/workflows/_build_rust_artifacts.yml index 56232ff1b0..e035b61207 100644 --- a/.github/workflows/_build_rust_artifacts.yml +++ b/.github/workflows/_build_rust_artifacts.yml @@ -46,7 +46,7 @@ on: connector_plugins: type: string required: false - default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_s3_sink,iggy_connector_stdout_sink,iggy_connector_surrealdb_sink" + default: "iggy_connector_elasticsearch_sink,iggy_connector_elasticsearch_source,iggy_connector_iceberg_sink,iggy_connector_postgres_sink,iggy_connector_postgres_source,iggy_connector_quickwit_sink,iggy_connector_random_source,iggy_connector_s3_sink,iggy_connector_stdout_sink,iggy_connector_surrealdb_sink, iggy_connector_rabbitmq_sink" description: "Comma-separated list of connector plugin crates to build as shared libraries" outputs: artifact_name: diff --git a/.github/workflows/edge-release.yml b/.github/workflows/edge-release.yml index f98f99c84d..15341c5109 100644 --- a/.github/workflows/edge-release.yml +++ b/.github/workflows/edge-release.yml @@ -111,6 +111,7 @@ jobs: - `iggy_connector_s3_sink` - `iggy_connector_stdout_sink` - `iggy_connector_surrealdb_sink` + - `iggy_connector_rabbitmq_sink` ## Downloads diff --git a/Cargo.lock b/Cargo.lock index 0ecf6332b7..e58e6eedd4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -349,6 +349,54 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "amq-protocol" +version = "7.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "587d313f3a8b4a40f866cc84b6059fe83133bf172165ac3b583129dd211d8e1c" +dependencies = [ + "amq-protocol-tcp", + "amq-protocol-types", + "amq-protocol-uri", + "cookie-factory", + "nom 7.1.3", + "serde", +] + +[[package]] +name = "amq-protocol-tcp" +version = "7.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc707ab9aa964a85d9fc25908a3fdc486d2e619406883b3105b48bf304a8d606" +dependencies = [ + "amq-protocol-uri", + "tcp-stream", + "tracing", +] + +[[package]] +name = "amq-protocol-types" +version = "7.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf99351d92a161c61ec6ecb213bc7057f5b837dd4e64ba6cb6491358efd770c4" +dependencies = [ + "cookie-factory", + "nom 7.1.3", + "serde", + "serde_json", +] + +[[package]] +name = "amq-protocol-uri" +version = "7.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f89f8273826a676282208e5af38461a07fe939def57396af6ad5997fcf56577d" +dependencies = [ + "amq-protocol-types", + "percent-encoding", + "url", +] + [[package]] name = "android_system_properties" version = "0.1.5" @@ -447,7 +495,7 @@ checksum = "797f94b6a53d7d10b56dc18290e0d40a2158352f108bb4ff32350825081a9f29" dependencies = [ "keyring-core", "log", - "security-framework", + "security-framework 3.7.0", ] [[package]] @@ -1004,7 +1052,7 @@ version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" dependencies = [ - "event-listener", + "event-listener 5.4.1", "event-listener-strategy", "futures-core", "pin-project-lite", @@ -1082,12 +1130,57 @@ checksum = "c96bf972d85afc50bf5ab8fe2d54d1586b4e0b46c97c50a0c9e71e2f7bcd812a" dependencies = [ "async-task", "concurrent-queue", - "fastrand", - "futures-lite", + "fastrand 2.5.0", + "futures-lite 2.6.1", "pin-project-lite", "slab", ] +[[package]] +name = "async-global-executor" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13f937e26114b93193065fd44f507aa2e9169ad0cdabbb996920b1fe1ddea7ba" +dependencies = [ + "async-channel", + "async-executor", + "async-io 2.6.0", + "async-lock 3.4.2", + "blocking", + "futures-lite 2.6.1", +] + +[[package]] +name = "async-global-executor-trait" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9af57045d58eeb1f7060e7025a1631cbc6399e0a1d10ad6735b3d0ea7f8346ce" +dependencies = [ + "async-global-executor", + "async-trait", + "executor-trait", +] + +[[package]] +name = "async-io" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fc5b45d93ef0529756f812ca52e44c221b35341892d3dcc34132ac02f3dd2af" +dependencies = [ + "async-lock 2.8.0", + "autocfg", + "cfg-if", + "concurrent-queue", + "futures-lite 1.13.0", + "log", + "parking", + "polling 2.8.0", + "rustix 0.37.28", + "slab", + "socket2 0.4.10", + "waker-fn", +] + [[package]] name = "async-io" version = "2.6.0" @@ -1098,21 +1191,30 @@ dependencies = [ "cfg-if", "concurrent-queue", "futures-io", - "futures-lite", + "futures-lite 2.6.1", "parking", - "polling", + "polling 3.11.0", "rustix 1.1.4", "slab", "windows-sys 0.61.2", ] +[[package]] +name = "async-lock" +version = "2.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b" +dependencies = [ + "event-listener 2.5.3", +] + [[package]] name = "async-lock" version = "3.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" dependencies = [ - "event-listener", + "event-listener 5.4.1", "event-listener-strategy", "pin-project-lite", ] @@ -1124,17 +1226,29 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc50921ec0055cdd8a16de48773bfeec5c972598674347252c0399676be7da75" dependencies = [ "async-channel", - "async-io", - "async-lock", + "async-io 2.6.0", + "async-lock 3.4.2", "async-signal", "async-task", "blocking", "cfg-if", - "event-listener", - "futures-lite", + "event-listener 5.4.1", + "futures-lite 2.6.1", "rustix 1.1.4", ] +[[package]] +name = "async-reactor-trait" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a6012d170ad00de56c9ee354aef2e358359deb1ec504254e0e5a3774771de0e" +dependencies = [ + "async-io 1.13.0", + "async-trait", + "futures-core", + "reactor-trait", +] + [[package]] name = "async-recursion" version = "1.1.1" @@ -1163,8 +1277,8 @@ version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52b5aaafa020cf5053a01f2a60e8ff5dccf550f0f77ec54a4e47285ac2bab485" dependencies = [ - "async-io", - "async-lock", + "async-io 2.6.0", + "async-lock 3.4.2", "atomic-waker", "cfg-if", "futures-core", @@ -1238,7 +1352,7 @@ checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6" dependencies = [ "async-compression", "crc32fast", - "futures-lite", + "futures-lite 2.6.1", "pin-project", "thiserror 2.0.19", "tokio", @@ -1372,7 +1486,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "hex", "http 1.4.2", "sha1 0.10.7", @@ -1460,7 +1574,7 @@ dependencies = [ "aws-types", "bytes", "bytes-utils", - "fastrand", + "fastrand 2.5.0", "http 1.4.2", "http-body 1.1.0", "percent-encoding", @@ -1488,7 +1602,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1514,7 +1628,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1540,7 +1654,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1567,7 +1681,7 @@ dependencies = [ "aws-smithy-types", "aws-smithy-xml", "aws-types", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1644,7 +1758,7 @@ dependencies = [ "hyper-util", "pin-project-lite", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-pki-types", "tokio", "tokio-rustls", @@ -1696,7 +1810,7 @@ dependencies = [ "aws-smithy-schema", "aws-smithy-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "http-body 0.4.6", @@ -1899,7 +2013,7 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" dependencies = [ - "fastrand", + "fastrand 2.5.0", "gloo-timers 0.3.0", "tokio", ] @@ -2213,7 +2327,7 @@ dependencies = [ "async-channel", "async-task", "futures-io", - "futures-lite", + "futures-lite 2.6.1", "piper", ] @@ -2255,7 +2369,7 @@ dependencies = [ "pin-project-lite", "rand 0.9.5", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-pki-types", "serde", "serde_derive", @@ -2851,6 +2965,18 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "cms" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b77c319abfd5219629c45c34c89ba945ed3c5e49fcde9d16b6c3885f118a730" +dependencies = [ + "const-oid 0.9.6", + "der", + "spki", + "x509-cert", +] + [[package]] name = "cobs" version = "0.3.0" @@ -2956,7 +3082,7 @@ dependencies = [ "compio-log", "compio-send-wrapper", "crossbeam-queue", - "flume", + "flume 0.12.0", "futures-util", "io-uring", "libc", @@ -2964,7 +3090,7 @@ dependencies = [ "mod_use", "once_cell", "pastey 0.2.3", - "polling", + "polling 3.11.0", "rustix 1.1.4", "smallvec", "socket2 0.6.5", @@ -3077,7 +3203,7 @@ dependencies = [ "compio-log", "compio-net", "compio-runtime", - "flume", + "flume 0.12.0", "futures-util", "libc", "quinn-proto", @@ -3335,6 +3461,12 @@ dependencies = [ "version_check", ] +[[package]] +name = "cookie-factory" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9885fa71e26b8ab7855e2ec7cae6e9b380edff76cd052e07c683a0319d51b3a2" + [[package]] name = "core-foundation" version = "0.9.4" @@ -4214,6 +4346,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ "const-oid 0.9.6", + "der_derive", + "flagset", "pem-rfc7468", "zeroize", ] @@ -4232,6 +4366,17 @@ dependencies = [ "rusticata-macros", ] +[[package]] +name = "der_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "deranged" version = "0.5.8" @@ -4328,6 +4473,15 @@ dependencies = [ "unicode-xid", ] +[[package]] +name = "des" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ffdd80ce8ce993de27e9f063a444a4d53ce8e8db4c1f00cc03af5ad5a9867a1e" +dependencies = [ + "cipher 0.4.4", +] + [[package]] name = "difflib" version = "0.4.0" @@ -4451,6 +4605,12 @@ dependencies = [ "const-random", ] +[[package]] +name = "doc-comment" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "780955b8b195a21ab8e4ac6b60dd1dbdcec1dc6c51c0617964b08c81785e12c9" + [[package]] name = "docker_credential" version = "1.4.0" @@ -4790,6 +4950,12 @@ dependencies = [ "num-traits", ] +[[package]] +name = "event-listener" +version = "2.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" + [[package]] name = "event-listener" version = "5.4.1" @@ -4807,10 +4973,19 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" dependencies = [ - "event-listener", + "event-listener 5.4.1", "pin-project-lite", ] +[[package]] +name = "executor-trait" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c39dff9342e4e0e16ce96be751eb21a94e94a87bb2f6e63ad1961c2ce109bf" +dependencies = [ + "async-trait", +] + [[package]] name = "expect-test" version = "1.5.1" @@ -4891,6 +5066,15 @@ dependencies = [ "serde", ] +[[package]] +name = "fastrand" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" +dependencies = [ + "instant", +] + [[package]] name = "fastrand" version = "2.5.0" @@ -4990,6 +5174,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flagset" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" + [[package]] name = "flatbuffers" version = "25.12.19" @@ -5026,13 +5216,24 @@ dependencies = [ "num-traits", ] +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "flume" version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ - "fastrand", + "fastrand 2.5.0", "futures-core", "futures-sink", "spin", @@ -5201,13 +5402,28 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" +[[package]] +name = "futures-lite" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce" +dependencies = [ + "fastrand 1.9.0", + "futures-core", + "futures-io", + "memchr", + "parking", + "pin-project-lite", + "waker-fn", +] + [[package]] name = "futures-lite" version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" dependencies = [ - "fastrand", + "fastrand 2.5.0", "futures-core", "futures-io", "parking", @@ -5995,6 +6211,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024" + [[package]] name = "hermit-abi" version = "0.5.2" @@ -6318,7 +6540,7 @@ dependencies = [ "hyper-util", "log", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "tokio", "tokio-rustls", "tower-service", @@ -6622,7 +6844,7 @@ dependencies = [ "bytemuck", "bytes", "dashmap", - "flume", + "flume 0.12.0", "futures", "futures-util", "iggy_binary_protocol", @@ -6752,7 +6974,7 @@ dependencies = [ "dotenvy", "figlet-rs", "figment", - "flume", + "flume 0.12.0", "futures", "iggy", "iggy_common", @@ -7119,6 +7341,19 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_rabbitmq_sink" +version = "0.4.1-edge.1" +dependencies = [ + "async-trait", + "dashmap", + "iggy_connector_sdk", + "lapin", + "serde", + "tokio", + "tracing", +] + [[package]] name = "iggy_connector_random_source" version = "0.4.1-edge.1" @@ -7406,6 +7641,15 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "instant" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0242819d153cba4b4b05a5a8f2a7e9bbf97b6055b2a002b395c96b5ff3c0222" +dependencies = [ + "cfg-if", +] + [[package]] name = "integer-encoding" version = "3.0.4" @@ -7443,6 +7687,7 @@ dependencies = [ "journal", "jsonwebtoken", "keyring-core", + "lapin", "lazy_static", "libc", "mongodb", @@ -7497,6 +7742,17 @@ dependencies = [ "rustversion", ] +[[package]] +name = "io-lifetimes" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2" +dependencies = [ + "hermit-abi 0.3.9", + "libc", + "windows-sys 0.48.0", +] + [[package]] name = "io-uring" version = "0.7.13" @@ -7787,6 +8043,28 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d4345964bb142484797b161f473a503a434de77149dd8c7427788c6e13379388" +[[package]] +name = "lapin" +version = "2.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02d2aa4725b9607915fa1a73e940710a3be6af508ce700e56897cbe8847fbb07" +dependencies = [ + "amq-protocol", + "async-global-executor-trait", + "async-reactor-trait", + "async-trait", + "executor-trait", + "flume 0.11.1", + "futures-core", + "futures-io", + "parking_lot", + "pinky-swear", + "reactor-trait", + "serde", + "tracing", + "waker-fn", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -8029,6 +8307,12 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8c7b0a3383c2a1002d11349c92c85a666a5fb679e96c79d782cf0dbe557fd6ee" +[[package]] +name = "linux-raw-sys" +version = "0.3.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519" + [[package]] name = "linux-raw-sys" version = "0.4.15" @@ -8537,12 +8821,12 @@ version = "0.12.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "957228ad12042ee839f93c8f257b62b4c0ab5eaae1d4fa60de53b27c9d7c5046" dependencies = [ - "async-lock", + "async-lock 3.4.2", "crossbeam-channel", "crossbeam-epoch", "crossbeam-utils", "equivalent", - "event-listener", + "event-listener 5.4.1", "futures-util", "parking_lot", "portable-atomic", @@ -8592,7 +8876,7 @@ dependencies = [ "md-5 0.11.0", "mongocrypt", "mongodb-internal-macros", - "pbkdf2", + "pbkdf2 0.13.0", "percent-encoding", "rand 0.9.5", "rustc_version_runtime", @@ -8932,7 +9216,7 @@ version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" dependencies = [ - "hermit-abi", + "hermit-abi 0.5.2", "libc", ] @@ -9146,6 +9430,12 @@ dependencies = [ "uuid", ] +[[package]] +name = "openssl-probe" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" + [[package]] name = "openssl-probe" version = "0.2.1" @@ -9298,6 +9588,28 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a80800c0488c3a21695ea981a54918fbb37abf04f4d0720c453632255e2ff0e" +[[package]] +name = "p12-keystore" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cae83056e7cb770211494a0ecf66d9fa7eba7d00977e5bb91f0e925b40b937f" +dependencies = [ + "cbc", + "cms", + "der", + "des", + "hex", + "hmac 0.12.1", + "pkcs12", + "pkcs5", + "rand 0.9.5", + "rc2", + "sha1 0.10.7", + "sha2 0.10.9", + "thiserror 2.0.19", + "x509-parser 0.17.0", +] + [[package]] name = "p256" version = "0.13.2" @@ -9518,6 +9830,16 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" +[[package]] +name = "pbkdf2" +version = "0.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2" +dependencies = [ + "digest 0.10.7", + "hmac 0.12.1", +] + [[package]] name = "pbkdf2" version = "0.13.0" @@ -9706,6 +10028,18 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "pinky-swear" +version = "6.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1ea6e230dd3a64d61bcb8b79e597d3ab6b4c94ec7a234ce687dd718b4f2e657" +dependencies = [ + "doc-comment", + "flume 0.11.1", + "parking_lot", + "tracing", +] + [[package]] name = "pinned" version = "0.1.0" @@ -9724,7 +10058,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" dependencies = [ "atomic-waker", - "fastrand", + "fastrand 2.5.0", "futures-io", ] @@ -9739,6 +10073,36 @@ dependencies = [ "spki", ] +[[package]] +name = "pkcs12" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "695b3df3d3cc1015f12d70235e35b6b79befc5fa7a9b95b951eab1dd07c9efc2" +dependencies = [ + "cms", + "const-oid 0.9.6", + "der", + "digest 0.10.7", + "spki", + "x509-cert", + "zeroize", +] + +[[package]] +name = "pkcs5" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6" +dependencies = [ + "aes 0.8.4", + "cbc", + "der", + "pbkdf2 0.12.2", + "scrypt", + "sha2 0.10.9", + "spki", +] + [[package]] name = "pkcs8" version = "0.10.2" @@ -9787,6 +10151,22 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5dcdc93847ad24990939cce6e1804361e903efcb5f99daa5abd87943a9d6d7ba" +[[package]] +name = "polling" +version = "2.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b2d323e8ca7996b3e23126511a523f7e62924d93ecd5ae73b333815b0eb3dce" +dependencies = [ + "autocfg", + "bitflags 1.3.2", + "cfg-if", + "concurrent-queue", + "libc", + "log", + "pin-project-lite", + "windows-sys 0.48.0", +] + [[package]] name = "polling" version = "3.11.0" @@ -9795,7 +10175,7 @@ checksum = "5d0e4f59085d47d8241c88ead0f274e8a0cb551f3625263c05eb8dd897c34218" dependencies = [ "cfg-if", "concurrent-queue", - "hermit-abi", + "hermit-abi 0.5.2", "pin-project-lite", "rustix 1.1.4", "windows-sys 0.61.2", @@ -10537,6 +10917,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rc2" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62c64daa8e9438b84aaae55010a93f396f8e60e3911590fcba770d04643fc1dd" +dependencies = [ + "cipher 0.4.4", +] + [[package]] name = "rcgen" version = "0.14.8" @@ -10547,10 +10936,21 @@ dependencies = [ "ring", "rustls-pki-types", "time", - "x509-parser", + "x509-parser 0.18.1", "yasna", ] +[[package]] +name = "reactor-trait" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "438a4293e4d097556730f4711998189416232f009c137389e0f961d2bc0ddc58" +dependencies = [ + "async-trait", + "futures-core", + "futures-io", +] + [[package]] name = "reborrow" version = "0.5.5" @@ -10713,7 +11113,7 @@ dependencies = [ "pin-project-lite", "quinn", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-pki-types", "serde", "serde_json", @@ -11176,6 +11576,20 @@ dependencies = [ "nom 7.1.3", ] +[[package]] +name = "rustix" +version = "0.37.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "519165d378b97752ca44bbe15047d5d3409e875f39327546b42ac81d7e18c1b6" +dependencies = [ + "bitflags 1.3.2", + "errno", + "io-lifetimes", + "libc", + "linux-raw-sys 0.3.8", + "windows-sys 0.48.0", +] + [[package]] name = "rustix" version = "0.38.44" @@ -11218,16 +11632,42 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-connector" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70cc376c6ba1823ae229bacf8ad93c136d93524eab0e4e5e0e4f96b9c4e5b212" +dependencies = [ + "log", + "rustls", + "rustls-native-certs 0.7.3", + "rustls-pki-types", + "rustls-webpki", +] + +[[package]] +name = "rustls-native-certs" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5bfb394eeed242e909609f56089eecfe5fda225042e8b171791b9c95f5931e5" +dependencies = [ + "openssl-probe 0.1.6", + "rustls-pemfile", + "rustls-pki-types", + "schannel", + "security-framework 2.11.1", +] + [[package]] name = "rustls-native-certs" version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" dependencies = [ - "openssl-probe", + "openssl-probe 0.2.1", "rustls-pki-types", "schannel", - "security-framework", + "security-framework 3.7.0", ] [[package]] @@ -11261,10 +11701,10 @@ dependencies = [ "log", "once_cell", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-platform-verifier-android", "rustls-webpki", - "security-framework", + "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", "windows-sys 0.61.2", @@ -11337,6 +11777,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "salsa20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213" +dependencies = [ + "cipher 0.4.4", +] + [[package]] name = "same-file" version = "1.0.6" @@ -11405,6 +11854,17 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "scrypt" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f" +dependencies = [ + "pbkdf2 0.12.2", + "salsa20", + "sha2 0.10.9", +] + [[package]] name = "sd-notify" version = "0.5.0" @@ -11474,6 +11934,19 @@ dependencies = [ "zbus", ] +[[package]] +name = "security-framework" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" +dependencies = [ + "bitflags 2.13.1", + "core-foundation 0.9.4", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -11776,7 +12249,7 @@ dependencies = [ "err_trail", "error_set", "figlet-rs", - "flume", + "flume 0.12.0", "fs2", "futures", "hash32 1.0.0", @@ -12247,6 +12720,16 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "199905e6153d6405f9728fe44daace35f8f837bbf830bb6e85fbd5828709a886" +[[package]] +name = "socket2" +version = "0.4.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7916fc008ca5542385b89a3d3ce689953c143e9304a9bf8beec1de48994c0d" +dependencies = [ + "libc", + "winapi", +] + [[package]] name = "socket2" version = "0.5.10" @@ -12355,7 +12838,7 @@ dependencies = [ "crc", "crossbeam-queue", "either", - "event-listener", + "event-listener 5.4.1", "futures-core", "futures-intrusive", "futures-io", @@ -12492,7 +12975,7 @@ checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" dependencies = [ "atoi", "chrono", - "flume", + "flume 0.12.0", "form_urlencoded", "futures-channel", "futures-core", @@ -12729,7 +13212,7 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9d6d5fbc4583cf3e5eee953506f13853a42ee6b4a21a983dffffb10c765cb80" dependencies = [ - "event-listener", + "event-listener 5.4.1", "futures-util", "local-event", "loom", @@ -12887,13 +13370,25 @@ dependencies = [ "xattr", ] +[[package]] +name = "tcp-stream" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "495b0abdce3dc1f8fd27240651c9e68890c14e9d9c61527b1ce44d8a5a7bd3d5" +dependencies = [ + "cfg-if", + "p12-keystore", + "rustls-connector", + "rustls-pemfile", +] + [[package]] name = "tempfile" version = "3.27.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ - "fastrand", + "fastrand 2.5.0", "getrandom 0.4.3", "once_cell", "rustix 1.1.4", @@ -14196,6 +14691,12 @@ dependencies = [ "libc", ] +[[package]] +name = "waker-fn" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "317211a0dc0ceedd78fb2ca9a44aed3d7b9b26f81870d485c07122b4350673b7" + [[package]] name = "walkdir" version = "2.5.0" @@ -14649,6 +15150,15 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets 0.48.5", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -14685,6 +15195,21 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm 0.48.5", + "windows_aarch64_msvc 0.48.5", + "windows_i686_gnu 0.48.5", + "windows_i686_msvc 0.48.5", + "windows_x86_64_gnu 0.48.5", + "windows_x86_64_gnullvm 0.48.5", + "windows_x86_64_msvc 0.48.5", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -14736,6 +15261,12 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -14748,6 +15279,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -14760,6 +15297,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -14784,6 +15327,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -14796,6 +15345,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -14808,6 +15363,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" @@ -14820,6 +15381,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" @@ -14909,6 +15476,34 @@ dependencies = [ "tap", ] +[[package]] +name = "x509-cert" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" +dependencies = [ + "const-oid 0.9.6", + "der", + "spki", +] + +[[package]] +name = "x509-parser" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4569f339c0c402346d4a75a9e39cf8dad310e287eef1ff56d4c68e5067f53460" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "rusticata-macros", + "thiserror 2.0.19", + "time", +] + [[package]] name = "x509-parser" version = "0.18.1" @@ -15077,17 +15672,17 @@ checksum = "fe18fb60dc696039e738717b76eaea21e7a4489bbb1885020b43c94236d7e98a" dependencies = [ "async-broadcast", "async-executor", - "async-io", - "async-lock", + "async-io 2.6.0", + "async-lock 3.4.2", "async-process", "async-recursion", "async-task", "async-trait", "blocking", "enumflags2", - "event-listener", + "event-listener 5.4.1", "futures-core", - "futures-lite", + "futures-lite 2.6.1", "hex", "libc", "ordered-stream", diff --git a/Cargo.toml b/Cargo.toml index 0147c62ed6..9935aff229 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,6 +43,7 @@ members = [ "core/connectors/sinks/mongodb_sink", "core/connectors/sinks/postgres_sink", "core/connectors/sinks/quickwit_sink", + "core/connectors/sinks/rabbitmq_sink", "core/connectors/sinks/s3_sink", "core/connectors/sinks/stdout_sink", "core/connectors/sinks/surrealdb_sink", @@ -210,6 +211,7 @@ journal = { path = "core/journal" } js-sys = "0.3" jsonwebtoken = { version = "10.4.0", features = ["rust_crypto"] } keyring-core = "1.0.0" +lapin = "2.5.1" lazy_static = "1.5.0" left-right = "0.11" lending-iterator = "0.1.7" diff --git a/core/connectors/README.md b/core/connectors/README.md index 64bfc9a1f8..70f77024c2 100644 --- a/core/connectors/README.md +++ b/core/connectors/README.md @@ -85,6 +85,7 @@ Each sink should have its own, custom configuration, which is passed along with - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog - **PostgreSQL Sink** - stores messages in PostgreSQL database tables - **Quickwit Sink** - indexes messages in Quickwit search engine +- **RabbitMQ Sink** - publishes messages to RabbitMQ exchanges via AMQP 0.9.1 - **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) - **Stdout Sink** - prints messages to standard output (useful for debugging/development) - **SurrealDB Sink** - writes messages into SurrealDB with deterministic record IDs for idempotent replay diff --git a/core/connectors/runtime/example_config/connectors/rabbitmq_sink.toml b/core/connectors/runtime/example_config/connectors/rabbitmq_sink.toml new file mode 100644 index 0000000000..faf4870ae8 --- /dev/null +++ b/core/connectors/runtime/example_config/connectors/rabbitmq_sink.toml @@ -0,0 +1,38 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +type = "sink" +key = "rabbitmq" +enabled = true +version = 0 +name = "RabbitMQ sink" +path = "target/release/libiggy_connector_rabbitmq_sink" +verbose = true + +[[streams]] +stream = "example_stream" +topics = ["example_topic"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "rabbitmq_sink_connector" + +[plugin_config] +amqp_url = "amqp://guest:guest@localhost:5672" +exchange = "iggy_events" +exchange_type = "topic" +routing_key = "iggy.messages" diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md index 57ea055490..c9933e5399 100644 --- a/core/connectors/sinks/README.md +++ b/core/connectors/sinks/README.md @@ -17,6 +17,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s | **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) | | **stdout_sink** | Prints messages to standard output (useful for debugging and development) | | **surrealdb_sink** | Writes messages into SurrealDB with deterministic record IDs for idempotent replay | +| **rabbitmq_sink** | Publishes messages to RabbitMQ exchanges via AMQP | The sink is represented by the single `Sink` trait, which defines the basic interface for all sink connectors. It provides methods for initializing the sink, writing data to external destination, and closing the sink. diff --git a/core/connectors/sinks/rabbitmq_sink/Cargo.toml b/core/connectors/sinks/rabbitmq_sink/Cargo.toml new file mode 100644 index 0000000000..8ba54d81ab --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/Cargo.toml @@ -0,0 +1,45 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "iggy_connector_rabbitmq_sink" +version = "0.4.1-edge.1" +description = "Iggy RabbitMQ sink connector for publishing stream messages to RabbitMQ exchanges via AMQP 0.9.1" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "rabbitmq", "amqp", "sink"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org" +documentation = "https://iggy.apache.org/docs" +repository = "https://github.com/apache/iggy" +readme = "../../README.md" +publish = false + +[package.metadata.cargo-machete] +ignored = ["dashmap"] + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +dashmap = { workspace = true } +iggy_connector_sdk = { workspace = true } +lapin = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } diff --git a/core/connectors/sinks/rabbitmq_sink/README.md b/core/connectors/sinks/rabbitmq_sink/README.md new file mode 100644 index 0000000000..798e212385 --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/README.md @@ -0,0 +1,18 @@ +# RabbitMQ Sink + +The RabbitMQ sink connector publishes messages from Iggy streams to RabbitMQ exchanges via AMQP 0.9.1. + +## Configuration + +- `amqp_url`: RabbitMQ connection URL (e.g. `amqp://guest:guest@localhost:5672`). +- `exchange`: Exchange name to publish to. +- `exchange_type`: Exchange type (`direct`, `topic`, `fanout`, `headers`). Defaults to `"topic"`. +- `routing_key`: Routing key for published messages. + +```toml +[plugin_config] +amqp_url = "amqp://guest:guest@localhost:5672" +exchange = "iggy_events" +exchange_type = "topic" +routing_key = "iggy.messages" +``` diff --git a/core/connectors/sinks/rabbitmq_sink/config.toml b/core/connectors/sinks/rabbitmq_sink/config.toml new file mode 100644 index 0000000000..3ed16ee2ee --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/config.toml @@ -0,0 +1,38 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +type = "sink" +key = "rabbitmq" +enabled = true +version = 0 +name = "RabbitMQ sink" +path = "../../target/release/libiggy_connector_rabbitmq_sink" +verbose = true + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "rabbitmq_sink_connector" + +[plugin_config] +amqp_url = "amqp://guest:guest@localhost:5672" +exchange = "iggy_events" +exchange_type = "topic" +routing_key = "iggy.messages" diff --git a/core/connectors/sinks/rabbitmq_sink/src/lib.rs b/core/connectors/sinks/rabbitmq_sink/src/lib.rs new file mode 100644 index 0000000000..239e40231e --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/src/lib.rs @@ -0,0 +1,388 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use async_trait::async_trait; +use iggy_connector_sdk::retry::{exponential_backoff, jitter}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use lapin::{ + BasicProperties, Channel, Connection, ConnectionProperties, ExchangeKind, + options::{ConfirmSelectOptions, ExchangeDeclareOptions}, + types::AMQPValue, + types::FieldTable, +}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +sink_connector!(RabbitMQSink); + +#[derive(Debug)] +struct RabbitMqState { + connection: Connection, + channel: Channel, +} + +#[derive(Debug)] +pub struct RabbitMQSink { + id: u32, + amqp_url: String, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + state: Mutex>, + reconnecting: AtomicBool, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + messages_published: AtomicU64, + publish_errors: AtomicU64, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RabbitMQSinkConfig { + #[serde(default = "default_amqp_url")] + amqp_url: String, + #[serde(default)] + exchange: Option, + #[serde(default = "default_exchange_type")] + exchange_type: Option, + #[serde(default)] + routing_key: Option, + #[serde(default = "default_true")] + include_metadata: Option, + #[serde(default)] + verbose_logging: Option, + #[serde(default = "default_max_retries")] + max_retries: Option, + #[serde(default = "default_retry_delay_secs")] + retry_delay_secs: Option, + #[serde(default = "default_max_retry_delay_secs")] + max_retry_delay_secs: Option, +} + +fn default_exchange_type() -> Option { + Some("topic".into()) +} + +fn default_amqp_url() -> String { + "amqp://guest:guest@localhost:5672".into() +} + +fn default_true() -> Option { + Some(true) +} + +fn default_max_retries() -> Option { + Some(3) +} +fn default_retry_delay_secs() -> Option { + Some(1) +} +fn default_max_retry_delay_secs() -> Option { + Some(5) +} + +impl RabbitMQSink { + pub fn new(id: u32, config: RabbitMQSinkConfig) -> Self { + RabbitMQSink { + id, + amqp_url: config.amqp_url, + exchange: config.exchange.unwrap_or_else(|| "iggy_events".into()), + exchange_type: config.exchange_type.unwrap_or_else(|| "topic".into()), + routing_key: config.routing_key.unwrap_or_else(|| "iggy.messages".into()), + include_metadata: config.include_metadata.unwrap_or(true), + verbose: config.verbose_logging.unwrap_or(false), + state: Mutex::new(None), + reconnecting: AtomicBool::new(false), + max_retries: config.max_retries.unwrap_or(3), + retry_delay: Duration::from_secs(config.retry_delay_secs.unwrap_or(1)), + max_retry_delay: Duration::from_secs(config.max_retry_delay_secs.unwrap_or(5)), + messages_published: AtomicU64::new(0), + publish_errors: AtomicU64::new(0), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result { + let mut attempts = 0u32; + + loop { + let channel = { + let guard = self.state.lock().await; + guard + .as_ref() + .map(|s| s.channel.clone()) + .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? + }; + + let mut last_error: Option = None; + let mut published: u64 = 0; + for message in messages { + let body = message.payload.clone().try_into_vec()?; + let mut props = BasicProperties::default(); + if self.include_metadata { + let mut headers = FieldTable::default(); + headers.insert( + "iggy_stream".into(), + AMQPValue::LongString(topic_metadata.stream.clone().into()), + ); + headers.insert( + "iggy_topic".into(), + AMQPValue::LongString(topic_metadata.topic.clone().into()), + ); + headers.insert( + "iggy_partition_id".into(), + AMQPValue::LongUInt(messages_metadata.partition_id), + ); + headers.insert( + "iggy_offset".into(), + AMQPValue::LongUInt(message.offset.try_into().unwrap_or(u32::MAX)), + ); + props = props.with_headers(headers); + } + + let confirm = channel + .basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ) + .await + .map_err(|e| Error::CannotStoreData(e.to_string()))?; + match confirm.await { + Ok(_) => published += 1, + Err(e) => { + last_error = Some(Error::CannotStoreData(format!("publish rejected: {e}"))); + break; + } + } + } + + if last_error.is_none() { + return Ok(published); + } + + let error = last_error.unwrap(); + attempts += 1; + + if !is_publish_retryable(&error) || attempts >= self.max_retries { + self.publish_errors + .fetch_add(messages.len() as u64 - published, Ordering::Relaxed); + return Err(Error::CannotStoreData(format!( + "batch publish failed after {attempts} attempts: {error}" + ))); + } + + match self.reconnect().await { + Ok(_) => {} + Err(reconnect_error) => { + self.publish_errors + .fetch_add(messages.len() as u64, Ordering::Relaxed); + return Err(Error::Connection(format!( + "failed to reconnect: {reconnect_error}" + ))); + } + } + + let delay = jitter(exponential_backoff( + self.retry_delay, + attempts.saturating_sub(1), + self.max_retry_delay, + )); + warn!( + "Transient RabbitMQ publish error for connector ID: {} (attempt {attempts}/{}): {error}. Retrying in {:?}.", + self.id, self.max_retries, delay + ); + tokio::time::sleep(delay).await; + } + } + + async fn reconnect(&self) -> Result<(), Error> { + if self + .reconnecting + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + tokio::time::sleep(self.retry_delay).await; + return Ok(()); + } + + warn!("Reconnecting RabbitMQ sink ID: {}", self.id); + let result = async { + let conn = Connection::connect(&self.amqp_url, ConnectionProperties::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let channel = conn + .create_channel() + .await + .map_err(|e| Error::Connection(e.to_string()))?; + channel + .confirm_select(ConfirmSelectOptions::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let exchange_kind = match self.exchange_type.as_str() { + "direct" => ExchangeKind::Direct, + "topic" => ExchangeKind::Topic, + "fanout" => ExchangeKind::Fanout, + "headers" => ExchangeKind::Headers, + other => { + return Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}" + ))); + } + }; + channel + .exchange_declare( + &self.exchange, + exchange_kind, + ExchangeDeclareOptions::default(), + FieldTable::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + *self.state.lock().await = Some(RabbitMqState { + connection: conn, + channel, + }); + Ok::<(), Error>(()) + } + .await; + self.reconnecting.store(false, Ordering::Release); + result + } +} + +#[async_trait] +impl Sink for RabbitMQSink { + async fn open(&mut self) -> Result<(), Error> { + let exchange_kind = match self.exchange_type.as_str() { + "direct" => ExchangeKind::Direct, + "topic" => ExchangeKind::Topic, + "fanout" => ExchangeKind::Fanout, + "headers" => ExchangeKind::Headers, + other => { + return Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}. Valid: direct, topic, fanout, headers" + ))); + } + }; + let conn = Connection::connect(&self.amqp_url, ConnectionProperties::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + let channel = conn + .create_channel() + .await + .map_err(|e| Error::Connection(e.to_string()))?; + channel + .confirm_select(lapin::options::ConfirmSelectOptions::default()) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + + channel + .exchange_declare( + &self.exchange, + exchange_kind, + ExchangeDeclareOptions::default(), + FieldTable::default(), + ) + .await + .map_err(|e| Error::Connection(e.to_string()))?; + *self.state.get_mut() = Some(RabbitMqState { + connection: conn, + channel, + }); + info!( + "Opened RabbitMQ sink ID: {}, connected to exchange: {}", + self.id, self.exchange + ); + + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec, + ) -> Result<(), Error> { + let published = self + .publish_batch_with_retry(topic_metadata, &messages_metadata, &messages) + .await?; + self.messages_published + .fetch_add(published, Ordering::Relaxed); + if self.verbose { + info!( + "Published {published} messages to exchange: {}", + self.exchange + ); + } else { + debug!( + "Published {published} messages to exchange: {}", + self.exchange + ); + } + Ok(()) + } + + async fn close(&mut self) -> Result<(), Error> { + let published = self.messages_published.load(Ordering::Relaxed); + let errors = self.publish_errors.load(Ordering::Relaxed); + info!( + "RabbitMQ sink ID: {} processed {} messages with {} errors", + self.id, published, errors + ); + + if let Some(state) = self.state.get_mut().take() { + state + .channel + .close(200, "OK") + .await + .map_err(|e| Error::Connection(e.to_string()))?; + state + .connection + .close(200, "OK") + .await + .map_err(|e| Error::Connection(e.to_string()))?; + } + Ok(()) + } +} + +fn is_publish_retryable(error: &Error) -> bool { + let msg = error.to_string().to_lowercase(); + msg.contains("connection") + || msg.contains("timeout") + || msg.contains("broken pipe") + || msg.contains("reset by peer") + || msg.contains("resource locked") + || msg.contains("channel closed") +} diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index d4d0bdbd53..34e97b15eb 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -69,6 +69,7 @@ iggy_connector_sdk = { workspace = true, features = ["api"] } journal = { workspace = true, optional = true } jsonwebtoken = { workspace = true } keyring-core = { workspace = true } +lapin = { workspace = true } lazy_static = { workspace = true } libc = { workspace = true } mongodb = { workspace = true } diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index 6d75fbf977..71ea35314f 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -27,6 +27,7 @@ mod influxdb; mod mongodb; mod postgres; mod quickwit; +mod rabbitmq; mod s3; mod surrealdb; mod wiremock; @@ -79,6 +80,10 @@ pub use postgres::{ PostgresSourceOps, }; pub use quickwit::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture}; +pub use rabbitmq::{ + RabbitMqOps, RabbitMqSinkDirectFixture, RabbitMqSinkFanoutFixture, RabbitMqSinkFixture, + RabbitMqSinkRawSchemaFixture, RabbitMqSinkWithoutMetadataFixture, +}; pub use s3::{S3SinkFixture, S3SinkOps, S3SinkRotationFixture}; pub use surrealdb::{ SurrealDbOps, SurrealDbSinkBatchFixture, SurrealDbSinkFixture, SurrealDbSinkJsonFixture, diff --git a/core/integration/tests/connectors/fixtures/rabbitmq/container.rs b/core/integration/tests/connectors/fixtures/rabbitmq/container.rs new file mode 100644 index 0000000000..65f9b92aeb --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/container.rs @@ -0,0 +1,302 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::connectors::fixtures; +use futures::StreamExt; +use integration::harness::TestBinaryError; +use lapin::{ + Connection, ConnectionProperties, ExchangeKind, + options::{BasicConsumeOptions, ExchangeDeclareOptions, QueueBindOptions, QueueDeclareOptions}, + types::FieldTable, +}; +use std::time::Duration; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage, ImageExt}; +use tokio::time::sleep; +use tracing::info; +use uuid::Uuid; + +const RABBITMQ_IMAGE: &str = "docker.io/rabbitmq"; +const RABBITMQ_TAG: &str = "4.0-management"; +const RABBITMQ_PORT: u16 = 5672; +const RABBITMQ_READY_MSG: &str = "Time to start RabbitMQ:"; +const RABBITMQ_BOOT_ATTEMPTS: usize = 60; +const RABBITMQ_BOOT_INTERVAL_MS: u64 = 1000; + +pub(super) const DEFAULT_TEST_STREAM: &str = "test_stream"; +pub(super) const DEFAULT_TEST_TOPIC: &str = "test_topic"; +pub(super) const DEFAULT_EXCHANGE: &str = "iggy_events"; +pub(super) const DEFAULT_EXCHANGE_TYPE: &str = "topic"; +pub(super) const DEFAULT_ROUTING_KEY: &str = "iggy.messages"; +pub(super) const DEFAULT_CONSUMER_GROUP: &str = "rabbitmq_sink_test_cg"; + +pub(super) const ENV_SINK_AMQP_URL: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_PLUGIN_CONFIG_AMQP_URL"; +pub(super) const ENV_SINK_EXCHANGE: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_PLUGIN_CONFIG_EXCHANGE"; +pub(super) const ENV_SINK_EXCHANGE_TYPE: &str = + "IGGY_CONNECTORS_SINK_RABBITMQ_PLUGIN_CONFIG_EXCHANGE_TYPE"; +pub(super) const ENV_SINK_ROUTING_KEY: &str = + "IGGY_CONNECTORS_SINK_RABBITMQ_PLUGIN_CONFIG_ROUTING_KEY"; +pub(super) const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_STREAMS_0_STREAM"; +pub(super) const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_STREAMS_0_TOPICS"; +pub(super) const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_STREAMS_0_SCHEMA"; +pub(super) const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_RABBITMQ_STREAMS_0_CONSUMER_GROUP"; +pub(super) const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_RABBITMQ_PATH"; +pub(super) const ENV_SINK_INCLUDE_METADATA: &str = + "IGGY_CONNECTORS_SINK_RABBITMQ_PLUGIN_CONFIG_INCLUDE_METADATA"; + +pub(super) enum RabbitMqExchangeSetup { + Topic, + Fanout, + Direct, +} +pub struct RabbitMqContainer { + #[allow(dead_code)] + container: ContainerAsync, + pub(super) amqp_url: String, + pub(super) queue_names: Vec, + pub(super) exchange_setup: RabbitMqExchangeSetup, +} + +impl RabbitMqContainer { + async fn start_container() -> Result<(ContainerAsync, String), TestBinaryError> { + let container = GenericImage::new(RABBITMQ_IMAGE, RABBITMQ_TAG) + .with_exposed_port(RABBITMQ_PORT.tcp()) + .with_wait_for(WaitFor::message_on_stdout(RABBITMQ_READY_MSG)) + .with_mapped_port(0, RABBITMQ_PORT.tcp()) + .with_container_name(fixtures::unique_container_name("rabbitmq")) + .start() + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to start container: {e}"), + })?; + + let mapped_port = container + .ports() + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to get ports: {e}"), + })? + .map_to_host_port_ipv4(RABBITMQ_PORT) + .ok_or_else(|| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: "No mapping for RabbitMQ port".to_string(), + })?; + let amqp_url = format!("amqp://guest:guest@127.0.0.1:{mapped_port}"); + Ok((container, amqp_url)) + } + + async fn start_with( + exchange_setup: RabbitMqExchangeSetup, + queue_count: usize, + ) -> Result { + let (container, amqp_url) = Self::start_container().await?; + let queue_names = (0..queue_count) + .map(|index| format!("test_queue_{}_{}", Uuid::new_v4().simple(), index)) + .collect(); + + let instance = Self { + container, + amqp_url, + exchange_setup, + queue_names, + }; + instance.wait_until_ready().await?; + + info!("RabbitMQ container available at {}", instance.amqp_url); + Ok(instance) + } + + pub(super) async fn start() -> Result { + Self::start_with(RabbitMqExchangeSetup::Topic, 1).await + } + + pub(super) async fn start_fanout() -> Result { + Self::start_with(RabbitMqExchangeSetup::Fanout, 2).await + } + + pub(super) async fn start_direct() -> Result { + Self::start_with(RabbitMqExchangeSetup::Direct, 1).await + } + + async fn wait_until_ready(&self) -> Result<(), TestBinaryError> { + let mut last_error = None; + + for _ in 0..RABBITMQ_BOOT_ATTEMPTS { + match Connection::connect(&self.amqp_url, ConnectionProperties::default()).await { + Ok(_) => { + return self.setup_exchange_and_queue().await; + } + Err(error) => last_error = Some(error.to_string()), + } + sleep(Duration::from_millis(RABBITMQ_BOOT_INTERVAL_MS)).await; + } + + let detail = last_error + .map(|error| format!(" Last error: {error}")) + .unwrap_or_default(); + Err(TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("RabbitMQ did not become ready.{detail}"), + }) + } + + async fn setup_exchange_and_queue(&self) -> Result<(), TestBinaryError> { + let conn = Connection::connect(&self.amqp_url, ConnectionProperties::default()) + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to create connection for consume: {e}"), + })?; + let channel = conn + .create_channel() + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to create channel for consume: {e}"), + })?; + + let exchange_kind = match self.exchange_setup { + RabbitMqExchangeSetup::Topic => ExchangeKind::Topic, + RabbitMqExchangeSetup::Fanout => ExchangeKind::Fanout, + RabbitMqExchangeSetup::Direct => ExchangeKind::Direct, + }; + channel + .exchange_declare( + DEFAULT_EXCHANGE, + exchange_kind, + ExchangeDeclareOptions::default(), + FieldTable::default(), + ) + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to declare exchange for consume: {e}"), + })?; + + for queue_name in &self.queue_names { + channel + .queue_declare( + queue_name, + QueueDeclareOptions { + auto_delete: true, + ..Default::default() + }, + FieldTable::default(), + ) + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to create queue for consume: {e}"), + })?; + + channel + .queue_bind( + queue_name, + DEFAULT_EXCHANGE, + DEFAULT_ROUTING_KEY, + QueueBindOptions::default(), + FieldTable::default(), + ) + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "RabbitMqContainer".to_string(), + message: format!("Failed to bind queue to exchange for consume: {e}"), + })?; + } + + Ok(()) + } +} + +pub struct ConsumedDelivery { + pub data: Vec, + pub headers: lapin::types::FieldTable, +} + +pub trait RabbitMqOps: Sync { + fn container(&self) -> &RabbitMqContainer; + + fn queue_names(&self) -> &[String] { + &self.container().queue_names + } + + async fn consume_messages( + &self, + count: usize, + ) -> Result, TestBinaryError> { + self.consume_messages_from(&self.container().queue_names[0], count) + .await + } + + async fn consume_messages_from( + &self, + queue_name: &str, + count: usize, + ) -> Result, TestBinaryError> { + let conn = Connection::connect(&self.container().amqp_url, ConnectionProperties::default()) + .await + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to connect to RabbitMQ for consume: {e}"), + })?; + let channel = conn + .create_channel() + .await + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to create channel for consume: {e}"), + })?; + + let mut consumer = channel + .basic_consume( + queue_name, + "", + BasicConsumeOptions::default(), + FieldTable::default(), + ) + .await + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to start consumer: {e}"), + })?; + + let mut messages = Vec::with_capacity(count); + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + while messages.len() < count && tokio::time::Instant::now() < deadline { + match tokio::time::timeout(Duration::from_secs(1), consumer.next()).await { + Ok(Some(delivery)) => { + let delivery = delivery.map_err(|e| TestBinaryError::InvalidState { + message: format!("Consumer error: {e}"), + })?; + let data = delivery.data.clone(); + let headers = delivery.properties.headers().clone().unwrap_or_default(); + delivery.ack(Default::default()).await.map_err(|e| { + TestBinaryError::InvalidState { + message: format!("Failed to ack message: {e}"), + } + })?; + messages.push(ConsumedDelivery { data, headers }); + } + Ok(None) => break, + Err(_) => continue, + } + } + + Ok(messages) + } +} diff --git a/core/integration/tests/connectors/fixtures/rabbitmq/mod.rs b/core/integration/tests/connectors/fixtures/rabbitmq/mod.rs new file mode 100644 index 0000000000..0e2ec2f80b --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/mod.rs @@ -0,0 +1,25 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod container; +mod sink; + +pub use container::RabbitMqOps; +pub use sink::{ + RabbitMqSinkDirectFixture, RabbitMqSinkFanoutFixture, RabbitMqSinkFixture, + RabbitMqSinkRawSchemaFixture, RabbitMqSinkWithoutMetadataFixture, +}; diff --git a/core/integration/tests/connectors/fixtures/rabbitmq/sink.rs b/core/integration/tests/connectors/fixtures/rabbitmq/sink.rs new file mode 100644 index 0000000000..e585ad2613 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/sink.rs @@ -0,0 +1,211 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use super::container::{ + DEFAULT_CONSUMER_GROUP, DEFAULT_EXCHANGE, DEFAULT_EXCHANGE_TYPE, DEFAULT_ROUTING_KEY, + DEFAULT_TEST_STREAM, DEFAULT_TEST_TOPIC, ENV_SINK_AMQP_URL, ENV_SINK_EXCHANGE, + ENV_SINK_EXCHANGE_TYPE, ENV_SINK_INCLUDE_METADATA, ENV_SINK_PATH, ENV_SINK_ROUTING_KEY, + ENV_SINK_STREAMS_0_CONSUMER_GROUP, ENV_SINK_STREAMS_0_SCHEMA, ENV_SINK_STREAMS_0_STREAM, + ENV_SINK_STREAMS_0_TOPICS, RabbitMqContainer, RabbitMqOps, +}; +use async_trait::async_trait; +use iggy_connector_sdk::Schema; +use integration::harness::{TestBinaryError, TestFixture}; +use std::collections::HashMap; + +pub struct RabbitMqSinkFixture { + container: RabbitMqContainer, + include_metadata: bool, + schema: Schema, +} + +impl RabbitMqOps for RabbitMqSinkFixture { + fn container(&self) -> &RabbitMqContainer { + &self.container + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start().await?; + Ok(Self { + container, + include_metadata: true, + schema: Schema::Json, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + let mut envs = HashMap::new(); + envs.insert( + ENV_SINK_AMQP_URL.to_string(), + self.container.amqp_url.clone(), + ); + envs.insert(ENV_SINK_EXCHANGE.to_string(), DEFAULT_EXCHANGE.into()); + envs.insert( + ENV_SINK_EXCHANGE_TYPE.to_string(), + DEFAULT_EXCHANGE_TYPE.into(), + ); + envs.insert(ENV_SINK_ROUTING_KEY.to_string(), DEFAULT_ROUTING_KEY.into()); + envs.insert( + ENV_SINK_STREAMS_0_STREAM.to_string(), + DEFAULT_TEST_STREAM.into(), + ); + envs.insert( + ENV_SINK_STREAMS_0_TOPICS.to_string(), + format!("[{}]", DEFAULT_TEST_TOPIC), + ); + envs.insert( + ENV_SINK_STREAMS_0_SCHEMA.to_string(), + self.schema.to_string(), + ); + envs.insert( + ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(), + DEFAULT_CONSUMER_GROUP.into(), + ); + envs.insert( + ENV_SINK_PATH.to_string(), + "../../target/debug/libiggy_connector_rabbitmq_sink".into(), + ); + envs.insert( + ENV_SINK_INCLUDE_METADATA.to_string(), + self.include_metadata.to_string(), + ); + envs + } +} + +pub struct RabbitMqSinkWithoutMetadataFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkWithoutMetadataFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkWithoutMetadataFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start().await?; + Ok(Self { + inner: RabbitMqSinkFixture { + container, + include_metadata: false, + schema: Schema::Json, + }, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + self.inner.connectors_runtime_envs() + } +} + +pub struct RabbitMqSinkRawSchemaFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkRawSchemaFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkRawSchemaFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start().await?; + Ok(Self { + inner: RabbitMqSinkFixture { + container, + include_metadata: true, + schema: Schema::Raw, + }, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + self.inner.connectors_runtime_envs() + } +} + +pub struct RabbitMqSinkFanoutFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkFanoutFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkFanoutFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start_fanout().await?; + Ok(Self { + inner: RabbitMqSinkFixture { + container, + include_metadata: true, + schema: Schema::Json, + }, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + let mut envs = self.inner.connectors_runtime_envs(); + envs.insert(ENV_SINK_EXCHANGE_TYPE.to_string(), "fanout".into()); + envs + } +} + +pub struct RabbitMqSinkDirectFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkDirectFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkDirectFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start_direct().await?; + Ok(Self { + inner: RabbitMqSinkFixture { + container, + include_metadata: true, + schema: Schema::Json, + }, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + let mut envs = self.inner.connectors_runtime_envs(); + envs.insert(ENV_SINK_EXCHANGE_TYPE.to_string(), "direct".into()); + envs + } +} diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index a1433160b9..abb7a37363 100644 --- a/core/integration/tests/connectors/mod.rs +++ b/core/integration/tests/connectors/mod.rs @@ -28,6 +28,7 @@ mod influxdb; mod mongodb; mod postgres; mod quickwit; +mod rabbitmq; mod random; mod random_source_liveness; mod runtime; diff --git a/core/integration/tests/connectors/rabbitmq/mod.rs b/core/integration/tests/connectors/rabbitmq/mod.rs new file mode 100644 index 0000000000..c4a30f2755 --- /dev/null +++ b/core/integration/tests/connectors/rabbitmq/mod.rs @@ -0,0 +1,18 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod rabbitmq_sink; diff --git a/core/integration/tests/connectors/rabbitmq/rabbitmq_sink.rs b/core/integration/tests/connectors/rabbitmq/rabbitmq_sink.rs new file mode 100644 index 0000000000..288c8998bf --- /dev/null +++ b/core/integration/tests/connectors/rabbitmq/rabbitmq_sink.rs @@ -0,0 +1,299 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use crate::connectors::fixtures::{ + RabbitMqOps, RabbitMqSinkDirectFixture, RabbitMqSinkFanoutFixture, RabbitMqSinkFixture, + RabbitMqSinkRawSchemaFixture, RabbitMqSinkWithoutMetadataFixture, +}; +use bytes::Bytes; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::Identifier; +use iggy_common::MessageClient; +use integration::harness::seeds; +use integration::iggy_harness; +use lapin::types::{AMQPValue, ShortString}; + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] + +async fn json_messages_are_published_to_rabbitmq_exchange( + harness: &TestHarness, + fixture: RabbitMqSinkFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let payloads = [ + serde_json::json!({"name": "Alice"}), + serde_json::json!({"name": "Bob"}), + serde_json::json!({"name": "Carol"}), + ]; + let mut messages: Vec = payloads + .iter() + .enumerate() + .map(|(idx, payload)| { + IggyMessage::builder() + .id((idx + 1) as u128) + .payload(Bytes::from(serde_json::to_vec(payload).unwrap())) + .build() + .unwrap() + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + let delivered = fixture.consume_messages(3).await.unwrap(); + assert_eq!(delivered.len(), 3); + for (idx, delivery) in delivered.iter().enumerate() { + let value: serde_json::Value = serde_json::from_slice(&delivery.data).unwrap(); + assert_eq!(value, payloads[idx]); + assert_eq!( + header_str(&delivery.headers, "iggy_stream").as_deref(), + Some("test_stream") + ); + assert_eq!( + header_str(&delivery.headers, "iggy_topic").as_deref(), + Some("test_topic") + ); + } +} + +fn header_str(headers: &lapin::types::FieldTable, key: &str) -> Option { + headers + .inner() + .get(&ShortString::from(key)) + .and_then(|v| match v { + AMQPValue::LongString(s) => Some(s.to_string()), + _ => None, + }) +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] + +async fn given_direct_exchange_when_published_should_deliver_to_bound_queue( + harness: &TestHarness, + fixture: RabbitMqSinkDirectFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let payloads = [ + serde_json::json!({"name": "Alice"}), + serde_json::json!({"name": "Bob"}), + serde_json::json!({"name": "Carol"}), + ]; + let mut messages: Vec = payloads + .iter() + .enumerate() + .map(|(idx, payload)| { + IggyMessage::builder() + .id((idx + 1) as u128) + .payload(Bytes::from(serde_json::to_vec(payload).unwrap())) + .build() + .unwrap() + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + let delivered = fixture.consume_messages(3).await.unwrap(); + assert_eq!(delivered.len(), 3); + for (idx, delivery) in delivered.iter().enumerate() { + let value: serde_json::Value = serde_json::from_slice(&delivery.data).unwrap(); + assert_eq!(value, payloads[idx]); + assert_eq!( + header_str(&delivery.headers, "iggy_stream").as_deref(), + Some("test_stream") + ); + assert_eq!( + header_str(&delivery.headers, "iggy_topic").as_deref(), + Some("test_topic") + ); + } +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_include_metadata_false_when_published_should_not_include_iggy_headers( + harness: &TestHarness, + fixture: RabbitMqSinkWithoutMetadataFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let payload = serde_json::json!({"name": "Alice"}); + let mut messages = vec![ + IggyMessage::builder() + .id(1) + .payload(Bytes::from(serde_json::to_vec(&payload).unwrap())) + .build() + .unwrap(), + ]; + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + let delivered = fixture.consume_messages(1).await.unwrap(); + assert_eq!(delivered.len(), 1); + let value: serde_json::Value = serde_json::from_slice(&delivered[0].data).unwrap(); + assert_eq!(value, payload); + assert!(delivered[0].headers.inner().is_empty()); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_raw_schema_when_published_should_preserve_raw_payload_bytes( + harness: &TestHarness, + fixture: RabbitMqSinkRawSchemaFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let raw_payloads: Vec> = vec![ + b"plain text message".to_vec(), + vec![0x00, 0x01, 0x02, 0xFF, 0xFE, 0xFD], + vec![0xDE, 0xAD, 0xBE, 0xEF], + ]; + + let mut messages: Vec = raw_payloads + .iter() + .enumerate() + .map(|(idx, payload)| { + IggyMessage::builder() + .id((idx + 1) as u128) + .payload(Bytes::from(payload.clone())) + .build() + .unwrap() + }) + .collect(); + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + let delivered = fixture.consume_messages(3).await.unwrap(); + assert_eq!(delivered.len(), 3); + for (idx, delivery) in delivered.iter().enumerate() { + assert_eq!(delivery.data, raw_payloads[idx]); + assert_eq!( + header_str(&delivery.headers, "iggy_stream").as_deref(), + Some("test_stream") + ); + assert_eq!( + header_str(&delivery.headers, "iggy_topic").as_deref(), + Some("test_topic") + ); + } +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_fanout_exchange_when_published_should_deliver_to_all_bound_queues( + harness: &TestHarness, + fixture: RabbitMqSinkFanoutFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let payload = serde_json::json!({"name": "Alice"}); + let mut messages = vec![ + IggyMessage::builder() + .id(1) + .payload(Bytes::from(serde_json::to_vec(&payload).unwrap())) + .build() + .unwrap(), + ]; + + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + + assert_eq!( + fixture.queue_names().len(), + 2, + "fanout fixture should bind two queues" + ); + for queue_name in fixture.queue_names() { + let delivered = fixture.consume_messages_from(queue_name, 1).await.unwrap(); + assert_eq!( + delivered.len(), + 1, + "fanout exchange must deliver to every bound queue" + ); + let value: serde_json::Value = serde_json::from_slice(&delivered[0].data).unwrap(); + assert_eq!(value, payload); + assert_eq!( + header_str(&delivered[0].headers, "iggy_stream").as_deref(), + Some("test_stream") + ); + assert_eq!( + header_str(&delivered[0].headers, "iggy_topic").as_deref(), + Some("test_topic") + ); + } +} diff --git a/core/integration/tests/connectors/rabbitmq/sink.toml b/core/integration/tests/connectors/rabbitmq/sink.toml new file mode 100644 index 0000000000..e425ed503b --- /dev/null +++ b/core/integration/tests/connectors/rabbitmq/sink.toml @@ -0,0 +1,20 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[connectors] +config_type = "local" +config_dir = "../connectors/sinks/rabbitmq_sink" diff --git a/scripts/bump-version.sh b/scripts/bump-version.sh index d1e44e334a..e291136cc3 100755 --- a/scripts/bump-version.sh +++ b/scripts/bump-version.sh @@ -87,7 +87,7 @@ EOF } RUST_COMPONENTS="rust-sdk rust-common rust-binary-protocol rust-server rust-cli rust-connector-sdk rust-mcp rust-bench rust-bench-dashboard-frontend rust-bench-dashboard-server rust-bench-report" -CONNECTOR_SINK_COMPONENTS="rust-connector-delta-sink rust-connector-elasticsearch-sink rust-connector-http-sink rust-connector-iceberg-sink rust-connector-influxdb-sink rust-connector-mongodb-sink rust-connector-postgres-sink rust-connector-quickwit-sink rust-connector-stdout-sink rust-connector-surrealdb-sink" +CONNECTOR_SINK_COMPONENTS="rust-connector-delta-sink rust-connector-elasticsearch-sink rust-connector-http-sink rust-connector-iceberg-sink rust-connector-influxdb-sink rust-connector-mongodb-sink rust-connector-postgres-sink rust-connector-quickwit-sink rust-connector-stdout-sink rust-connector-surrealdb-sink rust-connector-rabbitmq-sink" CONNECTOR_SOURCE_COMPONENTS="rust-connector-elasticsearch-source rust-connector-influxdb-source rust-connector-postgres-source rust-connector-random-source" CONNECTOR_COMPONENTS="rust-connector-runtime ${CONNECTOR_SINK_COMPONENTS} ${CONNECTOR_SOURCE_COMPONENTS}" SDK_COMPONENTS="sdk-python sdk-node sdk-go sdk-csharp sdk-java"