use emgauwa_lib::constants::WEBSOCKET_RETRY_TIMEOUT; use emgauwa_lib::db::errors::DatabaseError; use emgauwa_lib::db::{DbController, DbJunctionRelaySchedule, DbRelay, DbSchedule}; 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, ) -> DbController { DbController::create( conn, &ControllerUid::default(), &settings.name, settings.relays.len() as i64, ) .await .expect("Failed to create controller") } 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.unwrap(), 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) } #[tokio::main] async fn main() { let settings = settings::init(); init_logging(&settings.logging.level); let pool = db::init(&settings.database).await; let mut conn = pool.acquire().await.unwrap(); let db_controller = DbController::get_all(&mut conn) .await .expect("Failed to get controller from database") .pop() .unwrap_or_else(|| { futures::executor::block_on(create_this_controller(&mut conn, &settings)) }); for relay in &settings.relays { if DbRelay::get_by_controller_and_num(&mut conn, &db_controller, relay.number.unwrap()) .await .expect("Failed to get relay from database") .is_none() { create_this_relay(&mut conn, &db_controller, relay) .await .expect("Failed to create schedule."); } } let db_controller = db_controller .update(&mut conn, &db_controller.name, settings.relays.len() as i64) .await .unwrap(); let this = Controller::from_db_model(&mut conn, db_controller) .expect("Failed to convert database models"); let url = format!( "ws://{}:{}/api/v1/ws/controllers", settings.core.host, settings.core.port ); tokio::spawn(run_relay_loop(settings)); loop { log::info!( "Trying to connect in {} seconds...", WEBSOCKET_RETRY_TIMEOUT.as_secs() ); time::sleep(WEBSOCKET_RETRY_TIMEOUT).await; let connect_result = connect_async(&url).await; if let Err(err) = connect_result { log::warn!("Failed to connect to websocket: {}", err,); continue; } let (ws_stream, _) = connect_result.unwrap(); let (mut write, read) = ws_stream.split(); let ws_action = ControllerWsAction::Register(this.clone()); let ws_action_json = serde_json::to_string(&ws_action).unwrap(); write.send(Message::text(ws_action_json)).await.unwrap(); let read_handler = read.for_each(handle_message); read_handler.await; log::warn!("Lost connection to websocket"); } } 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), } }