p2panda_sync/manager/
session_map.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! Maintain mapping of generic topics to session ids and of session ids to a channel sender.
4use std::collections::{HashMap, HashSet};
5use std::hash::Hash;
6
7/// Map of generic topics to session ids and of session ids to a channel sender.
8#[derive(Clone, Debug)]
9pub struct SessionTopicMap<T, TX> {
10    pub(crate) session_tx_map: HashMap<u64, TX>,
11    pub(crate) session_topic_map: HashMap<u64, T>,
12    pub(crate) topic_session_map: HashMap<T, HashSet<u64>>,
13}
14
15impl<T, TX> Default for SessionTopicMap<T, TX> {
16    fn default() -> Self {
17        Self {
18            session_tx_map: Default::default(),
19            session_topic_map: Default::default(),
20            topic_session_map: Default::default(),
21        }
22    }
23}
24
25impl<T, TX> SessionTopicMap<T, TX>
26where
27    T: Clone + Hash + Eq,
28{
29    /// Insert a session id with their topic and tx channel.
30    pub fn insert_with_topic(&mut self, session_id: u64, topic: T, tx: TX) {
31        self.session_topic_map.insert(session_id, topic.clone());
32        self.topic_session_map
33            .entry(topic.clone())
34            .and_modify(|sessions| {
35                sessions.insert(session_id);
36            })
37            .or_insert(HashSet::from_iter([session_id]));
38        self.session_tx_map.insert(session_id, tx);
39    }
40
41    /// Drop a session from all mappings.
42    ///
43    /// Returns true if the session existed and was dropped, otherwise returns false when the
44    /// session was known
45    pub fn drop(&mut self, session_id: u64) -> bool {
46        let Some(topic) = self.session_topic_map.remove(&session_id) else {
47            return false;
48        };
49        self.topic_session_map
50            .entry(topic.clone())
51            .and_modify(|sessions| {
52                sessions.remove(&session_id);
53            });
54        self.session_tx_map.remove(&session_id);
55        true
56    }
57
58    /// Get the topic for a session id.
59    ///
60    /// Returns None of the session id was not known.
61    pub fn topic(&self, session_id: u64) -> Option<&T> {
62        self.session_topic_map.get(&session_id)
63    }
64
65    /// Get ids for all sessions associated with the given topic.
66    pub fn sessions(&self, topic: &T) -> HashSet<u64> {
67        self.topic_session_map
68            .get(topic)
69            .cloned()
70            .unwrap_or_default()
71    }
72
73    /// Get a reference to a session sender.
74    pub fn sender(&self, session_id: u64) -> Option<&TX> {
75        self.session_tx_map.get(&session_id)
76    }
77
78    /// Get a mutable reference to a session sender.
79    pub fn sender_mut(&mut self, session_id: u64) -> Option<&mut TX> {
80        self.session_tx_map.get_mut(&session_id)
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use std::collections::HashSet;
87
88    use futures_channel::mpsc;
89    use p2panda_core::Topic;
90
91    use crate::manager::SessionTopicMap;
92
93    const SESSION1: u64 = 1;
94    const SESSION2: u64 = 2;
95    const SESSION3: u64 = 3;
96
97    #[test]
98    fn default_is_empty() {
99        let map: SessionTopicMap<Topic, ()> = SessionTopicMap::default();
100        assert!(map.session_tx_map.is_empty());
101        assert!(map.session_topic_map.is_empty());
102        assert!(map.topic_session_map.is_empty());
103    }
104
105    #[test]
106    fn insert_with_topic() {
107        let (tx, _rx) = mpsc::channel::<()>(128);
108        let mut map = SessionTopicMap::default();
109        let topic_a = Topic::random();
110
111        map.insert_with_topic(SESSION1, topic_a, tx.clone());
112
113        // Check session→topic mapping
114        assert_eq!(map.topic(SESSION1), Some(&topic_a));
115
116        // Check topic→session mapping
117        assert_eq!(map.sessions(&topic_a), HashSet::from_iter([SESSION1]));
118
119        // Channel should be retrievable
120        assert!(map.sender(SESSION1).is_some());
121    }
122
123    #[test]
124    fn drop_session() {
125        let (tx, _rx) = mpsc::channel::<()>(128);
126        let mut map = SessionTopicMap::default();
127        let topic_a = Topic::random();
128        let topic_b = Topic::random();
129
130        map.insert_with_topic(SESSION1, topic_a, tx.clone());
131        map.insert_with_topic(SESSION2, topic_a, tx.clone());
132        map.insert_with_topic(SESSION3, topic_b, tx);
133
134        // Drop one from topic A
135        assert!(map.drop(SESSION1));
136
137        // Should be removed from all mappings
138        assert!(map.topic(SESSION1).is_none());
139        assert!(!map.session_tx_map.contains_key(&SESSION1));
140
141        // Remaining sessions for topic A
142        let sessions = map.sessions(&topic_a);
143        assert_eq!(sessions, HashSet::from([SESSION2]));
144
145        // Dropping a non-existent session returns false
146        assert!(!map.drop(10));
147    }
148
149    #[test]
150    fn insert_multiple_sessions_same_topic() {
151        let (tx1, _rx1) = mpsc::channel::<()>(128);
152        let (tx2, _rx2) = mpsc::channel::<()>(128);
153        let mut map = SessionTopicMap::default();
154
155        let topic_a = Topic::random();
156
157        map.insert_with_topic(SESSION1, topic_a, tx1);
158        map.insert_with_topic(SESSION2, topic_a, tx2);
159
160        let sessions = map.sessions(&topic_a);
161        assert_eq!(sessions, HashSet::from([SESSION1, SESSION2]));
162    }
163}
p2panda_sync/manager/session_map.rs

p2panda_sync/manager/
session_map.rs
