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
use serde::{Serialize, de::DeserializeOwned};
use crate::{
error::{AddJobErr, ObliterateError, PauseResumeError},
job::JobOptions,
luacommands::{
AddDelayedJob, AddPrioritizedJob, AddStandardJob, InvokeLuaScript, Obliterate,
ObliterateOk, Pause, PauseAction,
},
queue::Queue,
worker::{Worker, WorkerArgs},
};
impl<D, R> Queue<D, R> {
/// Create a worker instance for processing jobs of this queue.
/// This will immediately start preloading jobs to be processed.
///
/// Note that for picking up jobs, D must implement [serde::DeserializeOwned]
/// and R must implement [serde::Serialize].
pub fn worker(&self, worker_args: WorkerArgs) -> Worker<D, R>
where
R: Send + 'static,
D: Send + Sync + DeserializeOwned + std::fmt::Debug + 'static,
{
if self.pool.status().max_size < worker_args.parallel_connections * 2 {
self.pool.resize(worker_args.parallel_connections * 2);
}
Worker::new(self.pool.clone(), self.name.clone(), worker_args)
}
/// Stop processing jobs of a queue. Jobs already picked up will be
/// continued until completed or failed and delayed. Keep in mind,
/// that workers might pre-pull a few jobs depending on concurrency
/// settings. Those will still be processed as well.
pub async fn pause(&self) -> Result<(), PauseResumeError> {
let p = Pause {
queue: &self.name,
action: PauseAction::Pause,
};
let mut con = self.pool.get().await?;
p.call(&mut con).await?;
Ok(())
}
/// Continue processing jobs of a queue.
pub async fn resume(&self) -> Result<(), PauseResumeError> {
let p = Pause {
queue: &self.name,
action: PauseAction::Resume,
};
let mut con = self.pool.get().await?;
p.call(&mut con).await?;
Ok(())
}
/// Add a job with given options.
/// Go to [JobOptions] for documentation on the available options and default values.
pub async fn add_with(
&self,
job_name: &str,
data: &D,
job_options: &JobOptions,
) -> Result<String, AddJobErr>
where
D: Serialize,
{
let mut con = self.pool.get().await?;
if job_options.delay.is_some() {
let c = AddDelayedJob {
queue: &self.name,
job_name,
data,
job_options,
};
return c.call(&mut con).await;
}
if job_options.priority.is_some() {
let c = AddPrioritizedJob {
queue: &self.name,
job_name,
data,
job_options,
};
return c.call(&mut con).await;
}
let c = AddStandardJob {
queue: &self.name,
job_name,
data,
job_options,
};
c.call(&mut con).await
}
/// Add a job with default options. See documentation of [JobOptions] for default values.
pub async fn add(&self, job_name: &str, data: &D) -> Result<String, AddJobErr>
where
D: Serialize,
{
let job_options: JobOptions = Default::default();
self.add_with(job_name, data, &job_options).await
}
/**
Wipe this queue from existence.
This means all jobs, pending, done, failed, etc, as well as all markers and meta information.
It makes multiple trips to Redis, as it is unfeasable of deleting potentially millions of jobs in a single lua script execution.
*/
pub async fn obliterate(self) -> Result<(), ObliterateError> {
self.pause().await?;
let mut con = self.pool.get().await?;
loop {
let ob = Obliterate {
queue: &self.name,
batch_size: 1000,
force: true,
};
let res = ob.call(&mut con).await?;
match res {
ObliterateOk::Progress => (),
ObliterateOk::Obliterated => break,
}
}
Ok(())
}
}