use emgauwa_lib::constants::WEBSOCKET_RETRY_TIMEOUT; use emgauwa_lib::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; use emgauwa_lib::errors::EmgauwaError; use emgauwa_lib::models::{Controller, FromDbModel}; use emgauwa_lib::types::{ControllerUid, ControllerWsAction}; use emgauwa_lib::{db, utils}; use futures::{SinkExt, StreamExt}; use sqlx::pool::PoolConnection; use sqlx::Sqlite; use tokio::time; use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::protocol::Message; use tokio_tungstenite::tungstenite::Error; use utils::init_logging; use crate::relay_loop::run_relay_loop; use crate::settings::Settings; mod driver; mod relay_loop; mod settings; async fn create_this_controller( conn: &mut PoolConnection, settings: &Settings, ) -> Result { DbController::create( conn, &ControllerUid::default(), &settings.name, settings.relays.len() as i64, ) .await .map_err(EmgauwaError::from) } async fn create_this_relay( conn: &mut PoolConnection, this_controller: &DbController, settings_relay: &settings::Relay, ) -> Result { let relay = DbRelay::create( conn, &settings_relay.name, settings_relay.number.ok_or(EmgauwaError::Internal( "Relay number is missing".to_string(), ))?, this_controller, ) .await?; let off = DbSchedule::get_off(conn).await?; for weekday in 0..7 { DbJunctionRelaySchedule::set_schedule(conn, &relay, &off, weekday).await?; } Ok(relay) } async fn run_websocket(this: Controller, url: &str) -> Result<(), EmgauwaError> { match connect_async(url).await { Ok(connection) => { let (ws_stream, _) = connection; let (mut write, read) = ws_stream.split(); let ws_action = ControllerWsAction::Register(this.clone()); let ws_action_json = serde_json::to_string(&ws_action)?; if let Err(err) = write.send(Message::text(ws_action_json)).await { log::error!("Failed to register at websocket: {}", err); return Ok(()); } let read_handler = read.for_each(handle_message); read_handler.await; log::warn!("Lost connection to websocket"); } Err(err) => { log::warn!("Failed to connect to websocket: {}", err,); } } Ok(()) } #[tokio::main] async fn main() -> Result<(), std::io::Error> { let settings = settings::init()?; init_logging(&settings.logging.level)?; let pool = db::init(&settings.database) .await .map_err(EmgauwaError::from)?; let mut conn = pool.acquire().await.map_err(EmgauwaError::from)?; let db_controller = match DbController::get_all(&mut conn) .await .map_err(EmgauwaError::from)? .pop() { None => futures::executor::block_on(create_this_controller(&mut conn, &settings))?, Some(c) => c, }; for relay in &settings.relays { if DbRelay::get_by_controller_and_num( &mut conn, &db_controller, relay.number.ok_or(EmgauwaError::Internal( "Relay number is missing".to_string(), ))?, ) .await .map_err(EmgauwaError::from)? .is_none() { create_this_relay(&mut conn, &db_controller, relay) .await .map_err(EmgauwaError::from)?; } } let db_controller = db_controller .update(&mut conn, &db_controller.name, settings.relays.len() as i64) .await .map_err(EmgauwaError::from)?; let this = Controller::from_db_model(&mut conn, db_controller).map_err(EmgauwaError::from)?; let url = format!( "ws://{}:{}/api/v1/ws/controllers", settings.core.host, settings.core.port ); tokio::spawn(run_relay_loop(settings)); loop { let run_result = run_websocket(this.clone(), &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 handle_message(message_result: Result) { match message_result { Ok(message) => { if let Message::Text(msg_text) = message { log::debug!("{}", msg_text) } } Err(err) => log::debug!("Error: {}", err), } }