pondsocket 0.1.4

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
117
118
119
120
121
122
123
124
125
126
127

use std::sync::Arc;

use pondsocket_common::PondAssigns;
use serde_json::Value;

use crate::channel::Channel;
use crate::errors::{Result, bad_request};
use crate::transport::Transport;
use crate::types::{Event, Route};

pub struct JoinContext {
    pub channel: Arc<Channel>,
    pub event: Event,
    pub transport: Arc<dyn Transport>,
    pub route: Route,
    responded: bool,
    accepted: bool,
}

impl JoinContext {
    pub fn new(
        channel: Arc<Channel>,
        event: Event,
        transport: Arc<dyn Transport>,
        route: Route,
    ) -> Self {
        Self {
            channel,
            event,
            transport,
            route,
            responded: false,
            accepted: false,
        }
    }

    pub fn has_responded(&self) -> bool {
        self.responded
    }

    pub fn is_accepted(&self) -> bool {
        self.accepted
    }

    pub async fn accept(&mut self, assigns: PondAssigns) -> Result<&mut Self> {
        self.ensure_not_responded()?;
        for (key, value) in assigns {
            self.transport.set_assign(&key, value).await;
        }
        self.responded = true;
        self.accepted = true;
        self.channel.add_user(self.transport.clone()).await?;
        self.channel
            .send_acknowledge(&self.event.request_id, self.transport.id())
            .await?;
        Ok(self)
    }

    pub async fn decline(&mut self, code: u16, message: impl Into<String>) -> Result<()> {
        self.ensure_not_responded()?;
        self.responded = true;
        self.accepted = false;
        self.channel
            .send_unauthorized(
                &self.event.request_id,
                self.transport.id(),
                code,
                &message.into(),
            )
            .await
    }

    pub async fn reply(&self, event: &str, payload: Value) -> Result<()> {
        self.ensure_accepted()?;
        self.channel
            .send_system(
                event,
                payload,
                &self.event.request_id,
                &[self.transport.id().to_owned()],
            )
            .await
    }

    pub async fn track(&self, presence: Value) -> Result<()> {
        self.ensure_accepted()?;
        self.channel
            .track_presence(self.transport.id(), presence)
            .await
    }

    pub async fn broadcast(&self, event: &str, payload: Value) -> Result<()> {
        self.ensure_accepted()?;
        self.channel.broadcast(event, payload).await
    }

    pub async fn set_assign(&self, key: &str, value: Value) -> Result<()> {
        if self.accepted {
            self.channel
                .update_assign(self.transport.id(), key, value)
                .await
        } else {
            self.transport.set_assign(key, value).await;
            Ok(())
        }
    }

    fn ensure_not_responded(&self) -> Result<()> {
        if self.responded {
            return Err(bad_request(
                &self.channel.name,
                "already responded to join request",
            ));
        }
        Ok(())
    }

    fn ensure_accepted(&self) -> Result<()> {
        if !self.accepted {
            return Err(bad_request(
                &self.channel.name,
                "join request has not been accepted",
            ));
        }
        Ok(())
    }
}