apalis-core 0.4.9

Core for apalis: simple, extensible multithreaded background processing for Rust
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

use std::{marker::PhantomData, time::Duration};

use crate::worker::{HeartBeat, WorkerId};

use super::Storage;

#[derive(Debug, Clone)]
pub(super) struct KeepAlive<ST, Service>(WorkerId, ST, Duration, PhantomData<Service>);

impl<ST, Service> KeepAlive<ST, Service> {
    pub(super) fn new<J: 'static>(
        worker_id: &WorkerId,
        storage: ST,
        interval: Duration,
    ) -> KeepAlive<ST, Service>
    where
        ST: Storage<Output = J>,
    {
        KeepAlive(worker_id.clone(), storage, interval, PhantomData)
    }
}

#[async_trait::async_trait]
impl<S: Storage + Send + Sync, Service: Sync + Send> HeartBeat for KeepAlive<S, Service> {
    async fn heart_beat(&mut self) {
        match self.1.keep_alive::<Service>(&self.0).await {
            Err(e) => {
                tracing::warn!("An error occurred while attempting the keep alive heartbeat for worker {}. Error: {}", self.0, e)
            }
            _ => {
                tracing::trace!("keep alive heartbeat successful for worker: {}", self.0)
            }
        }
    }
    fn interval(&self) -> Duration {
        self.2
    }
}

#[derive(Debug, Clone)]
pub(super) struct ReenqueueOrphaned<ST>(ST, i32, Duration, Duration);

impl<ST> ReenqueueOrphaned<ST> {
    pub(super) fn new<J: 'static>(
        storage: ST,
        count: i32,
        interval: Duration,
        timeout_worker: Duration,
    ) -> ReenqueueOrphaned<ST>
    where
        ST: Storage<Output = J>,
    {
        ReenqueueOrphaned(storage, count, interval, timeout_worker)
    }
}

#[async_trait::async_trait]
impl<S: Storage + Send + Sync> HeartBeat for ReenqueueOrphaned<S> {
    async fn heart_beat(&mut self) {
        match self
            .0
            .heartbeat(super::StorageWorkerPulse::ReenqueueOrphaned {
                count: self.1,
                timeout_worker: self.3,
            })
            .await
        {
            Err(e) => {
                tracing::warn!("An error occurred while attempting the reenqueue orphaned heartbeat. Error: {}", e)
            }
            _ => {
                tracing::trace!("reenqueue orphaned heartbeat successful for storage")
            }
        }
    }
    fn interval(&self) -> Duration {
        self.2
    }
}

#[derive(Debug, Clone)]
pub(super) struct EnqueueScheduled<ST>(ST, i32, Duration);

impl<ST> EnqueueScheduled<ST> {
    pub(super) fn new<J: 'static>(
        storage: ST,
        count: i32,
        interval: Duration,
    ) -> EnqueueScheduled<ST>
    where
        ST: Storage<Output = J>,
    {
        EnqueueScheduled(storage, count, interval)
    }
}

#[async_trait::async_trait]
impl<S: Storage + Send + Sync> HeartBeat for EnqueueScheduled<S> {
    async fn heart_beat(&mut self) {
        match self
            .0
            .heartbeat(super::StorageWorkerPulse::EnqueueScheduled { count: self.1 })
            .await
        {
            Err(e) => {
                tracing::warn!(
                    "An error occurred while attempting the enqueue scheduled heartbeat. Error: {}",
                    e
                )
            }
            _ => {
                tracing::trace!("enqueue scheduled heartbeat successful for storage")
            }
        }
    }
    fn interval(&self) -> Duration {
        self.2
    }
}