diff --git a/.github/workflows/_build_rust_artifacts.yml b/.github/workflows/_build_rust_artifacts.yml index bd56b10716..a1592ad78f 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 2514c3227a..7dde3b7192 100644 --- a/.github/workflows/edge-release.yml +++ b/.github/workflows/edge-release.yml @@ -105,6 +105,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 4596a2e11b..9cd46447cc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -67,7 +67,7 @@ dependencies = [ "actix-rt", "actix-service", "actix-utils", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "brotli", "bytes", @@ -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]] @@ -659,7 +707,7 @@ dependencies = [ "arrow-schema 57.3.1", "arrow-select 57.3.1", "atoi", - "base64", + "base64 0.22.1", "chrono", "half", "lexical-core", @@ -680,7 +728,7 @@ dependencies = [ "arrow-schema 58.3.0", "arrow-select 58.3.0", "atoi", - "base64", + "base64 0.22.1", "chrono", "half", "lexical-core", @@ -1053,7 +1101,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", @@ -1131,12 +1179,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" @@ -1147,21 +1240,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", ] @@ -1173,17 +1275,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" @@ -1212,8 +1326,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", @@ -1287,7 +1401,7 @@ checksum = "0d8c50d65ce1b0e0cb65a785ff615f78860d7754290647d3b983208daa4f85e6" dependencies = [ "async-compression", "crc32fast", - "futures-lite", + "futures-lite 2.6.1", "pin-project", "thiserror 2.0.19", "tokio", @@ -1333,7 +1447,7 @@ version = "0.30.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "16e2cdb6d5ed835199484bb92bb8b3edd526effe995c61732580439c1a67e2e9" dependencies = [ - "base64", + "base64 0.22.1", "http 1.4.2", "log", "rustls", @@ -1421,7 +1535,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "hex", "http 1.4.2", "sha1 0.10.7", @@ -1510,7 +1624,7 @@ dependencies = [ "aws-types", "bytes", "bytes-utils", - "fastrand", + "fastrand 2.5.0", "http 1.4.2", "http-body 1.1.0", "percent-encoding", @@ -1538,7 +1652,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1564,7 +1678,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1590,7 +1704,7 @@ dependencies = [ "aws-smithy-types", "aws-types", "bytes", - "fastrand", + "fastrand 2.5.0", "http 0.2.12", "http 1.4.2", "regex-lite", @@ -1617,7 +1731,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", @@ -1694,7 +1808,7 @@ dependencies = [ "hyper-util", "pin-project-lite", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-pki-types", "tokio", "tokio-rustls", @@ -1746,7 +1860,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", @@ -1949,7 +2063,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", ] @@ -1966,6 +2080,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -2273,7 +2393,7 @@ dependencies = [ "async-channel", "async-task", "futures-io", - "futures-lite", + "futures-lite 2.6.1", "piper", ] @@ -2294,7 +2414,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee04c4c84f1f811b017f2fbb7dd8815c976e7ca98593de9c1e2afad0f636bff4" dependencies = [ "async-stream", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "bollard-buildkit-proto", "bollard-stubs", @@ -2315,7 +2435,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", @@ -2351,7 +2471,7 @@ version = "1.52.1-rc.29.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f0a8ca8799131c1837d1282c3f81f31e76ceb0ce426e04a7fe1ccee3287c066" dependencies = [ - "base64", + "base64 0.22.1", "bollard-buildkit-proto", "bytes", "prost", @@ -2447,7 +2567,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7969a9ba84b0ff843813e7249eed1678d9b6607ce5a3b8f0a47af3fcf7978e6e" dependencies = [ "ahash 0.8.12", - "base64", + "base64 0.22.1", "bitvec", "getrandom 0.2.17", "getrandom 0.3.4", @@ -2911,6 +3031,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" @@ -3016,7 +3148,7 @@ dependencies = [ "compio-log", "compio-send-wrapper", "crossbeam-queue", - "flume", + "flume 0.12.0", "futures-util", "io-uring", "libc", @@ -3024,7 +3156,7 @@ dependencies = [ "mod_use", "once_cell", "pastey 0.2.3", - "polling", + "polling 3.11.0", "rustix 1.1.4", "smallvec", "socket2 0.6.5", @@ -3137,7 +3269,7 @@ dependencies = [ "compio-log", "compio-net", "compio-runtime", - "flume", + "flume 0.12.0", "futures-util", "libc", "quinn-proto", @@ -3404,6 +3536,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" @@ -3787,7 +3925,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5f8c8abaf78cfe3cb7838d7cfcf75e4e3bf02fe3e54571e0a1f10bb485cf8fbf" dependencies = [ "async-stream", - "base64", + "base64 0.22.1", "cfg-if", "cfg_aliases", "compio", @@ -4283,6 +4421,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", ] @@ -4301,6 +4441,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" @@ -4397,6 +4548,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" @@ -4520,13 +4680,19 @@ 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" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "29547a1dc60885a552306986316bc9701ba120c1a8db6769fa68691529ad373d" dependencies = [ - "base64", + "base64 0.22.1", "serde", "serde_json", ] @@ -4638,7 +4804,7 @@ version = "9.1.0-alpha.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "12bb303aa6e1d28c0c86b6fbfe484fd0fd3f512629aeed1ac4f6b85f81d9834a" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "dyn-clone", "flate2", @@ -4859,6 +5025,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" @@ -4876,10 +5048,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" @@ -4937,6 +5118,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" @@ -5036,6 +5226,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" @@ -5072,13 +5268,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", @@ -5247,13 +5454,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", @@ -6041,6 +6263,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" @@ -6364,7 +6592,7 @@ dependencies = [ "hyper-util", "log", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "tokio", "tokio-rustls", "tower-service", @@ -6390,7 +6618,7 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-channel", "futures-util", @@ -6468,7 +6696,7 @@ dependencies = [ "as-any", "async-trait", "backon", - "base64", + "base64 0.22.1", "bimap", "bytes", "chrono", @@ -6668,7 +6896,7 @@ dependencies = [ "bytemuck", "bytes", "dashmap", - "flume", + "flume 0.12.0", "futures", "futures-util", "iggy_binary_protocol", @@ -6800,7 +7028,7 @@ dependencies = [ "dotenvy", "figlet-rs", "figment", - "flume", + "flume 0.12.0", "futures", "iggy", "iggy_common", @@ -6892,7 +7120,7 @@ dependencies = [ "aes-gcm", "async-broadcast", "async-trait", - "base64", + "base64 0.22.1", "blake3", "bon", "byte-unit", @@ -6960,7 +7188,7 @@ name = "iggy_connector_doris_sink" version = "0.2.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "blake3", "bytes", "humantime", @@ -6980,7 +7208,7 @@ name = "iggy_connector_elasticsearch_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "dashmap", "elasticsearch", "iggy_common", @@ -7016,7 +7244,7 @@ name = "iggy_connector_http_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "humantime", "iggy_connector_sdk", @@ -7058,7 +7286,7 @@ version = "0.5.0-edge.4" dependencies = [ "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "iggy_common", "iggy_connector_sdk", @@ -7080,7 +7308,7 @@ dependencies = [ "ahash 0.8.12", "async-trait", "axum", - "base64", + "base64 0.22.1", "chrono", "csv", "dashmap", @@ -7104,7 +7332,7 @@ name = "iggy_connector_meilisearch_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "iggy_common", "iggy_connector_sdk", "meilisearch-sdk", @@ -7157,7 +7385,7 @@ name = "iggy_connector_postgres_source" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "chrono", "dashmap", "futures", @@ -7188,6 +7416,23 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_rabbitmq_sink" +version = "0.4.1-edge.1" +dependencies = [ + "async-trait", + "dashmap", + "iggy", + "iggy_common", + "iggy_connector_sdk", + "lapin", + "secrecy", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "iggy_connector_random_source" version = "0.5.0-edge.4" @@ -7232,7 +7477,7 @@ name = "iggy_connector_s3_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "byte-unit", "chrono", "dashmap", @@ -7255,7 +7500,7 @@ dependencies = [ "anyhow", "apache-avro", "async-trait", - "base64", + "base64 0.22.1", "dashmap", "flatbuffers", "http 1.4.2", @@ -7301,7 +7546,7 @@ name = "iggy_connector_surrealdb_sink" version = "0.5.0-edge.4" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "iggy_common", "iggy_connector_sdk", @@ -7497,6 +7742,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" @@ -7510,7 +7764,7 @@ dependencies = [ "arrow 57.3.1", "assert_cmd", "async-trait", - "base64", + "base64 0.22.1", "bon", "bytemuck", "bytes", @@ -7536,6 +7790,7 @@ dependencies = [ "journal", "jsonwebtoken", "keyring-core", + "lapin", "lazy_static", "libc", "mongodb", @@ -7593,6 +7848,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" @@ -7808,7 +8074,7 @@ version = "10.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc" dependencies = [ - "base64", + "base64 0.22.1", "ed25519-dalek", "getrandom 0.2.17", "hmac 0.12.1", @@ -7892,11 +8158,33 @@ 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-regex" -version = "3.6.0" +version = "3.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6bae91019476d3ec7147de9aa291cadb6d870abf2f3015d2da73a90325ac1496" +checksum = "4994ba703f78b083e2f7946dac9251abd83fd43a0365f030e99b69be5b4b9ef9" dependencies = [ "lazy-regex-proc_macros", "once_cell", @@ -7905,9 +8193,9 @@ dependencies = [ [[package]] name = "lazy-regex-proc_macros" -version = "3.6.0" +version = "3.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4de9c1e1439d8b7b3061b2d209809f447ca33241733d9a3c01eabf2dc8d94358" +checksum = "fd97232314824e6dbef1918a871bb93f51070455e3715bf26e19a6d01aa977a0" dependencies = [ "proc-macro2", "quote", @@ -8132,6 +8420,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" @@ -8667,12 +8961,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", @@ -8705,7 +8999,7 @@ version = "3.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b814038f367d212f55de0a630cb35102a9b8ca23785a86955d62c0087c93846d" dependencies = [ - "base64", + "base64 0.22.1", "bitflags 2.13.1", "bson", "derive-where", @@ -8722,7 +9016,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", @@ -9035,7 +9329,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", ] @@ -9130,7 +9424,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "622acbc9100d3c10e2ee15804b0caa40e55c933d5aa53814cd520805b7958a49" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "chrono", "form_urlencoded", @@ -9171,7 +9465,7 @@ checksum = "27be39870c558e1fbc5a5f8d4a29aa916268c1dbcb9d467f2a9bbab35598060e" dependencies = [ "arc-swap", "async-trait", - "base64", + "base64 0.22.1", "bytes", "cargo_metadata", "cfg-if", @@ -9237,7 +9531,7 @@ checksum = "d075ab8a203a6ab4bc1bce0a4b9fe486a72bf8b939037f4b78d95386384bc80a" dependencies = [ "anyhow", "backon", - "base64", + "base64 0.22.1", "bytes", "crc32c", "futures", @@ -9258,6 +9552,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" @@ -9410,6 +9710,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" @@ -9487,7 +9809,7 @@ dependencies = [ "arrow-ipc 57.3.1", "arrow-schema 57.3.1", "arrow-select 57.3.1", - "base64", + "base64 0.22.1", "brotli", "bytes", "chrono", @@ -9522,7 +9844,7 @@ dependencies = [ "arrow-ipc 58.3.0", "arrow-schema 58.3.0", "arrow-select 58.3.0", - "base64", + "base64 0.22.1", "brotli", "bytes", "chrono", @@ -9633,6 +9955,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" @@ -9698,7 +10030,7 @@ version = "3.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" dependencies = [ - "base64", + "base64 0.22.1", "serde_core", ] @@ -9778,13 +10110,13 @@ dependencies = [ [[package]] name = "pgwire" -version = "0.40.4" +version = "0.40.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7981cfde34009be689a05a30c497ad5fbb552531d3d54230b3627264ff1bc384" +checksum = "8265901ede50d0879fe401c6fe282e7e4ff83ae10a48c1f8e4f89b92f7d5f604" dependencies = [ "async-trait", "aws-lc-rs", - "base64", + "base64 0.23.1", "bytes", "chrono", "derive-new", @@ -9884,6 +10216,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" @@ -9902,7 +10246,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c835479a4443ded371d6c535cbfd8d31ad92c5d23ae9770a61bc155e4992a3c1" dependencies = [ "atomic-waker", - "fastrand", + "fastrand 2.5.0", "futures-io", ] @@ -9917,6 +10261,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" @@ -9965,6 +10339,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" @@ -9973,7 +10363,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", @@ -10024,7 +10414,7 @@ version = "0.6.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" dependencies = [ - "base64", + "base64 0.22.1", "byteorder", "bytes", "fallible-iterator", @@ -10742,6 +11132,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" @@ -10752,10 +11151,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" @@ -10874,7 +11284,7 @@ checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701" dependencies = [ "anyhow", "async-trait", - "base64", + "base64 0.22.1", "chrono", "form_urlencoded", "getrandom 0.2.17", @@ -10901,7 +11311,7 @@ version = "0.12.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "futures-core", "futures-util", @@ -10918,7 +11328,7 @@ dependencies = [ "pin-project-lite", "quinn", "rustls", - "rustls-native-certs", + "rustls-native-certs 0.8.4", "rustls-pki-types", "serde", "serde_json", @@ -10944,7 +11354,7 @@ version = "0.13.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "encoding_rs", "futures-channel", @@ -11142,7 +11552,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "14db48ee17a9ba61810ab1a9c1beb7d06d8136ae39ac25a1137f10d357af01af" dependencies = [ "async-trait", - "base64", + "base64 0.22.1", "bytes", "chrono", "futures", @@ -11304,7 +11714,7 @@ dependencies = [ "async-trait", "aws-creds", "aws-region", - "base64", + "base64 0.22.1", "bytes", "cfg-if", "futures-util", @@ -11382,6 +11792,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" @@ -11424,16 +11848,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]] @@ -11467,10 +11917,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", @@ -11543,6 +11993,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" @@ -11611,6 +12070,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" @@ -11680,6 +12150,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" @@ -11895,7 +12378,7 @@ version = "3.21.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76a5c54c7310e7b8b9577c286d7e399ddd876c3e12b3ed917a8aabc4b96e9e8c" dependencies = [ - "base64", + "base64 0.22.1", "bs58", "chrono", "hex", @@ -12394,6 +12877,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" @@ -12517,14 +13010,14 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05b44e85bf579a8eeb4ceaa77a3a523baf2bf0e9bac7e40f405d537b5d2d5ccb" dependencies = [ - "base64", + "base64 0.22.1", "bytes", "cfg-if", "chrono", "crc", "crossbeam-queue", "either", - "event-listener", + "event-listener 5.4.1", "futures-core", "futures-intrusive", "futures-io", @@ -12623,7 +13116,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "87a2bdd6e83f6b3ea525ca9fee568030508b58355a43d0b2c1674d5f79dcd65e" dependencies = [ "atoi", - "base64", + "base64 0.22.1", "bitflags 2.13.1", "byteorder", "chrono", @@ -12661,7 +13154,7 @@ checksum = "488e99c397a62007e4229aec669a179816339afc6d2620ca6fa420dbee2e982c" dependencies = [ "atoi", "chrono", - "flume", + "flume 0.12.0", "form_urlencoded", "futures-channel", "futures-core", @@ -12898,7 +13391,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", @@ -13056,13 +13549,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", @@ -13567,7 +14072,7 @@ checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", "axum", - "base64", + "base64 0.22.1", "bytes", "h2 0.4.15", "http 1.4.2", @@ -14135,7 +14640,7 @@ version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dea7109cdcd5864d4eeb1b58a1648dc9bf520360d7af16ec26d0a9354bafcfc0" dependencies = [ - "base64", + "base64 0.22.1", "flate2", "log", "percent-encoding", @@ -14152,7 +14657,7 @@ version = "0.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e994ba84b0bd1b1b0cf92878b7ef898a5c1760108fe7b6010327e274917a808c" dependencies = [ - "base64", + "base64 0.22.1", "http 1.4.2", "httparse", "log", @@ -14183,7 +14688,7 @@ version = "0.45.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "80be9b06fbae3b8b303400ab20778c80bbaf338f563afe567cf3c9eea17b47ef" dependencies = [ - "base64", + "base64 0.22.1", "data-url", "flate2", "fontdb", @@ -14397,6 +14902,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" @@ -14876,6 +15387,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" @@ -14912,6 +15432,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" @@ -14963,6 +15498,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" @@ -14975,6 +15516,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" @@ -14987,6 +15534,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" @@ -15011,6 +15564,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" @@ -15023,6 +15582,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" @@ -15035,6 +15600,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" @@ -15047,6 +15618,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" @@ -15099,7 +15676,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "08db1edfb05d9b3c1542e521aea074442088292f00b5f28e435c714a98f85031" dependencies = [ "assert-json-diff", - "base64", + "base64 0.22.1", "deadpool", "futures", "http 1.4.2", @@ -15136,6 +15713,17 @@ 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-certificate" version = "0.25.0" @@ -15155,6 +15743,23 @@ dependencies = [ "zeroize", ] +[[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" @@ -15334,17 +15939,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 1544a23105..7affe00946 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,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/redshift_sink", "core/connectors/sinks/s3_sink", "core/connectors/sinks/stdout_sink", @@ -216,6 +217,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" libc = "0.2.188" diff --git a/core/connectors/README.md b/core/connectors/README.md index 890a263b5f..f88b3d2554 100644 --- a/core/connectors/README.md +++ b/core/connectors/README.md @@ -86,6 +86,7 @@ Each sink should have its own, custom configuration, which is passed along with - **Meilisearch Sink** - indexes messages in Meilisearch - **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 - **Reshift Sink** - stores messages in Redshift warehouse tables via S3 as staging - **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) 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..8d09f65ee7 --- /dev/null +++ b/core/connectors/runtime/example_config/connectors/rabbitmq_sink.toml @@ -0,0 +1,46 @@ +# 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" +durable_exchange = true +delivery_mode = "persistent" +include_metadata = true +verbose_logging = false +max_retries = 3 +retry_delay_secs = 1 +max_retry_delay_secs = 5 +timeout_secs = 30 diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md index 91de1fd084..1046edb4c3 100644 --- a/core/connectors/sinks/README.md +++ b/core/connectors/sinks/README.md @@ -19,6 +19,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..2fd4dd715e --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/Cargo.toml @@ -0,0 +1,51 @@ +# 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 = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +lapin = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +serde_json = { 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..2e8072ae74 --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/README.md @@ -0,0 +1,48 @@ +# RabbitMQ Sink + +The RabbitMQ sink connector publishes messages from Iggy streams to RabbitMQ exchanges via AMQP 0.9.1. + +## Configuration + +| Field | Type | Default | Description | +| --- | --- | --- | --- | +| `amqp_url` | string | `amqp://guest:guest@localhost:5672` | RabbitMQ connection URL, including credentials. Treated as a secret: never logged or serialized verbatim. | +| `exchange` | string | `iggy_events` | Exchange name to publish to. | +| `exchange_type` | string | `topic` | Exchange type: `direct`, `topic`, `fanout`, `headers`. | +| `routing_key` | string | `iggy.messages` | Routing key for published messages. | +| `durable_exchange` | bool | `true` | Declare the exchange as durable. Must match the durability of an operator-pre-created exchange, otherwise RabbitMQ closes the channel with `PRECONDITION_FAILED`. | +| `delivery_mode` | string | `persistent` | AMQP delivery mode: `persistent` (2) or `non_persistent` (1). Persistent messages survive broker restarts when the exchange and queue are durable; it forces an fsync on publish, so non-durable topologies may prefer `non_persistent`. | +| `include_metadata` | bool | `true` | Add `iggy_stream`, `iggy_topic`, `iggy_partition_id`, `iggy_offset` message headers. User-supplied headers are always preserved, regardless of this flag. | +| `verbose_logging` | bool | `false` | Log each published batch at `info` level instead of `debug`. | +| `max_retries` | u32 | `3` | Maximum transient publish retries before failing the batch. | +| `retry_delay_secs` | u64 | `1` | Base retry delay in seconds. | +| `max_retry_delay_secs` | u64 | `5` | Upper bound for exponential backoff. | +| `timeout_secs` | u64 | `30` | Timeout for connection, publish, and publisher-confirmation operations. | + +User headers on consumed Iggy messages are forwarded as AMQP headers: string values become AMQP `LongString`, raw binary values become `ByteArray`. This allows routing through a `headers` exchange on original user headers. + +Publishes are confirmed via `ConfirmSelect`. With `mandatory = true`, a message with a routing key that matches no binding is returned by RabbitMQ and the batch fails with a permanent error. + +**Delivery guarantee:** in-request retry only. Transient failures (connection loss, `Nack`, channel exceptions, +timeouts) are retried within a single `consume()` call up to `max_retries`, resuming at the first unconfirmed +message. Publishes are pipelined (all messages sent, then confirmed in order), so a failure mid-batch can already +have delivered the in-flight tail; resuming re-publishes those, so delivery is **at-least-once within a batch**. +However, the connectors runtime commits the consumer offset at poll time and discards `consume()`'s return value, +so there is no cross-poll redrive or DLQ: a failure that outlives the retry budget, or a crash mid-batch, is +**at-most-once** across polls. + +```toml +[plugin_config] +amqp_url = "amqp://guest:guest@localhost:5672" +exchange = "iggy_events" +exchange_type = "topic" +routing_key = "iggy.messages" +durable_exchange = true +delivery_mode = "persistent" +include_metadata = true +verbose_logging = false +max_retries = 3 +retry_delay_secs = 1 +max_retry_delay_secs = 5 +timeout_secs = 30 +``` diff --git a/core/connectors/sinks/rabbitmq_sink/config.toml b/core/connectors/sinks/rabbitmq_sink/config.toml new file mode 100644 index 0000000000..4c6bbfc00e --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/config.toml @@ -0,0 +1,46 @@ +# 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" +durable_exchange = true +delivery_mode = "persistent" +include_metadata = true +verbose_logging = false +max_retries = 3 +retry_delay_secs = 1 +max_retry_delay_secs = 5 +timeout_secs = 30 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..f32a311519 --- /dev/null +++ b/core/connectors/sinks/rabbitmq_sink/src/lib.rs @@ -0,0 +1,746 @@ +// 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::prelude::HeaderKind; +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}, + publisher_confirm::{Confirmation, PublisherConfirm}, + types::{AMQPValue, ByteArray, FieldTable, ShortString}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::Mutex; +use tokio::time::timeout; +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: SecretString, + exchange: String, + exchange_type: String, + routing_key: String, + include_metadata: bool, + verbose: bool, + durable_exchange: bool, + delivery_mode: u8, + timeout: Duration, + 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", + serialize_with = "iggy_common::serde_secret::serialize_secret" + )] + amqp_url: SecretString, + #[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, + #[serde(default = "default_true")] + durable_exchange: Option, + #[serde(default = "default_delivery_mode")] + delivery_mode: Option, + #[serde(default = "default_timeout_secs")] + timeout_secs: Option, +} + +fn default_exchange_type() -> Option { + Some("topic".into()) +} + +fn default_amqp_url() -> SecretString { + SecretString::from("amqp://guest:guest@localhost:5672") +} + +fn default_delivery_mode() -> Option { + Some("persistent".into()) +} + +fn default_timeout_secs() -> Option { + Some(30) +} + +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 { + let delivery_mode = match config.delivery_mode.as_deref() { + Some("non_persistent") => 1, + Some("persistent") => 2, + Some(other) => { + warn!( + "Unknown delivery_mode: {other}, defaulting to persistent for connector ID: {id}" + ); + 2 + } + None => 2, + }; + 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), + durable_exchange: config.durable_exchange.unwrap_or(true), + delivery_mode, + timeout: Duration::from_secs(config.timeout_secs.unwrap_or(30)), + 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), + } + } + + fn exchange_kind(&self) -> Result { + match self.exchange_type.as_str() { + "direct" => Ok(ExchangeKind::Direct), + "topic" => Ok(ExchangeKind::Topic), + "fanout" => Ok(ExchangeKind::Fanout), + "headers" => Ok(ExchangeKind::Headers), + other => Err(Error::InvalidConfigValue(format!( + "unknown exchange_type: {other}. Valid: direct, topic, fanout, headers" + ))), + } + } + + async fn publish_batch_with_retry( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + messages: &[ConsumedMessage], + ) -> Result { + let mut attempts = 0u32; + let mut confirmed: usize = 0; + + loop { + let channel = { + let guard = self.state.lock().await; + match guard.as_ref() { + Some(state) => state.channel.clone(), + None => { + drop(guard); + self.reconnect().await?; + self.state + .lock() + .await + .as_ref() + .map(|s| s.channel.clone()) + .ok_or_else(|| Error::Connection("RabbitMQ not connected".into()))? + } + } + }; + + let mut last_error: Option = None; + let mut last_retryable = false; + let mut in_flight: Vec<(PublisherConfirm, u64)> = + Vec::with_capacity(messages.len() - confirmed); + for message in &messages[confirmed..] { + let body = message.payload.try_to_bytes()?; + let mut props = BasicProperties::default() + .with_delivery_mode(self.delivery_mode) + .with_message_id(ShortString::from(message.offset.to_string())); + let headers = self.build_headers(topic_metadata, messages_metadata, message); + if !headers.inner().is_empty() { + props = props.with_headers(headers); + } + + let confirm = match timeout( + self.timeout, + channel.basic_publish( + &self.exchange, + &self.routing_key, + lapin::options::BasicPublishOptions { + mandatory: true, + ..Default::default() + }, + &body, + props, + ), + ) + .await + { + Ok(Ok(confirm)) => confirm, + Ok(Err(e)) => { + last_error = Some(Error::CannotStoreData(e.to_string())); + last_retryable = is_lapin_error_retryable(&e); + self.clear_state().await; + break; + } + Err(_) => { + last_error = Some(Error::CannotStoreData("publish timed out".into())); + last_retryable = true; + break; + } + }; + in_flight.push((confirm, message.offset)); + } + + // Confirms resolve in publish order over already-overlapped RTTs; returns are + // matched by message_id (lapin's Return-to-confirm FIFO has no delivery tag), so a + // post-publish failure re-publishes the already-delivered tail (at-least-once in batch). + for (confirm, offset) in in_flight { + match timeout(self.timeout, confirm).await { + Ok(Ok(Confirmation::Ack(None))) => confirmed += 1, + Ok(Ok(Confirmation::Ack(Some(returned)))) => { + let returned_id = returned.delivery.properties.message_id(); + last_error = Some(Error::InvalidRecordValue(format!( + "message offset {offset} (id {returned_id:?}) returned as unroutable by RabbitMQ" + ))); + last_retryable = false; + break; + } + Ok(Ok(Confirmation::Nack(_))) => { + last_error = + Some(Error::CannotStoreData("message nack'd by RabbitMQ".into())); + last_retryable = true; + break; + } + Ok(Ok(Confirmation::NotRequested)) => { + last_error = Some(Error::CannotStoreData( + "publisher confirms not enabled".into(), + )); + last_retryable = false; + break; + } + Ok(Err(e)) => { + last_error = Some(Error::CannotStoreData(format!("publish rejected: {e}"))); + last_retryable = is_lapin_error_retryable(&e); + self.clear_state().await; + break; + } + Err(_) => { + last_error = Some(Error::CannotStoreData( + "publisher confirmation timed out".into(), + )); + last_retryable = true; + break; + } + } + } + + if last_error.is_none() { + return Ok(confirmed as u64); + } + + let error = last_error.unwrap(); + attempts += 1; + + if !last_retryable || attempts >= self.max_retries { + self.publish_errors + .fetch_add((messages.len() - confirmed) as u64, 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() - confirmed) 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 clear_state(&self) { + *self.state.lock().await = None; + } + + fn build_headers( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + message: &ConsumedMessage, + ) -> FieldTable { + let mut headers = FieldTable::default(); + if let Some(user_headers) = &message.headers + && !user_headers.is_empty() + { + for (key, value) in user_headers { + let name = ShortString::from(key.to_string_value()); + let amqp_value = match value.kind() { + HeaderKind::String => AMQPValue::LongString(value.to_string_value().into()), + _ => AMQPValue::ByteArray(ByteArray::from(value.as_bytes())), + }; + headers.insert(name, amqp_value); + } + } + if self.include_metadata { + 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::LongLongInt(message.offset as i64), + ); + } + headers + } + + 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 = timeout( + self.timeout, + Connection::connect( + self.amqp_url.expose_secret(), + ConnectionProperties::default(), + ), + ) + .await + .map_err(|_| Error::Connection("connection timed out".into()))? + .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 = self.exchange_kind()?; + channel + .exchange_declare( + &self.exchange, + exchange_kind, + ExchangeDeclareOptions { + durable: self.durable_exchange, + ..Default::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 = self.exchange_kind()?; + let conn = timeout( + self.timeout, + Connection::connect( + self.amqp_url.expose_secret(), + ConnectionProperties::default(), + ), + ) + .await + .map_err(|_| Error::Connection("connection timed out".into()))? + .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 { + durable: self.durable_exchange, + ..Default::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_lapin_error_retryable(error: &lapin::Error) -> bool { + match error { + lapin::Error::InvalidChannelState(_) + | lapin::Error::InvalidConnectionState(_) + | lapin::Error::IOError(_) + | lapin::Error::MissingHeartbeatError => true, + // AMQP soft channel errors: 405 RESOURCE_LOCKED, 320 CONNECTION_FORCED. + lapin::Error::ProtocolError(amqp_error) => matches!(amqp_error.get_id(), 320 | 405), + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use iggy::prelude::{HeaderKey, HeaderValue}; + use iggy_connector_sdk::{Payload, Schema}; + use std::str::FromStr; + + fn test_config() -> RabbitMQSinkConfig { + RabbitMQSinkConfig { + amqp_url: SecretString::from("amqp://guest:guest@localhost:5672"), + exchange: None, + exchange_type: None, + routing_key: None, + include_metadata: Some(true), + verbose_logging: Some(false), + max_retries: Some(3), + retry_delay_secs: Some(1), + max_retry_delay_secs: Some(5), + durable_exchange: None, + delivery_mode: None, + timeout_secs: None, + } + } + + fn test_sink(config: RabbitMQSinkConfig) -> RabbitMQSink { + RabbitMQSink::new(1, config) + } + + fn test_message(offset: u64) -> ConsumedMessage { + ConsumedMessage { + id: 1, + offset, + checksum: 0, + timestamp: 0, + origin_timestamp: 0, + headers: None, + payload: Payload::Text("payload".into()), + } + } + + #[test] + fn given_offset_above_u32_max_when_build_headers_should_encode_full_value() { + let sink = test_sink(test_config()); + let topic = TopicMetadata { + stream: "test_stream".into(), + topic: "test_topic".into(), + }; + let metadata = MessagesMetadata { + partition_id: 0, + current_offset: 0, + schema: Schema::Json, + }; + let headers = sink.build_headers(&topic, &metadata, &test_message(u32::MAX as u64 + 1)); + match headers.inner().get(&ShortString::from("iggy_offset")) { + Some(AMQPValue::LongLongInt(value)) => assert_eq!(*value, u32::MAX as i64 + 1), + other => panic!("expected LongLongInt, got {other:?}"), + } + } + + #[test] + fn given_user_headers_when_build_headers_should_preserve_string_and_binary() { + let mut user_headers = std::collections::BTreeMap::new(); + user_headers.insert( + HeaderKey::from_str("content-type").unwrap(), + HeaderValue::from_str("application/json").unwrap(), + ); + user_headers.insert( + HeaderKey::from_str("trace-id").unwrap(), + HeaderValue::try_from(vec![0xDE, 0xAD, 0xBE, 0xEF]).unwrap(), + ); + let mut message = test_message(0); + message.headers = Some(user_headers); + + let sink = test_sink(test_config()); + let topic = TopicMetadata { + stream: "test_stream".into(), + topic: "test_topic".into(), + }; + let metadata = MessagesMetadata { + partition_id: 0, + current_offset: 0, + schema: Schema::Json, + }; + let headers = sink.build_headers(&topic, &metadata, &message); + let inner = headers.inner(); + assert!(matches!( + inner.get(&ShortString::from("content-type")), + Some(AMQPValue::LongString(s)) if s.as_bytes() == b"application/json" + )); + assert!(matches!( + inner.get(&ShortString::from("trace-id")), + Some(AMQPValue::ByteArray(b)) if b.as_slice() == [0xDE, 0xAD, 0xBE, 0xEF] + )); + } + + #[test] + fn given_persistent_delivery_mode_when_new_should_set_delivery_mode_2() { + let sink = test_sink(test_config()); + assert_eq!(sink.delivery_mode, 2); + } + + #[test] + fn given_non_persistent_delivery_mode_when_new_should_set_delivery_mode_1() { + let mut config = test_config(); + config.delivery_mode = Some("non_persistent".into()); + let sink = test_sink(config); + assert_eq!(sink.delivery_mode, 1); + } + + #[test] + fn given_durable_exchange_when_new_should_default_to_true() { + let sink = test_sink(test_config()); + assert!(sink.durable_exchange); + } + + #[test] + fn given_known_exchange_type_when_exchange_kind_should_map_to_lapin_kind() { + let sink = test_sink(test_config()); + assert!(matches!(sink.exchange_kind(), Ok(ExchangeKind::Topic))); + } + + #[test] + fn given_unknown_exchange_type_when_exchange_kind_should_error() { + let mut config = test_config(); + config.exchange_type = Some("mystery".into()); + let sink = test_sink(config); + assert!(matches!( + sink.exchange_kind(), + Err(Error::InvalidConfigValue(_)) + )); + } + + #[test] + fn given_transient_lapin_error_when_is_lapin_error_retryable_should_return_true() { + let errors = [ + lapin::Error::InvalidChannelState(lapin::ChannelState::Closed), + lapin::Error::InvalidConnectionState(lapin::ConnectionState::Error), + lapin::Error::IOError(std::sync::Arc::new(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + "reset by peer", + ))), + lapin::Error::MissingHeartbeatError, + lapin::Error::ProtocolError(lapin::protocol::AMQPError::new( + lapin::protocol::AMQPErrorKind::Soft( + lapin::protocol::AMQPSoftError::RESOURCELOCKED, + ), + "RESOURCE_LOCKED".into(), + )), + ]; + for error in errors { + assert!( + is_lapin_error_retryable(&error), + "expected {error:?} to be retryable" + ); + } + } + + #[test] + fn given_permanent_lapin_error_when_is_lapin_error_retryable_should_return_false() { + let errors = [ + lapin::Error::ProtocolError(lapin::protocol::AMQPError::new( + lapin::protocol::AMQPErrorKind::Soft( + lapin::protocol::AMQPSoftError::PRECONDITIONFAILED, + ), + "PRECONDITION_FAILED".into(), + )), + lapin::Error::InvalidChannel(0), + ]; + for error in errors { + assert!( + !is_lapin_error_retryable(&error), + "expected {error:?} to be permanent" + ); + } + } + + #[test] + fn given_timeout_secs_when_new_should_set_timeout() { + let mut config = test_config(); + config.timeout_secs = Some(7); + let sink = test_sink(config); + assert_eq!(sink.timeout, Duration::from_secs(7)); + } + + #[test] + fn given_minimal_config_when_deserialized_should_apply_defaults() { + let config: RabbitMQSinkConfig = serde_json::from_str("{}").unwrap(); + let sink = RabbitMQSink::new(1, config); + assert_eq!( + sink.amqp_url.expose_secret(), + "amqp://guest:guest@localhost:5672" + ); + assert_eq!(sink.exchange, "iggy_events"); + assert_eq!(sink.exchange_type, "topic"); + assert_eq!(sink.routing_key, "iggy.messages"); + assert!(sink.include_metadata); + assert!(sink.durable_exchange); + assert_eq!(sink.delivery_mode, 2); + assert_eq!(sink.max_retries, 3); + assert_eq!(sink.retry_delay, Duration::from_secs(1)); + assert_eq!(sink.max_retry_delay, Duration::from_secs(5)); + assert_eq!(sink.timeout, Duration::from_secs(30)); + } + + #[test] + fn given_unknown_delivery_mode_when_new_should_default_to_persistent() { + let mut config = test_config(); + config.delivery_mode = Some("fancy".into()); + let sink = test_sink(config); + assert_eq!(sink.delivery_mode, 2); + } +} diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index 2a4acaa090..a15d9aa890 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -67,6 +67,7 @@ iggy_connector_sdk = { workspace = true, features = ["api"] } journal = { workspace = 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 7eaf6fe510..d95ff7702a 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -28,6 +28,7 @@ mod meilisearch; mod mongodb; mod postgres; mod quickwit; +mod rabbitmq; mod redshift; mod s3; mod surrealdb; @@ -82,6 +83,11 @@ pub use postgres::{ PostgresSourceOps, }; pub use quickwit::{QuickwitFixture, QuickwitOps, QuickwitPreCreatedFixture}; +pub use rabbitmq::{ + RabbitMqOps, RabbitMqSinkDirectFixture, RabbitMqSinkFanoutFixture, RabbitMqSinkFixture, + RabbitMqSinkHeadersFixture, RabbitMqSinkRawSchemaFixture, RabbitMqSinkUnroutableFixture, + RabbitMqSinkWithoutMetadataFixture, +}; pub use redshift::{ RedshiftSinkFixture, RedshiftSinkJsonFixture, RedshiftSinkNoArchiveFixture, RedshiftSinkVarbyteFixture, 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..3864604490 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/container.rs @@ -0,0 +1,327 @@ +// 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::{AMQPValue, 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"; + +#[derive(PartialEq)] +pub(super) enum RabbitMqExchangeSetup { + Topic, + Fanout, + Direct, + Headers, +} +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 + } + + pub(super) async fn start_headers() -> Result { + Self::start_with(RabbitMqExchangeSetup::Headers, 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, + RabbitMqExchangeSetup::Headers => ExchangeKind::Headers, + }; + channel + .exchange_declare( + DEFAULT_EXCHANGE, + exchange_kind, + ExchangeDeclareOptions { + durable: true, + ..Default::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}"), + })?; + + let mut bind_arguments = FieldTable::default(); + if self.exchange_setup == RabbitMqExchangeSetup::Headers { + bind_arguments.insert("x-match".into(), AMQPValue::LongString("all".into())); + bind_arguments.insert("x-user".into(), AMQPValue::LongString("alice".into())); + } + channel + .queue_bind( + queue_name, + DEFAULT_EXCHANGE, + DEFAULT_ROUTING_KEY, + QueueBindOptions::default(), + bind_arguments, + ) + .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> { + self.consume_messages_from_with_timeout(queue_name, count, Duration::from_secs(60)) + .await + } + + async fn consume_messages_from_with_timeout( + &self, + queue_name: &str, + count: usize, + timeout: Duration, + ) -> 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() + timeout; + 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..a8154b4e15 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/mod.rs @@ -0,0 +1,26 @@ +// 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, + RabbitMqSinkHeadersFixture, RabbitMqSinkRawSchemaFixture, RabbitMqSinkUnroutableFixture, + 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..8301dacdc1 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/rabbitmq/sink.rs @@ -0,0 +1,273 @@ +// 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 + } +} + +pub struct RabbitMqSinkHeadersFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkHeadersFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkHeadersFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start_headers().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(), "headers".into()); + envs + } +} + +pub struct RabbitMqSinkUnroutableFixture { + inner: RabbitMqSinkFixture, +} + +impl std::ops::Deref for RabbitMqSinkUnroutableFixture { + type Target = RabbitMqSinkFixture; + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +#[async_trait] +impl TestFixture for RabbitMqSinkUnroutableFixture { + async fn setup() -> Result { + let container = RabbitMqContainer::start().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_ROUTING_KEY.to_string(), "unroutable.key".into()); + envs + } +} diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index f6346bcc20..54dd1437b9 100644 --- a/core/integration/tests/connectors/mod.rs +++ b/core/integration/tests/connectors/mod.rs @@ -29,6 +29,7 @@ mod meilisearch; mod mongodb; mod postgres; mod quickwit; +mod rabbitmq; mod random; mod random_source_liveness; mod redshift; 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..f9c2376fb1 --- /dev/null +++ b/core/integration/tests/connectors/rabbitmq/rabbitmq_sink.rs @@ -0,0 +1,429 @@ +// 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, + RabbitMqSinkHeadersFixture, RabbitMqSinkRawSchemaFixture, RabbitMqSinkUnroutableFixture, + RabbitMqSinkWithoutMetadataFixture, +}; +use bytes::Bytes; +use iggy::prelude::{HeaderKey, HeaderValue, IggyMessage, Partitioning}; +use iggy_common::Identifier; +use iggy_common::MessageClient; +use integration::harness::seeds; +use integration::iggy_harness; +use lapin::types::{AMQPValue, ShortString}; +use std::collections::BTreeMap; +use std::str::FromStr; +use std::time::Duration; + +#[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") + ); + } +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_predeclared_durable_exchange_when_published_should_deliver( + 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 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); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_unroutable_routing_key_when_published_should_not_deliver( + harness: &TestHarness, + fixture: RabbitMqSinkUnroutableFixture, +) { + 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(); + + tokio::time::sleep(Duration::from_secs(5)).await; + let delivered = fixture + .consume_messages_from_with_timeout(&fixture.queue_names()[0], 1, Duration::from_secs(5)) + .await + .unwrap(); + assert!( + delivered.is_empty(), + "unroutable mandatory publish must not reach any queue" + ); +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/rabbitmq/sink.toml")), + seed = seeds::connector_stream +)] +async fn given_user_header_when_published_through_headers_exchange_should_route( + harness: &TestHarness, + fixture: RabbitMqSinkHeadersFixture, +) { + 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 user_headers = BTreeMap::from([( + HeaderKey::from_str("x-user").unwrap(), + HeaderValue::from_str("alice").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())) + .user_headers(user_headers) + .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_eq!( + header_str(&delivered[0].headers, "x-user").as_deref(), + Some("alice"), + "user header must survive to AMQP headers for headers-exchange routing" + ); +} 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 88413af9d2..82d278eb62 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"