server/src/main.rs

   1 extern crate lemmy_server;
   2 #[macro_use]
   3 extern crate diesel_migrations;
   4
   5 use actix::prelude::*;
   6 use actix_files::NamedFile;
   7 use actix_web::*;
   8 use actix_web_actors::ws;
   9 use lemmy_server::db::establish_connection;
  10 use lemmy_server::feeds;
  11 use lemmy_server::nodeinfo;
  12 use lemmy_server::websocket::server::*;
  13 use lemmy_server::Settings;
  14 use std::env;
  15 use std::time::{Duration, Instant};
  16
  17 embed_migrations!();
  18
  19 /// How often heartbeat pings are sent
  20 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
  21 /// How long before lack of client response causes a timeout
  22 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
  23
  24 /// Entry point for our route
  25 fn chat_route(
  26   req: HttpRequest,
  27   stream: web::Payload,
  28   chat_server: web::Data<Addr<ChatServer>>,
  29 ) -> Result<HttpResponse, Error> {
  30   ws::start(
  31     WSSession {
  32       cs_addr: chat_server.get_ref().to_owned(),
  33       id: 0,
  34       hb: Instant::now(),
  35       ip: req
  36         .connection_info()
  37         .remote()
  38         .unwrap_or("127.0.0.1:12345")
  39         .split(":")
  40         .next()
  41         .unwrap_or("127.0.0.1")
  42         .to_string(),
  43     },
  44     &req,
  45     stream,
  46   )
  47 }
  48
  49 struct WSSession {
  50   cs_addr: Addr<ChatServer>,
  51   /// unique session id
  52   id: usize,
  53   ip: String,
  54   /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
  55   /// otherwise we drop connection.
  56   hb: Instant,
  57 }
  58
  59 impl Actor for WSSession {
  60   type Context = ws::WebsocketContext<Self>;
  61
  62   /// Method is called on actor start.
  63   /// We register ws session with ChatServer
  64   fn started(&mut self, ctx: &mut Self::Context) {
  65     // we'll start heartbeat process on session start.
  66     self.hb(ctx);
  67
  68     // register self in chat server. `AsyncContext::wait` register
  69     // future within context, but context waits until this future resolves
  70     // before processing any other events.
  71     // across all routes within application
  72     let addr = ctx.address();
  73     self
  74       .cs_addr
  75       .send(Connect {
  76         addr: addr.recipient(),
  77         ip: self.ip.to_owned(),
  78       })
  79       .into_actor(self)
  80       .then(|res, act, ctx| {
  81         match res {
  82           Ok(res) => act.id = res,
  83           // something is wrong with chat server
  84           _ => ctx.stop(),
  85         }
  86         fut::ok(())
  87       })
  88       .wait(ctx);
  89   }
  90
  91   fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
  92     // notify chat server
  93     self.cs_addr.do_send(Disconnect {
  94       id: self.id,
  95       ip: self.ip.to_owned(),
  96     });
  97     Running::Stop
  98   }
  99 }
 100
 101 /// Handle messages from chat server, we simply send it to peer websocket
 102 /// These are room messages, IE sent to others in the room
 103 impl Handler<WSMessage> for WSSession {
 104   type Result = ();
 105
 106   fn handle(&mut self, msg: WSMessage, ctx: &mut Self::Context) {
 107     // println!("id: {} msg: {}", self.id, msg.0);
 108     ctx.text(msg.0);
 109   }
 110 }
 111
 112 /// WebSocket message handler
 113 impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
 114   fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
 115     // println!("WEBSOCKET MESSAGE: {:?} from id: {}", msg, self.id);
 116     match msg {
 117       ws::Message::Ping(msg) => {
 118         self.hb = Instant::now();
 119         ctx.pong(&msg);
 120       }
 121       ws::Message::Pong(_) => {
 122         self.hb = Instant::now();
 123       }
 124       ws::Message::Text(text) => {
 125         let m = text.trim().to_owned();
 126         println!("WEBSOCKET MESSAGE: {:?} from id: {}", &m, self.id);
 127
 128         self
 129           .cs_addr
 130           .send(StandardMessage {
 131             id: self.id,
 132             msg: m,
 133           })
 134           .into_actor(self)
 135           .then(|res, _, ctx| {
 136             match res {
 137               Ok(res) => ctx.text(res),
 138               Err(e) => {
 139                 eprintln!("{}", &e);
 140               }
 141             }
 142             fut::ok(())
 143           })
 144           .wait(ctx);
 145       }
 146       ws::Message::Binary(_bin) => println!("Unexpected binary"),
 147       ws::Message::Close(_) => {
 148         ctx.stop();
 149       }
 150       _ => {}
 151     }
 152   }
 153 }
 154
 155 impl WSSession {
 156   /// helper method that sends ping to client every second.
 157   ///
 158   /// also this method checks heartbeats from client
 159   fn hb(&self, ctx: &mut ws::WebsocketContext<Self>) {
 160     ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
 161       // check client heartbeats
 162       if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
 163         // heartbeat timed out
 164         println!("Websocket Client heartbeat failed, disconnecting!");
 165
 166         // notify chat server
 167         act.cs_addr.do_send(Disconnect {
 168           id: act.id,
 169           ip: act.ip.to_owned(),
 170         });
 171
 172         // stop actor
 173         ctx.stop();
 174
 175         // don't try to send a ping
 176         return;
 177       }
 178
 179       ctx.ping("");
 180     });
 181   }
 182 }
 183
 184 fn main() {
 185   let _ = env_logger::init();
 186   let sys = actix::System::new("lemmy");
 187
 188   // Run the migrations from code
 189   let conn = establish_connection();
 190   embedded_migrations::run(&conn).unwrap();
 191
 192   // Start chat server actor in separate thread
 193   let server = ChatServer::default().start();
 194
 195   let settings = Settings::get();
 196
 197   // Create Http server with websocket support
 198   HttpServer::new(move || {
 199     App::new()
 200       .data(server.clone())
 201       // Front end routes
 202       .service(actix_files::Files::new("/static", front_end_dir()))
 203       .route("/", web::get().to(index))
 204       .route(
 205         "/home/type/{type}/sort/{sort}/page/{page}",
 206         web::get().to(index),
 207       )
 208       .route("/login", web::get().to(index))
 209       .route("/create_post", web::get().to(index))
 210       .route("/create_community", web::get().to(index))
 211       .route("/communities/page/{page}", web::get().to(index))
 212       .route("/communities", web::get().to(index))
 213       .route("/post/{id}/comment/{id2}", web::get().to(index))
 214       .route("/post/{id}", web::get().to(index))
 215       .route("/c/{name}/sort/{sort}/page/{page}", web::get().to(index))
 216       .route("/c/{name}", web::get().to(index))
 217       .route("/community/{id}", web::get().to(index))
 218       .route(
 219         "/u/{username}/view/{view}/sort/{sort}/page/{page}",
 220         web::get().to(index),
 221       )
 222       .route("/u/{username}", web::get().to(index))
 223       .route("/user/{id}", web::get().to(index))
 224       .route("/inbox", web::get().to(index))
 225       .route("/modlog/community/{community_id}", web::get().to(index))
 226       .route("/modlog", web::get().to(index))
 227       .route("/setup", web::get().to(index))
 228       .route(
 229         "/search/q/{q}/type/{type}/sort/{sort}/page/{page}",
 230         web::get().to(index),
 231       )
 232       .route("/search", web::get().to(index))
 233       .route("/sponsors", web::get().to(index))
 234       .route("/password_change/{token}", web::get().to(index))
 235       // Websocket
 236       .service(web::resource("/api/v1/ws").to(chat_route))
 237       // NodeInfo
 238       .route("/nodeinfo/2.0.json", web::get().to(nodeinfo::node_info))
 239       .route(
 240         "/.well-known/nodeinfo",
 241         web::get().to(nodeinfo::node_info_well_known),
 242       )
 243       // RSS
 244       .route("/feeds/{type}/{name}.xml", web::get().to(feeds::get_feed))
 245       .route("/feeds/all.xml", web::get().to(feeds::get_all_feed))
 246   })
 247   .bind((settings.bind, settings.port))
 248   .unwrap()
 249   .start();
 250
 251   println!("Started http server at {}:{}", settings.bind, settings.port);
 252   let _ = sys.run();
 253 }
 254
 255 fn index() -> Result<NamedFile, actix_web::error::Error> {
 256   Ok(NamedFile::open(front_end_dir() + "/index.html")?)
 257 }
 258
 259 fn front_end_dir() -> String {
 260   env::var("LEMMY_FRONT_END_DIR").unwrap_or("../ui/dist".to_string())
 261 }