resuma/realtime/
peer.rs

1use std::collections::HashMap;
2use std::time::{Duration, Instant};
3
4use tokio::sync::mpsc;
5
6use super::room::PeerId;
7
8/// One connected peer inside a [`Room`](super::Room): outbound channel, liveness,
9/// and app-defined per-peer state `S` (position, name, flags, …).
10///
11/// Resuma does not interpret `S` — it is stored and handed back so apps can build
12/// roster/snapshot messages without maintaining a second peer map of their own.
13pub struct Peer<S> {
14    pub id: PeerId,
15    pub state: S,
16    pub last_seen: Instant,
17    tx: mpsc::UnboundedSender<String>,
18    rate: HashMap<&'static str, Instant>,
19}
20
21impl<S> Peer<S> {
22    pub(super) fn new(id: PeerId, state: S, tx: mpsc::UnboundedSender<String>) -> Self {
23        Self {
24            id,
25            state,
26            last_seen: Instant::now(),
27            tx,
28            rate: HashMap::new(),
29        }
30    }
31
32    /// Send a pre-serialized text message to this peer only. Silently drops if the
33    /// peer's writer task already exited (client disconnected mid-send) — callers
34    /// should not treat this as fatal, the read loop will observe the close soon.
35    pub fn send(&self, msg: impl Into<String>) {
36        let _ = self.tx.send(msg.into());
37    }
38
39    /// Refresh liveness. Call on every inbound message, not just heartbeats, so a
40    /// chatty-but-silent-on-`ping` client is not swept as stale.
41    pub fn touch(&mut self) {
42        self.last_seen = Instant::now();
43    }
44
45    /// Throttle a noisy message `kind` (e.g. `"pose"`, `"place"`, `"remove"`) to at
46    /// most once per `min_interval`. Returns `true` the first time and again after
47    /// each interval elapses, `false` otherwise.
48    ///
49    /// Generalizes the `last_pose` / `last_place` / `last_remove` / `last_sync`
50    /// fields a hand-rolled multiplayer relay needs one-per-message-type — here it
51    /// is a single map keyed by a `&'static str` tag apps choose per call site.
52    pub fn allow_rate(&mut self, kind: &'static str, min_interval: Duration) -> bool {
53        let now = Instant::now();
54        match self.rate.get(kind) {
55            Some(last) if now.duration_since(*last) < min_interval => false,
56            _ => {
57                self.rate.insert(kind, now);
58                true
59            }
60        }
61    }
62}
63
64#[cfg(test)]
65mod tests {
66    use super::*;
67
68    fn peer_with_rx(id: &str) -> (Peer<()>, mpsc::UnboundedReceiver<String>) {
69        let (tx, rx) = mpsc::unbounded_channel();
70        (Peer::new(id.to_string(), (), tx), rx)
71    }
72
73    #[test]
74    fn allow_rate_gates_first_call_only_within_window() {
75        let (mut p, _rx) = peer_with_rx("a");
76        assert!(p.allow_rate("pose", Duration::from_secs(60)));
77        assert!(!p.allow_rate("pose", Duration::from_secs(60)));
78        // Different kind is independent.
79        assert!(p.allow_rate("place", Duration::from_secs(60)));
80    }
81
82    #[test]
83    fn send_does_not_panic_after_receiver_dropped() {
84        let (p, rx) = peer_with_rx("a");
85        drop(rx);
86        p.send("hello"); // unbounded_channel send after drop just errors silently
87    }
88}
resuma/realtime/peer.rs

resuma/realtime/
peer.rs
