emgauwa/common
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add WIP websocket to controller and refactor settings

Browse source
This commit is contained in:
Tobias Reisinger 2023-11-23 16:00:24 +01:00
parent 452454f9e8
commit 32c75ad73a
Signed by: serguzim
GPG key ID: 13AD60C237A28DFE
11 changed files with 128 additions and 37 deletions

31
Cargo.lock generated
View file

@ -650,9 +650,9 @@ dependencies = [
[[package]]
name = "config"
version = "0.13.3"
version = "0.13.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d379af7f68bfc21714c6c7dea883544201741d2ce8274bb12fa54f89507f52a7"
checksum = "23738e11972c7643e4ec947840fc463b6a571afcd3e735bdfce7d03c7a784aca"
dependencies = [
"async-trait",
"json5",
@ -845,7 +845,6 @@ name = "emgauwa-controller"
version = "0.5.0"
dependencies = [
"chrono",
"config",
"emgauwa-lib",
"futures",
"log",
@ -853,6 +852,7 @@ dependencies = [
"serde_derive",
"serde_json",
"simple_logger",
"tokio",
"tokio-tungstenite",
"uuid",
]
@ -865,7 +865,6 @@ dependencies = [
"actix-web",
"actix-web-actors",
"chrono",
"config",
"emgauwa-lib",
"futures",
"log",
@ -884,6 +883,7 @@ dependencies = [
"actix-web",
"actix-web-actors",
"chrono",
"config",
"futures",
"libsqlite3-sys",
"log",
@ -1636,6 +1636,16 @@ dependencies = [
"libm",
]
[[package]]
name = "num_cpus"
version = "1.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
dependencies = [
"hermit-abi",
"libc",
]
[[package]]
name = "num_threads"
version = "0.1.6"
@ -2530,13 +2540,26 @@ dependencies = [
"bytes",
"libc",
"mio",
"num_cpus",
"parking_lot",
"pin-project-lite",
"signal-hook-registry",
"socket2 0.5.5",
"tokio-macros",
"windows-sys 0.48.0",
]
[[package]]
name = "tokio-macros"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.38",
]
[[package]]
name = "tokio-tungstenite"
version = "0.20.1"

11
Makefile Normal file
View file

@ -0,0 +1,11 @@
build:
cargo build
sqlx: build
cargo sqlx database create
cargo sqlx migrate run
cargo sqlx prepare
build-rpi:
cross build --target arm-unknown-linux-gnueabihf

9
emgauwa-controller.toml Normal file
View file

@ -0,0 +1,9 @@
database = "sqlite://emgauwa-controller.sqlite"
[core]
port = 4419
host = "127.0.0.1"
[logging]
level = "DEBUG"
file = "stdout"

5
emgauwa-controller/Cargo.toml
View file

@ -7,8 +7,7 @@ authors = ["Tobias Reisinger <tobias@msrg.cc>"]
[dependencies]
emgauwa-lib = { path = "../emgauwa-lib" }
config = "0.13"
tokio = { version = "1.34", features = ["io-std", "macros", "rt-multi-thread"] }
tokio-tungstenite = "0.20"
simple_logger = "4.2"
@ -21,4 +20,4 @@ serde = "1.0"
serde_json = "1.0"
serde_derive = "1.0"
futures = "0.3.29"
futures = "0.3"

52
emgauwa-controller/src/main.rs
View file

@ -1,5 +1,53 @@
use std::str;
use futures::{future, pin_mut, StreamExt};
use futures::channel::mpsc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message};
mod settings;
fn main() {
let _settings = settings::init();
#[tokio::main]
async fn main() {
let settings = settings::init();
let url = format!(
"ws://{}:{}/api/v1/ws/controllers",
settings.core.host,
settings.core.port
);
let (stdin_tx, stdin_rx) = mpsc::unbounded();
tokio::spawn(read_stdin(stdin_tx));
let (ws_stream, _) = connect_async(url).await.expect("Failed to connect");
println!("WebSocket handshake has been successfully completed");
let (write, read) = ws_stream.split();
let stdin_to_ws = stdin_rx.map(Ok).forward(write);
let ws_to_stdout = {
read.for_each(|message| async {
let data = message.unwrap().into_text().unwrap();
println!("{}", data);
})
};
pin_mut!(stdin_to_ws, ws_to_stdout);
future::select(stdin_to_ws, ws_to_stdout).await;
}
// Our helper method which will read data from stdin and send it along the
// sender provided.
async fn read_stdin(tx: mpsc::UnboundedSender<Message>) {
let mut stdin = tokio::io::stdin();
loop {
let mut buf = vec![0; 1024];
let n = match stdin.read(&mut buf).await {
Err(_) | Ok(0) => break,
Ok(n) => n,
};
buf.truncate(n);
tx.unbounded_send(Message::text(str::from_utf8(&buf).unwrap())).unwrap();
}
}

17
emgauwa-controller/src/settings.rs
View file

@ -1,5 +1,4 @@
use config::Config;
use emgauwa_lib::constants;
use emgauwa_lib::{constants, utils};
use serde_derive::Deserialize;
#[derive(Clone, Debug, Deserialize)]
@ -22,6 +21,7 @@ pub struct Logging {
#[serde(default)]
#[allow(unused)]
pub struct Settings {
pub core: Core,
pub database: String,
pub logging: Logging,
}
@ -29,6 +29,7 @@ pub struct Settings {
impl Default for Settings {
fn default() -> Self {
Settings {
core: Core::default(),
database: String::from("sqlite://emgauwa-controller.sqlite"),
logging: Logging::default(),
}
@ -54,15 +55,5 @@ impl Default for Logging {
}
pub fn init() -> Settings {
Config::builder()
.add_source(config::File::with_name("emgauwa-controller"))
.add_source(
config::Environment::with_prefix("EMGAUWA_CONTROLLER")
.prefix_separator("_")
.separator("__"),
)
.build()
.unwrap()
.try_deserialize::<Settings>()
.unwrap_or_else(|_| panic!("Error reading settings."))
utils::load_settings("controller", "CONTROLLER")
}

2
emgauwa-core/Cargo.toml
View file

@ -11,8 +11,6 @@ actix = "0.13"
actix-web = "4.4"
actix-web-actors = "4.2"
config = "0.13"
simple_logger = "4.2"
log = "0.4"

15
emgauwa-core/src/settings.rs
View file

@ -1,5 +1,4 @@
use config::Config;
use emgauwa_lib::constants;
use emgauwa_lib::{constants, utils};
use serde_derive::Deserialize;
#[derive(Clone, Debug, Deserialize)]
@ -41,15 +40,5 @@ impl Default for Logging {
}
pub fn init() -> Settings {
Config::builder()
.add_source(config::File::with_name("emgauwa-core"))
.add_source(
config::Environment::with_prefix("EMGAUWA_CORE")
.prefix_separator("__")
.separator("__"),
)
.build()
.unwrap()
.try_deserialize::<Settings>()
.expect("Error reading settings.")
utils::load_settings("core", "CORE")
}

2
emgauwa-lib/Cargo.toml
View file

@ -14,6 +14,8 @@ serde = "1.0"
serde_json = "1.0"
serde_derive = "1.0"
config = "0.13"
chrono = { version = "0.4", features = ["serde"] }
sqlx = { version = "0.7", features = ["sqlite", "runtime-async-std", "macros", "chrono"] }

1
emgauwa-lib/src/handlers/v1/ws/controllers.rs
View file

@ -25,6 +25,7 @@ impl StreamHandler<Result<Message, ProtocolError>> for ControllerWs {
fn handle(&mut self, msg: Result<Message, ProtocolError>, ctx: &mut Self::Context) {
let schedules = futures::executor::block_on(get_schedules(&mut self.pool)).unwrap();
let schedules_json = serde_json::to_string(&schedules).unwrap();
println!("{:?}", msg);
match msg {
Ok(Message::Ping(msg)) => ctx.pong(&msg),
Ok(Message::Text(_text)) => ctx.text(schedules_json),

20
emgauwa-lib/src/utils.rs
View file

@ -1,3 +1,23 @@
pub fn vec_has_error<T, E>(target: &[Result<T, E>]) -> bool {
target.iter().any(|t| t.is_err())
}
pub fn load_settings<T>(config_name: &str, env_prefix: &str) -> T
where
for<'de> T: serde::Deserialize<'de>
{
let default_file = config::File::with_name(&format!("emgauwa-{}", config_name))
.required(false);
config::Config::builder()
.add_source(default_file)
.add_source(
config::Environment::with_prefix(&format!("EMGAUWA_{}", env_prefix))
.prefix_separator("__")
.separator("__"),
)
.build()
.expect("Error building settings")
.try_deserialize::<T>()
.expect("Error reading settings")
}