diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..e74d5f40ad458bb8e2cf4caa67731add8141edce Binary files /dev/null and b/.DS_Store differ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000000000000000000000000000000000000..c3da10d13ce60fefa3e773de0a600d7e64499e63 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +target/ +.vscode/ +Cargo.lock +.idea/ \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000000000000000000000000000000000000..6a0d516b82ca5f6111ec077d1adca67306423487 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,9 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000000000000000000000000000000000000..35eb1ddfbbc029bcab630581847471d7f238ec53 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index 56cb13f99baf4389d2a3e1dbf1f710f7eda8f69f..a68cec87abde3068f8b87a3c7df833a3b739519f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,10 +7,17 @@ edition = "2018" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] +actix = "0.9.0" +actix-codec = "0.2.0" +actix-web-actors = "2.0.0" +awc = "1.0.1" +env_logger = "0.6" +bytes = "0.5.3" actix-web = "2.0" actix-rt = "1.0" tera = "1.0" -env_logger = "0.6" + +rand = "0.6" mysql = "17.0" serde = "1.0" serde_derive = "1.0" diff --git a/setting.toml.default b/setting.toml similarity index 45% rename from setting.toml.default rename to setting.toml index 6f413e6747e84f717d6dd8495685f127a3768abc..3864a01f11479ef60307ccd86e85391917803714 100644 --- a/setting.toml.default +++ b/setting.toml @@ -1,13 +1,13 @@ [app] -host = "127.0.0.1" -port = 8081 +host = "0.0.0.0" +port = 18081 [database] -host = "localhost" -name = "rust_admin" -user = "rust_admin" -password = "rust-x-lsl" -port = 3306 +host = "106.14.68.253" +name = "honeyworld" +user = "root" +password = "honeyworld" +port = 13306 [oss] access_key_id = "" diff --git a/src/main.rs b/src/main.rs index 3b84e781c71f4ae6d48e2f9f3cddcd52c7c58b6d..7597ae2fc57449114270bda2b770c866479b098b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,42 +1,252 @@ -#[macro_use] extern crate fluffy; -#[macro_use] extern crate lazy_static; -#[macro_use] extern crate serde_json; +#[macro_use] +extern crate fluffy; +#[macro_use] +extern crate lazy_static; +#[macro_use] +extern crate serde_json; -use actix_web::{App, HttpServer, middleware, web}; -use fluffy::{db}; -use actix_session::{CookieSession}; use actix_files::Files; +use actix_session::CookieSession; +use fluffy::db; +use std::time::{Duration, Instant}; + +use actix::*; +use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer, middleware}; +use actix_web_actors::ws; + + +mod caches; +mod common; mod config; +mod controllers; mod filters; -mod validations; mod models; -mod common; -mod controllers; -mod caches; +mod validations; +mod server; use controllers::{ - Controller, - index::Index, - admins::Admins, - admin_roles::AdminRoles, - menus::Menus, - users::Users, - videos::Videos, - video_categories::VideoCategories, - video_replies::VideoReplies, - video_tags::VideoTags, - user_levels::UserLevels, - watch_records::WatchRecords, - ads::Ads, - navs::Navs, - configs::Configs, - video_authors::VideoAuthors, + admin_roles::AdminRoles, admins::Admins, ads::Ads, configs::Configs, index::Index, + menus::Menus, navs::Navs, user_levels::UserLevels, users::Users, video_authors::VideoAuthors, + video_categories::VideoCategories, video_replies::VideoReplies, video_tags::VideoTags, + videos::Videos, watch_records::WatchRecords, Controller, }; + +/// 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 +pub async fn chat_route( + req: HttpRequest, + stream: web::Payload, + srv: web::Data>, +) -> Result { + ws::start( + WsChatSession { + id: 0, + hb: Instant::now(), + room: "Main".to_owned(), + name: None, + addr: srv.get_ref().clone(), + }, + &req, + stream, + ) +} + +struct WsChatSession { + /// unique session id + id: usize, + /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT), + /// otherwise we drop connection. + hb: Instant, + /// joined room + room: String, + /// peer name + name: Option, + /// Chat server + addr: Addr, +} + +impl Actor for WsChatSession { + type Context = ws::WebsocketContext; + + /// 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. + // HttpContext::state() is instance of WsChatSessionState, state is shared + // across all routes within application + let addr = ctx.address(); + self.addr + .send(server::Connect { + addr: addr.recipient(), + }) + .into_actor(self) + .then(|res, act, ctx| { + match res { + Ok(res) => act.id = res, + // something is wrong with chat server + _ => ctx.stop(), + } + fut::ready(()) + }) + .wait(ctx); + } + + fn stopping(&mut self, _: &mut Self::Context) -> Running { + // notify chat server + self.addr.do_send(server::Disconnect { id: self.id }); + Running::Stop + } +} + +/// Handle messages from chat server, we simply send it to peer websocket +impl Handler for WsChatSession { + type Result = (); + + fn handle(&mut self, msg: server::Message, ctx: &mut Self::Context) { + ctx.text(msg.0); + } +} + +/// WebSocket message handler +impl StreamHandler> for WsChatSession { + fn handle( + &mut self, + msg: Result, + ctx: &mut Self::Context, + ) { + let msg = match msg { + Err(_) => { + ctx.stop(); + return; + } + Ok(msg) => msg, + }; + + println!("WEBSOCKET MESSAGE: {:?}", msg); + 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(); + // we check for /sss type of messages + if m.starts_with('/') { + let v: Vec<&str> = m.splitn(2, ' ').collect(); + match v[0] { + "/list" => { + // Send ListRooms message to chat server and wait for + // response + println!("List rooms"); + self.addr + .send(server::ListRooms) + .into_actor(self) + .then(|res, _, ctx| { + match res { + Ok(rooms) => { + for room in rooms { + ctx.text(room); + } + } + _ => println!("Something is wrong"), + } + fut::ready(()) + }) + .wait(ctx) + // .wait(ctx) pauses all events in context, + // so actor wont receive any new messages until it get list + // of rooms back + } + "/join" => { + if v.len() == 2 { + self.room = v[1].to_owned(); + self.addr.do_send(server::Join { + id: self.id, + name: self.room.clone(), + }); + + ctx.text("joined"); + } else { + ctx.text("!!! room name is required"); + } + } + "/name" => { + if v.len() == 2 { + self.name = Some(v[1].to_owned()); + } else { + ctx.text("!!! name is required"); + } + } + _ => ctx.text(format!("!!! unknown command: {:?}", m)), + } + } else { + let msg = if let Some(ref name) = self.name { + format!("{}: {}", name, m) + } else { + m.to_owned() + }; + // send message to chat server + self.addr.do_send(server::ClientMessage { + id: self.id, + msg, + room: self.room.clone(), + }) + } + } + ws::Message::Binary(_) => println!("Unexpected binary"), + ws::Message::Close(_) => { + ctx.stop(); + } + ws::Message::Continuation(_) => { + ctx.stop(); + } + ws::Message::Nop => (), + } + } +} + +impl WsChatSession { + /// helper method that sends ping to client every second. + /// + /// also this method checks heartbeats from client + fn hb(&self, ctx: &mut ws::WebsocketContext) { + 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.addr.do_send(server::Disconnect { id: act.id }); + + // stop actor + ctx.stop(); + + // don't try to send a ping + return; + } + + ctx.ping(b""); + }); + } +} + #[actix_rt::main] async fn main() -> std::io::Result<()> { - // 正式环境可以去掉日志显示 std::env::set_var("RUST_LOG", "actix_web=info"); //正式环境可以注释此行 *** env_logger::init(); //正式环境可以注释此行 *** @@ -48,9 +258,10 @@ async fn main() -> std::io::Result<()> { let host_port = &format!("{}:{}", &info.host, &info.port); //地址/端口 println!("Started At: {}", host_port); - //let table_fields = caches::TABLE_FIELDS.lock().unwrap(); - HttpServer::new(|| { + let server = server::ChatServer::default().start(); + //let table_fields = caches::TABLE_FIELDS.lock().unwrap(); + HttpServer::new(move || { let mut tpl = tmpl!("/templates/**/*"); //模板引擎 tpl.register_filter("state_name", filters::state_name); tpl.register_filter("menu_name", filters::menus::menu_name); @@ -61,10 +272,10 @@ async fn main() -> std::io::Result<()> { tpl.register_filter("author_name", filters::video_authors::author_name); //let generated = generate(); - App::new() .wrap(CookieSession::signed(&[0; 32]).secure(false)) .data(tpl) + .data(server.clone()) .wrap(middleware::Logger::default()) //正式环境可以注释此行 *** .service(Files::new("/static", "public/static/")) //静态文件目录 .service(Files::new("/upload", "public/upload/")) //上传文件目录 @@ -85,7 +296,7 @@ async fn main() -> std::io::Result<()> { .service(get!("/admins/edit/{id}", Admins::edit)) .service(post!("/admins/save/{id}", Admins::save)) .service(get!("/admins/delete/{ids}", Admins::delete)) - //角色管理 + //角色管理 .service(get!("/admin_roles", AdminRoles::index)) .service(get!("/admin_roles/edit/{id}", AdminRoles::edit)) .service(post!("/admin_roles/save/{id}", AdminRoles::save)) @@ -104,7 +315,10 @@ async fn main() -> std::io::Result<()> { .service(get!("/video_categories", VideoCategories::index)) .service(get!("/video_categories/edit/{id}", VideoCategories::edit)) .service(post!("/video_categories/save/{id}", VideoCategories::save)) - .service(get!("/video_categories/delete/{ids}", VideoCategories::delete)) + .service(get!( + "/video_categories/delete/{ids}", + VideoCategories::delete + )) //视频管理 .service(get!("/videos", Videos::index)) .service(get!("/videos/edit/{id}", Videos::edit)) @@ -148,9 +362,10 @@ async fn main() -> std::io::Result<()> { //网站设置 .service(get!("/configs/edit/{id}", Configs::edit)) .service(post!("/configs/save/{id}", Configs::save)) - +// .service(web::resource("/ws/").route(web::get().to(websocket::ws_index))) + .service(web::resource("/ws/").to(chat_route)) }) - .bind(host_port)? - .run() - .await + .bind(host_port)? + .run() + .await } diff --git a/src/server.rs b/src/server.rs new file mode 100644 index 0000000000000000000000000000000000000000..3a59b9209bb4af8107a937de1d9809a6772e7928 --- /dev/null +++ b/src/server.rs @@ -0,0 +1,202 @@ +//! `ChatServer` is an actor. It maintains list of connection client session. +//! And manages available rooms. Peers send messages to other peers in same +//! room through `ChatServer`. + +use actix::prelude::*; +use rand::{self, rngs::ThreadRng, Rng}; +use std::collections::{HashMap, HashSet}; + +/// Chat server sends this messages to session +#[derive(Message)] +#[rtype(result = "()")] +pub struct Message(pub String); + +/// Message for chat server communications + +/// New chat session is created +#[derive(Message)] +#[rtype(usize)] +pub struct Connect { + pub addr: Recipient, +} + +/// Session is disconnected +#[derive(Message)] +#[rtype(result = "()")] +pub struct Disconnect { + pub id: usize, +} + +/// Send message to specific room +#[derive(Message)] +#[rtype(result = "()")] +pub struct ClientMessage { + /// Id of the client session + pub id: usize, + /// Peer message + pub msg: String, + /// Room name + pub room: String, +} + +/// List of available rooms +pub struct ListRooms; + +impl actix::Message for ListRooms { + type Result = Vec; +} + +/// Join room, if room does not exists create new one. +#[derive(Message)] +#[rtype(result = "()")] +pub struct Join { + /// Client id + pub id: usize, + /// Room name + pub name: String, +} + +/// `ChatServer` manages chat rooms and responsible for coordinating chat +/// session. implementation is super primitive +pub struct ChatServer { + sessions: HashMap>, + rooms: HashMap>, + rng: ThreadRng, +} + +impl Default for ChatServer { + fn default() -> ChatServer { + // default room + let mut rooms = HashMap::new(); + rooms.insert("Main".to_owned(), HashSet::new()); + + ChatServer { + sessions: HashMap::new(), + rooms, + rng: rand::thread_rng(), + } + } +} + +impl ChatServer { + /// Send message to all users in the room + fn send_message(&self, room: &str, message: &str, skip_id: usize) { + if let Some(sessions) = self.rooms.get(room) { + for id in sessions { + if *id != skip_id { + if let Some(addr) = self.sessions.get(id) { + let _ = addr.do_send(Message(message.to_owned())); + } + } + } + } + } +} + +/// Make actor from `ChatServer` +impl Actor for ChatServer { + /// We are going to use simple Context, we just need ability to communicate + /// with other actors. + type Context = Context; +} + +/// Handler for Connect message. +/// +/// Register new session and assign unique id to this session +impl Handler for ChatServer { + type Result = usize; + + fn handle(&mut self, msg: Connect, _: &mut Context) -> Self::Result { + println!("Someone joined"); + + // notify all users in same room +// self.send_message(&"Main".to_owned(), "Someone joined", 0); + + // register session with random id + let id = self.rng.gen::(); + self.sessions.insert(id, msg.addr); + + // auto join session to Main room + self.rooms.get_mut(&"Main".to_owned()).unwrap().insert(id); + + // send id back + id + } +} + +/// Handler for Disconnect message. +impl Handler for ChatServer { + type Result = (); + + fn handle(&mut self, msg: Disconnect, _: &mut Context) { + println!("Someone disconnected"); + + let mut rooms: Vec = Vec::new(); + + // remove address + if self.sessions.remove(&msg.id).is_some() { + // remove session from all rooms + for (name, sessions) in &mut self.rooms { + if sessions.remove(&msg.id) { + rooms.push(name.to_owned()); + } + } + } + // send message to other users + for room in rooms { + self.send_message(&room, "Someone disconnected", 0); + } + } +} + +/// Handler for Message message. +impl Handler for ChatServer { + type Result = (); + + fn handle(&mut self, msg: ClientMessage, _: &mut Context) { + self.send_message(&msg.room, msg.msg.as_str(), msg.id); + } +} + +/// Handler for `ListRooms` message. +impl Handler for ChatServer { + type Result = MessageResult; + + fn handle(&mut self, _: ListRooms, _: &mut Context) -> Self::Result { + let mut rooms = Vec::new(); + + for key in self.rooms.keys() { + rooms.push(key.to_owned()) + } + + MessageResult(rooms) + } +} + +/// Join room, send disconnect message to old room +/// send join message to new room +impl Handler for ChatServer { + type Result = (); + + fn handle(&mut self, msg: Join, _: &mut Context) { + let Join { id, name } = msg; + let mut rooms = Vec::new(); + + // remove session from all rooms + for (n, sessions) in &mut self.rooms { + if sessions.remove(&id) { + rooms.push(n.to_owned()); + } + } + // send message to other users + for room in rooms { + self.send_message(&room, "Someone disconnected", 0); + } + + if self.rooms.get_mut(&name).is_none() { + self.rooms.insert(name.clone(), HashSet::new()); + } + self.send_message(&name, "Someone connected", id); + self.rooms.get_mut(&name).unwrap().insert(id); + } +} diff --git a/src/websocket/mod.rs b/src/websocket/mod.rs new file mode 100644 index 0000000000000000000000000000000000000000..9a6e424e8bec76a348c2053fa01d6c8c127ca165 --- /dev/null +++ b/src/websocket/mod.rs @@ -0,0 +1,250 @@ +use std::time::{Duration, Instant}; + +use actix::*; +use actix_files as fs; +use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer}; +use actix_web_actors::ws; + +/// 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 +pub async fn chat_route( + req: HttpRequest, + stream: web::Payload, + srv: web::Data>, +) -> Result { + ws::start( + WsChatSession { + id: 0, + hb: Instant::now(), + room: "Main".to_owned(), + name: None, + addr: srv.get_ref().clone(), + }, + &req, + stream, + ) +} + +struct WsChatSession { + /// unique session id + id: usize, + /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT), + /// otherwise we drop connection. + hb: Instant, + /// joined room + room: String, + /// peer name + name: Option, + /// Chat server + addr: Addr, +} + +impl Actor for WsChatSession { + type Context = ws::WebsocketContext; + + /// 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. + // HttpContext::state() is instance of WsChatSessionState, state is shared + // across all routes within application + let addr = ctx.address(); + self.addr + .send(server::Connect { + addr: addr.recipient(), + }) + .into_actor(self) + .then(|res, act, ctx| { + match res { + Ok(res) => act.id = res, + // something is wrong with chat server + _ => ctx.stop(), + } + fut::ready(()) + }) + .wait(ctx); + } + + fn stopping(&mut self, _: &mut Self::Context) -> Running { + // notify chat server + self.addr.do_send(server::Disconnect { id: self.id }); + Running::Stop + } +} + +/// Handle messages from chat server, we simply send it to peer websocket +impl Handler for WsChatSession { + type Result = (); + + fn handle(&mut self, msg: server::Message, ctx: &mut Self::Context) { + ctx.text(msg.0); + } +} + +/// WebSocket message handler +impl StreamHandler> for WsChatSession { + fn handle( + &mut self, + msg: Result, + ctx: &mut Self::Context, + ) { + let msg = match msg { + Err(_) => { + ctx.stop(); + return; + } + Ok(msg) => msg, + }; + + println!("WEBSOCKET MESSAGE: {:?}", msg); + 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(); + // we check for /sss type of messages + if m.starts_with('/') { + let v: Vec<&str> = m.splitn(2, ' ').collect(); + match v[0] { + "/list" => { + // Send ListRooms message to chat server and wait for + // response + println!("List rooms"); + self.addr + .send(server::ListRooms) + .into_actor(self) + .then(|res, _, ctx| { + match res { + Ok(rooms) => { + for room in rooms { + ctx.text(room); + } + } + _ => println!("Something is wrong"), + } + fut::ready(()) + }) + .wait(ctx) + // .wait(ctx) pauses all events in context, + // so actor wont receive any new messages until it get list + // of rooms back + } + "/join" => { + if v.len() == 2 { + self.room = v[1].to_owned(); + self.addr.do_send(server::Join { + id: self.id, + name: self.room.clone(), + }); + + ctx.text("joined"); + } else { + ctx.text("!!! room name is required"); + } + } + "/name" => { + if v.len() == 2 { + self.name = Some(v[1].to_owned()); + } else { + ctx.text("!!! name is required"); + } + } + _ => ctx.text(format!("!!! unknown command: {:?}", m)), + } + } else { + let msg = if let Some(ref name) = self.name { + format!("{}: {}", name, m) + } else { + m.to_owned() + }; + // send message to chat server + self.addr.do_send(server::ClientMessage { + id: self.id, + msg, + room: self.room.clone(), + }) + } + } + ws::Message::Binary(_) => println!("Unexpected binary"), + ws::Message::Close(_) => { + ctx.stop(); + } + ws::Message::Continuation(_) => { + ctx.stop(); + } + ws::Message::Nop => (), + } + } +} + +impl WsChatSession { + /// helper method that sends ping to client every second. + /// + /// also this method checks heartbeats from client + fn hb(&self, ctx: &mut ws::WebsocketContext) { + 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.addr.do_send(server::Disconnect { id: act.id }); + + // stop actor + ctx.stop(); + + // don't try to send a ping + return; + } + + ctx.ping(b""); + }); + } +} + + +#[actix_rt::main] +async fn main() -> std::io::Result<()> { + env_logger::init(); + + // Start chat server actor + let server = server::ChatServer::default().start(); + + // Create Http server with websocket support + HttpServer::new(move || { + App::new() + .data(server.clone()) + // redirect to websocket.html + .service(web::resource("/").route(web::get().to(|| { + HttpResponse::Found() + .header("LOCATION", "/static/websocket.html") + .finish() + }))) + // websocket + .service(web::resource("/ws/").to(chat_route)) + // static resources + .service(fs::Files::new("/static/", "static/")) + }) + .bind("127.0.0.1:8080")? + .run() + .await +} + + + + diff --git "a/\346\215\220\350\265\240\345\220\215\345\215\225.md" "b/\346\215\220\350\265\240\345\220\215\345\215\225.md" deleted file mode 100644 index bf1e8e86d14a74e167600797b22d2e4f71005b7f..0000000000000000000000000000000000000000 --- "a/\346\215\220\350\265\240\345\220\215\345\215\225.md" +++ /dev/null @@ -1,15 +0,0 @@ -#### 非常感谢各位朋友的支持, 让我们一起努力推广rust! - -| 昵称 | 日期 | 捐赠金额 | -| -- | -- | -- | -| 小卓 | 2020-02-08 | 66.66 元 | -| fengan | 2020-02-09 | 100.0 0 元 | -| larry | 2020-02-10 | 1.00 元 | -| **称 | 2020-02-10 | 10.00 元 | -| **EFER | 2020-02-12 | 30.00 元 | -| 宏景1371 | 2020-02-13 | 50.0 元 | -| **fy | 2020-02-14 | 51.2 元 | -| huanghuang | 2020-02-15 | 33.0 元 | -| zy010101 | 2020-02-15 | 33.0 元 | -| **超 | 2020-02-17 | 100.0 元 | -| **nte | 2020-02-17 | 200.0 元 |