summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/main.rs12
-rw-r--r--src/routes/start.rs10
-rw-r--r--src/services/ping.rs28
3 files changed, 26 insertions, 24 deletions
diff --git a/src/main.rs b/src/main.rs
index 9c31ec8..124c44e 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -4,7 +4,7 @@ use axum::{Router, routing::post};
4use axum::routing::{get, put}; 4use axum::routing::{get, put};
5use sqlx::PgPool; 5use sqlx::PgPool;
6use time::util::local_offset; 6use time::util::local_offset;
7use tokio::sync::mpsc::{self, Sender}; 7use tokio::sync::broadcast::{channel, Sender};
8use tracing::{info, level_filters::LevelFilter}; 8use tracing::{info, level_filters::LevelFilter};
9use tracing_subscriber::{EnvFilter, fmt::{self, time::LocalTime}, prelude::*}; 9use tracing_subscriber::{EnvFilter, fmt::{self, time::LocalTime}, prelude::*};
10use crate::config::SETTINGS; 10use crate::config::SETTINGS;
@@ -46,15 +46,8 @@ async fn main() {
46 let db = init_db_pool().await; 46 let db = init_db_pool().await;
47 sqlx::migrate!().run(&db).await.unwrap(); 47 sqlx::migrate!().run(&db).await.unwrap();
48 48
49 let (tx, mut rx) = mpsc::channel(32); 49 let (tx, _) = channel(32);
50 50
51 // FIXME: once_cell? or just static mutable
52 tokio::spawn( async move {
53 while let Some(message) = rx.recv().await {
54 println!("GOT = {}", message);
55 }
56 });
57
58 let shared_state = Arc::new(AppState { db, ping_send: tx }); 51 let shared_state = Arc::new(AppState { db, ping_send: tx });
59 52
60 let app = Router::new() 53 let app = Router::new()
@@ -76,5 +69,4 @@ async fn main() {
76pub struct AppState { 69pub struct AppState {
77 db: PgPool, 70 db: PgPool,
78 ping_send: Sender<String>, 71 ping_send: Sender<String>,
79 // ping_receive: Receiver<String>
80} 72}
diff --git a/src/routes/start.rs b/src/routes/start.rs
index b45fe5b..863ef16 100644
--- a/src/routes/start.rs
+++ b/src/routes/start.rs
@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
4use std::sync::Arc; 4use std::sync::Arc;
5use axum::extract::State; 5use axum::extract::State;
6use serde_json::{json, Value}; 6use serde_json::{json, Value};
7use tracing::info; 7use tracing::{debug, info};
8use crate::auth::auth; 8use crate::auth::auth;
9use crate::config::SETTINGS; 9use crate::config::SETTINGS;
10use crate::wol::{create_buffer, send_packet}; 10use crate::wol::{create_buffer, send_packet};
@@ -39,8 +39,12 @@ pub async fn start(State(state): State<Arc<crate::AppState>>, headers: HeaderMap
39 )?; 39 )?;
40 40
41 if payload.ping.is_some_and(|ping| ping) { 41 if payload.ping.is_some_and(|ping| ping) {
42 tokio::spawn(async move {crate::services::ping::spawn(state.ping_send.clone()).await}); 42 debug!("ping true");
43 } 43 tokio::spawn(async move {
44 debug!("Init ping service");
45 crate::services::ping::spawn(state.ping_send.clone()).await
46 });
47 };
44 Ok(Json(json!(StartResponse { id: device.id, boot: true }))) 48 Ok(Json(json!(StartResponse { id: device.id, boot: true })))
45 } else { 49 } else {
46 Err(WebolError::Generic) 50 Err(WebolError::Generic)
diff --git a/src/services/ping.rs b/src/services/ping.rs
index 6e710ec..ff328a5 100644
--- a/src/services/ping.rs
+++ b/src/services/ping.rs
@@ -1,8 +1,9 @@
1use std::borrow::Cow;
1use std::sync::Arc; 2use std::sync::Arc;
2 3
3use axum::{extract::{WebSocketUpgrade, ws::WebSocket, State}, response::Response}; 4use axum::{extract::{WebSocketUpgrade, ws::WebSocket, State}, response::Response};
4use tokio::sync::mpsc::Sender; 5use tokio::sync::broadcast::{Sender};
5use tracing::{debug, error}; 6use tracing::{debug, error, trace};
6 7
7use crate::{error::WebolError, AppState}; 8use crate::{error::WebolError, AppState};
8 9
@@ -12,7 +13,7 @@ pub async fn spawn(tx: Sender<String>) -> Result<(), WebolError> {
12 let mut cont = true; 13 let mut cont = true;
13 while cont { 14 while cont {
14 let ping = surge_ping::ping( 15 let ping = surge_ping::ping(
15 "192.168.178.28".parse().map_err(WebolError::IpParse)?, 16 "127.0.0.1".parse().map_err(WebolError::IpParse)?,
16 &payload 17 &payload
17 ).await; 18 ).await;
18 19
@@ -30,24 +31,29 @@ pub async fn spawn(tx: Sender<String>) -> Result<(), WebolError> {
30 debug!("Ping took {:?}", duration); 31 debug!("Ping took {:?}", duration);
31 cont = false; 32 cont = false;
32 // FIXME: remove unwrap 33 // FIXME: remove unwrap
33 tx.send("Got ping".to_string()).await.unwrap(); 34 tx.send("Got ping".to_string()).unwrap();
34 }; 35 };
35 } 36 }
36 37
37 Ok(()) 38 Ok(())
38} 39}
39 40
40pub async fn ws_ping(ws: WebSocketUpgrade, State(_state): State<Arc<AppState>>) -> Response { 41// TODO: Status to routes, websocket here
41 ws.on_upgrade(handle_socket) 42pub async fn ws_ping(State(state): State<Arc<AppState>>, ws: WebSocketUpgrade) -> Response {
43 ws.on_upgrade(move |socket| handle_socket(socket, state.ping_send.clone()))
42} 44}
43 45
44// FIXME: Handle commands through enum 46// FIXME: Handle commands through enum
45async fn handle_socket(mut socket: WebSocket) { 47async fn handle_socket(mut socket: WebSocket, tx: Sender<String>) {
46 // TODO: Understand Cow 48 // TODO: Understand Cow
47 49 while let message = tx.subscribe().recv().await.unwrap() {
48 // match socket.send(axum::extract::ws::Message::Close(Some(CloseFrame { code: 4000, reason: Cow::Owned("started".to_owned()) }))).await.map_err(WebolError::Axum) { 50 trace!("GOT = {}", message);
49 match socket.send(axum::extract::ws::Message::Text("started".to_string())).await.map_err(WebolError::Axum) { 51 if &message == "Got ping" {
52 break;
53 }
54 };
55 match socket.send(axum::extract::ws::Message::Close(Some(axum::extract::ws::CloseFrame { code: 4000, reason: Cow::Owned("started".to_owned()) }))).await.map_err(WebolError::Axum) {
50 Ok(..) => (), 56 Ok(..) => (),
51 Err(err) => { error!("Server Error: {:?}", err) } 57 Err(err) => { error!("Server Error: {:?}", err) }
52 }; 58 };
53} 59} \ No newline at end of file