crabbyq 1.0.0

A declarative async Rust framework for message-driven microservices.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use std::pin::Pin;

use async_trait::async_trait;
use futures_util::{Stream, StreamExt};

use crate::brokers::base::{Broker, BrokerMessage, HeaderMap};

#[derive(Clone)]
pub struct NatsBroker {
    client: async_nats::Client,
}

impl NatsBroker {
    pub fn new(client: async_nats::Client) -> Self {
        Self { client: client }
    }

    pub fn client(&self) -> async_nats::Client {
        self.client.clone()
    }
}

#[async_trait]
impl Broker for NatsBroker {
    type MessageStream = Pin<Box<dyn Stream<Item = BrokerMessage> + Send + Unpin>>;

    async fn subscribe(
        &self,
        subjects: &[String],
    ) -> Result<Self::MessageStream, crate::brokers::base::BrokerError> {
        let mut streams = Vec::new();

        for subject in subjects {
            let subscriber = self
                .client
                .subscribe(subject.clone())
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;

            let stream = subscriber.map(|msg| BrokerMessage {
                subject: msg.subject.to_string(),
                payload: msg.payload.to_vec(),
                headers: msg.headers.map(|headers| {
                    headers
                        .iter()
                        .filter_map(|(key, values)| {
                            values
                                .first()
                                .map(|value| (key.to_string(), value.to_string()))
                        })
                        .collect()
                }),
                reply_to: msg.reply.map(|reply| reply.to_string()),
                acknowledger: None,
            });

            streams.push(stream);
        }

        let merged = futures_util::stream::select_all(streams);
        Ok(Box::pin(merged))
    }

    async fn publish(
        &self,
        subject: &str,
        payload: &[u8],
        headers: Option<&HeaderMap>,
    ) -> Result<(), crate::brokers::base::BrokerError> {
        if let Some(headers) = headers {
            let mut nats_headers = async_nats::HeaderMap::new();
            for (key, value) in headers {
                nats_headers.insert(key.as_str(), value.as_str());
            }

            self.client
                .publish_with_headers(subject.to_string(), nats_headers, payload.to_vec().into())
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;
        } else {
            self.client
                .publish(subject.to_string(), payload.to_vec().into())
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?;
        }
        Ok(())
    }

    async fn request(
        &self,
        subject: &str,
        payload: &[u8],
        headers: Option<&HeaderMap>,
    ) -> Result<BrokerMessage, crate::brokers::base::BrokerError> {
        let message = if let Some(headers) = headers {
            let mut nats_headers = async_nats::HeaderMap::new();
            for (key, value) in headers {
                nats_headers.insert(key.as_str(), value.as_str());
            }

            self.client
                .request_with_headers(subject.to_string(), nats_headers, payload.to_vec().into())
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?
        } else {
            self.client
                .request(subject.to_string(), payload.to_vec().into())
                .await
                .map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)?
        };

        Ok(BrokerMessage {
            subject: message.subject.to_string(),
            payload: message.payload.to_vec(),
            headers: message.headers.map(|headers| {
                headers
                    .iter()
                    .filter_map(|(key, values)| {
                        values
                            .first()
                            .map(|value| (key.to_string(), value.to_string()))
                    })
                    .collect()
            }),
            reply_to: message.reply.map(|reply| reply.to_string()),
            acknowledger: None,
        })
    }
}