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

use chrono::Duration;
use futures::{future::BoxFuture, stream::BoxStream, Stream};
use serde::Serialize;

use crate::{
    error::StorageError,
    queue::Heartbeat,
    request::{JobRequest, JobState},
};

pub type StorageResult<I> = BoxFuture<'static, Result<I, StorageError>>;
pub type JobStream<T> = BoxStream<'static, Result<Option<JobRequest<T>>, StorageError>>;

/// Represents a [Storage] that can be passed to a [crate::builder::QueueBuilder]
pub trait Storage: Clone {
    type Output: Serialize;

    /// Pushes a job to a storage
    ///
    /// TODO: return id
    fn push(&mut self, job: Self::Output) -> StorageResult<()>;

    /// Get the stream of jobs
    fn consume(&mut self) -> JobStream<Self::Output>;

    fn len(&self) -> StorageResult<i64> {
        let fut = async { Ok(0) };
        Box::pin(fut)
    }

    fn ack(&mut self, job_id: String) -> StorageResult<()> {
        let fut = async { Ok(()) };
        Box::pin(fut)
    }

    fn retry(&mut self, job_id: String) -> StorageResult<()> {
        let fut = async { Ok(()) };
        Box::pin(fut)
    }

    fn heartbeat(&mut self, beat: Heartbeat) -> StorageResult<bool> {
        let fut = async { Ok(true) };
        Box::pin(fut)
    }

    fn kill(&mut self, job_id: String) -> StorageResult<()> {
        let fut = async { Ok(()) };
        Box::pin(fut)
    }

    fn reschedule(&mut self, job_id: String, wait: Duration) -> StorageResult<()> {
        let fut = async { Ok(()) };
        Box::pin(fut)
    }
}

pub trait StorageJobExt<Output>: Storage<Output = Output> {
    fn find_by_id(&mut self, job_id: String) -> StorageResult<Output>;
    fn list_by_page(&mut self, status: JobState, page: i32) -> StorageResult<Vec<Output>>;
}