2 chat_server::ChatServer,
3 messages::{Connect, Disconnect, StandardMessage, WsMessage},
7 use actix_web::{web, Error, HttpRequest, HttpResponse};
8 use actix_web_actors::ws;
9 use lemmy_utils::{rate_limit::RateLimitCell, utils::get_ip, ConnectionId, IpAddr};
10 use std::time::{Duration, Instant};
11 use tracing::{debug, error, info};
13 /// How often heartbeat pings are sent
14 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
15 /// How long before lack of client response causes a timeout
16 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
18 /// Entry point for our route
19 pub async fn chat_route(
22 context: web::Data<LemmyContext>,
23 rate_limiter: web::Data<RateLimitCell>,
24 ) -> Result<HttpResponse, Error> {
27 cs_addr: context.chat_server().clone(),
30 ip: get_ip(&req.connection_info()),
31 rate_limiter: rate_limiter.as_ref().clone(),
39 cs_addr: Addr<ChatServer>,
43 /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
44 /// otherwise we drop connection.
46 /// A rate limiter for websocket joins
47 rate_limiter: RateLimitCell,
50 impl Actor for WsSession {
51 type Context = ws::WebsocketContext<Self>;
53 /// Method is called on actor start.
54 /// We register ws session with ChatServer
55 fn started(&mut self, ctx: &mut Self::Context) {
56 // we'll start heartbeat process on session start.
59 // register self in chat server. `AsyncContext::wait` register
60 // future within context, but context waits until this future resolves
61 // before processing any other events.
62 // across all routes within application
63 let addr = ctx.address();
65 if !self.rate_limit_check(ctx) {
72 addr: addr.recipient(),
76 .then(|res, act, ctx| {
78 Ok(res) => act.id = res,
79 // something is wrong with chat server
87 fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
89 self.cs_addr.do_send(Disconnect {
97 /// Handle messages from chat server, we simply send it to peer websocket
98 /// These are room messages, IE sent to others in the room
99 impl Handler<WsMessage> for WsSession {
102 fn handle(&mut self, msg: WsMessage, ctx: &mut Self::Context) {
107 /// WebSocket message handler
108 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
109 fn handle(&mut self, result: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
110 if !self.rate_limit_check(ctx) {
114 let message = match result {
122 ws::Message::Ping(msg) => {
123 self.hb = Instant::now();
126 ws::Message::Pong(_) => {
127 self.hb = Instant::now();
129 ws::Message::Text(text) => {
130 let m = text.trim().to_owned();
134 .send(StandardMessage {
139 .then(|res, _, ctx| {
141 Ok(Ok(res)) => ctx.text(res),
143 Err(e) => error!("{}", &e),
145 actix::fut::ready(())
149 ws::Message::Binary(_bin) => info!("Unexpected binary"),
150 ws::Message::Close(_) => {
159 /// helper method that sends ping to client every second.
161 /// also this method checks heartbeats from client
162 fn hb(ctx: &mut ws::WebsocketContext<Self>) {
163 ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
164 // check client heartbeats
165 if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
166 // heartbeat timed out
167 debug!("Websocket Client heartbeat failed, disconnecting!");
169 // notify chat server
170 act.cs_addr.do_send(Disconnect {
178 // don't try to send a ping
186 /// Check the rate limit, and stop the ctx if it fails
187 fn rate_limit_check(&mut self, ctx: &mut ws::WebsocketContext<Self>) -> bool {
188 let check = self.rate_limiter.message().check(self.ip.clone());
190 debug!("Websocket join with IP: {} has been rate limited.", self.ip);