diff --git a/f9/src/app.rs b/f9/src/app.rs index 874814b..95a914b 100644 --- a/f9/src/app.rs +++ b/f9/src/app.rs @@ -20,38 +20,20 @@ use serde_derive::Deserialize; use reqwest::blocking; use actix_web_actors::ws; use actix::prelude::*; +use std::{sync::OnceLock}; + use r9_macro::SharedFromSelf; use r9_macro_derive::SharedFromSelf; pub trait UserApp { fn get_pkg_name(&self) -> String; - fn init(&mut self); - fn update(&mut self); - fn uninit(&mut self); + fn init(&self); + fn update(&self); + fn uninit(&self); fn get_http_listen_port(&self) -> i32; } -pub struct HttpContext { - id: u64, - query_str: String, - add_tick: i64, - pub handled: bool, - //not_found: bool, - pub rsp: String -} - -#[derive(Message)] -#[rtype(result = "()")] -pub struct Message(pub String); - -/// Session is disconnected -#[derive(Message)] -#[rtype(result = "()")] -pub struct Disconnect { - pub id: usize, -} - #[derive(SharedFromSelf)] pub struct App { _self_wp: Weak::>, @@ -59,16 +41,10 @@ pub struct App { node_id: i32, instance_id: i32, //tokio_rt: Runtime, - user_app: Option>>, + user_app: Option<&'static dyn UserApp>, im_msgs: Rc::>>, im_work_msgs: Rc::>>, im_mutex: Mutex, - http_handlers: HashMap>>, - webapp_state: Arc::, -} - -struct AppState { - request: Arc>>>>, } #[derive(Default)] @@ -79,46 +55,6 @@ struct IMMsgNode { entry: Rc::>>, } -async fn index(data: Data>, req: HttpRequest) -> impl Responder { - println!("http.thread.id {:?}", std::thread::current().id()); - let context = Arc::new( - Mutex::new(HttpContext{ - id: 0, - query_str: String::from(req.query_string()), - add_tick: 0, - handled: false, - rsp: "".to_string() - }) - ); - { - let v = &mut data.request.lock().unwrap(); - (*v).push(context.clone()); - } - { - while !context.lock().unwrap().handled { - tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; - } - } - let data = context.lock().unwrap().rsp.clone(); - HttpResponse::Ok().body(data) -} - -async fn run_app(port: u16, app_state: Arc::, - tx: std::sync::mpsc::Sender) -> std::io::Result<()> { - let server = HttpServer::new(move || { - WebApp::new() - .data(app_state.clone()) - .route("/webapp/index.php",web::get().to(index)) - }) - .bind(("0.0.0.0", port))? - .workers(1) - .run(); - - let _ = tx.send(server.handle()); - - server.await -} - impl App { pub fn instance() -> Rc::> { @@ -139,12 +75,6 @@ impl App { im_msgs: r9::ListHead::::new_head(), im_work_msgs: r9::ListHead::::new_head(), im_mutex: Mutex::new(1), - http_handlers: Default::default(), - webapp_state: Arc::new( - AppState{ - request: Default::default() - } - ), _self_wp: Default::default(), } ))); @@ -156,17 +86,17 @@ impl App { } } - pub fn init(user_app: Rc::>) { - App::instance().borrow_mut().user_app = Some(user_app.clone()); + pub fn init(user_app: &'static dyn UserApp) { + App::instance().borrow_mut().user_app = Some(user_app); //App::instance().borrow_mut().tokio_rt.enter(); //App::instance().borrow_mut().init_http_server(); crate::Timer::instance().borrow_mut().init(); - user_app.borrow_mut().init(); + user_app.init(); } pub fn uninit() { let user_app = App::instance().borrow_mut().user_app.clone(); - user_app.as_ref().unwrap().borrow_mut().uninit(); + //user_app.as_ref().unwrap().borrow_mut().uninit(); crate::Timer::instance().borrow_mut().uninit(); } @@ -174,14 +104,13 @@ impl App { loop { //crate::Timer::update(); App::dispatch_immsg(); - App::dispatch_httprequest(); App::update_user_app(); std::thread::sleep(Duration::from_millis(1)); } } pub fn get_pkg_name(&self) -> String { - return self.user_app.as_ref().unwrap().borrow().get_pkg_name(); + return self.user_app.unwrap().get_pkg_name(); } pub fn new_uuid(&self) -> i64 { @@ -246,8 +175,7 @@ impl App { } fn update_user_app() { - let user_app = App::instance().borrow_mut().user_app.clone(); - user_app.as_ref().unwrap().borrow_mut().update(); + App::instance().borrow_mut().user_app.unwrap().update(); } fn fetch_immsg(&mut self) -> Rc::>> { @@ -255,74 +183,4 @@ impl App { return self.im_work_msgs.clone(); } - fn dispatch_httprequest() { - #[derive(Deserialize)] - struct Ca { - c: String, - a: String - } - let pending_list = App::instance().borrow_mut().fetch_request(); - match pending_list { - Some(list) => { - for req in list { - let ctx = &mut req.lock().unwrap(); - let ca = web::Query::::from_query(&ctx.query_str).unwrap(); - let key = format!("{}${}", &ca.c, &ca.a); - let h = App::instance().borrow_mut().get_http_handle(&key); - match h { - Some(handle) => { - (handle.borrow_mut())(ctx); - ctx.handled = true; - } - None => { - ctx.rsp = "not implement".to_string(); - ctx.handled = true; - } - }; - } - } - None => { - - } - } - } - - fn fetch_request(&mut self) -> Option>>> { - let pending_list = &mut self.webapp_state.request.lock().unwrap(); - if pending_list.len() <= 0 { - return None; - } - let result = pending_list.clone(); - pending_list.clear(); - return Some(result); - } - - fn get_http_handle(&mut self, key: &String) -> - Option>> { - match self.http_handlers.get(key) { - Some(v) => { - return Some(v.clone()); - } - None => { - return None; - } - } - } - - fn init_http_server(&mut self) { - let (tx, _) = std::sync::mpsc::channel(); - - let port = self.user_app.as_ref().unwrap().borrow().get_http_listen_port() as u16; - let app_state = self.webapp_state.clone(); - std::thread::spawn(move || { - let server_future = run_app(port, app_state, tx); - rt::System::new().block_on(server_future) - }); - } - - pub fn add_http_handler(&mut self, key: String, - cb: Rc::>) { - self.http_handlers.insert(key, cb); - } - }