apalis-sql 1.0.0-rc.7

SQL utilities for background job processing with apalis
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

use std::time::Duration;

use apalis_core::backend::{
    poll_strategy::{BackoffConfig, IntervalStrategy, MultiStrategy, StrategyBuilder},
    queue::Queue,
};

/// Configuration for the SQL backend
#[derive(Debug, Clone)]
pub struct Config {
    keep_alive: Duration,
    buffer_size: usize,
    poll_strategy: MultiStrategy,
    reenqueue_orphaned_after: Duration,
    queue: Queue,
    ack: bool,
}

impl Default for Config {
    fn default() -> Self {
        Self {
            keep_alive: Duration::from_secs(30),
            buffer_size: 10,
            poll_strategy: StrategyBuilder::new()
                .apply(
                    IntervalStrategy::new(Duration::from_millis(100))
                        .with_backoff(BackoffConfig::default()),
                )
                .build(),
            reenqueue_orphaned_after: Duration::from_secs(300), // 5 minutes
            queue: Queue::from("default"),
            ack: true,
        }
    }
}

impl Config {
    /// Create a new config with a jobs queue
    #[must_use]
    pub fn new(queue: &str) -> Self {
        Self {
            queue: Queue::from(queue),
            ..Default::default()
        }
    }

    /// Interval between database poll queries
    ///
    /// Defaults to 100ms
    #[must_use]
    pub fn with_poll_interval(mut self, strategy: MultiStrategy) -> Self {
        self.poll_strategy = strategy;
        self
    }

    /// Interval between worker keep-alive database updates
    ///
    /// Defaults to 30s
    #[must_use]
    pub fn set_keep_alive(mut self, keep_alive: Duration) -> Self {
        self.keep_alive = keep_alive;
        self
    }

    /// Buffer size to use when querying for jobs
    ///
    /// Defaults to 10
    #[must_use]
    pub fn set_buffer_size(mut self, buffer_size: usize) -> Self {
        self.buffer_size = buffer_size;
        self
    }

    /// Gets a reference to the keep_alive duration.
    #[must_use]
    pub fn keep_alive(&self) -> &Duration {
        &self.keep_alive
    }

    /// Gets a mutable reference to the keep_alive duration.
    pub fn keep_alive_mut(&mut self) -> &mut Duration {
        &mut self.keep_alive
    }

    /// Gets the buffer size.
    #[must_use]
    pub fn buffer_size(&self) -> usize {
        self.buffer_size
    }

    /// Gets a reference to the poll_strategy.
    #[must_use]
    pub fn poll_strategy(&self) -> &MultiStrategy {
        &self.poll_strategy
    }

    /// Gets a mutable reference to the poll_strategy.
    pub fn poll_strategy_mut(&mut self) -> &mut MultiStrategy {
        &mut self.poll_strategy
    }

    /// Gets a reference to the queue.
    #[must_use]
    pub fn queue(&self) -> &Queue {
        &self.queue
    }

    /// Gets a mutable reference to the queue.
    pub fn queue_mut(&mut self) -> &mut Queue {
        &mut self.queue
    }

    /// Gets the reenqueue_orphaned_after duration.
    #[must_use]
    pub fn reenqueue_orphaned_after(&self) -> Duration {
        self.reenqueue_orphaned_after
    }

    /// Gets a mutable reference to the reenqueue_orphaned_after.
    pub fn reenqueue_orphaned_after_mut(&mut self) -> &mut Duration {
        &mut self.reenqueue_orphaned_after
    }

    /// Occasionally some workers die, or abandon jobs because of panics.
    /// This is the time a task takes before its back to the queue
    ///
    /// Defaults to 5 minutes
    #[must_use]
    pub fn set_reenqueue_orphaned_after(mut self, after: Duration) -> Self {
        self.reenqueue_orphaned_after = after;
        self
    }

    /// Whether to automatically acknowledge jobs after processing
    /// Defaults to true
    #[must_use]
    pub fn ack(&self) -> bool {
        self.ack
    }

    /// Set whether to automatically acknowledge jobs after processing
    #[must_use]
    pub fn set_ack(mut self, auto_ack: bool) -> Self {
        self.ack = auto_ack;
        self
    }
}