Improve active handling for controllers

This commit is contained in:
Tobias Reisinger 2023-11-28 00:19:15 +01:00
parent ec461a1a14
commit 6459804e1f
Signed by: serguzim
GPG key ID: 13AD60C237A28DFE
7 changed files with 87 additions and 35 deletions

1
.gitattributes vendored
View file

@ -1,3 +1,4 @@
* text=auto * text=auto
Cargo.lock -diff Cargo.lock -diff
.sqlx/query-*.json -diff

View file

@ -3,10 +3,16 @@ build:
cargo build cargo build
sqlx: sqlx:
rm ./emgauwa-dev.sqlite rm ./emgauwa-dev.sqlite || true
cargo sqlx database create cargo sqlx database create
cargo sqlx migrate run cargo sqlx migrate run
cargo sqlx prepare --workspace cargo sqlx prepare --workspace
build-rpi: build-rpi:
cross build --target arm-unknown-linux-gnueabihf cross build --target arm-unknown-linux-gnueabihf
clean-db:
rm ./emgauwa-dev.sqlite || true
rm ./emgauwa-core.sqlite || true
rm ./emgauwa-controller.sqlite || true
$(MAKE) sqlx

View file

@ -28,7 +28,6 @@ async fn create_this_controller(
&ControllerUid::default(), &ControllerUid::default(),
&settings.name, &settings.name,
i64::try_from(settings.relays.len()).expect("Too many relays"), i64::try_from(settings.relays.len()).expect("Too many relays"),
true,
) )
.await .await
.expect("Failed to create controller") .expect("Failed to create controller")
@ -86,7 +85,7 @@ async fn main() {
.collect(); .collect();
let db_controller = db_controller let db_controller = db_controller
.update(&mut conn, &db_controller.name, db_relays.len() as i64, true) .update(&mut conn, &db_controller.name, db_relays.len() as i64)
.await .await
.unwrap(); .unwrap();

View file

@ -5,6 +5,7 @@ use std::str::FromStr;
use crate::utils::drop_privileges; use crate::utils::drop_privileges;
use actix_web::middleware::TrailingSlash; use actix_web::middleware::TrailingSlash;
use actix_web::{middleware, web, App, HttpServer}; use actix_web::{middleware, web, App, HttpServer};
use emgauwa_lib::db::DbController;
use emgauwa_lib::handlers; use emgauwa_lib::handlers;
use log::{trace, LevelFilter}; use log::{trace, LevelFilter};
use simple_logger::SimpleLogger; use simple_logger::SimpleLogger;
@ -32,6 +33,14 @@ async fn main() -> std::io::Result<()> {
let pool = emgauwa_lib::db::init(&settings.database).await; let pool = emgauwa_lib::db::init(&settings.database).await;
// This block is to ensure that the connection is dropped after use.
{
let mut conn = pool.acquire().await.unwrap();
DbController::all_inactive(&mut conn)
.await
.expect("Error setting all controllers inactive");
}
log::info!("Starting server on {}:{}", settings.host, settings.port); log::info!("Starting server on {}:{}", settings.host, settings.port);
HttpServer::new(move || { HttpServer::new(move || {
let cors = Cors::default().allow_any_method().allow_any_header(); let cors = Cors::default().allow_any_method().allow_any_header();

View file

@ -58,11 +58,10 @@ impl DbController {
uid: &ControllerUid, uid: &ControllerUid,
new_name: &str, new_name: &str,
new_relay_count: i64, new_relay_count: i64,
new_active: bool,
) -> Result<DbController, DatabaseError> { ) -> Result<DbController, DatabaseError> {
match DbController::get_by_uid(conn, uid).await? { match DbController::get_by_uid(conn, uid).await? {
Some(tag) => Ok(tag), Some(tag) => Ok(tag),
None => DbController::create(conn, uid, new_name, new_relay_count, new_active).await, None => DbController::create(conn, uid, new_name, new_relay_count).await,
} }
} }
@ -94,15 +93,13 @@ impl DbController {
new_uid: &ControllerUid, new_uid: &ControllerUid,
new_name: &str, new_name: &str,
new_relay_count: i64, new_relay_count: i64,
new_active: bool,
) -> Result<DbController, DatabaseError> { ) -> Result<DbController, DatabaseError> {
sqlx::query_as!( sqlx::query_as!(
DbController, DbController,
"INSERT INTO controllers (uid, name, relay_count, active) VALUES (?, ?, ?, ?) RETURNING *", "INSERT INTO controllers (uid, name, relay_count) VALUES (?, ?, ?) RETURNING *",
new_uid, new_uid,
new_name, new_name,
new_relay_count, new_relay_count,
new_active,
) )
.fetch_optional(conn.deref_mut()) .fetch_optional(conn.deref_mut())
.await? .await?
@ -114,12 +111,28 @@ impl DbController {
conn: &mut PoolConnection<Sqlite>, conn: &mut PoolConnection<Sqlite>,
new_name: &str, new_name: &str,
new_relay_count: i64, new_relay_count: i64,
) -> Result<DbController, DatabaseError> {
sqlx::query!(
"UPDATE controllers SET name = ?, relay_count = ? WHERE id = ?",
new_name,
new_relay_count,
self.id,
)
.execute(conn.deref_mut())
.await?;
Self::get(conn, self.id)
.await?
.ok_or(DatabaseError::UpdateGetError)
}
pub async fn update_active(
&self,
conn: &mut PoolConnection<Sqlite>,
new_active: bool, new_active: bool,
) -> Result<DbController, DatabaseError> { ) -> Result<DbController, DatabaseError> {
sqlx::query!( sqlx::query!(
"UPDATE controllers SET name = ?, relay_count = ?, active = ? WHERE id = ?", "UPDATE controllers SET active = ? WHERE id = ?",
new_name,
new_relay_count,
new_active, new_active,
self.id, self.id,
) )
@ -130,4 +143,11 @@ impl DbController {
.await? .await?
.ok_or(DatabaseError::UpdateGetError) .ok_or(DatabaseError::UpdateGetError)
} }
pub async fn all_inactive(conn: &mut PoolConnection<Sqlite>) -> Result<(), DatabaseError> {
sqlx::query!("UPDATE controllers SET active = 0")
.execute(conn.deref_mut())
.await?;
Ok(())
}
} }

View file

@ -18,10 +18,18 @@ pub enum ControllerWsAction {
struct ControllerWs { struct ControllerWs {
pub pool: Pool<Sqlite>, pub pool: Pool<Sqlite>,
pub controller: Option<DbController>,
} }
impl Actor for ControllerWs { impl Actor for ControllerWs {
type Context = ws::WebsocketContext<Self>; type Context = ws::WebsocketContext<Self>;
fn stopped(&mut self, _ctx: &mut Self::Context) {
if let Some(controller) = &self.controller {
let mut pool_conn = futures::executor::block_on(self.pool.acquire()).unwrap();
futures::executor::block_on(controller.update_active(&mut pool_conn, false)).unwrap();
}
}
} }
impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs { impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs {
@ -32,7 +40,8 @@ impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs {
Ok(Message::Ping(msg)) => ctx.pong(&msg), Ok(Message::Ping(msg)) => ctx.pong(&msg),
Ok(Message::Text(text)) => { Ok(Message::Text(text)) => {
let action: ControllerWsAction = serde_json::from_str(&text).unwrap(); let action: ControllerWsAction = serde_json::from_str(&text).unwrap();
let action_res = futures::executor::block_on(handle_action(&mut pool_conn, action)); let action_res =
futures::executor::block_on(self.handle_action(&mut pool_conn, action));
if let Err(e) = action_res { if let Err(e) = action_res {
log::error!("Error handling action: {:?}", e); log::error!("Error handling action: {:?}", e);
ctx.text(serde_json::to_string(&e).unwrap()); ctx.text(serde_json::to_string(&e).unwrap());
@ -47,33 +56,39 @@ impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs {
} }
} }
pub async fn handle_action( impl ControllerWs {
conn: &mut PoolConnection<Sqlite>, pub async fn handle_action(
action: ControllerWsAction, &mut self,
) -> Result<(), DatabaseError> { conn: &mut PoolConnection<Sqlite>,
match action { action: ControllerWsAction,
ControllerWsAction::Register(controller) => { ) -> Result<(), DatabaseError> {
log::info!("Registering controller: {:?}", controller); match action {
let c = &controller.controller; ControllerWsAction::Register(controller) => {
let controller_db = log::info!("Registering controller: {:?}", controller);
DbController::get_by_uid_or_create(conn, &c.uid, &c.name, c.relay_count, c.active) let c = &controller.controller;
let controller_db =
DbController::get_by_uid_or_create(conn, &c.uid, &c.name, c.relay_count)
.await?;
controller_db.update_active(conn, true).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?; .await?;
println!("Controller relay: {:?}", relay_db);
}
println!("Controller: {:?}", controller_db); self.controller = Some(controller_db);
for relay in &controller.relays { Ok(())
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(())
} }
} }
} }
@ -87,6 +102,7 @@ pub async fn index(
let resp = ws::start( let resp = ws::start(
ControllerWs { ControllerWs {
pool: pool.get_ref().clone(), pool: pool.get_ref().clone(),
controller: None,
}, },
&req, &req,
stream, stream,

View file

@ -18,6 +18,7 @@ CREATE TABLE controllers
active active
BOOLEAN BOOLEAN
NOT NULL NOT NULL
DEFAULT false
); );
CREATE TABLE relays CREATE TABLE relays