Browse Source

Use actix config to handle routes in seperate folders (#378)

feature/settings-cleanup
Felix Ableitner 3 years ago
parent
commit
1978180a0d
  1. 4
      server/src/lib.rs
  2. 266
      server/src/main.rs
  3. 18
      server/src/routes/federation.rs
  4. 18
      server/src/routes/feeds.rs
  5. 45
      server/src/routes/index.rs
  6. 6
      server/src/routes/mod.rs
  7. 9
      server/src/routes/nodeinfo.rs
  8. 12
      server/src/routes/webfinger.rs
  9. 179
      server/src/routes/websocket.rs

4
server/src/lib.rs

@ -25,12 +25,10 @@ pub extern crate strum;
pub mod api;
pub mod apub;
pub mod db;
pub mod feeds;
pub mod nodeinfo;
pub mod routes;
pub mod schema;
pub mod settings;
pub mod version;
pub mod webfinger;
pub mod websocket;
use crate::settings::Settings;

266
server/src/main.rs

@ -2,186 +2,18 @@ extern crate lemmy_server;
#[macro_use]
extern crate diesel_migrations;
use actix::prelude::*;
use actix_files::NamedFile;
use actix_web::*;
use actix_web_actors::ws;
use lemmy_server::apub;
use lemmy_server::db::establish_connection;
use lemmy_server::feeds;
use lemmy_server::nodeinfo;
use lemmy_server::routes::federation;
use lemmy_server::routes::feeds;
use lemmy_server::routes::index;
use lemmy_server::routes::nodeinfo;
use lemmy_server::routes::webfinger;
use lemmy_server::routes::websocket;
use lemmy_server::settings::Settings;
use lemmy_server::webfinger;
use lemmy_server::websocket::server::*;
use std::time::{Duration, Instant};
embed_migrations!();
/// How often heartbeat pings are sent
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);
/// Entry point for our route
fn chat_route(
req: HttpRequest,
stream: web::Payload,
chat_server: web::Data<Addr<ChatServer>>,
) -> Result<HttpResponse, Error> {
ws::start(
WSSession {
cs_addr: chat_server.get_ref().to_owned(),
id: 0,
hb: Instant::now(),
ip: req
.connection_info()
.remote()
.unwrap_or("127.0.0.1:12345")
.split(":")
.next()
.unwrap_or("127.0.0.1")
.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,
}
impl Actor for WSSession {
type Context = ws::WebsocketContext<Self>;
/// Method is called on actor start.
/// We register ws session with ChatServer
fn started(&mut self, ctx: &mut Self::Context) {
// we'll start heartbeat process on session start.
self.hb(ctx);
// register self in chat server. `AsyncContext::wait` register
// future within context, but context waits until this future resolves
// before processing any other events.
// across all routes within application
let addr = ctx.address();
self
.cs_addr
.send(Connect {
addr: addr.recipient(),
ip: self.ip.to_owned(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
Ok(res) => act.id = res,
// something is wrong with chat server
_ => ctx.stop(),
}
fut::ok(())
})
.wait(ctx);
}
fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
// notify chat server
self.cs_addr.do_send(Disconnect {
id: self.id,
ip: self.ip.to_owned(),
});
Running::Stop
}
}
/// Handle messages from chat server, we simply send it to peer websocket
/// These are room messages, IE sent to others in the room
impl Handler<WSMessage> for WSSession {
type Result = ();
fn handle(&mut self, msg: WSMessage, ctx: &mut Self::Context) {
// println!("id: {} msg: {}", self.id, msg.0);
ctx.text(msg.0);
}
}
/// WebSocket message handler
impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
// println!("WEBSOCKET MESSAGE: {:?} from id: {}", msg, self.id);
match msg {
ws::Message::Ping(msg) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
ws::Message::Pong(_) => {
self.hb = Instant::now();
}
ws::Message::Text(text) => {
let m = text.trim().to_owned();
println!("WEBSOCKET MESSAGE: {:?} from id: {}", &m, self.id);
self
.cs_addr
.send(StandardMessage {
id: self.id,
msg: m,
})
.into_actor(self)
.then(|res, _, ctx| {
match res {
Ok(res) => ctx.text(res),
Err(e) => {
eprintln!("{}", &e);
}
}
fut::ok(())
})
.wait(ctx);
}
ws::Message::Binary(_bin) => println!("Unexpected binary"),
ws::Message::Close(_) => {
ctx.stop();
}
_ => {}
}
}
}
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>) {
ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
// check client heartbeats
if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
// heartbeat timed out
println!("Websocket Client heartbeat failed, disconnecting!");
// notify chat server
act.cs_addr.do_send(Disconnect {
id: act.id,
ip: act.ip.to_owned(),
});
// stop actor
ctx.stop();
// don't try to send a ping
return;
}
ctx.ping("");
});
}
}
fn main() {
let _ = env_logger::init();
let sys = actix::System::new("lemmy");
@ -190,87 +22,21 @@ fn main() {
let conn = establish_connection();
embedded_migrations::run(&conn).unwrap();
// Start chat server actor in separate thread
let server = ChatServer::default().start();
let settings = Settings::get();
// Create Http server with websocket support
HttpServer::new(move || {
let app = App::new()
.data(server.clone())
// Front end routes
App::new()
.configure(federation::config)
.configure(feeds::config)
.configure(index::config)
.configure(nodeinfo::config)
.configure(webfinger::config)
.configure(websocket::config)
.service(actix_files::Files::new(
"/static",
settings.front_end_dir.to_owned(),
))
.route("/", web::get().to(index))
.route(
"/home/type/{type}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/login", web::get().to(index))
.route("/create_post", web::get().to(index))
.route("/create_community", web::get().to(index))
.route("/communities/page/{page}", web::get().to(index))
.route("/communities", web::get().to(index))
.route("/post/{id}/comment/{id2}", web::get().to(index))
.route("/post/{id}", web::get().to(index))
.route("/c/{name}/sort/{sort}/page/{page}", web::get().to(index))
.route("/c/{name}", web::get().to(index))
.route("/community/{id}", web::get().to(index))
.route(
"/u/{username}/view/{view}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/u/{username}", web::get().to(index))
.route("/user/{id}", web::get().to(index))
.route("/inbox", web::get().to(index))
.route("/modlog/community/{community_id}", web::get().to(index))
.route("/modlog", web::get().to(index))
.route("/setup", web::get().to(index))
.route(
"/search/q/{q}/type/{type}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/search", web::get().to(index))
.route("/sponsors", web::get().to(index))
.route("/password_change/{token}", web::get().to(index))
// Websocket
.service(web::resource("/api/v1/ws").to(chat_route))
// NodeInfo
.route("/nodeinfo/2.0.json", web::get().to(nodeinfo::node_info))
.route(
"/.well-known/nodeinfo",
web::get().to(nodeinfo::node_info_well_known),
)
// RSS
.route("/feeds/{type}/{name}.xml", web::get().to(feeds::get_feed))
.route("/feeds/all.xml", web::get().to(feeds::get_all_feed))
// Federation
.route(
"/federation/c/{community_name}",
web::get().to(apub::community::get_apub_community),
)
.route(
"/federation/c/{community_name}/followers",
web::get().to(apub::community::get_apub_community_followers),
)
.route(
"/federation/u/{user_name}",
web::get().to(apub::user::get_apub_user),
)
.route("/feeds/all.xml", web::get().to(feeds::get_all_feed));
// Federation
if settings.federation_enabled {
app.route(
".well-known/webfinger",
web::get().to(webfinger::get_webfinger_response),
)
} else {
app
}
})
.bind((settings.bind, settings.port))
.unwrap()
@ -279,9 +45,3 @@ fn main() {
println!("Started http server at {}:{}", settings.bind, settings.port);
let _ = sys.run();
}
fn index() -> Result<NamedFile, actix_web::error::Error> {
Ok(NamedFile::open(
Settings::get().front_end_dir.to_owned() + "/index.html",
)?)
}

18
server/src/routes/federation.rs

@ -0,0 +1,18 @@
use crate::apub;
use actix_web::web;
pub fn config(cfg: &mut web::ServiceConfig) {
cfg
.route(
"/federation/c/{community_name}",
web::get().to(apub::community::get_apub_community),
)
.route(
"/federation/c/{community_name}/followers",
web::get().to(apub::community::get_apub_community_followers),
)
.route(
"/federation/u/{user_name}",
web::get().to(apub::user::get_apub_user),
);
}

server/src/feeds.rs → server/src/routes/feeds.rs

45
server/src/routes/index.rs

@ -0,0 +1,45 @@
use crate::settings::Settings;
use actix_files::NamedFile;
use actix_web::web;
pub fn config(cfg: &mut web::ServiceConfig) {
cfg
.route("/", web::get().to(index))
.route(
"/home/type/{type}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/login", web::get().to(index))
.route("/create_post", web::get().to(index))
.route("/create_community", web::get().to(index))
.route("/communities/page/{page}", web::get().to(index))
.route("/communities", web::get().to(index))
.route("/post/{id}/comment/{id2}", web::get().to(index))
.route("/post/{id}", web::get().to(index))
.route("/c/{name}/sort/{sort}/page/{page}", web::get().to(index))
.route("/c/{name}", web::get().to(index))
.route("/community/{id}", web::get().to(index))
.route(
"/u/{username}/view/{view}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/u/{username}", web::get().to(index))
.route("/user/{id}", web::get().to(index))
.route("/inbox", web::get().to(index))
.route("/modlog/community/{community_id}", web::get().to(index))
.route("/modlog", web::get().to(index))
.route("/setup", web::get().to(index))
.route(
"/search/q/{q}/type/{type}/sort/{sort}/page/{page}",
web::get().to(index),
)
.route("/search", web::get().to(index))
.route("/sponsors", web::get().to(index))
.route("/password_change/{token}", web::get().to(index));
}
fn index() -> Result<NamedFile, actix_web::error::Error> {
Ok(NamedFile::open(
Settings::get().front_end_dir.to_owned() + "/index.html",
)?)
}

6
server/src/routes/mod.rs

@ -0,0 +1,6 @@
pub mod federation;
pub mod feeds;
pub mod index;
pub mod nodeinfo;
pub mod webfinger;
pub mod websocket;

server/src/nodeinfo.rs → server/src/routes/nodeinfo.rs

server/src/webfinger.rs → server/src/routes/webfinger.rs

179
server/src/routes/websocket.rs

@ -0,0 +1,179 @@
use crate::websocket::server::*;
use actix::prelude::*;
use actix_web::web;
use actix_web::*;
use actix_web_actors::ws;
use std::time::{Duration, Instant};
pub fn config(cfg: &mut web::ServiceConfig) {
// Start chat server actor in separate thread
let server = ChatServer::default().start();
cfg
.data(server.clone())
.service(web::resource("/api/v1/ws").to(chat_route));
}
/// How often heartbeat pings are sent
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);
/// Entry point for our route
fn chat_route(
req: HttpRequest,
stream: web::Payload,
chat_server: web::Data<Addr<ChatServer>>,
) -> Result<HttpResponse, Error> {
ws::start(
WSSession {
cs_addr: chat_server.get_ref().to_owned(),
id: 0,
hb: Instant::now(),
ip: req
.connection_info()
.remote()
.unwrap_or("127.0.0.1:12345")
.split(":")
.next()
.unwrap_or("127.0.0.1")
.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,
}
impl Actor for WSSession {
type Context = ws::WebsocketContext<Self>;
/// Method is called on actor start.
/// We register ws session with ChatServer
fn started(&mut self, ctx: &mut Self::Context) {
// we'll start heartbeat process on session start.
self.hb(ctx);
// register self in chat server. `AsyncContext::wait` register
// future within context, but context waits until this future resolves
// before processing any other events.
// across all routes within application
let addr = ctx.address();
self
.cs_addr
.send(Connect {
addr: addr.recipient(),
ip: self.ip.to_owned(),
})
.into_actor(self)
.then(|res, act, ctx| {
match res {
Ok(res) => act.id = res,
// something is wrong with chat server
_ => ctx.stop(),
}
fut::ok(())
})
.wait(ctx);
}
fn stopping(&mut self, _ctx: &mut Self::Context) -> Running {
// notify chat server
self.cs_addr.do_send(Disconnect {
id: self.id,
ip: self.ip.to_owned(),
});
Running::Stop
}
}
/// Handle messages from chat server, we simply send it to peer websocket
/// These are room messages, IE sent to others in the room
impl Handler<WSMessage> for WSSession {
type Result = ();
fn handle(&mut self, msg: WSMessage, ctx: &mut Self::Context) {
// println!("id: {} msg: {}", self.id, msg.0);
ctx.text(msg.0);
}
}
/// WebSocket message handler
impl StreamHandler<ws::Message, ws::ProtocolError> for WSSession {
fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
// println!("WEBSOCKET MESSAGE: {:?} from id: {}", msg, self.id);
match msg {
ws::Message::Ping(msg) => {
self.hb = Instant::now();
ctx.pong(&msg);
}
ws::Message::Pong(_) => {
self.hb = Instant::now();
}
ws::Message::Text(text) => {
let m = text.trim().to_owned();
println!("WEBSOCKET MESSAGE: {:?} from id: {}", &m, self.id);
self
.cs_addr
.send(StandardMessage {
id: self.id,
msg: m,
})
.into_actor(self)
.then(|res, _, ctx| {
match res {
Ok(res) => ctx.text(res),
Err(e) => {
eprintln!("{}", &e);
}
}
fut::ok(())
})
.wait(ctx);
}
ws::Message::Binary(_bin) => println!("Unexpected binary"),
ws::Message::Close(_) => {
ctx.stop();
}
_ => {}
}
}
}
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>) {
ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
// check client heartbeats
if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
// heartbeat timed out
println!("Websocket Client heartbeat failed, disconnecting!");
// notify chat server
act.cs_addr.do_send(Disconnect {
id: act.id,
ip: act.ip.to_owned(),
});
// stop actor
ctx.stop();
// don't try to send a ping
return;
}
ctx.ping("");
});
}
}
Loading…
Cancel
Save