sz_orm_queue/
nats.rs

1use crate::error::MqError;
2use crate::queue::{InMemoryQueue, Message, MessageQueue};
3use async_trait::async_trait;
4
5pub struct InMemoryNatsQueue {
6    inner: InMemoryQueue,
7}
8
9impl InMemoryNatsQueue {
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 InMemoryNatsQueue {
30    fn default() -> Self {
31        Self::new()
32    }
33}
34
35#[async_trait]
36impl MessageQueue for InMemoryNatsQueue {
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_nats_publish_and_consume() {
60        let queue = InMemoryNatsQueue::new();
61        queue.publish("nats.subject", b"hello nats").await.unwrap();
62        let msg = queue
63            .consume("nats.subject")
64            .await
65            .unwrap()
66            .expect("message should exist");
67        assert_eq!(msg.payload, b"hello nats");
68        queue.ack(&msg.id).await.unwrap();
69    }
70
71    #[tokio::test]
72    async fn test_nats_consume_empty() {
73        let queue = InMemoryNatsQueue::new();
74        assert!(queue.consume("empty").await.unwrap().is_none());
75    }
76
77    #[tokio::test]
78    async fn test_nats_subscribe() {
79        let queue = InMemoryNatsQueue::new();
80        queue.subscribe("nats.subject").await.unwrap();
81        assert_eq!(queue.subscriber_count("nats.subject").await, 1);
82    }
83}
sz_orm_queue/nats.rs

sz_orm_queue/
nats.rs
