azums 0.1.3

High-performance job queue & streaming engine for Rust — from embedded to cloud
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

use azums::{quickstart, Job, MemoryBackend, StorageBackend};
use std::sync::Arc;

#[tokio::test]
async fn test_memory_backend_lifecycle() -> anyhow::Result<()> {
    let backend = MemoryBackend::new();
    backend.run_migrations().await?;
    backend.health_check().await?;

    let job_id = backend
        .enqueue(Job::new("in_ram_task", serde_json::json!({"val": 42})).into())
        .await?;

    let leased = backend
        .lease_jobs_batch("default", "mem-worker", 10, 5)
        .await?;
    assert_eq!(leased.len(), 1);
    assert_eq!(leased[0].id, job_id);

    let attempts = backend
        .start_attempts_batch(&["default".into()], &[job_id], "mem-worker")
        .await?;
    assert_eq!(attempts.len(), 1);
    let (_jid, attempt_id, attempt_no) = attempts[0];
    assert_eq!(attempt_no, 1);

    backend
        .mark_succeeded(job_id, attempt_id, "mem-worker", 5)
        .await?;

    let job = backend.get_job(job_id).await?.unwrap();
    assert_eq!(job.status, "succeeded");

    backend.clear();
    assert!(backend.get_job(job_id).await?.is_none());

    Ok(())
}

#[tokio::test]
async fn test_quickstart_in_memory() -> anyhow::Result<()> {
    // Developers can run cargo test with zero Docker and zero disk I/O!
    let flow = quickstart("memory").await?;

    let _id = flow
        .enqueue(Job::new(
            "ephemeral_job",
            serde_json::json!({"data": "test"}),
        ))
        .await?;

    let executed = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let executed_clone = executed.clone();

    flow.register_handler("ephemeral_job", move |job| {
        let ex = executed_clone.clone();
        async move {
            assert_eq!(job.payload["data"], "test");
            ex.store(true, std::sync::atomic::Ordering::SeqCst);
            Ok(())
        }
    })
    .await;

    let count = flow.run_until_empty().await?;
    assert_eq!(count, 1);
    assert!(executed.load(std::sync::atomic::Ordering::SeqCst));

    Ok(())
}