use url instead of listen_addr and add tests

Author: marcomq

src/endpoints/http.rs

@@ -41,9 +41,9 @@ impl HttpConsumer {
             .with_state(request_tx);
 
         let listen_address = config
-            .listen_address
+            .url
             .as_deref()
-            .ok_or_else(|| anyhow!("'listen_address' is required for http source connection"))?;
+            .ok_or_else(|| anyhow!("'url' is required for http source connection"))?;
         let addr: SocketAddr = listen_address
             .parse()
             .with_context(|| format!("Invalid listen address: {}", listen_address))?;

src/middleware/deduplication.rs

@@ -137,3 +137,54 @@ impl MessageConsumer for DeduplicationConsumer {
         self
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::endpoints::memory::MemoryConsumer;
+    use crate::models::{DeduplicationMiddleware, MemoryConfig};
+    use tempfile::tempdir;
+
+    #[tokio::test]
+    async fn test_deduplication_logic() {
+        let dir = tempdir().unwrap();
+        let db_path = dir.path().join("dedup_test").to_str().unwrap().to_string();
+
+        let config = DeduplicationMiddleware {
+            sled_path: db_path,
+            ttl_seconds: 60,
+        };
+
+        let mem_cfg = MemoryConfig {
+            topic: "dedup_topic".to_string(),
+            capacity: Some(10),
+        };
+        let mem_consumer = MemoryConsumer::new(&mem_cfg).unwrap();
+        let channel = mem_consumer.channel();
+
+        // 1. Send a message
+        let msg1 = CanonicalMessage::new(b"data1".to_vec(), Some(100));
+        channel.send_message(msg1).await.unwrap();
+
+        // 2. Send a duplicate message
+        let msg2 = CanonicalMessage::new(b"data1_dup".to_vec(), Some(100));
+        channel.send_message(msg2).await.unwrap();
+
+        // 3. Send a new message
+        let msg3 = CanonicalMessage::new(b"data2".to_vec(), Some(101));
+        channel.send_message(msg3).await.unwrap();
+
+        let mut dedup_consumer =
+            DeduplicationConsumer::new(Box::new(mem_consumer), &config, "test_route").unwrap();
+
+        // First receive: Should be msg1 (ID 100)
+        let (rec1, commit1) = dedup_consumer.receive().await.unwrap();
+        assert_eq!(rec1.message_id, Some(100));
+        commit1(None).await;
+
+        // Second receive: Should be msg3 (ID 101). msg2 (ID 100) is skipped internally.
+        let (rec2, commit2) = dedup_consumer.receive().await.unwrap();
+        assert_eq!(rec2.message_id, Some(101));
+        commit2(None).await;
+    }
+}

src/middleware/retry.rs

@@ -106,3 +106,96 @@ impl MessagePublisher for RetryPublisher {
         self
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::traits::MessagePublisher;
+    use crate::CanonicalMessage;
+    use anyhow::anyhow;
+    use async_trait::async_trait;
+    use std::any::Any;
+    use std::sync::{Arc, Mutex};
+
+    #[derive(Clone)]
+    struct MockPublisher {
+        attempts: Arc<Mutex<usize>>,
+        succeed_after: usize,
+    }
+
+    #[async_trait]
+    impl MessagePublisher for MockPublisher {
+        async fn send(&self, _msg: CanonicalMessage) -> anyhow::Result<Option<CanonicalMessage>> {
+            let mut attempts = self.attempts.lock().unwrap();
+            *attempts += 1;
+            if *attempts > self.succeed_after {
+                Ok(None)
+            } else {
+                Err(anyhow!("Simulated error"))
+            }
+        }
+
+        async fn send_batch(
+            &self,
+            _messages: Vec<CanonicalMessage>,
+        ) -> anyhow::Result<(Option<Vec<CanonicalMessage>>, Vec<CanonicalMessage>)> {
+            let mut attempts = self.attempts.lock().unwrap();
+            *attempts += 1;
+            if *attempts > self.succeed_after {
+                Ok((None, Vec::new()))
+            } else {
+                Err(anyhow!("Simulated batch error"))
+            }
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    #[tokio::test]
+    async fn test_retry_success() {
+        let attempts = Arc::new(Mutex::new(0));
+        let mock = MockPublisher {
+            attempts: attempts.clone(),
+            succeed_after: 2, // Fails 2 times, succeeds on 3rd
+        };
+
+        let config = RetryMiddleware {
+            max_attempts: 5,
+            initial_interval_ms: 1,
+            max_interval_ms: 10,
+            multiplier: 1.0,
+        };
+
+        let retry_publisher = RetryPublisher::new(Box::new(mock), config);
+        let msg = CanonicalMessage::new(vec![], None);
+
+        let result = retry_publisher.send(msg).await;
+        assert!(result.is_ok());
+        assert_eq!(*attempts.lock().unwrap(), 3);
+    }
+
+    #[tokio::test]
+    async fn test_retry_exhaustion() {
+        let attempts = Arc::new(Mutex::new(0));
+        let mock = MockPublisher {
+            attempts: attempts.clone(),
+            succeed_after: 10,
+        };
+
+        let config = RetryMiddleware {
+            max_attempts: 3,
+            initial_interval_ms: 1,
+            max_interval_ms: 10,
+            multiplier: 1.0,
+        };
+
+        let retry_publisher = RetryPublisher::new(Box::new(mock), config);
+        let msg = CanonicalMessage::new(vec![], None);
+
+        let result = retry_publisher.send(msg).await;
+        assert!(result.is_err());
+        assert_eq!(*attempts.lock().unwrap(), 3);
+    }
+}

src/models.rs

@@ -451,7 +451,6 @@ pub struct HttpEndpoint {
 #[serde(deny_unknown_fields)]
 pub struct HttpConfig {
     pub url: Option<String>,
-    pub listen_address: Option<String>,
     #[serde(default)]
     pub tls: TlsConfig,
     pub response_sink: Option<String>,