rabbit_borough 0.1.7

Create a RabbitMQ consumer project with minimal effort, by bypassing templating, configuration and complicated resiliency logic
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 crate::configuration::config_model::BindingProperties;
use crate::configuration::config_model::ConnectionProperties;
use crate::configuration::config_model::DeclareProperties;
use crate::consumer::connection_manager::build_url;
use crate::consumer::connection_manager::create_channel;
use lapin::options::ExchangeDeclareOptions;
use lapin::options::*;
use lapin::types::FieldTable;
use lapin::Channel;
use lapin::Error;
use lapin::ExchangeKind;
use lapin::Queue;

#[derive(Clone)]
pub struct SetupModel {
    pub channel: Channel,
    pub queue: Option<Result<Queue, Error>>,
    pub exchange: Option<Result<(), Error>>,
    pub binding: Option<Result<(), Error>>,
}

///  # Executes configuration defined in the configuration
///  * Creates a channel
///  * Declares the Queue
///  * Declares the Exchange
///  * Declares the Binding between the Exchange and the Queue
///
pub async fn setup_consumer(
    connection: ConnectionProperties,
    bind: BindingProperties,
    declare: DeclareProperties,
) -> SetupModel {
    let channel = create_channel(build_url(connection.clone()).as_str(), connection.retry)
        .await
        .expect("channel to be created");
    let mut queue = None;
    if declare.queue {
        queue = Some(
            channel
                .queue_declare(
                    &bind.queue,
                    QueueDeclareOptions::default(),
                    FieldTable::default(),
                )
                .await,
        );
    }
    let mut exchange = None;
    if declare.exchange {
        exchange = Some(
            create_exchange(
                &bind.exchange,
                channel.clone(),
                ExchangeDeclareOptions {
                    passive: bind.exchange_declaration_options.passive,
                    durable: bind.exchange_declaration_options.durable,
                    auto_delete: bind.exchange_declaration_options.auto_delete,
                    internal: false,
                    nowait: false,
                },
            )
            .await,
        );
    }

    let mut binding = None;
    if declare.binding {
        binding = Some(
            create_exchange_queue_binding(
                channel.clone(),
                &bind.queue,
                &bind.exchange,
                &bind.routing_key,
            )
            .await,
        );
    }

    SetupModel {
        channel,
        queue,
        exchange,
        binding,
    }
}

/// # Creates an exchange
/// * Returns an exchange
async fn create_exchange(
    exchange: &str,
    channel: Channel,
    options: ExchangeDeclareOptions,
) -> Result<(), Error> {
    channel
        .exchange_declare(
            exchange,
            ExchangeKind::Fanout,
            options,
            FieldTable::default(),
        )
        .await
}

/// # Binds the exchange with a queue
/// * Returns the binding
async fn create_exchange_queue_binding(
    channel: Channel,
    queue: &str,
    exchange: &str,
    routing_key: &str,
) -> Result<(), Error> {
    channel
        .queue_bind(
            queue,
            exchange,
            routing_key,
            QueueBindOptions { nowait: false },
            FieldTable::default(),
        )
        .await
}