taba_client/client/handlers/
mod.rs

1//! Consuming with a *handler* (D140, doc 19 §18.3): `consume_with` (and, with the `serde` *feature*,
2//! `consume_typed`) reads from the queue with long polling, hands each message to a [`Delivery`], deletes the
3//! message if the *handler* returns `Ok` and `delete` is still `true`, and **renews the
4//! invisibility by itself** (a second *thread*, every third of it) while the *handler* runs. With the
5//! `serde` *feature*, `publish_object` serializes the object before publishing.
6
7use std::fmt;
8use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
9use std::sync::{Condvar, Mutex};
10use std::time::Duration;
11
12use super::{Client, ConsumeOptions, Error, Message, VisibilityResponse};
13
14/// One message at a time (the `maxMessages` of consuming with a *handler*).
15pub const DEFAULT_MAX_MESSAGES: u32 = 1;
16/// The long polling of consuming with a *handler*, in seconds.
17pub const DEFAULT_WAIT_TIME: u32 = 20;
18/// The invisibility of consuming with a *handler*, in seconds, which the renewal keeps.
19pub const DEFAULT_VISIBILITY_TIMEOUT: u32 = 120;
20
21/// The error the *handler* returns; its message is **not** deleted.
22pub type HandlerError = Box<dyn std::error::Error>;
23
24/// What consuming with a *handler* returns when it goes wrong: the API (or the client itself) or the *handler*.
25#[derive(Debug)]
26pub enum ConsumeError {
27    Api(Error),
28    Handler(HandlerError),
29}
30
31impl fmt::Display for ConsumeError {
32    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
33        match self {
34            ConsumeError::Api(e) => write!(f, "{e}"),
35            ConsumeError::Handler(e) => write!(f, "handler: {e}"),
36        }
37    }
38}
39
40impl std::error::Error for ConsumeError {}
41
42impl From<Error> for ConsumeError {
43    fn from(e: Error) -> ConsumeError {
44        ConsumeError::Api(e)
45    }
46}
47
48/// What the *handler* receives: the whole message, the body (already deserialized in `consume_typed`; `()`
49/// in `consume_with`), `delete` (default `true`) and [`Delivery::change_visibility`].
50pub struct Delivery<'a, T = ()> {
51    pub message: &'a Message,
52    pub body: T,
53    /// `true` (the default): the message is deleted (`ack`) if the *handler* returns `Ok`. With `false`, it
54    /// stays in the queue and becomes visible again at the end of the invisibility.
55    pub delete: bool,
56    client: &'a Client,
57    topic: &'a str,
58    lease: &'a AtomicU32,
59    lost: &'a AtomicBool,
60}
61
62impl<T> Delivery<'_, T> {
63    /// Changes the invisibility to `seconds` from now on; the value becomes the renewal's.
64    pub fn change_visibility(&self, seconds: u32) -> Result<VisibilityResponse, Error> {
65        let reply = self.client.change_visibility(
66            self.topic,
67            &self.message.id,
68            &self.message.receipt_token,
69            seconds,
70        )?;
71        self.lease.store(seconds, Ordering::SeqCst);
72        Ok(reply)
73    }
74
75    /// The seconds the automatic renewal requests on each round.
76    pub fn lease_seconds(&self) -> u32 {
77        self.lease.load(Ordering::SeqCst)
78    }
79
80    /// `true` if the server refused a renewal (the message already went back to the queue or was deleted).
81    pub fn lease_lost(&self) -> bool {
82        self.lost.load(Ordering::SeqCst)
83    }
84}
85
86/// The defaults of consuming with a *handler*; what the caller set applies.
87fn with_defaults(options: &ConsumeOptions) -> ConsumeOptions {
88    let mut options = options.clone();
89    options.max_messages.get_or_insert(DEFAULT_MAX_MESSAGES);
90    options.wait_time.get_or_insert(DEFAULT_WAIT_TIME);
91    options
92        .visibility_timeout
93        .get_or_insert(DEFAULT_VISIBILITY_TIMEOUT);
94    options
95}
96
97/// The renewal interval: a third of the invisibility, so there is time left to retry if a
98/// renewal fails before the message is released to another consumer.
99fn renewal_interval(lease_seconds: u32) -> Duration {
100    Duration::from_secs_f64(f64::from(lease_seconds.max(1)) / 3.0)
101}
102
103impl Client {
104    /// Reads from the queue and calls `handler` for each message (see the module). Returns how many were
105    /// handled. If `handler` returns `Err`, the message is not deleted and the error propagates.
106    pub fn consume_with<H>(
107        &self,
108        topic: &str,
109        options: &ConsumeOptions,
110        mut handler: H,
111    ) -> Result<usize, ConsumeError>
112    where
113        H: FnMut(&mut Delivery<'_, ()>) -> Result<(), HandlerError>,
114    {
115        let options = with_defaults(options);
116        let lease = options
117            .visibility_timeout
118            .unwrap_or(DEFAULT_VISIBILITY_TIMEOUT);
119        let messages = self.consume(topic, &options)?;
120        let mut handled = 0;
121        for message in &messages {
122            self.handle_one(topic, lease, message, (), &mut handler)?;
123            handled += 1;
124        }
125        Ok(handled)
126    }
127
128    fn handle_one<T, H>(
129        &self,
130        topic: &str,
131        lease_seconds: u32,
132        message: &Message,
133        body: T,
134        handler: &mut H,
135    ) -> Result<(), ConsumeError>
136    where
137        H: FnMut(&mut Delivery<'_, T>) -> Result<(), HandlerError>,
138    {
139        let lease = AtomicU32::new(lease_seconds);
140        let lost = AtomicBool::new(false);
141        let stop = (Mutex::new(false), Condvar::new());
142        let mut delivery = Delivery {
143            message,
144            body,
145            delete: true,
146            client: self,
147            topic,
148            lease: &lease,
149            lost: &lost,
150        };
151        let result = std::thread::scope(|scope| {
152            scope.spawn(|| self.keep_alive(topic, message, &lease, &lost, &stop));
153            let result = handler(&mut delivery);
154            *stop.0.lock().unwrap() = true;
155            stop.1.notify_all();
156            result
157        });
158        result.map_err(ConsumeError::Handler)?;
159        if delivery.delete {
160            match self.ack(topic, &message.id, &message.receipt_token) {
161                // The message already went back to the queue (deadline expired, `409`) or is already gone (`404`).
162                Err(e) if e.status == 409 || e.status == 404 => {}
163                other => other?,
164            }
165        }
166        Ok(())
167    }
168
169    /// The second *thread*: renews the invisibility until the *handler* finishes. Never fails.
170    fn keep_alive(
171        &self,
172        topic: &str,
173        message: &Message,
174        lease: &AtomicU32,
175        lost: &AtomicBool,
176        stop: &(Mutex<bool>, Condvar),
177    ) {
178        loop {
179            let interval = renewal_interval(lease.load(Ordering::SeqCst));
180            let guard = stop.0.lock().unwrap();
181            let (guard, _) = stop
182                .1
183                .wait_timeout_while(guard, interval, |stopped| !*stopped)
184                .unwrap();
185            if *guard {
186                return;
187            }
188            drop(guard);
189            if let Err(e) = self.change_visibility(
190                topic,
191                &message.id,
192                &message.receipt_token,
193                lease.load(Ordering::SeqCst),
194            ) {
195                if e.status == 409 || e.status == 404 {
196                    // It already went back to the queue or was deleted: renewing is no longer of any use.
197                    lost.store(true, Ordering::SeqCst);
198                    return;
199                }
200                // Transient failure (network, 5xx): tries again on the next round.
201            }
202        }
203    }
204}
205
206#[cfg(feature = "serde")]
207impl Client {
208    /// Serializes `value` to JSON and publishes it as `application/json`. To serialize on your
209    /// own, use `publish` with the bytes.
210    pub fn publish_object<T: serde::Serialize>(
211        &self,
212        topic: &str,
213        value: &T,
214    ) -> Result<super::PublishResponse, Error> {
215        let json = serde_json::to_vec(value)
216            .map_err(|e| Error::client("client.invalid-request", format!("objeto: {e}")))?;
217        self.publish(topic, &json, &super::PublishOptions::default())
218    }
219
220    /// [`Client::consume_with`] with the body already deserialized as `T` in `delivery.body`. If the
221    /// body is not a `T`, it returns `client.invalid-response` before calling the *handler*, and the
222    /// message is not deleted.
223    pub fn consume_typed<T, H>(
224        &self,
225        topic: &str,
226        options: &ConsumeOptions,
227        mut handler: H,
228    ) -> Result<usize, ConsumeError>
229    where
230        T: serde::de::DeserializeOwned,
231        H: FnMut(&mut Delivery<'_, T>) -> Result<(), HandlerError>,
232    {
233        let options = with_defaults(options);
234        let lease = options
235            .visibility_timeout
236            .unwrap_or(DEFAULT_VISIBILITY_TIMEOUT);
237        let messages = self.consume(topic, &options)?;
238        let mut handled = 0;
239        for message in &messages {
240            let body: T = serde_json::from_slice(&message.body).map_err(|e| {
241                Error::client(
242                    "client.invalid-response",
243                    format!("message body {}: {e}", message.id),
244                )
245            })?;
246            self.handle_one(topic, lease, message, body, &mut handler)?;
247            handled += 1;
248        }
249        Ok(handled)
250    }
251}
252
253#[cfg(test)]
254mod tests;
taba_client/client/handlers/mod.rs

taba_client/client/handlers/
mod.rs
