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

extern crate redis;
use async_trait::async_trait;
use serde::{de::DeserializeOwned, Serialize};

use crate::backend::{DequeuBackend, EnqueuBackend};
use crate::backends::redis::RedisBackend;

/// Error type for `JsonRedisBackend`.
/// It wraps `redis::RedisError` and `serde_json::Error`.
#[derive(Debug)]
pub enum JsonRedisError {
    Redis(redis::RedisError),
    Serde(serde_json::Error),
}

/// Redis backend with JSON serialization.
/// It implements both `DequeuBackend` and `EnqueuBackend` traits.
/// It requires `serde::Serialize` and `serde::DeserializeOwned` traits for task type.
/// It overrides `RedisBackend` with `serde_json::to_string` and `serde_json::from_str` calls.
/// The logic of `RedisBackend` is not changed.
#[derive(Clone)]
pub struct JsonRedisBackend<T> {
    backend: RedisBackend,
    _phantom: std::marker::PhantomData<T>,
}

impl<T> JsonRedisBackend<T> {
    pub fn new(backend: RedisBackend) -> Self {
        Self {
            backend,
            _phantom: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<T> EnqueuBackend<T, f64, JsonRedisError> for JsonRedisBackend<T>
where
    T: Serialize + Send + Sync,
{
    /// Serializes data to JSON and calls `RedisBackend::enqueue`.
    async fn enqueue(&self, data: T, score: f64) -> Result<(), JsonRedisError> {
        let data = match serde_json::to_string(&data) {
            Ok(data) => data,
            Err(e) => return Err(JsonRedisError::Serde(e)),
        };
        match self.backend.enqueue(data, score).await {
            Ok(_) => Ok(()),
            Err(e) => Err(JsonRedisError::Redis(e)),
        }
    }
}

#[async_trait]
impl<T> DequeuBackend<serde_json::Result<T>, f64, redis::RedisError> for JsonRedisBackend<T>
where
    T: DeserializeOwned + Send + Sync,
{
    /// Returns `Vec<serde_json::Result<T>>` because it is possible that some tasks will be corrupted.
    async fn dequeue(&self, score: f64) -> Result<Vec<serde_json::Result<T>>, redis::RedisError> {
        let data = self.backend.dequeue(score).await?;
        Ok(data.into_iter().map(|d| serde_json::from_str(&d)).collect())
    }
}