pondsocket 0.1.1

Channel-based realtime server framework with distributed pubsub support.
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

use std::collections::HashMap;
use std::sync::Arc;

use tokio::sync::RwLock;

use crate::channel::{
    Channel, ChannelConfig, EventHandler, EventRegistration, LeaveHandler, OutgoingHandler,
    OutgoingRegistration,
};
use crate::errors::Result;
use crate::pubsub::PubSub;
use crate::types::Options;

pub struct Lobby {
    path: String,
    endpoint_path: String,
    options: Options,
    pubsub: Option<Arc<dyn PubSub>>,
    channels: RwLock<HashMap<String, Arc<Channel>>>,
    event_handlers: RwLock<Vec<EventRegistration>>,
    outgoing_handlers: RwLock<Vec<OutgoingRegistration>>,
    leave_handler: RwLock<Option<Arc<dyn LeaveHandler>>>,
}

impl Lobby {
    pub fn new(
        path: impl Into<String>,
        endpoint_path: impl Into<String>,
        options: Options,
        pubsub: Option<Arc<dyn PubSub>>,
    ) -> Arc<Self> {
        Arc::new(Self {
            path: path.into(),
            endpoint_path: endpoint_path.into(),
            options,
            pubsub,
            channels: RwLock::new(HashMap::new()),
            event_handlers: RwLock::new(Vec::new()),
            outgoing_handlers: RwLock::new(Vec::new()),
            leave_handler: RwLock::new(None),
        })
    }

    pub fn path(&self) -> &str {
        &self.path
    }

    pub async fn on_message<H>(&self, pattern: impl Into<String>, handler: H)
    where
        H: EventHandler + 'static,
    {
        self.event_handlers.write().await.push(EventRegistration {
            pattern: pattern.into(),
            handler: Arc::new(handler),
        });
    }

    pub async fn on_outgoing<H>(&self, pattern: impl Into<String>, handler: H)
    where
        H: OutgoingHandler + 'static,
    {
        self.outgoing_handlers
            .write()
            .await
            .push(OutgoingRegistration {
                pattern: pattern.into(),
                handler: Arc::new(handler),
            });
    }

    pub async fn on_leave<H>(&self, handler: H)
    where
        H: LeaveHandler + 'static,
    {
        *self.leave_handler.write().await = Some(Arc::new(handler));
    }

    pub async fn get_channel(&self, name: &str) -> Option<Arc<Channel>> {
        self.channels.read().await.get(name).cloned()
    }

    pub async fn get_or_create_channel(&self, name: &str) -> Result<Arc<Channel>> {
        if let Some(channel) = self.get_channel(name).await {
            return Ok(channel);
        }
        let mut channels = self.channels.write().await;
        if let Some(channel) = channels.get(name) {
            return Ok(channel.clone());
        }
        let channel = Channel::new(ChannelConfig {
            name: name.to_owned(),
            endpoint_path: self.endpoint_path.clone(),
            options: self.options.clone(),
            pubsub: self.pubsub.clone(),
            event_handlers: self.event_handlers.read().await.clone(),
            outgoing_handlers: self.outgoing_handlers.read().await.clone(),
            leave_handler: self.leave_handler.read().await.clone(),
        });
        channel.start().await?;
        channels.insert(name.to_owned(), channel.clone());
        Ok(channel)
    }

    pub async fn list_channels(&self) -> Vec<Arc<Channel>> {
        self.channels.read().await.values().cloned().collect()
    }

    pub async fn close(&self) -> Result<()> {
        let channels = self.list_channels().await;
        for channel in channels {
            let _ = channel.close().await;
        }
        self.channels.write().await.clear();
        Ok(())
    }
}