arifa 0.3.44

Redis-based realtime pub/sub engine for WebSocket applications
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

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

#[derive(Default)]
struct Inner {
    sessions_active: AtomicU64,
    messages_routed: AtomicU64,
    messages_dropped: AtomicU64,
    redis_reconnects: AtomicU64,
}

/// Cheap, lock-free counters for the router's health. Clone freely —
/// it's an `Arc` under the hood, so every clone shares the same counters.
///
/// Wire `snapshot()` up to a `/metrics` or `/debug/arifa` endpoint, or
/// log it periodically, to actually see problems (queue buildup, drops,
/// reconnect churn) before they become outages.
#[derive(Clone, Default)]
pub struct Metrics(Arc<Inner>);

#[derive(Debug, Clone, Copy, serde::Serialize)]
pub struct MetricsSnapshot {
    pub sessions_active: u64,
    pub messages_routed: u64,
    pub messages_dropped: u64,
    pub redis_reconnects: u64,
}

impl Metrics {
    pub fn snapshot(&self) -> MetricsSnapshot {
        MetricsSnapshot {
            sessions_active: self.0.sessions_active.load(Ordering::Relaxed),
            messages_routed: self.0.messages_routed.load(Ordering::Relaxed),
            messages_dropped: self.0.messages_dropped.load(Ordering::Relaxed),
            redis_reconnects: self.0.redis_reconnects.load(Ordering::Relaxed),
        }
    }

    pub(crate) fn session_started(&self) {
        self.0.sessions_active.fetch_add(1, Ordering::Relaxed);
    }

    pub(crate) fn session_ended(&self) {
        self.0.sessions_active.fetch_sub(1, Ordering::Relaxed);
    }

    pub(crate) fn record_routed(&self, delivered: u64, dropped: u64) {
        self.0
            .messages_routed
            .fetch_add(delivered, Ordering::Relaxed);
        self.0
            .messages_dropped
            .fetch_add(dropped, Ordering::Relaxed);
    }

    pub(crate) fn record_reconnect(&self) {
        self.0.redis_reconnects.fetch_add(1, Ordering::Relaxed);
    }
}