From 8785186dfa8a9f1d1f8c2be37719d870a0ac4781 Mon Sep 17 00:00:00 2001 From: Tobias Reisinger Date: Thu, 7 Dec 2023 04:30:33 +0100 Subject: [PATCH] Add notifier to break relay loop --- emgauwa-controller/src/app_state.rs | 28 ++++++++++++- emgauwa-controller/src/main.rs | 23 ++++------- emgauwa-controller/src/relay_loop.rs | 41 +++++++++++++++---- emgauwa-controller/src/utils.rs | 22 ++++++++-- emgauwa-controller/src/ws/mod.rs | 28 +++++++++++-- emgauwa-core/src/handlers/v1/controllers.rs | 8 ++-- emgauwa-core/src/handlers/v1/relays.rs | 4 +- .../handlers/v1/ws/controllers/handlers.rs | 2 +- .../src/handlers/v1/ws/controllers/mod.rs | 4 +- emgauwa-core/src/handlers/v1/ws/mod.rs | 4 +- emgauwa-core/src/main.rs | 4 +- emgauwa-lib/src/constants.rs | 1 + 12 files changed, 124 insertions(+), 45 deletions(-) diff --git a/emgauwa-controller/src/app_state.rs b/emgauwa-controller/src/app_state.rs index ff4f90f..aa9553c 100644 --- a/emgauwa-controller/src/app_state.rs +++ b/emgauwa-controller/src/app_state.rs @@ -1,8 +1,11 @@ +use std::sync::Arc; + use actix::{Actor, Context, Handler, Message}; use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::models::Controller; use futures::executor::block_on; use sqlx::{Pool, Sqlite}; +use tokio::sync::Notify; #[derive(Message)] #[rtype(result = "Result<(), EmgauwaError>")] @@ -12,14 +15,27 @@ pub struct Reload {} #[rtype(result = "Controller")] pub struct GetThis {} +#[derive(Message)] +#[rtype(result = "Arc")] +pub struct GetNotifier {} + pub struct AppState { pub pool: Pool, pub this: Controller, + pub notifier: Arc, } impl AppState { pub fn new(pool: Pool, this: Controller) -> AppState { - AppState { pool, this } + AppState { + pool, + this, + notifier: Arc::new(Notify::new()), + } + } + + pub fn notify_change(&self) { + self.notifier.notify_one(); } } @@ -35,6 +51,8 @@ impl Handler for AppState { self.this.reload(&mut pool_conn)?; + self.notify_change(); + Ok(()) } } @@ -46,3 +64,11 @@ impl Handler for AppState { self.this.clone() } } + +impl Handler for AppState { + type Result = Arc; + + fn handle(&mut self, _msg: GetNotifier, _ctx: &mut Self::Context) -> Self::Result { + Arc::clone(&self.notifier) + } +} diff --git a/emgauwa-controller/src/main.rs b/emgauwa-controller/src/main.rs index 9d820e6..c6920fa 100644 --- a/emgauwa-controller/src/main.rs +++ b/emgauwa-controller/src/main.rs @@ -1,5 +1,4 @@ use actix::Actor; -use emgauwa_lib::constants::WEBSOCKET_RETRY_TIMEOUT; use emgauwa_lib::db; use emgauwa_lib::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; use emgauwa_lib::errors::EmgauwaError; @@ -8,11 +7,10 @@ use emgauwa_lib::types::ControllerUid; use emgauwa_lib::utils::init_logging; use sqlx::pool::PoolConnection; use sqlx::Sqlite; -use tokio::time; -use crate::relay_loop::run_relay_loop; +use crate::relay_loop::run_relays_loop; use crate::settings::Settings; -use crate::ws::run_websocket; +use crate::ws::run_ws_loop; mod app_state; mod driver; @@ -110,18 +108,11 @@ async fn main() -> Result<(), std::io::Error> { settings.core.host, settings.core.port ); - tokio::spawn(run_relay_loop(settings)); - loop { - let run_result = run_websocket(pool.clone(), &app_state, &url).await; - if let Err(err) = run_result { - log::error!("Error running websocket: {}", err); - } - - log::info!( - "Retrying to connect in {} seconds...", - WEBSOCKET_RETRY_TIMEOUT.as_secs() - ); - time::sleep(WEBSOCKET_RETRY_TIMEOUT).await; + tokio::select! { + _ = run_relays_loop(app_state.clone()) => {}, + _ = run_ws_loop(pool.clone(), app_state.clone(), url) => {}, } + + Ok(()) } diff --git a/emgauwa-controller/src/relay_loop.rs b/emgauwa-controller/src/relay_loop.rs index 42b5009..f3bb8cc 100644 --- a/emgauwa-controller/src/relay_loop.rs +++ b/emgauwa-controller/src/relay_loop.rs @@ -1,17 +1,42 @@ use std::time::Duration; +use actix::Addr; use chrono::Local; +use emgauwa_lib::constants::RELAYS_RETRY_TIMEOUT; +use emgauwa_lib::errors::EmgauwaError; +use futures::pin_mut; use tokio::time; -use tokio::time::Instant; +use tokio::time::timeout; +use utils::app_state_get_notifier; -use crate::settings::Settings; +use crate::app_state::AppState; +use crate::utils; -#[allow(unused_variables)] -pub async fn run_relay_loop(settings: Settings) { - let default_duration = Duration::from_millis(10000); +pub async fn run_relays_loop(app_state: Addr) { + log::debug!("Spawned relays loop"); loop { - let next_timestamp = Instant::now() + default_duration; - time::sleep_until(next_timestamp).await; - log::debug!("Relay loop: {}", Local::now().naive_local().time()) + let run_result = run_relays(&app_state).await; + if let Err(err) = run_result { + log::error!("Error running relays: {}", err); + } + time::sleep(RELAYS_RETRY_TIMEOUT).await; + } +} + +async fn run_relays(app_state: &Addr) -> Result<(), EmgauwaError> { + let default_duration = Duration::new(10, 0); + let notifier = &*app_state_get_notifier(app_state).await?; + + loop { + let notifier_future = notifier.notified(); + pin_mut!(notifier_future); + let timeout_result = timeout(default_duration, &mut notifier_future).await; + let changed = timeout_result.is_ok(); + + log::debug!( + "Relay loop at {} (changed: {})", + Local::now().naive_local().time(), + changed + ); } } diff --git a/emgauwa-controller/src/utils.rs b/emgauwa-controller/src/utils.rs index df17faa..0822272 100644 --- a/emgauwa-controller/src/utils.rs +++ b/emgauwa-controller/src/utils.rs @@ -1,9 +1,25 @@ +use std::sync::Arc; + use actix::Addr; use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::models::Controller; +use tokio::sync::Notify; -use crate::app_state::{AppState, GetThis}; +use crate::app_state; +use crate::app_state::AppState; -pub async fn get_this(app_state: &Addr) -> Result { - app_state.send(GetThis {}).await.map_err(EmgauwaError::from) +pub async fn app_state_get_this(app_state: &Addr) -> Result { + app_state + .send(app_state::GetThis {}) + .await + .map_err(EmgauwaError::from) +} + +pub async fn app_state_get_notifier( + app_state: &Addr, +) -> Result, EmgauwaError> { + app_state + .send(app_state::GetNotifier {}) + .await + .map_err(EmgauwaError::from) } diff --git a/emgauwa-controller/src/ws/mod.rs b/emgauwa-controller/src/ws/mod.rs index 0c67fab..edead72 100644 --- a/emgauwa-controller/src/ws/mod.rs +++ b/emgauwa-controller/src/ws/mod.rs @@ -1,4 +1,5 @@ use actix::Addr; +use emgauwa_lib::constants::WEBSOCKET_RETRY_TIMEOUT; use emgauwa_lib::db::DbController; use emgauwa_lib::errors::{DatabaseError, EmgauwaError}; use emgauwa_lib::models::Controller; @@ -6,25 +7,44 @@ use emgauwa_lib::types::ControllerWsAction; use futures::{SinkExt, StreamExt}; use sqlx::pool::PoolConnection; use sqlx::{Pool, Sqlite}; +use tokio::time; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::{connect_async, tungstenite}; use crate::app_state; use crate::app_state::AppState; -use crate::utils::get_this; +use crate::utils::app_state_get_this; -pub async fn run_websocket( +pub async fn run_ws_loop(pool: Pool, app_state: Addr, url: String) { + log::debug!("Spawned ws loop"); + loop { + let run_result = run_websocket(pool.clone(), &app_state, &url).await; + if let Err(err) = run_result { + log::error!("Error running websocket: {}", err); + } + + log::info!( + "Retrying to connect in {} seconds...", + WEBSOCKET_RETRY_TIMEOUT.as_secs() + ); + time::sleep(WEBSOCKET_RETRY_TIMEOUT).await; + } +} + +async fn run_websocket( pool: Pool, app_state: &Addr, url: &str, ) -> Result<(), EmgauwaError> { + log::debug!("Trying to connect to {}", url); match connect_async(url).await { Ok(connection) => { + log::debug!("Websocket connected"); let (ws_stream, _) = connection; let (mut write, read) = ws_stream.split(); - let ws_action = ControllerWsAction::Register(get_this(app_state).await?); + let ws_action = ControllerWsAction::Register(app_state_get_this(app_state).await?); let ws_action_json = serde_json::to_string(&ws_action)?; if let Err(err) = write.send(Message::text(ws_action_json)).await { @@ -98,7 +118,7 @@ pub async fn handle_controller( app_state: &Addr, controller: Controller, ) -> Result<(), EmgauwaError> { - let this = get_this(app_state).await?; + let this = app_state_get_this(app_state).await?; if controller.c.uid != this.c.uid { return Err(EmgauwaError::Other(String::from( "Controller UID mismatch during update", diff --git a/emgauwa-core/src/handlers/v1/controllers.rs b/emgauwa-core/src/handlers/v1/controllers.rs index 171cb5b..1fdd6cd 100644 --- a/emgauwa-core/src/handlers/v1/controllers.rs +++ b/emgauwa-core/src/handlers/v1/controllers.rs @@ -41,7 +41,7 @@ pub async fn show( #[put("/api/v1/controllers/{controller_id}")] pub async fn update( pool: web::Data>, - app_server: web::Data>, + app_state: web::Data>, path: web::Path<(String,)>, data: web::Json, ) -> Result { @@ -60,7 +60,7 @@ pub async fn update( let return_controller = Controller::from_db_model(&mut pool_conn, controller)?; - app_server + app_state .send(app_state::Action { controller_uid: uid.clone(), action: ControllerWsAction::Controller(return_controller.clone()), @@ -73,7 +73,7 @@ pub async fn update( #[delete("/api/v1/controllers/{controller_id}")] pub async fn delete( pool: web::Data>, - app_server: web::Data>, + app_state: web::Data>, path: web::Path<(String,)>, ) -> Result { let mut pool_conn = pool.acquire().await?; @@ -81,7 +81,7 @@ pub async fn delete( let (controller_uid,) = path.into_inner(); let uid = ControllerUid::try_from(controller_uid.as_str())?; - app_server + app_state .send(app_state::DisconnectController { controller_uid: uid.clone(), }) diff --git a/emgauwa-core/src/handlers/v1/relays.rs b/emgauwa-core/src/handlers/v1/relays.rs index 08c728b..6f2ddfa 100644 --- a/emgauwa-core/src/handlers/v1/relays.rs +++ b/emgauwa-core/src/handlers/v1/relays.rs @@ -84,7 +84,7 @@ pub async fn show_for_controller( #[put("/api/v1/controllers/{controller_id}/relays/{relay_num}")] pub async fn update_for_controller( pool: web::Data>, - app_server: web::Data>, + app_state: web::Data>, path: web::Path<(String, i64)>, data: web::Json, ) -> Result { @@ -140,7 +140,7 @@ pub async fn update_for_controller( let return_relay = Relay::from_db_model(&mut pool_conn, relay)?; - app_server + app_state .send(app_state::Action { controller_uid: uid, action: ControllerWsAction::Relays(vec![return_relay.clone()]), diff --git a/emgauwa-core/src/handlers/v1/ws/controllers/handlers.rs b/emgauwa-core/src/handlers/v1/ws/controllers/handlers.rs index a2e12de..6148c5e 100644 --- a/emgauwa-core/src/handlers/v1/ws/controllers/handlers.rs +++ b/emgauwa-core/src/handlers/v1/ws/controllers/handlers.rs @@ -64,7 +64,7 @@ impl ControllerWs { let addr = ctx.address(); self.controller_uid = Some(controller_uid.clone()); - block_on(self.app_server.send(ConnectController { + block_on(self.app_state.send(ConnectController { address: addr.recipient(), controller, }))??; diff --git a/emgauwa-core/src/handlers/v1/ws/controllers/mod.rs b/emgauwa-core/src/handlers/v1/ws/controllers/mod.rs index 0a131d6..445e408 100644 --- a/emgauwa-core/src/handlers/v1/ws/controllers/mod.rs +++ b/emgauwa-core/src/handlers/v1/ws/controllers/mod.rs @@ -19,7 +19,7 @@ use crate::utils::flatten_result; pub struct ControllerWs { pub pool: Pool, pub controller_uid: Option, - pub app_server: Addr, + pub app_state: Addr, pub hb: Instant, } @@ -33,7 +33,7 @@ impl Actor for ControllerWs { fn stopped(&mut self, _ctx: &mut Self::Context) { if let Some(controller_uid) = &self.controller_uid { let flat_res = flatten_result( - block_on(self.app_server.send(DisconnectController { + block_on(self.app_state.send(DisconnectController { controller_uid: controller_uid.clone(), })) .map_err(EmgauwaError::from), diff --git a/emgauwa-core/src/handlers/v1/ws/mod.rs b/emgauwa-core/src/handlers/v1/ws/mod.rs index dd66ad5..23a545a 100644 --- a/emgauwa-core/src/handlers/v1/ws/mod.rs +++ b/emgauwa-core/src/handlers/v1/ws/mod.rs @@ -14,7 +14,7 @@ pub mod controllers; #[get("/api/v1/ws/controllers")] pub async fn ws_controllers( pool: web::Data>, - app_server: web::Data>, + app_state: web::Data>, req: HttpRequest, stream: web::Payload, ) -> Result { @@ -22,7 +22,7 @@ pub async fn ws_controllers( ControllerWs { pool: pool.get_ref().clone(), controller_uid: None, - app_server: app_server.get_ref().clone(), + app_state: app_state.get_ref().clone(), hb: Instant::now(), }, &req, diff --git a/emgauwa-core/src/main.rs b/emgauwa-core/src/main.rs index 00f2561..10320cc 100644 --- a/emgauwa-core/src/main.rs +++ b/emgauwa-core/src/main.rs @@ -33,7 +33,7 @@ async fn main() -> Result<(), std::io::Error> { .map_err(EmgauwaError::from)?; conn.close().await.map_err(EmgauwaError::from)?; - let app_server = AppState::new(pool.clone()).start(); + let app_state = AppState::new(pool.clone()).start(); log::info!("Starting server on {}:{}", settings.host, settings.port); @@ -54,7 +54,7 @@ async fn main() -> Result<(), std::io::Error> { .wrap(middleware::NormalizePath::new(TrailingSlash::Trim)) .app_data(web::JsonConfig::default().error_handler(handlers::json_error_handler)) .app_data(web::Data::new(pool.clone())) - .app_data(web::Data::new(app_server.clone())) + .app_data(web::Data::new(app_state.clone())) .service(handlers::v1::controllers::index) .service(handlers::v1::controllers::show) .service(handlers::v1::controllers::update) diff --git a/emgauwa-lib/src/constants.rs b/emgauwa-lib/src/constants.rs index acf6777..b4e93cb 100644 --- a/emgauwa-lib/src/constants.rs +++ b/emgauwa-lib/src/constants.rs @@ -5,3 +5,4 @@ pub const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); pub const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(15); pub const WEBSOCKET_RETRY_TIMEOUT: Duration = Duration::from_secs(5); +pub const RELAYS_RETRY_TIMEOUT: Duration = Duration::from_secs(5);