1 extern crate lemmy_server;
2 #[macro_use] extern crate diesel_migrations;
4 use std::time::{Instant, Duration};
6 use lemmy_server::actix::*;
7 use lemmy_server::actix_web::server::HttpServer;
8 use lemmy_server::actix_web::{ws, App, Error, HttpRequest, HttpResponse, fs::NamedFile, fs};
9 use lemmy_server::websocket::server::*;
10 use lemmy_server::establish_connection;
14 /// How often heartbeat pings are sent
15 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
16 /// How long before lack of client response causes a timeout
17 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
19 /// This is our websocket route state, this state is shared with all route
20 /// instances via `HttpContext::state()`
21 struct WsChatSessionState {
22 addr: Addr<ChatServer>,
25 /// Entry point for our route
26 fn chat_route(req: &HttpRequest<WsChatSessionState>) -> Result<HttpResponse, Error> {
32 ip: req.connection_info().remote().unwrap_or("127.0.0.1:12345").split(":").next().unwrap_or("127.0.0.1").to_string()
41 /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
42 /// otherwise we drop connection.
46 impl Actor for WSSession {
47 type Context = ws::WebsocketContext<Self, WsChatSessionState>;
49 /// Method is called on actor start.
50 /// We register ws session with ChatServer
51 fn started(&mut self, ctx: &mut Self::Context) {
52 // we'll start heartbeat process on session start.
55 // register self in chat server. `AsyncContext::wait` register
56 // future within context, but context waits until this future resolves
57 // before processing any other events.
58 // HttpContext::state() is instance of WsChatSessionState, state is shared
59 // across all routes within application
60 let addr = ctx.address();
64 addr: addr.recipient(),
65 ip: self.ip.to_owned(),
68 .then(|res, act, ctx| {
70 Ok(res) => act.id = res,
71 // something is wrong with chat server
79 fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
81 ctx.state().addr.do_send(Disconnect {
83 ip: self.ip.to_owned(),
89 /// Handle messages from chat server, we simply send it to peer websocket
90 /// These are room messages, IE sent to others in the room
91 impl Handler<WSMessage> for WSSession {
94 fn handle(&mut self, msg: WSMessage, ctx: &mut Self::Context) {
95 // println!("id: {} msg: {}", self.id, msg.0);
100 /// WebSocket message handler
101 impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
102 fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
103 // println!("WEBSOCKET MESSAGE: {:?} from id: {}", msg, self.id);
105 ws::Message::Ping(msg) => {
106 self.hb = Instant::now();
109 ws::Message::Pong(_) => {
110 self.hb = Instant::now();
112 ws::Message::Text(text) => {
113 let m = text.trim().to_owned();
117 .send(StandardMessage {
122 .then(|res, _, ctx| {
124 Ok(res) => ctx.text(res),
130 // Ok(res) => ctx.text(res),
131 // // something is wrong with chat server
137 // we check for /sss type of messages
138 // if m.starts_with('/') {
139 // let v: Vec<&str> = m.splitn(2, ' ').collect();
142 // // Send ListRooms message to chat server and wait for
144 // println!("List rooms");
149 // .then(|res, _, ctx| {
152 // for room in rooms {
156 // _ => println!("Something is wrong"),
161 // .wait(ctx) pauses all events in context,
162 // so actor wont receive any new messages until it get list
167 // self.room = v[1].to_owned();
168 // ctx.state().addr.do_send(Join {
170 // name: self.room.clone(),
173 // ctx.text("joined");
175 // ctx.text("!!! room name is required");
180 // self.name = Some(v[1].to_owned());
182 // ctx.text("!!! name is required");
185 // _ => ctx.text(format!("!!! unknown command: {:?}", m)),
188 // let msg = if let Some(ref name) = self.name {
189 // format!("{}: {}", name, m)
193 // send message to chat server
194 // ctx.state().addr.do_send(ClientMessage {
197 // room: self.room.clone(),
201 ws::Message::Binary(_bin) => println!("Unexpected binary"),
202 ws::Message::Close(_) => {
210 /// helper method that sends ping to client every second.
212 /// also this method checks heartbeats from client
213 fn hb(&self, ctx: &mut ws::WebsocketContext<Self, WsChatSessionState>) {
214 ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
215 // check client heartbeats
216 if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
217 // heartbeat timed out
218 println!("Websocket Client heartbeat failed, disconnecting!");
220 // notify chat server
223 .do_send(Disconnect { id: act.id, ip: act.ip.to_owned() });
228 // don't try to send a ping
238 let _ = env_logger::init();
239 let sys = actix::System::new("lemmy");
241 // Run the migrations from code
242 let conn = establish_connection();
243 embedded_migrations::run(&conn).unwrap();
245 // Start chat server actor in separate thread
246 let server = Arbiter::start(|_| ChatServer::default());
248 // Create Http server with websocket support
249 HttpServer::new(move || {
250 // Websocket sessions state
251 let state = WsChatSessionState {
252 addr: server.clone(),
255 App::with_state(state)
256 // redirect to websocket.html
257 // .resource("/", |r| r.method(http::Method::GET).f(|_| {
258 // HttpResponse::Found()
259 // .header("LOCATION", "/static/websocket.html")
262 .resource("/service/ws", |r| r.route().f(chat_route))
264 .resource("/", |r| r.route().f(index))
267 fs::StaticFiles::new(front_end_dir()).unwrap()
270 }).bind("0.0.0.0:8536")
274 println!("Started http server: 0.0.0.0:8536");
278 fn index(_req: &HttpRequest<WsChatSessionState>) -> Result<NamedFile, actix_web::error::Error> {
279 Ok(NamedFile::open(front_end_dir() + "/index.html")?)
282 fn front_end_dir() -> String {
283 env::var("LEMMY_FRONT_END_DIR").unwrap_or("../ui/dist".to_string())