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
//! AMQP requests.
use std::sync::Arc;
use lapin::options::{BasicAckOptions, BasicRejectOptions};
use lapin::protocol::basic::AMQPProperties;
use lapin::{message::Delivery, Channel};
use tracing::{debug, error, warn};
use crate::extract::ReqId;
/// An AMQP request.
#[derive(Debug)]
pub struct Request<S> {
/// The app state. This is added to the app at construction in [`crate::App::new`] and given to each request.
state: Arc<S>,
/// Request ID. This is a unique ID for every request. Either a newly created UUID or whatever
/// is found in the `req_id` header of the incoming AMQP message.
req_id: ReqId,
/// Has this message been (n)ack'ed?
// This has to be pub within kanin so that the acker extractor can set it.
pub(crate) acked: bool,
/// The channel the message was received on.
channel: Channel,
/// The message delivery.
delivery: Delivery,
}
impl<S> Request<S> {
/// Constructs a new request from a [`Channel`] and [`Delivery`].
pub fn new(channel: Channel, delivery: Delivery, state: Arc<S>) -> Self {
Self {
state,
channel,
acked: false,
req_id: ReqId::from_delivery(&delivery),
delivery,
}
}
/// Returns a reference to the request ID of this request.
pub fn req_id(&self) -> &ReqId {
&self.req_id
}
/// Returns a reference to the delivery of this request.
pub fn delivery(&self) -> &Delivery {
&self.delivery
}
/// Returns a mutable reference to the delivery of this request.
///
/// For now, this is a private interface. It could potentially be made public in the future.
pub(crate) fn delivery_mut(&mut self) -> &mut Delivery {
&mut self.delivery
}
/// Returns the app state for the given type.
pub fn state<T>(&self) -> T
where
T: for<'a> From<&'a S>,
{
self.state.as_ref().into()
}
/// Returns a reference to the [`Channel`] the message was delivered on.
pub fn channel(&self) -> &Channel {
&self.channel
}
/// Returns the AMQP properties of the request, unless the request was already extracted.
pub fn properties(&self) -> &AMQPProperties {
&self.delivery.properties
}
/// Returns the `app_id` AMQP property of the request.
pub fn app_id(&self) -> Option<&str> {
self.properties()
.app_id()
.as_ref()
.map(|app_id| app_id.as_str())
}
/// Acks the request, letting the AMQP broker know that it was received and processed successfully.
pub(crate) async fn ack(&mut self, options: BasicAckOptions) -> Result<(), lapin::Error> {
self.delivery.ack(options).await?;
self.acked = true;
Ok(())
}
}
/// We implement [`Drop`] on [`Request`] to ensure that requests that were not explicitly acknowledged will be rejected.
impl<S> Drop for Request<S> {
fn drop(&mut self) {
// If we already acked, do nothing.
if self.acked {
return;
}
// We haven't acked and the request is being dropped.
// This almost certainly indicates a panic during request handling.
// We will reject the request to tell the AMQP broker to requeue this message ASAP.
warn!("Rejecting unacked request {} due to drop.", self.req_id);
let req_id = self.req_id.clone();
// Yoink the acker from the delivery so we can give it to a future to reject the message.
// This is a bit of a hack. Hopefully lapin improves the interface in the future, see also https://github.com/amqp-rs/lapin/issues/402.
let acker = std::mem::take(&mut self.delivery.acker);
// Rejecting is async so we have to spawn a task to do it.
// Unfortunately we can't really be sure that this ever completes.
tokio::spawn(async move {
match acker.reject(BasicRejectOptions { requeue: true }).await {
Ok(()) => debug!("Successfully rejected request {} during drop.", req_id),
Err(e) => error!("Failed to reject request {} during drop: {e}", req_id),
}
});
// Strictly speaking not necessary but nice to indicate that we have at least tried (even if we only try in the future).
self.acked = true;
}
}