oxanus 0.9.8

A simple & fast job queue system.
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

use crate::shared::*;
use serde::Serialize;
use testresult::TestResult;

#[derive(Serialize)]
struct QueueDynamic(i32);

#[derive(Serialize)]
struct QueueStatic;

impl oxanus::Queue for QueueDynamic {
    fn to_config() -> oxanus::QueueConfig {
        oxanus::QueueConfig::as_dynamic("dynamic").concurrency(6)
    }
}

impl oxanus::Queue for QueueStatic {
    fn to_config() -> oxanus::QueueConfig {
        oxanus::QueueConfig::as_static("static").concurrency(2)
    }
}

#[tokio::test]
pub async fn test_stats() -> TestResult {
    let redis_pool = setup();
    let ctx = oxanus::Context::value(());
    let storage = oxanus::Storage::builder()
        .namespace(random_string())
        .build_from_pool(redis_pool)?;
    let config = oxanus::Config::new(&storage)
        .register_queue::<QueueDynamic>()
        .register_queue::<QueueStatic>()
        .register_worker::<WorkerNoop>()
        .exit_when_processed(8);

    storage.enqueue(QueueDynamic(1), WorkerNoop {}).await?;
    storage.enqueue(QueueDynamic(2), WorkerNoop {}).await?;
    storage.enqueue(QueueStatic, WorkerNoop {}).await?;
    storage.enqueue(QueueStatic, WorkerNoop {}).await?;
    storage.enqueue(QueueDynamic(1), WorkerNoop {}).await?;
    storage.enqueue(QueueDynamic(2), WorkerNoop {}).await?;
    storage.enqueue(QueueDynamic(3), WorkerNoop {}).await?;
    storage.enqueue(QueueDynamic(4), WorkerNoop {}).await?;

    let stats = storage.stats().await?;

    assert_eq!(stats.queues.len(), 2);
    assert_eq!(stats.queues[0].key, "dynamic");
    assert_eq!(stats.queues[0].queues.len(), 4);

    assert_eq!(stats.queues[0].queues[0].suffix, "1");
    assert_eq!(stats.queues[0].queues[0].enqueued, 2);
    assert_eq!(stats.queues[0].queues[0].processed, 0);
    assert_eq!(stats.queues[0].queues[0].succeeded, 0);
    assert_eq!(stats.queues[0].queues[0].panicked, 0);
    assert_eq!(stats.queues[0].queues[0].failed, 0);
    assert!(stats.queues[0].queues[0].latency_s > 0.0);

    assert_eq!(stats.queues[0].queues[1].suffix, "2");
    assert_eq!(stats.queues[0].queues[1].enqueued, 2);
    assert_eq!(stats.queues[0].queues[1].processed, 0);
    assert_eq!(stats.queues[0].queues[1].succeeded, 0);
    assert_eq!(stats.queues[0].queues[1].panicked, 0);
    assert_eq!(stats.queues[0].queues[1].failed, 0);
    assert!(stats.queues[0].queues[1].latency_s > 0.0);

    assert_eq!(stats.queues[0].queues[2].suffix, "3");
    assert_eq!(stats.queues[0].queues[2].enqueued, 1);
    assert_eq!(stats.queues[0].queues[2].processed, 0);
    assert_eq!(stats.queues[0].queues[2].succeeded, 0);
    assert_eq!(stats.queues[0].queues[2].panicked, 0);
    assert_eq!(stats.queues[0].queues[2].failed, 0);
    assert!(stats.queues[0].queues[2].latency_s > 0.0);

    assert_eq!(stats.queues[0].queues[3].suffix, "4");
    assert_eq!(stats.queues[0].queues[3].enqueued, 1);
    assert_eq!(stats.queues[0].queues[3].processed, 0);
    assert_eq!(stats.queues[0].queues[3].succeeded, 0);
    assert_eq!(stats.queues[0].queues[3].panicked, 0);
    assert_eq!(stats.queues[0].queues[3].failed, 0);
    assert!(stats.queues[0].queues[3].latency_s > 0.0);

    assert_eq!(stats.queues[0].enqueued, 6);
    assert_eq!(stats.queues[0].processed, 0);
    assert_eq!(stats.queues[0].succeeded, 0);
    assert_eq!(stats.queues[0].panicked, 0);
    assert_eq!(stats.queues[0].failed, 0);
    assert!(stats.queues[0].latency_s > 0.0);

    assert_eq!(stats.queues[1].key, "static");
    assert_eq!(stats.queues[1].queues.len(), 0);

    assert_eq!(stats.queues[1].enqueued, 2);
    assert_eq!(stats.queues[1].processed, 0);
    assert_eq!(stats.queues[1].succeeded, 0);
    assert_eq!(stats.queues[1].panicked, 0);
    assert_eq!(stats.queues[1].failed, 0);
    assert!(stats.queues[1].latency_s > 0.0);

    let stats = oxanus::run(config, ctx).await?;

    assert_eq!(stats.processed, 8);

    let stats = storage.stats().await?;

    assert_eq!(stats.global.processed, 8);
    assert_eq!(stats.global.failed, 0);

    assert_eq!(stats.queues.len(), 2);
    assert_eq!(stats.queues[0].key, "dynamic");
    assert_eq!(stats.queues[0].queues.len(), 4);

    assert_eq!(stats.queues[0].enqueued, 0);
    assert_eq!(stats.queues[0].processed, 6);
    assert_eq!(stats.queues[0].succeeded, 6);
    assert_eq!(stats.queues[0].panicked, 0);
    assert_eq!(stats.queues[0].failed, 0);
    assert_eq!(stats.queues[0].latency_s, 0.0);

    assert_eq!(stats.queues[0].queues[0].suffix, "1");
    assert_eq!(stats.queues[0].queues[0].enqueued, 0);
    assert_eq!(stats.queues[0].queues[0].processed, 2);
    assert_eq!(stats.queues[0].queues[0].succeeded, 2);
    assert_eq!(stats.queues[0].queues[0].panicked, 0);
    assert_eq!(stats.queues[0].queues[0].failed, 0);
    assert_eq!(stats.queues[0].queues[0].latency_s, 0.0);

    assert_eq!(stats.queues[0].queues[1].suffix, "2");
    assert_eq!(stats.queues[0].queues[1].enqueued, 0);
    assert_eq!(stats.queues[0].queues[1].processed, 2);
    assert_eq!(stats.queues[0].queues[1].succeeded, 2);
    assert_eq!(stats.queues[0].queues[1].panicked, 0);
    assert_eq!(stats.queues[0].queues[1].failed, 0);
    assert_eq!(stats.queues[0].queues[1].latency_s, 0.0);

    assert_eq!(stats.queues[0].queues[2].suffix, "3");
    assert_eq!(stats.queues[0].queues[2].enqueued, 0);
    assert_eq!(stats.queues[0].queues[2].processed, 1);
    assert_eq!(stats.queues[0].queues[2].succeeded, 1);
    assert_eq!(stats.queues[0].queues[2].panicked, 0);
    assert_eq!(stats.queues[0].queues[2].failed, 0);
    assert_eq!(stats.queues[0].queues[2].latency_s, 0.0);

    assert_eq!(stats.queues[0].queues[3].suffix, "4");
    assert_eq!(stats.queues[0].queues[3].enqueued, 0);
    assert_eq!(stats.queues[0].queues[3].processed, 1);
    assert_eq!(stats.queues[0].queues[3].succeeded, 1);
    assert_eq!(stats.queues[0].queues[3].panicked, 0);
    assert_eq!(stats.queues[0].queues[3].failed, 0);
    assert_eq!(stats.queues[0].queues[3].latency_s, 0.0);

    assert_eq!(stats.queues[1].key, "static");
    assert_eq!(stats.queues[1].queues.len(), 0);

    assert_eq!(stats.queues[1].enqueued, 0);
    assert_eq!(stats.queues[1].processed, 2);
    assert_eq!(stats.queues[1].succeeded, 2);
    assert_eq!(stats.queues[1].panicked, 0);
    assert_eq!(stats.queues[1].failed, 0);
    assert_eq!(stats.queues[1].latency_s, 0.0);

    Ok(())
}