feat(sink): add RabbitMQ sink for AMQP messaging

Add a RabbitMQ sink that publishes events to AMQP exchanges.

Features:
- Configurable exchange and routing key
- Dynamic routing via metadata
- Publisher confirms for delivery guarantees
- Single-pass processing with borrowing for efficiency

Author: psteinroe

Cargo.lock

@@ -6,6 +6,7 @@ version = "0.1.0"
 [features]
 default            = []
 sink-nats          = ["dep:async-nats"]
+sink-rabbitmq      = ["dep:lapin"]
 sink-redis-streams = ["dep:redis"]
 sink-redis-strings = ["dep:redis"]
 sink-webhook       = ["dep:reqwest"]
@@ -52,6 +53,7 @@ uuid         = { version = "1.19.0", default-features = false, features = ["v4"]
 
 # Optional sink dependencies.
 async-nats = { version = "0.38", optional = true }
+lapin = { version = "2.5", optional = true }
 redis = { version = "0.27", default-features = false, features = [
   "tokio-comp",
   "connection-manager",
@@ -67,9 +69,15 @@ tikv-jemallocator = { version = "0.6.1", default-features = false, features = [
 ] }
 
 
-ctor                   = { version = "0.4", optional = true }
-testcontainers         = { version = "0.23", optional = true, features = ["blocking"] }
-testcontainers-modules = { version = "0.11", optional = true, features = ["postgres", "redis", "nats", "blocking"] }
+ctor = { version = "0.4", optional = true }
+testcontainers = { version = "0.23", optional = true, features = ["blocking"] }
+testcontainers-modules = { version = "0.11", optional = true, features = [
+  "postgres",
+  "redis",
+  "nats",
+  "rabbitmq",
+  "blocking",
+] }
 
 [dev-dependencies]
 temp-env = "0.3"

Cargo.toml

@@ -9,6 +9,9 @@ use crate::sink::redis_strings::RedisStringsSinkConfig;
 #[cfg(feature = "sink-redis-streams")]
 use crate::sink::redis_streams::RedisStreamsSinkConfig;
 
+#[cfg(feature = "sink-rabbitmq")]
+use crate::sink::rabbitmq::RabbitmqSinkConfig;
+
 #[cfg(feature = "sink-webhook")]
 use crate::sink::webhook::WebhookSinkConfig;
 
@@ -36,6 +39,11 @@ pub enum SinkConfig {
     #[serde(rename = "nats")]
     Nats(NatsSinkConfig),
 
+    /// RabbitMQ sink for AMQP messaging.
+    #[cfg(feature = "sink-rabbitmq")]
+    #[serde(rename = "rabbitmq")]
+    Rabbitmq(RabbitmqSinkConfig),
+
     /// Webhook sink for HTTP POST delivery.
     #[cfg(feature = "sink-webhook")]
     #[serde(rename = "webhook")]

src/config/sink.rs

@@ -117,6 +117,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> {
             AnySink::Nats(s)
         }
 
+        #[cfg(feature = "sink-rabbitmq")]
+        SinkConfig::Rabbitmq(cfg) => {
+            use crate::sink::rabbitmq::RabbitmqSink;
+            let s = RabbitmqSink::new(cfg.clone()).await.map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to create RabbitMQ sink",
+                    e.to_string()
+                )
+            })?;
+            AnySink::Rabbitmq(s)
+        }
+
         #[cfg(feature = "sink-webhook")]
         SinkConfig::Webhook(cfg) => {
             use crate::sink::webhook::WebhookSink;
@@ -190,6 +203,11 @@ fn log_sink_config(config: &SinkConfig) {
             debug!("using nats sink");
         }
 
+        #[cfg(feature = "sink-rabbitmq")]
+        SinkConfig::Rabbitmq(_cfg) => {
+            debug!("using rabbitmq sink");
+        }
+
         #[cfg(feature = "sink-webhook")]
         SinkConfig::Webhook(_cfg) => {
             debug!("using webhook sink");

src/core.rs

@@ -10,6 +10,9 @@ pub mod redis_strings;
 #[cfg(feature = "sink-redis-streams")]
 pub mod redis_streams;
 
+#[cfg(feature = "sink-rabbitmq")]
+pub mod rabbitmq;
+
 #[cfg(feature = "sink-webhook")]
 pub mod webhook;
 
@@ -27,6 +30,9 @@ use redis_strings::RedisStringsSink;
 #[cfg(feature = "sink-redis-streams")]
 use redis_streams::RedisStreamsSink;
 
+#[cfg(feature = "sink-rabbitmq")]
+use rabbitmq::RabbitmqSink;
+
 #[cfg(feature = "sink-webhook")]
 use webhook::WebhookSink;
 
@@ -53,6 +59,10 @@ pub enum AnySink {
     #[cfg(feature = "sink-nats")]
     Nats(NatsSink),
 
+    /// RabbitMQ sink for AMQP messaging.
+    #[cfg(feature = "sink-rabbitmq")]
+    Rabbitmq(RabbitmqSink),
+
     /// Webhook sink for HTTP POST delivery.
     #[cfg(feature = "sink-webhook")]
     Webhook(WebhookSink),
@@ -76,6 +86,9 @@ impl Sink for AnySink {
             #[cfg(feature = "sink-nats")]
             AnySink::Nats(sink) => sink.publish_events(events).await,
 
+            #[cfg(feature = "sink-rabbitmq")]
+            AnySink::Rabbitmq(sink) => sink.publish_events(events).await,
+
             #[cfg(feature = "sink-webhook")]
             AnySink::Webhook(sink) => sink.publish_events(events).await,
         }