Merge branch 'Jonnnh-upgrade-actor-web-1.0.x' into dev

server/Cargo.toml

@@ -6,22 +6,23 @@ edition = "2018"
 
 [dependencies]
 diesel = { version = "1.4.2", features = ["postgres","chrono"] }
-diesel_migrations = "*"
+diesel_migrations = "1.4.0"
-dotenv = "0.9.0"
+dotenv = "0.14.1"
-bcrypt = "0.3"
+bcrypt = "0.5.0"
-activitypub = "0.1.4"
+activitypub = "0.1.5"
-chrono = { version = "0.4", features = ["serde"] }
+chrono = { version = "0.4.7", features = ["serde"] }
 failure = "0.1.5"
-serde_json = "*"
+serde_json = "1.0.40"
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "1.0.94", features = ["derive"] }
-actix = "*"
+actix = "0.8.3"
-actix-web = "*"
+actix-web = "1.0"
-env_logger = "*"
+actix-files = "0.1.3"
-rand = "0.6.5"
+actix-web-actors = "1.0"
-strum = "0.14.0"
+env_logger = "0.6.2"
-strum_macros = "0.14.0"
+rand = "0.7.0"
-jsonwebtoken = "*"
+strum = "0.15.0"
-regex = "*"
+strum_macros = "0.15.0"
-lazy_static = "*"
+jsonwebtoken = "6.0.1"
-reqwest = "*"
+regex = "1.1.9"
-openssl = { version = "0.10", features = ["vendored"] }
+lazy_static = "1.3.0"
+

server/src/main.rs

@@ -1,13 +1,15 @@
 extern crate lemmy_server;
-#[macro_use] extern crate diesel_migrations;
+#[macro_use]
+extern crate diesel_migrations;
 
-use std::time::{Instant, Duration};
+use actix::prelude::*;
-use std::env;
+use actix_files::NamedFile;
-use lemmy_server::actix::*;
+use actix_web::*;
-use lemmy_server::actix_web::server::HttpServer;
+use actix_web_actors::ws;
-use lemmy_server::actix_web::{ws, App, Error, HttpRequest, HttpResponse, fs::NamedFile, fs};
-use lemmy_server::websocket::server::*;
 use lemmy_server::db::establish_connection;
+use lemmy_server::websocket::server::*;
+use std::env;
+use std::time::{Duration, Instant};
 
 embed_migrations!();
 
@@ -16,41 +18,43 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 /// How long before lack of client response causes a timeout
 const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
 
-/// This is our websocket route state, this state is shared with all route
-/// instances via `HttpContext::state()`
-struct WsChatSessionState {
-  addr: Addr<ChatServer>,
-}
-
 /// Entry point for our route
