pulses 0.2.0

A robust, high-performance background job processing library for Rust.
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

//! Periodically reclaims pending messages abandoned by crashed/stalled
//! consumers and re-routes them to handler pools.

use std::time::Duration;

use tokio::time::MissedTickBehavior;
use tokio_util::sync::CancellationToken;

use crate::core::Broker;
use crate::runtime::router::Router;

pub(crate) struct Reclaimer<B: Broker> {
    broker: B,
    subscription: B::Subscription,
    router: Router<B::AckToken>,
    interval: Duration,
    cancellation: CancellationToken,
}

impl<B: Broker> Reclaimer<B> {
    pub(crate) fn new(
        broker: B, subscription: B::Subscription, router: Router<B::AckToken>, interval: Duration,
        cancellation: CancellationToken,
    ) -> Self {
        Self { broker, subscription, router, interval, cancellation }
    }

    pub(crate) async fn run(self) {
        let mut ticker = tokio::time::interval(self.interval.max(Duration::from_millis(1)));
        ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);

        loop {
            tokio::select! {
                _ = self.cancellation.cancelled() => return,
                _ = ticker.tick() => {}
            }

            let reclaim_result = tokio::select! {
                _ = self.cancellation.cancelled() => return,
                result = self.broker.reclaim(&self.subscription) => result,
            };

            match reclaim_result {
                Ok(batch) => {
                    for (envelope, token) in batch {
                        if let Err(error) = self.router.route(envelope, token, &self.cancellation).await {
                            tracing::error!(%error, "failed to route reclaimed message");
                        }
                        if self.cancellation.is_cancelled() {
                            return;
                        }
                    }
                }
                Err(error) => {
                    tracing::warn!(%error, "broker reclaim failed");
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use tokio::sync::mpsc;
    use tokio::time::timeout;
    use tokio_util::sync::CancellationToken;

    use super::Reclaimer;
    use crate::handler_set::RoutingTable;
    use crate::runtime::router::Router;
    use crate::test_support::MockBroker;
    use crate::test_support::MockSubscription;
    use crate::test_support::make_envelope;
    use crate::test_support::token;

    #[tokio::test]
    async fn reclaims_and_routes() {
        let broker = MockBroker::default();
        broker.enqueue_reclaim(vec![(make_envelope("orders", "9-9"), token("9-9"))]);

        let (tx, mut rx) = mpsc::channel(8);
        let mut table = RoutingTable::new();
        table.insert(Arc::from("orders"), 1 << 0);
        let router = Router::new(table, vec![tx]);
        let cancellation = CancellationToken::new();

        let reclaimer =
            Reclaimer::new(broker, MockSubscription::default(), router, Duration::from_millis(5), cancellation.clone());
        let handle = tokio::spawn(reclaimer.run());

        let task = timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap();
        assert_eq!(task.envelope.id.as_ref(), "9-9");

        cancellation.cancel();
        timeout(Duration::from_secs(1), handle).await.unwrap().unwrap();
    }
}