apalis-redis 1.0.0

Background task processing for rust using apalis and redis
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

use apalis_core::worker::context::WorkerContext;
use redis::{Script, aio::ConnectionLike};

use crate::{Config, RedisTask, error::Error, queries::current_timestamp};

pub(crate) async fn acquire_leases<C>(
    conn: &mut C,
    worker: &WorkerContext,
    config: &Config,
    task_ids: &[String],
) -> Result<u32, Error>
where
    C: ConnectionLike,
{
    if task_ids.is_empty() {
        return Ok(0);
    }

    let script = redis::Script::new(include_str!("../../lua/acquire_leases.lua"));
    let worker_inflight_set = config.inflight_set_for(worker);
    let workers_set = config.workers_set();
    let job_meta_hash = config.job_meta_hash();
    let active_jobs_list = config.active_jobs_list();
    let now: u64 = current_timestamp();

    let mut invocation = script.prepare_invoke();
    invocation
        .key(worker_inflight_set)
        .key(&workers_set)
        .key(&job_meta_hash)
        .key(&active_jobs_list)
        .arg(worker.name())
        .arg(now)
        .arg(config.emit_events);

    for task_id in task_ids {
        invocation.arg(task_id);
    }

    let acquired = invocation.invoke_async::<u32>(conn).await?;

    Ok(acquired)
}

pub(crate) async fn renew_leases<C>(
    conn: &mut C,
    worker: &WorkerContext,
    config: &Config,
) -> Result<u32, Error>
where
    C: ConnectionLike,
{
    if worker.task_count() == 0 {
        return Ok(0);
    }

    let script = redis::Script::new(include_str!("../../lua/renew_leases.lua"));
    let worker_inflight_set = config.inflight_set_for(worker);
    let workers_set = config.workers_set();
    let job_meta_hash = config.job_meta_hash();
    let now: u64 = current_timestamp();

    let mut invocation = script.prepare_invoke();
    invocation
        .key(worker_inflight_set)
        .key(&workers_set)
        .key(&job_meta_hash)
        .arg(worker.name())
        .arg(now)
        .arg(config.emit_events);

    #[cfg(feature = "task-tracking")]
    let tasks = worker
        .tasks()
        .map_err(|_| Error::WorkerOutOfSync)?
        .iter()
        .map(|t| t.task_id().to_string())
        .collect::<Vec<_>>();

    #[cfg(not(feature = "task-tracking"))]
    let tasks: Vec<String> = vec![];

    for task in tasks {
        invocation.arg(task);
    }

    let renewed = invocation.invoke_async::<u32>(conn).await?;

    Ok(renewed)
}

pub(crate) async fn release_leases<C: ConnectionLike>(
    conn: &mut C,
    config: &Config,
    worker: &WorkerContext,
    tasks: &Vec<RedisTask>,
) -> Result<u64, Error> {
    let worker_inflight_set = config.inflight_set_for(worker);
    let active_jobs_list = config.active_jobs_list();
    let signal_list = config.signal_list();
    let workers_set = config.workers_set();
    let job_meta_hash = config.job_meta_hash();

    let script = Script::new(include_str!("../../lua/release_leases.lua"));
    let mut invocation = script.prepare_invoke();

    invocation
        .key(worker_inflight_set)
        .key(active_jobs_list)
        .key(signal_list)
        .key(workers_set)
        .key(job_meta_hash)
        .arg(worker.name())
        .arg(config.emit_events);

    for task in tasks {
        invocation.arg(task.task_id().unwrap().to_string());
    }

    Ok(invocation.invoke_async::<u64>(conn).await?)
}