2 chat_server::{ChatServer, SessionInfo},
8 GetCommunityUsersOnline,
17 SendCommunityRoomMessage,
25 use actix::{Actor, Context, Handler, ResponseFuture};
26 use lemmy_db_schema::utils::naive_now;
27 use lemmy_utils::ConnectionId;
28 use opentelemetry::trace::TraceContextExt;
31 use tracing::{error, info};
32 use tracing_opentelemetry::OpenTelemetrySpanExt;
34 /// Make actor from `ChatServer`
35 impl Actor for ChatServer {
36 /// We are going to use simple Context, we just need ability to communicate
37 /// with other actors.
38 type Context = Context<Self>;
41 /// Handler for Connect message.
43 /// Register new session and assign unique id to this session
44 impl Handler<Connect> for ChatServer {
45 type Result = ConnectionId;
47 fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
48 // register session with random id
49 let id = self.rng.gen::<usize>();
50 info!("{} joined", &msg.ip);
64 /// Handler for Disconnect message.
65 impl Handler<Disconnect> for ChatServer {
68 fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
69 // Remove connections from sessions and all 3 scopes
70 if self.sessions.remove(&msg.id).is_some() {
71 for sessions in self.user_rooms.values_mut() {
72 sessions.remove(&msg.id);
75 for sessions in self.post_rooms.values_mut() {
76 sessions.remove(&msg.id);
79 for sessions in self.community_rooms.values_mut() {
80 sessions.remove(&msg.id);
86 fn root_span() -> tracing::Span {
87 let span = tracing::info_span!(
90 trace_id = tracing::field::Empty,
93 let trace_id = span.context().span().span_context().trace_id().to_string();
94 span.record("trace_id", &tracing::field::display(trace_id));
100 /// Handler for Message message.
101 impl Handler<StandardMessage> for ChatServer {
102 type Result = ResponseFuture<Result<String, std::convert::Infallible>>;
104 fn handle(&mut self, msg: StandardMessage, ctx: &mut Context<Self>) -> Self::Result {
105 use tracing::Instrument;
106 let fut = self.parse_json_message(msg, ctx);
107 let span = root_span();
113 // info!("Message Sent: {}", m);
117 error!("Error during message handling {}", e);
120 .unwrap_or_else(|_| String::from(r#"{"error":"failed to serialize json"}"#)),
130 impl<OP, Response> Handler<SendAllMessage<OP, Response>> for ChatServer
132 OP: OperationType + ToString,
137 fn handle(&mut self, msg: SendAllMessage<OP, Response>, _: &mut Context<Self>) {
139 .send_all_message(&msg.op, &msg.response, msg.websocket_id)
144 impl<OP, Response> Handler<SendUserRoomMessage<OP, Response>> for ChatServer
146 OP: OperationType + ToString,
151 fn handle(&mut self, msg: SendUserRoomMessage<OP, Response>, _: &mut Context<Self>) {
153 .send_user_room_message(
156 msg.local_recipient_id,
163 impl<OP, Response> Handler<SendCommunityRoomMessage<OP, Response>> for ChatServer
165 OP: OperationType + ToString,
170 fn handle(&mut self, msg: SendCommunityRoomMessage<OP, Response>, _: &mut Context<Self>) {
172 .send_community_room_message(&msg.op, &msg.response, msg.community_id, msg.websocket_id)
177 impl<Response> Handler<SendModRoomMessage<Response>> for ChatServer
183 fn handle(&mut self, msg: SendModRoomMessage<Response>, _: &mut Context<Self>) {
185 .send_mod_room_message(&msg.op, &msg.response, msg.community_id, msg.websocket_id)
190 impl<OP> Handler<SendPost<OP>> for ChatServer
192 OP: OperationType + ToString,
196 fn handle(&mut self, msg: SendPost<OP>, _: &mut Context<Self>) {
197 self.send_post(&msg.op, &msg.post, msg.websocket_id).ok();
201 impl<OP> Handler<SendComment<OP>> for ChatServer
203 OP: OperationType + ToString,
207 fn handle(&mut self, msg: SendComment<OP>, _: &mut Context<Self>) {
209 .send_comment(&msg.op, &msg.comment, msg.websocket_id)
214 impl Handler<JoinUserRoom> for ChatServer {
217 fn handle(&mut self, msg: JoinUserRoom, _: &mut Context<Self>) {
218 self.join_user_room(msg.local_user_id, msg.id).ok();
222 impl Handler<JoinCommunityRoom> for ChatServer {
225 fn handle(&mut self, msg: JoinCommunityRoom, _: &mut Context<Self>) {
226 self.join_community_room(msg.community_id, msg.id).ok();
230 impl Handler<JoinModRoom> for ChatServer {
233 fn handle(&mut self, msg: JoinModRoom, _: &mut Context<Self>) {
234 self.join_mod_room(msg.community_id, msg.id).ok();
238 impl Handler<JoinPostRoom> for ChatServer {
241 fn handle(&mut self, msg: JoinPostRoom, _: &mut Context<Self>) {
242 self.join_post_room(msg.post_id, msg.id).ok();
246 impl Handler<GetUsersOnline> for ChatServer {
249 fn handle(&mut self, _msg: GetUsersOnline, _: &mut Context<Self>) -> Self::Result {
254 impl Handler<GetPostUsersOnline> for ChatServer {
257 fn handle(&mut self, msg: GetPostUsersOnline, _: &mut Context<Self>) -> Self::Result {
258 if let Some(users) = self.post_rooms.get(&msg.post_id) {
266 impl Handler<GetCommunityUsersOnline> for ChatServer {
269 fn handle(&mut self, msg: GetCommunityUsersOnline, _: &mut Context<Self>) -> Self::Result {
270 if let Some(users) = self.community_rooms.get(&msg.community_id) {
278 impl Handler<CaptchaItem> for ChatServer {
281 fn handle(&mut self, msg: CaptchaItem, _: &mut Context<Self>) {
282 self.captchas.push(msg);
286 impl Handler<CheckCaptcha> for ChatServer {
289 fn handle(&mut self, msg: CheckCaptcha, _: &mut Context<Self>) -> Self::Result {
290 // Remove all the ones that are past the expire time
291 self.captchas.retain(|x| x.expires.gt(&naive_now()));
296 .any(|r| r.uuid == msg.uuid && r.answer.to_lowercase() == msg.answer.to_lowercase());
298 // Remove this uuid so it can't be re-checked (Checks only work once)
299 self.captchas.retain(|x| x.uuid != msg.uuid);