amqprs 2.1.5

AMQP 0-9-1 client implementation for RabbitMQ
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

use crate::{
    api::error::Error,
    frame::{Frame, Select, SelectOk},
};

use super::{Channel, Result};

/// Arguments for [`confirm_select`]
///
/// See [AMQP_0-9-1 Reference](https://github.com/rabbitmq/amqp-0.9.1-spec/blob/main/docs/amqp-0-9-1-reference.md#confirm.select).
///
/// [`confirm_select`]: struct.Channel.html#method.confirm_select
#[derive(Debug, Clone, Default)]
pub struct ConfirmSelectArguments {
    /// Default: `false`
    pub no_wait: bool,
}

impl ConfirmSelectArguments {
    /// Create new arguments with defaults.
    pub fn new(no_wait: bool) -> Self {
        Self { no_wait }
    }
}

/// APIs for AMQP confirm class.
impl Channel {
    /// See [AMQP_0-9-1 Reference](https://github.com/rabbitmq/amqp-0.9.1-spec/blob/main/docs/amqp-0-9-1-reference.md#confirm.select).
    ///
    /// # Errors
    ///
    /// Returns error if any failure in comunication with server.
    pub async fn confirm_select(&self, args: ConfirmSelectArguments) -> Result<()> {
        let select = Select::new(args.no_wait);
        if args.no_wait {
            self.shared
                .outgoing_tx
                .send((self.shared.channel_id, select.into_frame()))
                .await?;
            Ok(())
        } else {
            let responder_rx = self.register_responder(SelectOk::header()).await?;

            let _method = synchronous_request!(
                self.shared.outgoing_tx,
                (self.shared.channel_id, select.into_frame()),
                responder_rx,
                Frame::SelectOk,
                Error::ChannelUseError
            )?;
            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use tokio::time;

    use crate::{
        callbacks::{DefaultChannelCallback, DefaultConnectionCallback},
        channel::BasicPublishArguments,
        connection::{Connection, OpenConnectionArguments},
        test_utils::setup_logging,
        BasicProperties,
    };

    use super::ConfirmSelectArguments;

    #[tokio::test]
    async fn test_publish_confirm_mode() {
        setup_logging();

        let args = OpenConnectionArguments::new("localhost", 5672, "user", "bitnami");

        let connection = Connection::open(&args).await.unwrap();
        connection
            .register_callback(DefaultConnectionCallback)
            .await
            .unwrap();

        let channel = connection.open_channel(None).await.unwrap();
        channel
            .register_callback(DefaultChannelCallback)
            .await
            .unwrap();

        // set to publish confirm mode
        channel
            .confirm_select(ConfirmSelectArguments::default())
            .await
            .unwrap();

        let args = BasicPublishArguments::new("amq.topic", "amqprs.test.transaction");

        let basic_properties = BasicProperties::default().with_persistence(true).finish();

        let content = String::from("AMQPRS test publish confirm").into_bytes();

        channel
            .basic_publish(basic_properties, content, args)
            .await
            .unwrap();

        time::sleep(Duration::from_millis(100)).await;
        channel.close().await.unwrap();
        connection.close().await.unwrap();
    }
}