amqp-client-rust 0.0.6

An asynchronous AMQP client library for Rust, designed for high-performance communication with RabbitMQ. Features include automatic queue and exchange management, message publishing, subscribing, and RPC support.
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
130
131
132
133
134
135
136
137
138
139
140
141
142

use amqprs::{
    callbacks::{ChannelCallback, ConnectionCallback},
    channel::Channel,
    connection::Connection,
    error::Error as AMQPError,
    Ack, BasicProperties, Cancel, Close, CloseChannel, Nack, Return
};
use async_trait::async_trait;
use tokio::sync::mpsc::UnboundedSender;
use tracing::{info, debug, error, warn};
use crate::api::{connection_manager::ConnectionCommand, utils::ChannelCmd};

pub type AMQPResult<T> = std::result::Result<T, AMQPError>;
pub struct MyChannelCallback{
    pub channel_tx: UnboundedSender<ChannelCmd>,
}

#[async_trait]
impl ChannelCallback for MyChannelCallback {
    async fn close(&mut self, channel: &Channel, close: CloseChannel) -> AMQPResult<()> {
        match close.reply_code() {
            200 => {
                info!("Channel closed normally: {}", close);
            },
            403_u16..=406_u16 => {
                error!("Channel closed due to channel exception (Safe to Reopen): {}", close);
                if let Err(e) = self.channel_tx.send(ChannelCmd::ReOpen(channel.channel_id())) {
                    error!("Failed to send ReOpen command to connection manager: {}", e);
                }
            },
            320_u16|500_u16..=599_u16 => {
                error!("Channel closed due to fatal connection exception: {}", close);
            },
            _ => {
                error!(
                    "Unhandled close request for channel {}, cause: {}",
                    channel, close
                );
            }
        }
        Ok(())
    }
    async fn cancel(&mut self, _channel: &Channel, _cancel: Cancel) -> AMQPResult<()> {
        warn!(
            "handle cancel request for consumer {} on channel {}",
            _cancel.consumer_tag(),
            _channel
        );
        Ok(())
    }
    async fn flow(&mut self, _channel: &Channel, _active: bool) -> AMQPResult<bool> {
        info!(
            "handle flow request active={} for channel {}",
            _active, _channel
        );
        Ok(true)
    }
    async fn publish_ack(&mut self, _channel: &Channel, ack: Ack) {
        info!(
            "handle publish ack delivery_tag={} on channel {}",
            ack.delivery_tag(),
            _channel
        );
        let tag = ack.delivery_tag();
        let multiple = ack.mutiple();
        debug!("Received ACK: tag={}, multiple={}", tag, multiple);

        if let Err(e) = self.channel_tx.send(ChannelCmd::PublishAck((tag, multiple))) {
            error!("Failed to send ACK to connection manager: {}", e);
        }
        
    }
    async fn publish_nack(&mut self, _channel: &Channel, nack: Nack) {
        warn!(
            "handle publish nack delivery_tag={} on channel {}",
            nack.delivery_tag(),
            _channel
        );
        let tag = nack.delivery_tag();
        let multiple = nack.multiple();

        if let Err(e) = self.channel_tx.send(ChannelCmd::PublishNack((tag, multiple))) {
            error!("Failed to send NACK to connection manager: {}", e);
        }
        
    }
    async fn publish_return(
        &mut self,
        _channel: &Channel,
        _ret: Return,
        _basic_properties: BasicProperties,
        _content: Vec<u8>,
    ) {
        warn!(
            "handle publish return {} on channel {}, content size: {}",
            _ret,
            _channel,
            _content.len()
        );
    }
}


pub struct MyConnectionCallback{
    pub sender: UnboundedSender<ConnectionCommand>,
}


#[async_trait]
impl ConnectionCallback for MyConnectionCallback {

    async fn close(&mut self, _connection: &Connection, _close: Close) -> AMQPResult<()> {
        error!(
            "handle close request for connection {}, cause: {}",
            _connection, _close
        );
        let _ = self.sender.send(ConnectionCommand::CheckConnection{});
        Ok(())
    }

    async fn blocked(&mut self, _connection: &Connection, _reason: String) {
        debug!(
            "handle blocked notification for connection {}, reason: {}",
            _connection, _reason
        );
    }

    async fn unblocked(&mut self, _connection: &Connection) {
        debug!(
            "handle unblocked notification for connection {}",
            _connection
        );
    }
    
    async fn secret_updated(&mut self, connection: &Connection){
        debug!(
            "handle secret updated notification for connection {}",
            connection
        );
        let _ = self.sender.send(ConnectionCommand::CheckConnection{});
    }
}