clasp-router 4.0.0

CLASP message router and server
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

//! SUBSCRIBE / UNSUBSCRIBE message handlers.
//!
//! Manages per-session subscriptions with glob-pattern matching, enforces
//! per-session subscription limits, and sends filtered snapshots on subscribe.

use clasp_core::{codec, ErrorMessage, Message, SecurityMode};
use tracing::{debug, warn};

use super::{send_chunked_snapshot, HandlerContext, MessageResult};
use crate::subscription::Subscription;

pub(crate) async fn handle_subscribe(
    sub: &clasp_core::SubscribeMessage,
    ctx: &HandlerContext<'_>,
) -> Option<MessageResult> {
    let session = ctx.session.as_ref()?;

    let current_subs = session.subscriptions().len();
    let max_subs = 1000;
    if current_subs >= max_subs {
        warn!(
            "Session {} subscription limit reached ({}/{})",
            session.id, current_subs, max_subs
        );
        #[cfg(feature = "metrics")]
        metrics::counter!("clasp_errors_total", "code" => "429").increment(1);
        let error = Message::Error(ErrorMessage {
            code: 429,
            message: format!("Subscription limit reached (max {})", max_subs),
            address: Some(sub.pattern.clone()),
            correlation_id: None,
        });
        let bytes = codec::encode(&error).ok()?;
        return Some(MessageResult::Send(bytes));
    }

    if ctx.security_mode == SecurityMode::Authenticated
        && !session.has_strict_read_scope(&sub.pattern)
    {
        warn!(
            "Session {} denied SUBSCRIBE to {} - insufficient scope",
            session.id, sub.pattern
        );
        let error = Message::Error(ErrorMessage {
            code: 301,
            message: "Insufficient scope for subscription".to_string(),
            address: Some(sub.pattern.clone()),
            correlation_id: None,
        });
        let bytes = codec::encode(&error).ok()?;
        return Some(MessageResult::Send(bytes));
    }

    match Subscription::new(
        sub.id,
        session.id.clone(),
        &sub.pattern,
        sub.types.clone(),
        sub.options.clone().unwrap_or_default(),
    ) {
        Ok(subscription) => {
            ctx.subscriptions.add(subscription);
            session.add_subscription(sub.id);
            #[cfg(feature = "metrics")]
            metrics::gauge!("clasp_subscriptions_active").increment(1.0);
            debug!("Session {} subscribed to {}", session.id, sub.pattern);

            let mut snapshot = ctx.state.snapshot(&sub.pattern);
            if let Some(ref filter) = ctx.snapshot_filter {
                snapshot.params = filter.filter_snapshot(snapshot.params, session, ctx.state);
            }
            if !snapshot.params.is_empty() {
                send_chunked_snapshot(ctx.sender, snapshot).await;
            }
        }
        Err(e) => {
            warn!("Invalid subscription pattern: {}", e);
            let error = Message::Error(ErrorMessage {
                code: 202,
                message: e.to_string(),
                address: Some(sub.pattern.clone()),
                correlation_id: None,
            });
            let bytes = codec::encode(&error).ok()?;
            return Some(MessageResult::Send(bytes));
        }
    }

    Some(MessageResult::None)
}

pub(crate) async fn handle_unsubscribe(
    unsub: &clasp_core::UnsubscribeMessage,
    ctx: &HandlerContext<'_>,
) -> Option<MessageResult> {
    let session = ctx.session.as_ref()?;
    ctx.subscriptions.remove(&session.id, unsub.id);
    session.remove_subscription(unsub.id);
    #[cfg(feature = "metrics")]
    metrics::gauge!("clasp_subscriptions_active").decrement(1.0);
    Some(MessageResult::None)
}