amq-rpc 0.1.0

RabbitMQ RPC library
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

use lapin::{
    options::*, types::FieldTable, BasicProperties,
    Connection, ConnectionProperties, Consumer, Channel, Queue
};
use std::sync::Arc;
use tokio::sync::RwLock;
use crate::error::{RpcError, RpcResult};

#[derive(Clone)]
pub struct AmqpConnection {
    connection: Arc<RwLock<Option<Connection>>>,
    channel: Arc<RwLock<Option<Channel>>>,
    url: String,
}

impl AmqpConnection {
    pub fn new(url: impl Into<String>) -> Self {
        Self {
            connection: Arc::new(RwLock::new(None)),
            channel: Arc::new(RwLock::new(None)),
            url: url.into(),
        }
    }

    pub async fn connect(&self) -> RpcResult<()> {
        let conn = Connection::connect(&self.url, ConnectionProperties::default()).await?;
        let channel = conn.create_channel().await?;

        *self.connection.write().await = Some(conn);
        *self.channel.write().await = Some(channel);

        Ok(())
    }

    pub async fn get_channel(&self) -> RpcResult<Channel> {
        let channel_guard = self.channel.read().await;
        match channel_guard.as_ref() {
            Some(channel) => Ok(channel.clone()),
            None => Err(RpcError::InvalidCommand { message: "No connection available".to_string() })
        }
    }

    pub async fn declare_queue(&self, queue_name: &str, exclusive: bool) -> RpcResult<Queue> {
        let channel = self.get_channel().await?;
        let queue = channel
            .queue_declare(
                queue_name,
                QueueDeclareOptions {
                    exclusive,
                    ..Default::default()
                },
                FieldTable::default(),
            )
            .await?;
        Ok(queue)
    }

    pub async fn publish(
        &self,
        queue_name: &str,
        payload: &[u8],
        correlation_id: &str,
        reply_to: Option<&str>,
        expiration: Option<u64>,
    ) -> RpcResult<()> {
        let channel = self.get_channel().await?;

        let mut properties = BasicProperties::default()
            .with_correlation_id(correlation_id.into());

        if let Some(reply_queue) = reply_to {
            properties = properties.with_reply_to(reply_queue.into());
        }

        if let Some(exp) = expiration {
            properties = properties.with_expiration(exp.to_string().into());
        }

        channel
            .basic_publish(
                "",
                queue_name,
                BasicPublishOptions {
                    mandatory: true,
                    ..Default::default()
                },
                payload,
                properties,
            )
            .await?
            .await?;

        Ok(())
    }

    pub async fn consume(&self, queue_name: &str, consumer_tag: &str) -> RpcResult<Consumer> {
        let channel = self.get_channel().await?;
        let consumer = channel
            .basic_consume(
                queue_name,
                consumer_tag,
                BasicConsumeOptions::default(),
                FieldTable::default(),
            )
            .await?;
        Ok(consumer)
    }

    pub async fn ack(&self, delivery_tag: u64) -> RpcResult<()> {
        let channel = self.get_channel().await?;
        channel.basic_ack(delivery_tag, BasicAckOptions::default()).await?;
        Ok(())
    }

    pub async fn close(&self) -> RpcResult<()> {
        if let Some(connection) = self.connection.write().await.take() {
            connection.close(200, "Normal shutdown").await?;
        }
        Ok(())
    }
}