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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
/*
This file is part of keks-meet (https://codeberg.org/metamuffin/keks-meet)
which is licensed under the GNU Affero General Public License (version 3); see /COPYING.
Copyright (C) 2023 metamuffin <metamuffin@disroot.org>
*/
use crate::{
idgen::IdGenerator,
protocol::{ClientboundPacket, ServerboundPacket},
};
use futures_util::{stream::SplitStream, StreamExt};
use log::{debug, error, warn};
use serde::{Deserialize, Serialize};
use std::{
collections::{HashMap, HashSet},
sync::{Arc, LazyLock},
};
use tokio::sync::{mpsc::Sender, RwLock};
use warp::ws::WebSocket;
static CLIENTS: LazyLock<RwLock<HashMap<Client, Sender<ClientboundPacket>>>> =
LazyLock::new(|| Default::default());
#[repr(transparent)]
#[derive(Debug, Hash, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct Client(u64);
#[derive(Default)]
pub struct State {
idgen: IdGenerator,
rooms: RwLock<HashMap<String, Arc<Room>>>,
}
#[derive(Debug, Default)]
pub struct Room {
pub users: RwLock<HashSet<Client>>,
}
#[derive(Debug, Default)]
pub struct ClientState {
current_room: Option<Arc<Room>>,
}
impl State {
pub async fn connect(&self, rx: SplitStream<WebSocket>, tx: Sender<ClientboundPacket>) {
debug!("new client connected");
let client = Client(self.idgen.generate().await);
CLIENTS.write().await.insert(client, tx);
self.connect_inner(client, rx).await;
CLIENTS.write().await.remove(&client);
}
async fn connect_inner(&self, client: Client, mut rx: SplitStream<WebSocket>) {
let mut cstate = ClientState::default();
client
.send(ClientboundPacket::Init {
your_id: client,
version: format!("keks-meet {}", env!("CARGO_PKG_VERSION")),
})
.await;
while let Some(result) = rx.next().await {
let msg = match result {
Ok(msg) => msg,
Err(e) => {
error!("websocket error: {e}");
break;
}
};
if let Ok(s) = msg.to_str() {
let packet = match serde_json::from_str::<ServerboundPacket>(s) {
Ok(p) => p,
Err(e) => {
error!("client sent invalid packet: {e:?}");
break;
}
};
debug!("<- {packet:?}");
self.on_recv(client, &mut cstate, packet).await;
}
}
if let Some(room) = cstate.current_room {
room.leave(client).await;
// TODO dont leak room
}
}
async fn on_recv(&self, client: Client, cstate: &mut ClientState, packet: ServerboundPacket) {
match packet {
ServerboundPacket::Ping => (),
ServerboundPacket::Join { hash } => {
if let Some(room) = &cstate.current_room {
room.leave(client).await;
// TODO dont leak room
// if room.should_remove().await {
// self.rooms.write().await.remove(üw);
// }
}
if let Some(hash) = hash {
let room = self.rooms.write().await.entry(hash).or_default().clone();
room.join(client).await;
cstate.current_room = Some(room.clone())
} else {
cstate.current_room = None
}
}
ServerboundPacket::Relay { recipient, message } => {
if let Some(room) = &cstate.current_room {
let packet = ClientboundPacket::Message {
sender: client,
message,
};
if let Some(recipient) = recipient {
room.send_to_client(recipient, packet).await;
} else {
room.broadcast(Some(client), packet).await;
}
}
}
ServerboundPacket::WatchRooms(_) => todo!(),
}
}
}
impl Client {
pub async fn send(&self, packet: ClientboundPacket) {
if let Some(s) = CLIENTS.read().await.get(&self) {
s.send(packet).await.unwrap();
} else {
warn!("invalid recipient {self:?}")
}
}
}
impl Room {
pub async fn join(&self, client: Client) {
debug!("client join {client:?}");
self.users.write().await.insert(client);
// send join of this client to all clients
self.broadcast(Some(client), ClientboundPacket::ClientJoin { id: client })
.await;
// send join of all other clients to this one
for rc in self.users.read().await.iter() {
self.send_to_client(client, ClientboundPacket::ClientJoin { id: *rc })
.await;
}
}
pub async fn leave(&self, client: Client) {
debug!("client leave {client:?}");
for c in self.users.read().await.iter() {
if *c != client {
self.send_to_client(*c, ClientboundPacket::ClientLeave { id: client })
.await;
}
}
self.users.write().await.remove(&client);
self.broadcast(Some(client), ClientboundPacket::ClientLeave { id: client })
.await;
}
pub async fn broadcast(&self, sender: Option<Client>, packet: ClientboundPacket) {
for c in self.users.read().await.iter() {
if sender != Some(*c) {
c.send(packet.clone()).await;
}
}
}
pub async fn send_to_client(&self, recipient: Client, packet: ClientboundPacket) {
if let Some(c) = self.users.read().await.get(&recipient) {
c.send(packet).await;
}
}
pub async fn should_remove(&self) -> bool {
self.users.read().await.len() == 0
}
}
|
