use crate::db::errors::DatabaseError; use crate::db::{DbController, DbRelay}; use crate::handlers::errors::ApiError; use crate::models::Controller; use actix::{Actor, StreamHandler}; use actix_web::{get, web, HttpRequest, HttpResponse}; use actix_web_actors::ws; use actix_web_actors::ws::ProtocolError; use futures::FutureExt; use serde_derive::{Deserialize, Serialize}; use sqlx::pool::PoolConnection; use sqlx::{Pool, Sqlite}; use ws::Message; #[derive(Debug, Serialize, Deserialize)] pub enum ControllerWsAction { Register(Controller), } struct ControllerWs { pub pool: Pool<Sqlite>, } impl Actor for ControllerWs { type Context = ws::WebsocketContext<Self>; } impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs { fn handle(&mut self, msg: Result<Message, ProtocolError>, ctx: &mut Self::Context) { let mut pool_conn = futures::executor::block_on(self.pool.acquire()).unwrap(); match msg { Ok(Message::Ping(msg)) => ctx.pong(&msg), Ok(Message::Text(text)) => { let action: ControllerWsAction = serde_json::from_str(&text).unwrap(); let action_res = futures::executor::block_on(handle_action(&mut pool_conn, action)); if let Err(e) = action_res { log::error!("Error handling action: {:?}", e); ctx.text(serde_json::to_string(&e).unwrap()); } } _ => {} } //let schedules = futures::executor::block_on(DbSchedule::get_all(&mut pool_conn)).unwrap(); //let schedules_json = serde_json::to_string(&schedules).unwrap(); //ctx.text(schedules_json); } } pub async fn handle_action( conn: &mut PoolConnection<Sqlite>, action: ControllerWsAction, ) -> Result<(), DatabaseError> { match action { ControllerWsAction::Register(controller) => { log::info!("Registering controller: {:?}", controller); let c = &controller.controller; let controller_db = DbController::get_by_uid_or_create(conn, &c.uid, &c.name, c.relay_count, c.active) .await?; println!("Controller: {:?}", controller_db); for relay in &controller.relays { let r = &relay.relay; let relay_db = DbRelay::get_by_controller_and_num_or_create( conn, &controller_db, r.number, &r.name, ) .await?; println!("Controller relay: {:?}", relay_db); } Ok(()) } } } #[get("/api/v1/ws/controllers")] pub async fn index( pool: web::Data<Pool<Sqlite>>, req: HttpRequest, stream: web::Payload, ) -> Result<HttpResponse, ApiError> { let resp = ws::start( ControllerWs { pool: pool.get_ref().clone(), }, &req, stream, ) .map_err(|_| ApiError::InternalError(String::from("error starting websocket"))); println!("{:?}", resp); resp }