distkit 0.3.0

A toolkit of distributed systems primitives for Rust, backed by 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
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

use std::sync::{Arc, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};

use redis::aio::ConnectionManager;

use crate::DistkitRedisKey;
use crate::icounter::{StrictInstanceAwareCounter, StrictInstanceAwareCounterOptions};

static RUN_ID: OnceLock<u128> = OnceLock::new();

pub fn run_id() -> u128 {
    *RUN_ID.get_or_init(|| {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    })
}

pub async fn make_connection() -> ConnectionManager {
    let url = std::env::var("REDIS_URL").expect("REDIS_URL must be set — run via `make test`");
    let client = redis::Client::open(url).expect("valid Redis URL");
    client
        .get_connection_manager()
        .await
        .expect("Redis must be reachable")
}

pub async fn make_counter(prefix: &str) -> Arc<StrictInstanceAwareCounter> {
    let conn = make_connection().await;
    let unique_prefix = format!("{}_{}", run_id(), prefix);
    StrictInstanceAwareCounter::new(StrictInstanceAwareCounterOptions::new(
        DistkitRedisKey::from(unique_prefix),
        conn,
    ))
}

#[allow(dead_code)]
pub async fn make_counter_with_opts(
    prefix: &str,
    threshold_ms: u64,
) -> Arc<StrictInstanceAwareCounter> {
    let conn = make_connection().await;
    let unique_prefix = format!("{}_{}", run_id(), prefix);
    StrictInstanceAwareCounter::new(StrictInstanceAwareCounterOptions {
        prefix: DistkitRedisKey::from(unique_prefix),
        connection_manager: conn,
        dead_instance_threshold_ms: threshold_ms,
    })
}

/// Two counters sharing the same Redis prefix (different instance IDs).
pub async fn make_pair(
    prefix: &str,
) -> (
    Arc<StrictInstanceAwareCounter>,
    Arc<StrictInstanceAwareCounter>,
) {
    let conn1 = make_connection().await;
    let conn2 = make_connection().await;
    let unique_prefix = format!("{}_{}", run_id(), prefix);

    let c1 = StrictInstanceAwareCounter::new(StrictInstanceAwareCounterOptions::new(
        DistkitRedisKey::from(unique_prefix.clone()),
        conn1,
    ));
    let c2 = StrictInstanceAwareCounter::new(StrictInstanceAwareCounterOptions::new(
        DistkitRedisKey::from(unique_prefix),
        conn2,
    ));
    (c1, c2)
}

pub async fn make_pair_with_opts(
    prefix: &str,
    threshold_ms: u64,
) -> (
    Arc<StrictInstanceAwareCounter>,
    Arc<StrictInstanceAwareCounter>,
) {
    let conn1 = make_connection().await;
    let conn2 = make_connection().await;
    let unique_prefix = format!("{}_{}", run_id(), prefix);

    let opts = |conn| StrictInstanceAwareCounterOptions {
        prefix: DistkitRedisKey::from(unique_prefix.clone()),
        connection_manager: conn,
        dead_instance_threshold_ms: threshold_ms,
    };

    let c1 = StrictInstanceAwareCounter::new(opts(conn1));
    let c2 = StrictInstanceAwareCounter::new(opts(conn2));
    (c1, c2)
}

/// Creates `n` counters sharing the same Redis prefix (each with a unique instance ID).
pub async fn make_n_counters(prefix: &str, n: usize) -> Vec<Arc<StrictInstanceAwareCounter>> {
    let unique_prefix = format!("{}_{}", run_id(), prefix);
    let mut counters = Vec::with_capacity(n);
    for _ in 0..n {
        let conn = make_connection().await;
        counters.push(StrictInstanceAwareCounter::new(
            StrictInstanceAwareCounterOptions::new(
                DistkitRedisKey::from(unique_prefix.clone()),
                conn,
            ),
        ));
    }
    counters
}

pub async fn make_n_counters_with_opts(
    prefix: &str,
    n: usize,
    threshold_ms: u64,
) -> Vec<Arc<StrictInstanceAwareCounter>> {
    let unique_prefix = format!("{}_{}", run_id(), prefix);
    let mut counters = Vec::with_capacity(n);
    for _ in 0..n {
        let conn = make_connection().await;
        counters.push(StrictInstanceAwareCounter::new(
            StrictInstanceAwareCounterOptions {
                prefix: DistkitRedisKey::from(unique_prefix.clone()),
                connection_manager: conn,
                dead_instance_threshold_ms: threshold_ms,
            },
        ));
    }
    counters
}

pub fn key(name: &str) -> DistkitRedisKey {
    DistkitRedisKey::from(name.to_string())
}