apalis-amqp 0.2.0

Message queuing utilities for Rust using apalis and Amqp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

use apalis_core::{
    job::{Job},
    layers::ack::{Ack, AckError},
    worker::WorkerId,
};
use lapin::options::BasicAckOptions;
use serde::{de::DeserializeOwned, Serialize};

use crate::{AmqpBackend, DeliveryTag};

#[async_trait::async_trait]
impl<J: Send + Sync + Serialize + DeserializeOwned + Job + 'static> Ack<J> for AmqpBackend<J> {
    type Acknowledger = DeliveryTag;
    async fn ack(&self, _worker_id: &WorkerId, tag: &DeliveryTag) -> Result<(), AckError> {
        let channel = self.channel().clone();
        channel
            .basic_ack(tag.0, BasicAckOptions::default())
            .await
            .map_err(|e| AckError::NoAck(e.into()))
    }
}