apalis-core 1.0.0

Core for apalis: simple, extensible multithreaded background processing 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

use std::{
    sync::Arc,
    task::{Context, Poll},
    time::Duration,
};

use futures_core::Stream;

#[cfg(feature = "sleep")]
use crate::backend::ext::poll_strategy::{BackoffConfig, IntervalStrategy};
use crate::backend::ext::poll_strategy::{
    BoxedPollStrategy, PollSnapshot, PollStrategy, StreamStrategy,
};

/// A polling strategy that combines multiple strategies
/// The strategies are polled in the order they were added to the builder
/// In case of multiple strategies being ready at the same time, the first one added will be chosen
#[derive(Clone, Default)]
pub struct Strategy {
    strategies: Arc<std::sync::Mutex<Vec<BoxedPollStrategy>>>,
}

impl Strategy {
    /// Generate a builder for combining strategies
    #[must_use]
    pub fn new() -> Self {
        Self {
            strategies: Default::default(),
        }
    }

    /// Apply a polling strategy to the builder
    /// Strategies are executed in the order they are added, with the first strategy having the highest priority
    /// In case of multiple strategies being ready at the same time, the first one added will be chosen
    #[must_use]
    pub fn apply<S>(self, strategy: S) -> Self
    where
        S: PollStrategy + Send + Sync + 'static,
    {
        self.strategies.lock().unwrap().push(Box::new(strategy));
        self
    }

    /// Apply a stream strategy
    #[must_use]
    pub fn stream<S>(self, stream: S) -> Self
    where
        S: Stream + Unpin + Send + Sync + 'static,
    {
        self.apply(StreamStrategy::new(stream))
    }

    /// Apply an interval strategy
    #[must_use]
    #[cfg(feature = "sleep")]
    pub fn interval(self, period: Duration) -> Self {
        self.apply(IntervalStrategy::new(period))
    }

    /// Apply an interval with backoff
    #[must_use]
    #[cfg(feature = "sleep")]
    pub fn interval_with_backoff(self, period: Duration, backoff: BackoffConfig) -> Self {
        self.apply(IntervalStrategy::new(period).with_backoff(backoff))
    }
}

impl std::fmt::Debug for Strategy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MultiStrategy")
            .field("strategies", &self.strategies.lock().unwrap().len())
            .finish()
    }
}

impl PollStrategy for Strategy {
    fn poll_drive(&mut self, cx: &mut Context<'_>, worker: &PollSnapshot) -> Poll<Option<()>> {
        // Priority order: first-added strategy wins if multiple are ready.
        for strategy in &mut self.strategies.lock().unwrap().iter_mut() {
            if strategy.poll_drive(cx, worker).is_ready() {
                return Poll::Ready(Some(()));
            }
        }
        Poll::Pending
    }
}