Add RabbitMQ Stream source connector

ArroyoSystems · Dec 1, 2024 · f699f00 · f699f00
1 parent a9ac8f7
commit f699f00
Show file tree

Hide file tree

Showing 8 changed files with 638 additions and 0 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/arroyo-connectors/Cargo.toml b/crates/arroyo-connectors/Cargo.toml
@@ -65,6 +65,9 @@ tokio-tungstenite = { version = "0.24", features = ["native-tls"] }
 # Webhook
 reqwest = { workspace = true, features = ["stream"] }
 
+# RabbitMQ Stream
+rabbitmq-stream-client = "0.7"
+
 # Redis
 redis = { version = "0.27", features = ["default", "tokio-rustls-comp", "cluster-async", "connection-manager"] }
 

diff --git a/crates/arroyo-connectors/src/lib.rs b/crates/arroyo-connectors/src/lib.rs
@@ -22,6 +22,7 @@ use fluvio::FluvioConnector;
 use impulse::ImpulseConnector;
 use nats::NatsConnector;
 use nexmark::NexmarkConnector;
+use rabbitmq_stream::RabbitmqStreamConnector;
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
@@ -46,6 +47,7 @@ pub mod nats;
 pub mod nexmark;
 pub mod polling_http;
 pub mod preview;
+pub mod rabbitmq_stream;
 pub mod redis;
 pub mod single_file;
 pub mod sse;
@@ -68,6 +70,7 @@ pub fn connectors() -> HashMap<&'static str, Box<dyn ErasedConnector>> {
         Box::new(NexmarkConnector {}),
         Box::new(PollingHTTPConnector {}),
         Box::new(PreviewConnector {}),
+        Box::new(RabbitmqStreamConnector {}),
         Box::new(RedisConnector {}),
         Box::new(SingleFileConnector {}),
         Box::new(SSEConnector {}),