-fn chat_route(req: &HttpRequest<WsChatSessionState>) -> Result<HttpResponse, Error> {
+fn chat_route(
+  req: HttpRequest,
+  stream: web::Payload,
+  chat_server: web::Data<Addr<ChatServer>>,
+  ) -> Result<HttpResponse, Error> {
   ws::start(
-    req,
     WSSession {
+      cs_addr: chat_server.get_ref().to_owned(),
       id: 0,
       hb: Instant::now(),
-      ip: req.connection_info()
+      ip: req
+        .connection_info()
         .remote()
         .unwrap_or("127.0.0.1:12345")
         .split(":")
         .next()
         .unwrap_or("127.0.0.1")
-        .to_string()
+        .to_string(),
     },
+    &req,
+    stream,
     )
 }
 
 struct WSSession {
+  cs_addr: Addr<ChatServer>,
   /// unique session id
   id: usize,
   ip: String,
   /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
   /// otherwise we drop connection.
-  hb: Instant
+  hb: Instant,
 }
 
 impl Actor for WSSession {
-  type Context = ws::WebsocketContext<Self, WsChatSessionState>;
+  type Context = ws::WebsocketContext<Self>;
 
   /// Method is called on actor start.
   /// We register ws session with ChatServer
@@ -61,11 +65,9 @@ impl Actor for WSSession {
     // register self in chat server. `AsyncContext::wait` register
     // future within context, but context waits until this future resolves
     // before processing any other events.
-    // HttpContext::state() is instance of WsChatSessionState, state is shared
     // across all routes within application
     let addr = ctx.address();
-    ctx.state()
+    self.cs_addr
-      .addr
       .send(Connect {
         addr: addr.recipient(),
         ip: self.ip.to_owned(),
@@ -82,9 +84,9 @@ impl Actor for WSSession {
     .wait(ctx);
   }
 
-  fn stopping(&mut self, ctx: &mut Self::Context) -> Running {
+  fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
     // notify chat server
-    ctx.state().addr.do_send(Disconnect { 
+    self.cs_addr.do_send(Disconnect {
       id: self.id,
       ip: self.ip.to_owned(),
     });
@@ -119,8 +121,7 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
         let m = text.trim().to_owned();
         println!("WEBSOCKET MESSAGE: {:?} from id: {}", &m, self.id);
 
-        ctx.state()
+        self.cs_addr
-          .addr
           .send(StandardMessage {
             id: self.id,
             msg: m,
@@ -140,7 +141,8 @@ impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
       ws::Message::Binary(_bin) => println!("Unexpected binary"),
       ws::Message::Close(_) => {
         ctx.stop();
-      },
+      }
+      _ => {}
     }
   }
 }
@@ -149,7 +151,7 @@ impl WSSession {
   /// helper method that sends ping to client every second.
   ///
   /// also this method checks heartbeats from client
-  fn hb(&self, ctx: &mut ws::WebsocketContext<Self, WsChatSessionState>) {
+  fn hb(&self, ctx: &mut ws::WebsocketContext<Self>) {
     ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
       // check client heartbeats
       if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
@@ -157,9 +159,10 @@ impl WSSession {
         println!("Websocket Client heartbeat failed, disconnecting!");
 
         // notify chat server
-        ctx.state()
+        act.cs_addr.do_send(Disconnect {
-          .addr
+          id: act.id,
-          .do_send(Disconnect { id: act.id, ip: act.ip.to_owned() });
+          ip: act.ip.to_owned(),
+        });
 
         // stop actor
         ctx.stop();
@@ -182,34 +185,26 @@ fn main() {
   embedded_migrations::run(&conn).unwrap();
 
   // Start chat server actor in separate thread
-  let server = Arbiter::start(|_| ChatServer::default());
+  let server = ChatServer::default().start();
-
   // Create Http server with websocket support
   HttpServer::new(move || {
-    // Websocket sessions state
+    App::new()
-    let state = WsChatSessionState {
+      .data(server.clone())
-      addr: server.clone(),
+      .service(web::resource("/api/v1/ws").to(chat_route))
-    };
+      //            .service(web::resource("/api/v1/rest").route(web::post().to(||{})))
-
+      .service(web::resource("/").to(index))
-    App::with_state(state)
-      // .resource("/api/v1/rest", |r| r.method(http::Method::POST).f(|_| {})
-      .resource("/api/v1/ws", |r| r.route().f(chat_route))
       // static resources
-      .resource("/", |r| r.route().f(index))
+      .service(actix_files::Files::new("/static", front_end_dir()))
-      .handler(
+  })
-        "/static",
+  .bind("0.0.0.0:8536")
-        fs::StaticFiles::new(front_end_dir()).unwrap()
+    .unwrap()
-      )
-      .finish()
-  }).bind("0.0.0.0:8536")
-  .unwrap()
     .start();
 
   println!("Started http server: 0.0.0.0:8536");
   let _ = sys.run();
 }
 
-fn index(_req: &HttpRequest<WsChatSessionState>) -> Result<NamedFile, actix_web::error::Error> {
+fn index() -> Result<NamedFile, actix_web::error::Error> {
   Ok(NamedFile::open(front_end_dir() + "/index.html")?)
 }