serwus 0.2.4

Helpers for building actix-web/diesel based services
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
143
144
145
146
147
148
149

use actix::{Actor, Addr, Handler, Message, dev::ToEnvelope};
use actix_web::rt::task::spawn_blocking;
use amiquip::{
    Channel, ConfirmSmoother, ConsumerMessage, ConsumerOptions, Exchange, Publish,
    QueueDeclareOptions,
};
use log::{error, info};
use serde::{Serialize, de::DeserializeOwned};

pub fn spawn_rabbit_consumer<T, A>(
    act: Addr<A>,
    channel: Channel,
    consume_queue: &'static str,
    publish_queue: Option<&'static str>,
) where
    T: DeserializeOwned + Message + Send,
    A: Handler<T>,
    <T as Message>::Result: Send,
    <A as Actor>::Context: ToEnvelope<A, T>,
{
    info!("Rabbit consumer starting...");

    spawn_blocking(move || {
        if let Some(publish_queue) = publish_queue {
            match channel.queue_declare(publish_queue, QueueDeclareOptions::default()) {
                Ok(_) => (),
                Err(err) => panic!("Error while declaring publisher queue: {err}"),
            };
        }

        let queue = match channel.queue_declare(consume_queue, QueueDeclareOptions::default()) {
            Ok(q) => q,
            Err(err) => panic!("Error while declaring consumer queue: {err}"),
        };

        // Start a consumer.
        let consumer = match queue.consume(ConsumerOptions::default()) {
            Ok(c) => c,
            Err(err) => panic!("Error while creating consumer: {err}"),
        };

        info!("Rabbit consumer started. Waiting for messages...");

        for message in consumer.receiver().iter() {
            match message {
                ConsumerMessage::Delivery(delivery) => {
                    let body = &delivery.body[..];
                    let msg: T = match serde_json::from_slice(body) {
                        Ok(msg) => msg,
                        Err(err) => {
                            error!("Rabbit deserialize error: {}", err);
                            continue;
                        }
                    };

                    act.do_send(msg);

                    match consumer.ack(delivery) {
                        Ok(d) => d,
                        Err(err) => error!("Error while delivery ack: {}", err),
                    };
                }
                other => {
                    info!("Consumer ended: {:?}", other);
                    break;
                }
            }
        }
    });
}

pub enum SendError {
    Channel(amiquip::Error),
    Serde(serde_json::Error),
    Publish(amiquip::Error),
    Confirm(crossbeam_channel::RecvError),
}

pub fn send_and_wait_for_ack(
    msg: impl Serialize,
    channel: &Channel,
    routing_key: &'static str,
) -> Result<(), SendError> {
    let exchange = Exchange::direct(channel);

    // register a pub confirm listener before putting the channel into confirm mode
    let confirm_listener = match channel.listen_for_publisher_confirms() {
        Ok(c_l) => c_l,
        Err(err) => {
            error!(
                "Error while registering confirm listener in rabbitmq: {}",
                err
            );
            return Err(SendError::Channel(err));
        }
    };

    // put channel in confirm mode
    channel.enable_publisher_confirms().unwrap(); // TODO:

    // create a confirm smoother so we can process perfectly sequential confirmations
    let mut confirm_smoother = ConfirmSmoother::new();

    // Serialize struct
    let data = match serde_json::to_string(&msg) {
        Ok(data) => data,
        Err(err) => return Err(SendError::Serde(err)),
    };

    // Publish message to the queue.
    match exchange.publish(Publish::new(data.as_bytes(), routing_key)) {
        Ok(_) => {
            info!("Exchange {}: Message published.", routing_key);
        }
        Err(err) => {
            error!(
                "Exchange {}: Error while publishing message to rabbitmq: {}",
                routing_key, err
            );
            return Err(SendError::Publish(err));
        }
    };

    info!("Exchange {}: Waiting for confirmation...", routing_key);
    // wait for confirmation from the server for those 1 messages
    let mut confirmed = 0;
    while confirmed == 0 {
        let confirm = match confirm_listener.recv() {
            Ok(confirm) => {
                info!("Confirmed!");
                confirm
            }
            Err(err) => {
                error!(
                    "Exchange {}: Error while confirming recv: {:?}",
                    routing_key, err
                );
                return Err(SendError::Confirm(err));
            }
        };
        println!("got raw confirm {confirm:?} from server");
        for confirm in confirm_smoother.process(confirm) {
            info!("Exchange {}: Message confirmed: {:?}", routing_key, confirm);
            confirmed += 1;
        }
    }

    Ok(())
}