apalis-postgres 1.0.0-rc.7

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

use std::time::Duration;

use futures::{FutureExt, Stream, stream};
use sqlx::{PgPool, postgres::types::PgInterval};

use crate::Config;

pub fn reenqueue_orphaned(
    pool: PgPool,
    config: Config,
) -> impl Future<Output = Result<u64, sqlx::Error>> + Send {
    let dead_for = config.reenqueue_orphaned_after().as_secs() as i64;
    let queue = config.queue().to_string();
    let dead_for = PgInterval {
        months: 0,
        days: 0,
        microseconds: dead_for * 1_000_000,
    };
    async move {
        match sqlx::query_file!("queries/backend/reenqueue_orphaned.sql", dead_for, queue,)
            .execute(&pool)
            .await
        {
            Ok(res) => {
                if res.rows_affected() > 0 {
                    // log::info!(
                    //     "Re-enqueued {} orphaned tasks that were being processed by dead workers",
                    //     res.rows_affected()
                    // );
                }
                Ok(res.rows_affected())
            }
            Err(e) => {
                // log::error!("Failed to re-enqueue orphaned tasks: {e}");
                Err(e)
            }
        }
    }
}

pub fn reenqueue_orphaned_stream(
    pool: PgPool,
    config: Config,
    interval: Duration,
) -> impl Stream<Item = Result<u64, sqlx::Error>> + Send {
    let config = config.clone();
    stream::unfold((), move |_| {
        let pool = pool.clone();
        let config = config.clone();
        let interval = apalis_core::timer::Delay::new(interval);
        let fut = async move {
            interval.await;
            reenqueue_orphaned(pool, config).await
        };
        fut.map(|res| Some((res, ())))
    })
}