fswtch 0.1.8

Rust bindings and helpers for writing FreeSWITCH modules
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

use std::{
    collections::HashMap,
    sync::{LazyLock, Mutex},
    time::{Duration, Instant},
};

static LIMITERS: LazyLock<Mutex<HashMap<String, Bucket>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));
const MAX_BUCKETS: usize = 10_000;

fswtch::module_exports! {
    module = mod_rate_limiter,
    load = switch_module_load,
}

#[derive(Debug, Clone)]
struct Bucket {
    remaining: u32,
    reset_at: Instant,
}

#[derive(Debug, Clone)]
struct LimitRequest {
    key: String,
    limit: u32,
    window: Duration,
}

impl LimitRequest {
    fn parse(text: &str) -> Option<Self> {
        let mut parts = text.split_whitespace();
        let key = parts.next()?.to_owned();
        let limit = parts
            .next()
            .and_then(|value| value.parse().ok())
            .unwrap_or(10);
        let window_secs = parts
            .next()
            .and_then(|value| value.parse().ok())
            .unwrap_or(60);
        Some(Self {
            key,
            limit,
            window: Duration::from_secs(window_secs),
        })
    }
}

fswtch::api_callback! {
    fn allow_api(cmd, _session, stream) {
        fswtch::log_info("mod_rate_limiter", "rust_rate_limit invoked");
        let Some(request) = cmd.as_deref().and_then(LimitRequest::parse) else {
            let status = stream.write("usage: rust_rate_limit <key> [limit] [window-secs]\n");
            return fswtch::false_on_success(status);
        };

        let now = Instant::now();
        let mut limiters = LIMITERS
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        if !limiters.contains_key(&request.key) && limiters.len() >= MAX_BUCKETS {
            fswtch::log_error("mod_rate_limiter", "rate limiter bucket limit reached");
            return stream.write("rate limiter bucket limit reached\n");
        }
        let bucket = limiters
            .entry(request.key.clone())
            .or_insert_with(|| Bucket {
                remaining: request.limit,
                reset_at: now + request.window,
            });

        if now >= bucket.reset_at {
            bucket.remaining = request.limit;
            bucket.reset_at = now + request.window;
        }

        let allowed = bucket.remaining > 0;
        if allowed {
            bucket.remaining -= 1;
        }
        fswtch::log_info(
            "mod_rate_limiter",
            format!(
                "key={} allowed={} remaining={}",
                request.key, allowed, bucket.remaining
            ),
        );

        stream.write(
            &format!(
                "key={} allowed={} remaining={}\n",
                request.key, allowed, bucket.remaining
            ),
        )
    }
}

fswtch::api_callback! {
    fn reset_api(_cmd, _session, stream) {
        fswtch::log_info("mod_rate_limiter", "rust_rate_limit_reset invoked");
        LIMITERS
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clear();
        stream.write("rate limiters reset\n")
    }
}

fswtch::module_load! {
    fn switch_module_load(module) for "mod_rate_limiter" {
        fswtch::log_info("mod_rate_limiter", "loading module");
        module
            .api(
                "rust_rate_limit",
                "checks a token-bucket rate limit",
                "rust_rate_limit <key> [limit] [window-secs]",
                allow_api,
            )
            .and_then(|module| {
                module.api(
                    "rust_rate_limit_reset",
                    "clears all rate limiter buckets",
                    "rust_rate_limit_reset",
                    reset_api,
                )
            })
    }
}