feat(sink): add AWS SNS sink for topic publishing

Author: psteinroe

Cargo.lock

@@ -10,6 +10,7 @@ sink-nats          = ["dep:async-nats"]
 sink-rabbitmq      = ["dep:lapin"]
 sink-redis-streams = ["dep:redis"]
 sink-redis-strings = ["dep:redis"]
+sink-sns           = ["dep:aws-sdk-sns", "dep:aws-config"]
 sink-sqs           = ["dep:aws-sdk-sqs", "dep:aws-config"]
 sink-webhook       = ["dep:reqwest"]
 test-utils         = ["dep:ctor", "dep:testcontainers", "dep:testcontainers-modules"]
@@ -56,6 +57,7 @@ uuid         = { version = "1.19.0", default-features = false, features = ["v4"]
 # Optional sink dependencies.
 async-nats = { version = "0.38", optional = true }
 aws-config = { version = "1.5", optional = true }
+aws-sdk-sns = { version = "1.60", optional = true }
 aws-sdk-sqs = { version = "1.60", optional = true }
 lapin = { version = "2.5", optional = true }
 rdkafka = { version = "0.36", optional = true, default-features = false, features = ["tokio", "cmake-build"] }
@@ -83,6 +85,7 @@ testcontainers-modules = { version = "0.11", optional = true, features = [
   "rabbitmq",
   "kafka",
   "elasticmq",
+  "localstack",
   "blocking",
 ] }
 

Cargo.toml

@@ -21,6 +21,9 @@ use crate::sink::webhook::WebhookSinkConfig;
 #[cfg(feature = "sink-sqs")]
 use crate::sink::sqs::SqsSinkConfig;
 
+#[cfg(feature = "sink-sns")]
+use crate::sink::sns::SnsSinkConfig;
+
 /// Sink destination configuration.
 ///
 /// Determines where replicated events are sent.
@@ -64,4 +67,9 @@ pub enum SinkConfig {
     #[cfg(feature = "sink-sqs")]
     #[serde(rename = "sqs")]
     Sqs(SqsSinkConfig),
+
+    /// AWS SNS sink for topic publishing.
+    #[cfg(feature = "sink-sns")]
+    #[serde(rename = "sns")]
+    Sns(SnsSinkConfig),
 }

src/config/sink.rs

@@ -168,6 +168,19 @@ async fn run_pipeline(config: &PipelineConfig) -> EtlResult<()> {
             })?;
             AnySink::Sqs(s)
         }
+
+        #[cfg(feature = "sink-sns")]
+        SinkConfig::Sns(cfg) => {
+            use crate::sink::sns::SnsSink;
+            let s = SnsSink::new(cfg.clone()).await.map_err(|e| {
+                etl::etl_error!(
+                    etl::error::ErrorKind::InvalidData,
+                    "Failed to create SNS sink",
+                    e.to_string()
+                )
+            })?;
+            AnySink::Sns(s)
+        }
     };
 
     // Create PgStream as an ETL destination
@@ -248,6 +261,11 @@ fn log_sink_config(config: &SinkConfig) {
         SinkConfig::Sqs(_cfg) => {
             debug!("using sqs sink");
         }
+
+        #[cfg(feature = "sink-sns")]
+        SinkConfig::Sns(_cfg) => {
+            debug!("using sns sink");
+        }
     }
 }
 

src/core.rs

@@ -22,6 +22,9 @@ pub mod kafka;
 #[cfg(feature = "sink-sqs")]
 pub mod sqs;
 
+#[cfg(feature = "sink-sns")]
+pub mod sns;
+
 pub use base::Sink;
 
 use etl::error::EtlResult;
@@ -48,6 +51,9 @@ use kafka::KafkaSink;
 #[cfg(feature = "sink-sqs")]
 use sqs::SqsSink;
 
+#[cfg(feature = "sink-sns")]
+use sns::SnsSink;
+
 use crate::types::TriggeredEvent;
 
 /// Wrapper enum for all supported sink types.
@@ -86,6 +92,10 @@ pub enum AnySink {
     /// AWS SQS sink for queue messaging.
     #[cfg(feature = "sink-sqs")]
     Sqs(SqsSink),
+
+    /// AWS SNS sink for topic publishing.
+    #[cfg(feature = "sink-sns")]
+    Sns(SnsSink),
 }
 
 impl Sink for AnySink {
@@ -117,6 +127,9 @@ impl Sink for AnySink {
 
             #[cfg(feature = "sink-sqs")]
             AnySink::Sqs(sink) => sink.publish_events(events).await,
+
+            #[cfg(feature = "sink-sns")]
+            AnySink::Sns(sink) => sink.publish_events(events).await,
         }
     }
 }