Add notifier to break relay loop

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<Notify>")]
+pub struct GetNotifier {}
+
 pub struct AppState {
 	pub pool: Pool<Sqlite>,
 	pub this: Controller,
+	pub notifier: Arc<Notify>,
 }
 
 impl AppState {
 	pub fn new(pool: Pool<Sqlite>, 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<Reload> for AppState {
 
 		self.this.reload(&mut pool_conn)?;
 
+		self.notify_change();
+
 		Ok(())
 	}
 }
@@ -46,3 +64,11 @@ impl Handler<GetThis> for AppState {
 		self.this.clone()
 	}
 }
+
+impl Handler<GetNotifier> for AppState {
+	type Result = Arc<Notify>;
+
+	fn handle(&mut self, _msg: GetNotifier, _ctx: &mut Self::Context) -> Self::Result {
+		Arc::clone(&self.notifier)
+	}
+}

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 {
+	tokio::select! {
-		let run_result = run_websocket(pool.clone(), &app_state, &url).await;
+		_ = run_relays_loop(app_state.clone()) => {},
-		if let Err(err) = run_result {
+		_ = run_ws_loop(pool.clone(), app_state.clone(), url) => {},
-			log::error!("Error running websocket: {}", err);
 	}
 
-		log::info!(
+	Ok(())
-			"Retrying to connect in {} seconds...",
-			WEBSOCKET_RETRY_TIMEOUT.as_secs()
-		);
-		time::sleep(WEBSOCKET_RETRY_TIMEOUT).await;
-	}
 }

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_relays_loop(app_state: Addr<AppState>) {
-pub async fn run_relay_loop(settings: Settings) {
+	log::debug!("Spawned relays loop");
-	let default_duration = Duration::from_millis(10000);
 	loop {
-		let next_timestamp = Instant::now() + default_duration;
+		let run_result = run_relays(&app_state).await;
-		time::sleep_until(next_timestamp).await;
+		if let Err(err) = run_result {
-		log::debug!("Relay loop: {}", Local::now().naive_local().time())
+			log::error!("Error running relays: {}", err);
+		}
+		time::sleep(RELAYS_RETRY_TIMEOUT).await;
+	}
+}
+
+async fn run_relays(app_state: &Addr<AppState>) -> 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
+		);
 	}
 }

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<AppState>) -> Result<Controller, EmgauwaError> {
+pub async fn app_state_get_this(app_state: &Addr<AppState>) -> Result<Controller, EmgauwaError> {
-	app_state.send(GetThis {}).await.map_err(EmgauwaError::from)
+	app_state
+		.send(app_state::GetThis {})
+		.await
+		.map_err(EmgauwaError::from)
+}
+
+pub async fn app_state_get_notifier(
+	app_state: &Addr<AppState>,
+) -> Result<Arc<Notify>, EmgauwaError> {
+	app_state
+		.send(app_state::GetNotifier {})
+		.await
+		.map_err(EmgauwaError::from)
 }

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<Sqlite>, app_state: Addr<AppState>, 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<Sqlite>,
 	app_state: &Addr<AppState>,
 	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<AppState>,
 	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",

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<Pool<Sqlite>>,
-	app_server: web::Data<Addr<AppState>>,
+	app_state: web::Data<Addr<AppState>>,
 	path: web::Path<(String,)>,
 	data: web::Json<RequestUpdateController>,
 ) -> Result<HttpResponse, EmgauwaError> {
@@ -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<Pool<Sqlite>>,
-	app_server: web::Data<Addr<AppState>>,
+	app_state: web::Data<Addr<AppState>>,
 	path: web::Path<(String,)>,
 ) -> Result<HttpResponse, EmgauwaError> {
 	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(),
 		})

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<Pool<Sqlite>>,
-	app_server: web::Data<Addr<AppState>>,
+	app_state: web::Data<Addr<AppState>>,
 	path: web::Path<(String, i64)>,
 	data: web::Json<RequestUpdateRelay>,
 ) -> Result<HttpResponse, EmgauwaError> {
@@ -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()]),

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,
 		}))??;

emgauwa-core/src/handlers/v1/ws/controllers/mod.rs

@@ -19,7 +19,7 @@ use crate::utils::flatten_result;
 pub struct ControllerWs {
 	pub pool: Pool<Sqlite>,
 	pub controller_uid: Option<ControllerUid>,
-	pub app_server: Addr<AppState>,
+	pub app_state: Addr<AppState>,
 	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),

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<Pool<Sqlite>>,
-	app_server: web::Data<Addr<AppState>>,
+	app_state: web::Data<Addr<AppState>>,
 	req: HttpRequest,
 	stream: web::Payload,
 ) -> Result<HttpResponse, EmgauwaError> {
@@ -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,

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)

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);