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
143
144
145
146

use crate::error::DispatchError;
use futures::stream::StreamExt;
use lapin::{
    options::{BasicAckOptions, BasicConsumeOptions},
    types::FieldTable,
};
use log::{debug, error, info, trace};
use minerva_broker as broker;
use minerva_broker::LapinPool;
use minerva_data::db::DBPool;
use mongodb::Client as MongoClient;
use redis::Client as RedisClient;
use tokio::time::Duration;

mod session_management;

/// Number of seconds to wait for each task to reprocess
const QUEUE_PROCESSING_WAIT_SECS: u64 = 3;

/// Primary entry point for the consuming of messages from the message broker.
///
/// # Note
/// This procedure contain A LOT of `unwrap`'s. This is by design. This procedure
/// is supposed to run asynchronously for each queue that is supposed to be listened
/// to by the DISPATCH service, and by tenant as well. If any step on this fails,
/// the listener should also fail, and maybe be restarted.
pub async fn queue_consume(
    tenant: String,
    rabbitmq: LapinPool,
    _postgresql: DBPool,
    mongodb: MongoClient,
    _redis: RedisClient,
) -> Result<(), Box<dyn std::error::Error>> {
    trace!("Start queue consumer for {}", tenant);
    let mut handlers = vec![];
    for queue in broker::QUEUES {
        let tenant = tenant.clone();
        let rabbitmq = rabbitmq.clone();
        let mongodb = mongodb.clone();

        handlers.push(tokio::spawn(async move {
            let consumer_name = format!("{}.{}.consumer", tenant, queue);
            debug!("Starting consumer {}", consumer_name);
            loop {
                trace!("Begin broker connection loop");
                let conn = rabbitmq
                    .get()
                    .await
                    .map_err(|_| {
                        let error = DispatchError::Connection {
                            consumer_name: consumer_name.clone(),
                            service_name: "RabbitMQ".to_string(),
                        };
                        error!("{}", error);
                        error
                    })
                    .unwrap();

                debug!("Create channel");
                let channel = conn
                    .create_channel()
                    .await
                    .map_err(|_| {
                        let error = DispatchError::Connection {
                            consumer_name: consumer_name.clone(),
                            service_name: "RabbitMQ channel".to_string(),
                        };
                        error!("{}", error);
                        error
                    })
                    .unwrap();

                debug!("Create basic consumer");
                let mut consumer = channel
                    .basic_consume(
                        queue,
                        &consumer_name,
                        BasicConsumeOptions::default(),
                        FieldTable::default(),
                    )
                    .await
                    .map_err(|_| {
                        let error = DispatchError::Connection {
                            consumer_name: consumer_name.clone(),
                            service_name: "queue using a new consumer".to_string(),
                        };
                        error!("{}", error);
                        error
                    })
                    .unwrap();

                debug!("Await next delivery");
                while let Some(delivery) = consumer.next().await {
                    let delivery = delivery
                        .map_err(|_| {
                            let error = DispatchError::Delivery {
                                consumer_name: consumer_name.clone(),
                            };
                            error!("{}", error);
                            error
                        })
                        .unwrap();

                    debug!("Process delivered message");
                    if (queue == &"session_management")
                        && session_management::dispatch(
                            &tenant,
                            &consumer_name,
                            &mongodb,
                            &delivery.data,
                        )
                        .await
                        .unwrap()
                    {
                        trace!("Ack session_management delivery");
                        // If message is known, then we send back an
                        // ack signal. If not, well... leave it to
                        // another consumer
                        delivery
                            .ack(BasicAckOptions::default())
                            .await
                            .map_err(|_| {
                                let error = DispatchError::Ack {
                                    consumer_name: consumer_name.clone(),
                                };
                                error!("{}", error);
                                error
                            })
                            .unwrap();
                    }
                }

                // End of queue processing. Let's wait for a while.
                info!("{}: Processed.", consumer_name);
                tokio::time::sleep(Duration::from_secs(QUEUE_PROCESSING_WAIT_SECS)).await;
            }
        }))
    }

    for handle in handlers {
        handle.await?;
    }

    info!("All handlers terminated.");
    Ok(())
}