apalis-pgmq 0.1.0-rc.4

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

use std::fmt::Debug;

use apalis_core::{error::BoxDynError, task::Parts, worker::ext::ack::Acknowledge};
use futures::{
    FutureExt,
    future::{self, BoxFuture},
};

use crate::{PGMQueue, context::PgMqContext, errors::PgmqError, query};

impl<T, C, Res> Acknowledge<Res, PgMqContext, i64> for PGMQueue<T, C>
where
    T: Send,
    Res: Debug + Send + Sync,
    C: Send,
{
    type Error = PgmqError;

    type Future = BoxFuture<'static, Result<(), Self::Error>>;

    fn ack(
        &mut self,
        res: &Result<Res, BoxDynError>,
        parts: &Parts<PgMqContext, i64>,
    ) -> Self::Future {
        if res.is_ok() {
            let task_id = parts.task_id.as_ref().unwrap().inner().to_owned();
            let queue_name = self.config.queue().to_owned();
            let conn = self.connection.clone();

            let fut = async move {
                let query = query::archive_batch(queue_name.as_ref())?;
                let row = sqlx::query(&query).bind([task_id]).execute(&conn).await?;

                let num_archived = row.rows_affected();
                if num_archived == 1 {
                    Ok(())
                } else {
                    Err(PgmqError::DatabaseError(sqlx::Error::RowNotFound))
                }
            };
            return fut.boxed();
        }
        future::ready(Ok(())).boxed()
    }
}