sz_orm_queue/
rabbitmq.rs

1use crate::error::MqError;
2use crate::queue::{InMemoryQueue, Message, MessageQueue};
3use async_trait::async_trait;
4
5pub struct InMemoryRabbitmqQueue {
6    inner: InMemoryQueue,
7}
8
9impl InMemoryRabbitmqQueue {
10    pub fn new() -> Self {
11        Self {
12            inner: InMemoryQueue::new(),
13        }
14    }
15
16    pub async fn message_count(&self, topic: &str) -> usize {
17        self.inner.message_count(topic).await
18    }
19
20    pub async fn subscriber_count(&self, topic: &str) -> usize {
21        self.inner.subscriber_count(topic).await
22    }
23
24    pub async fn in_flight_count(&self) -> usize {
25        self.inner.in_flight_count().await
26    }
27}
28
29impl Default for InMemoryRabbitmqQueue {
30    fn default() -> Self {
31        Self::new()
32    }
33}
34
35#[async_trait]
36impl MessageQueue for InMemoryRabbitmqQueue {
37    async fn publish(&self, topic: &str, message: &[u8]) -> Result<(), MqError> {
38        self.inner.publish(topic, message).await
39    }
40
41    async fn consume(&self, topic: &str) -> Result<Option<Message>, MqError> {
42        self.inner.consume(topic).await
43    }
44
45    async fn ack(&self, message_id: &str) -> Result<(), MqError> {
46        self.inner.ack(message_id).await
47    }
48
49    async fn subscribe(&self, topic: &str) -> Result<(), MqError> {
50        self.inner.subscribe(topic).await
51    }
52}
53
54#[cfg(test)]
55mod tests {
56    use super::*;
57
58    #[tokio::test]
59    async fn test_rabbitmq_publish_and_consume() {
60        let queue = InMemoryRabbitmqQueue::new();
61        queue.publish("amq-topic", b"hello rabbit").await.unwrap();
62        assert_eq!(queue.message_count("amq-topic").await, 1);
63
64        let msg = queue
65            .consume("amq-topic")
66            .await
67            .unwrap()
68            .expect("message should exist");
69        assert_eq!(msg.payload, b"hello rabbit");
70        assert!(!msg.id.is_empty());
71
72        queue.ack(&msg.id).await.unwrap();
73        assert_eq!(queue.in_flight_count().await, 0);
74    }
75
76    #[tokio::test]
77    async fn test_rabbitmq_consume_empty() {
78        let queue = InMemoryRabbitmqQueue::new();
79        let result = queue.consume("empty").await.unwrap();
80        assert!(result.is_none());
81    }
82
83    #[tokio::test]
84    async fn test_rabbitmq_ack_unknown() {
85        let queue = InMemoryRabbitmqQueue::new();
86        assert!(queue.ack("no-such-id").await.is_err());
87    }
88
89    #[tokio::test]
90    async fn test_rabbitmq_subscribe_increments() {
91        let queue = InMemoryRabbitmqQueue::new();
92        queue.subscribe("amq-topic").await.unwrap();
93        assert_eq!(queue.subscriber_count("amq-topic").await, 1);
94    }
95}
sz_orm_queue/rabbitmq.rs

sz_orm_queue/
rabbitmq.rs
