4 chat_server::ChatServer,
5 messages::{Connect, Disconnect, StandardMessage, WsMessage},
9 use actix_web::{web, Error, HttpRequest, HttpResponse};
10 use actix_web_actors::ws;
11 use lemmy_utils::{rate_limit::RateLimitCell, utils::get_ip, ConnectionId, IpAddr};
12 use std::time::{Duration, Instant};
13 use tracing::{debug, error, info};
15 /// How often heartbeat pings are sent
16 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
17 /// How long before lack of client response causes a timeout
18 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
20 /// Entry point for our route
21 pub async fn chat_route(
24 context: web::Data<LemmyContext>,
25 rate_limiter: web::Data<RateLimitCell>,
26 ) -> Result<HttpResponse, Error> {
29 cs_addr: context.chat_server().clone(),
32 ip: get_ip(&req.connection_info()),
33 rate_limiter: rate_limiter.as_ref().clone(),
41 cs_addr: Addr<ChatServer>,
45 /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
46 /// otherwise we drop connection.
48 /// A rate limiter for websocket joins
49 rate_limiter: RateLimitCell,
52 impl Actor for WsSession {
53 type Context = ws::WebsocketContext<Self>;
55 /// Method is called on actor start.
56 /// We register ws session with ChatServer
57 fn started(&mut self, ctx: &mut Self::Context) {
58 // we'll start heartbeat process on session start.
61 // register self in chat server. `AsyncContext::wait` register
62 // future within context, but context waits until this future resolves
63 // before processing any other events.
64 // across all routes within application
65 let addr = ctx.address();
67 if !self.rate_limit_check(ctx) {
74 addr: addr.recipient(),
78 .then(|res, act, ctx| {
80 Ok(res) => act.id = res,
81 // something is wrong with chat server
89 fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
91 self.cs_addr.do_send(Disconnect {
99 /// Handle messages from chat server, we simply send it to peer websocket
100 /// These are room messages, IE sent to others in the room
101 impl Handler<WsMessage> for WsSession {
104 fn handle(&mut self, msg: WsMessage, ctx: &mut Self::Context) {
109 /// WebSocket message handler
110 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsSession {
111 fn handle(&mut self, result: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
112 if !self.rate_limit_check(ctx) {
116 let message = match result {
124 ws::Message::Ping(msg) => {
125 self.hb = Instant::now();
128 ws::Message::Pong(_) => {
129 self.hb = Instant::now();
131 ws::Message::Text(text) => {
132 let m = text.trim().to_owned();
136 .send(StandardMessage {
141 .then(|res, _, ctx| {
143 Ok(Ok(res)) => ctx.text(res),
145 Err(e) => error!("{}", &e),
147 actix::fut::ready(())
151 ws::Message::Binary(_bin) => info!("Unexpected binary"),
152 ws::Message::Close(_) => {
161 /// helper method that sends ping to client every second.
163 /// also this method checks heartbeats from client
164 fn hb(ctx: &mut ws::WebsocketContext<Self>) {
165 ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
166 // check client heartbeats
167 if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
168 // heartbeat timed out
169 debug!("Websocket Client heartbeat failed, disconnecting!");
171 // notify chat server
172 act.cs_addr.do_send(Disconnect {
180 // don't try to send a ping
188 /// Check the rate limit, and stop the ctx if it fails
189 fn rate_limit_check(&mut self, ctx: &mut ws::WebsocketContext<Self>) -> bool {
190 let check = self.rate_limiter.message().check(self.ip.clone());
192 debug!("Websocket join with IP: {} has been rate limited.", self.ip);