jono_core 0.1.2

Shared core utilities for Jono queue libraries
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
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
120
121
122
123
124
125
126
127
128
129
130
131

//! Provides functionality to query details jobs in the Jono queue system.

use redis::{Commands, Connection};
use serde_json::Value;
use std::collections::HashMap;

use crate::{Context, Error, JobMetadata, JobStatus, Result};

/// Interface for querying job details
pub struct Inspector {
    context: Context,
}

impl Inspector {
    pub fn with_context(context: Context) -> Self {
        Self { context }
    }

    pub fn job_exists(&self, job_id: &str) -> Result<bool> {
        let mut conn = self.get_connection()?;
        let keys = self.context.keys();

        let exists: bool = conn
            .exists(keys.job_metadata_hash(job_id))
            .map_err(Error::Redis)?;

        Ok(exists)
    }

    pub fn job_is_canceled(&self, job_id: &str) -> Result<bool> {
        let mut conn = self.get_connection()?;
        let keys = self.context.keys();

        let score: Option<i64> = conn
            .zscore(keys.canceled_set(), job_id)
            .map_err(Error::Redis)?;

        Ok(score.is_some())
    }

    pub fn get_job_status(&self, job_id: &str) -> Result<JobStatus> {
        let mut conn = self.get_connection()?;
        let keys = self.context.keys();

        let exists: bool = conn
            .exists(keys.job_metadata_hash(job_id))
            .map_err(Error::Redis)?;
        if !exists {
            return Err(Error::NotFound(format!("Job {} not found", job_id)));
        }

        let in_running_set: bool = conn
            .zscore::<_, _, Option<i64>>(keys.running_set(), job_id)
            .map_err(Error::Redis)?
            .is_some();
        if in_running_set {
            return Ok(JobStatus::Running);
        }

        let in_queued_set: bool = conn
            .zscore::<_, _, Option<i64>>(keys.queued_set(), job_id)
            .map_err(Error::Redis)?
            .is_some();
        if in_queued_set {
            return Ok(JobStatus::Queued);
        }

        let in_scheduled_set: bool = conn
            .zscore::<_, _, Option<i64>>(keys.scheduled_set(), job_id)
            .map_err(Error::Redis)?
            .is_some();
        if in_scheduled_set {
            return Ok(JobStatus::Scheduled);
        }

        let in_canceled_set: bool = conn
            .zscore::<_, _, Option<i64>>(keys.canceled_set(), job_id)
            .map_err(Error::Redis)?
            .is_some();
        if in_canceled_set {
            return Ok(JobStatus::Canceled);
        }

        let has_completed_at_field: Option<String> = conn
            .hget(keys.job_metadata_hash(job_id), "completed_at")
            .map_err(Error::Redis)?;
        if has_completed_at_field.is_some() {
            return Ok(JobStatus::Completed);
        }

        let attempt_history: Option<String> = conn
            .hget(keys.job_metadata_hash(job_id), "attempt_history")
            .map_err(Error::Redis)?;
        if let Some(history_str) = attempt_history {
            let history: Value =
                serde_json::from_str(&history_str).map_err(Error::Serialization)?;

            if let Value::Array(attempts) = history {
                let max_attempts: u32 = conn
                    .hget::<_, _, Option<String>>(keys.job_metadata_hash(job_id), "max_attempts")
                    .map_err(Error::Redis)?
                    .and_then(|m| m.parse().ok())
                    .unwrap_or(3);

                if attempts.len() >= max_attempts as usize {
                    return Ok(JobStatus::Failed);
                }
            }
        }

        Ok(JobStatus::Failed)
    }

    pub fn get_job_metadata(&self, job_id: &str) -> Result<JobMetadata> {
        let mut conn = self.get_connection()?;
        let keys = self.context.keys();

        let metadata_key = keys.job_metadata_hash(job_id);
        let exists: bool = conn.exists(&metadata_key).map_err(Error::Redis)?;
        if !exists {
            return Err(Error::NotFound(format!("Job {} not found", job_id)));
        }

        let hash: HashMap<String, String> = conn.hgetall(&metadata_key).map_err(Error::Redis)?;
        JobMetadata::from_hash(hash)
    }

    fn get_connection(&self) -> Result<Connection> {
        self.context.get_connection()
    }
}