azums 1.0.1

Embedded durable execution runtime for Rust, from Memory and SQLite to PostgreSQL and Redis
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

mod common;

use azums::{
    make_sqlite_pool, quickstart, BackendCapabilities, BackpressureCapability, OrderingCapability,
    PostgresBackend, SqliteBackend, StorageBackend,
};

#[tokio::test]
async fn memory_declares_process_local_capabilities() -> anyhow::Result<()> {
    let client = quickstart("memory").await?;

    assert_eq!(
        client.capabilities(),
        BackendCapabilities {
            transactional_enqueue: false,
            durable_jobs: false,
            notifications: true,
            streams: true,
            consumer_groups: true,
            distributed_workers: false,
            ordering: OrderingCapability::FifoAndFastestLeasing,
            backpressure: BackpressureCapability::BacklogOnly,
        }
    );

    Ok(())
}

#[tokio::test]
async fn sqlite_declares_embedded_sql_capabilities() -> anyhow::Result<()> {
    let db_url = format!(
        "sqlite://file:test_sqlite_capabilities_{}?mode=memory&cache=shared",
        uuid::Uuid::new_v4()
    );
    let pool = make_sqlite_pool(&db_url).await?;
    let backend = SqliteBackend::new(pool);

    assert_eq!(
        backend.capabilities(),
        BackendCapabilities {
            transactional_enqueue: true,
            durable_jobs: true,
            notifications: true,
            streams: true,
            consumer_groups: true,
            distributed_workers: false,
            ordering: OrderingCapability::FifoAndFastestLeasing,
            backpressure: BackpressureCapability::BacklogOnly,
        }
    );

    Ok(())
}

#[tokio::test]
async fn postgres_declares_distributed_sql_capabilities_when_available() -> anyhow::Result<()> {
    let Some(pool) = common::setup_db().await else {
        return Ok(());
    };
    let backend = PostgresBackend::new(pool);

    assert_eq!(
        backend.capabilities(),
        BackendCapabilities {
            transactional_enqueue: true,
            durable_jobs: true,
            notifications: true,
            streams: true,
            consumer_groups: true,
            distributed_workers: true,
            ordering: OrderingCapability::FifoAndFastestLeasing,
            backpressure: BackpressureCapability::ExecutionRateLimit,
        }
    );

    Ok(())
}

#[tokio::test]
async fn redis_declares_atomic_distributed_capabilities_when_available() -> anyhow::Result<()> {
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());

    let client = match quickstart(&redis_url).await {
        Ok(c) => c,
        Err(_) => {
            eprintln!(
                "Skipping live Redis capabilities test: No Redis server reachable at {redis_url}"
            );
            return Ok(());
        }
    };

    assert_eq!(
        client.capabilities(),
        BackendCapabilities {
            transactional_enqueue: false,
            durable_jobs: true,
            notifications: true,
            streams: true,
            consumer_groups: true,
            distributed_workers: true,
            ordering: OrderingCapability::FifoLeasing,
            backpressure: BackpressureCapability::BacklogOnly,
        }
    );

    Ok(())
}