diff --git a/Cargo.lock b/Cargo.lock index d61f03c..fc34e9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3887,8 +3887,11 @@ dependencies = [ "hypersdk", "pulse-ui", "pulse-wire", + "rand 0.8.7", + "serde", "serde_json", "tokio", + "toml", ] [[package]] @@ -4728,6 +4731,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -5239,6 +5251,21 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "1.1.3+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c" +dependencies = [ + "indexmap 2.14.0", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + [[package]] name = "toml_datetime" version = "1.1.1+spec-1.1.0" @@ -5269,6 +5296,12 @@ dependencies = [ "winnow", ] +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + [[package]] name = "tonic" version = "0.14.6" diff --git a/Cargo.toml b/Cargo.toml index d516a74..f70095e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,10 +8,20 @@ pulse-ui = { workspace = true } pulse-wire = { workspace = true } chrono = "0.4.45" -tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "fs", "io-util", "time"] } +tokio = { workspace = true, features = [ + "rt-multi-thread", + "macros", + "net", + "fs", + "io-util", + "time", +] } crossterm = { workspace = true } hypersdk = "0.2.14" serde_json = "1" +rand = "0.8.7" +toml = "1.1.3" +serde = { version = "1.0.229", features = ["serde_derive"] } [workspace] members = ["pulse-macros", "pulse-ui", "pulse-wire"] diff --git a/src/engine/config.rs b/src/engine/config.rs new file mode 100644 index 0000000..8555d75 --- /dev/null +++ b/src/engine/config.rs @@ -0,0 +1,54 @@ +use std::path::PathBuf; + +pub fn home_dir() -> tokio::io::Result { + std::env::home_dir().ok_or_else(|| { + tokio::io::Error::new(std::io::ErrorKind::NotFound, "Unable to get home directory") + }) +} + +pub fn pulse_directory() -> tokio::io::Result { + Ok(home_dir()?.join(".config/pulse-trader")) +} + +pub fn pulse_config_directory() -> tokio::io::Result { + Ok(home_dir()?.join(".config/pulse-trader/config.toml")) +} + +#[derive(Debug, Default, serde::Serialize, serde::Deserialize)] +pub struct WatchList { + pub symbols: Vec, +} + +#[derive(Debug, Default, serde::Serialize, serde::Deserialize)] +pub struct Config { + pub watchlist: WatchList, +} + +impl Config { + pub async fn new() -> tokio::io::Result { + let path = pulse_config_directory()?; + + if !path.exists() { + let default = Self::default(); + + tokio::fs::create_dir_all(path.parent().unwrap()).await?; + tokio::fs::write(&path, default.to_string()?).await?; + + return Ok(default); + } + + let output = tokio::fs::read_to_string(path).await?; + + Self::from_str(&output) + } + + pub fn from_str(s: &str) -> tokio::io::Result { + toml::from_str(s) + .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::InvalidInput, v.to_string())) + } + + pub fn to_string(&self) -> tokio::io::Result { + toml::to_string(self) + .map_err(|v| tokio::io::Error::new(std::io::ErrorKind::Other, v.to_string())) + } +} diff --git a/src/engine/engine.rs b/src/engine/engine.rs index ed734b6..0aa96c7 100644 --- a/src/engine/engine.rs +++ b/src/engine/engine.rs @@ -1,60 +1,102 @@ -use crate::terminal::TerminalServer; +use std::sync::Arc; -const WATCH_LIST_SYMBOLS: &[&str] = &["BTC", "ETH", "SOL", "XRP"]; +use tokio::{sync::Mutex, task::JoinHandle}; +use crate::{config::Config, terminal::TerminalServer}; + +#[derive(Debug, Clone)] pub struct Engine { - pub terminal_server: TerminalServer, + pub terminal_server: Arc, + pub config: Arc>, } impl Engine { - pub fn new() -> Self { - Self { - terminal_server: TerminalServer::new(), + pub async fn new() -> tokio::io::Result> { + let config = Arc::new(Mutex::new(Config::new().await?)); + + Ok(Arc::new_cyclic(|engine| Self { + terminal_server: TerminalServer::new(engine.clone()), + config, + })) + } + + pub async fn spawn_terminal_server(&self) -> JoinHandle> { + let terminal_server = self.terminal_server.clone(); + + tokio::spawn(async move { terminal_server.run().await }) + } + + pub async fn spawn_broadcaster(&self) -> JoinHandle> { + let s = self.clone(); + + tokio::spawn(async move { s.run_broadcaster().await }) + } + + pub async fn run_broadcaster(&self) -> tokio::io::Result<()> { + let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5)); + + loop { + refresh.tick().await; + + let watch_list = &self.config.lock().await.watchlist.symbols; + + match crate::fetch::fetch_watch_list(watch_list).await { + Ok(watch_list) => { + if let Err(error) = self + .terminal_server + .broadcast( + pulse_wire::terminal::TerminalServerMessage::WatchListUpdated( + watch_list, + ), + ) + .await + { + eprintln!("Failed to broadcast Hyperliquid watch list: {error}"); + } + } + Err(error) => eprintln!("Failed to refresh Hyperliquid watch list: {error}"), + } } } - pub fn spawn_terminal_server(&self) { - let terminal_server = self.terminal_server.clone(); - - tokio::spawn(async move { - terminal_server - .run() - .await - .expect("Failed to run terminal server"); - }); - } - - pub fn spawn_broadcaster(&mut self) { - let terminal_server = self.terminal_server.clone(); - - tokio::spawn(async move { - let mut refresh = tokio::time::interval(tokio::time::Duration::from_secs(5)); - - loop { - refresh.tick().await; - - match crate::fetch::fetch_watch_list(WATCH_LIST_SYMBOLS).await { - Ok(watch_list) => { - if let Err(error) = terminal_server - .broadcast( - pulse_wire::terminal::TerminalServerMessage::WatchListUpdated( - watch_list, - ), - ) - .await - { - eprintln!("Failed to broadcast Hyperliquid watch list: {error}"); - } - } - Err(error) => eprintln!("Failed to refresh Hyperliquid watch list: {error}"), - } - } - }); - } - - pub async fn run_engine(&mut self) -> tokio::io::Result<()> { + pub async fn run_engine(&self) -> tokio::io::Result<()> { loop { tokio::time::sleep(tokio::time::Duration::from_millis(5000)).await; } } + + pub async fn execute_command(&self, command: &str, args: Vec<&str>) -> tokio::io::Result<()> { + match command { + "config" => { + const MESSAGE: &str = "Invalid command arguments, usage: config "; + + if args.len() != 1 { + self.terminal_server.error("config", MESSAGE).await?; + + return Ok(()); + } + + match args[0] { + "reload" => { + *self.config.lock().await = Config::new().await?; + } + + _ => { + self.terminal_server.error("config", MESSAGE).await?; + } + } + } + + _ => { + self.terminal_server + .error( + "Command executor", + &format!("Command '{}' not found", command), + ) + .await?; + } + } + + Ok(()) + } } diff --git a/src/engine/fetch.rs b/src/engine/fetch.rs index 1c6ae18..9f7b271 100644 --- a/src/engine/fetch.rs +++ b/src/engine/fetch.rs @@ -12,7 +12,7 @@ fn number(value: &Value, field: &str) -> Result { .map_err(|error| format!("could not parse asset context field {field} ({raw}): {error}")) } -pub async fn fetch_watch_list(symbols: &[&str]) -> Result, String> { +pub async fn fetch_watch_list(symbols: &[String]) -> Result, String> { let response = hypersdk::hypercore::mainnet() .meta_and_asset_ctxs(None) .await @@ -73,7 +73,7 @@ pub async fn fetch_watch_list(symbols: &[&str]) -> Result, St .iter() .map(|symbol| { by_symbol - .remove(*symbol) + .remove(symbol.as_str()) .ok_or_else(|| format!("{symbol} is not in the Hyperliquid perpetual universe")) }) .collect() diff --git a/src/engine/main.rs b/src/engine/main.rs index bb7f06f..2e7cd52 100644 --- a/src/engine/main.rs +++ b/src/engine/main.rs @@ -1,16 +1,22 @@ +pub mod config; pub mod engine; pub mod fetch; pub mod terminal; #[tokio::main] async fn main() -> tokio::io::Result<()> { - let mut engine = engine::Engine::new(); + let engine = engine::Engine::new().await?; - engine.spawn_terminal_server(); + let terminal_server = engine.spawn_terminal_server().await; - engine.spawn_broadcaster(); + let broadcaster = engine.spawn_broadcaster().await; engine.run_engine().await?; + let (terminal_server, broadcaster) = tokio::join!(terminal_server, broadcaster); + + terminal_server??; + broadcaster??; + Ok(()) } diff --git a/src/engine/terminal.rs b/src/engine/terminal.rs index 85a9fb5..4e3a503 100644 --- a/src/engine/terminal.rs +++ b/src/engine/terminal.rs @@ -1,6 +1,12 @@ -use std::sync::Arc; +use std::{ + collections::HashMap, + sync::{Arc, Weak}, +}; -use pulse_wire::{PulseWire, terminal::TerminalClientMessage}; +use pulse_wire::{ + PulseWire, + terminal::{EventLog, LogKind, TerminalClientMessage}, +}; use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::{ @@ -10,19 +16,25 @@ use tokio::{ sync::Mutex, }; -#[derive(Debug, Clone)] +use crate::engine::Engine; + +#[derive(Debug)] pub struct TerminalServer { - clients: Arc>>, + clients: Mutex>, + logs: Mutex>, + engine: Weak, } impl TerminalServer { - pub fn new() -> Self { - Self { - clients: Arc::new(Mutex::new(Vec::new())), - } + pub fn new(engine: Weak) -> Arc { + Arc::new(Self { + clients: Mutex::new(HashMap::new()), + logs: Mutex::new(Vec::new()), + engine, + }) } - pub async fn run(&self) -> tokio::io::Result<()> { + pub async fn run(self: &Arc) -> tokio::io::Result<()> { let path = pulse_wire::server_path(); if path.exists() { @@ -38,19 +50,31 @@ impl TerminalServer { let (reader, writer) = stream.into_split(); - self.clients.lock().await.push(writer); + let id = rand::random(); + + self.clients.lock().await.insert(id, writer); let s = self.clone(); tokio::spawn(async move { - if let Err(err) = s.handle_client(reader).await { + if let Err(err) = s.handle_client(&id, reader).await { eprintln!("Terminal connection error: {err}"); } }); } } - async fn handle_client(&self, mut reader: OwnedReadHalf) -> tokio::io::Result<()> { + async fn handle_client( + self: &Arc, + id: &usize, + mut reader: OwnedReadHalf, + ) -> tokio::io::Result<()> { + self.send_to( + id, + pulse_wire::terminal::TerminalServerMessage::SetLogs(self.logs.lock().await.clone()), + ) + .await?; + loop { let mut len_buf = [0u8; size_of::()]; let size = reader.read_exact(&mut len_buf).await?; @@ -69,24 +93,13 @@ impl TerminalServer { TerminalClientMessage::ExecuteCommand(command) => { let command = command.as_str(); - let (command, _args) = if let Some((command, args)) = command.split_once(" ") { + let (command, args) = if let Some((command, args)) = command.split_once(" ") { (command, args.split(" ").collect()) } else { (command, Vec::new()) }; - match command { - _ => { - self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog( - pulse_wire::terminal::EventLog { - kind: pulse_wire::terminal::LogKind::Err, - name: "Command executor".to_string(), - message: format!("Command '{}' not found", command), - }, - )) - .await?; - } - } + self.get_engine().execute_command(command, args).await?; } } } @@ -95,32 +108,91 @@ impl TerminalServer { } pub async fn broadcast( - &self, + self: &Arc, message: pulse_wire::terminal::TerminalServerMessage, ) -> tokio::io::Result<()> { let msg = message.to_com(); let mut clients = self.clients.lock().await; - for i in (0..clients.len()).rev() { - if let Err(e) = Self::send_to_client(&mut clients, i, &msg).await { - clients.remove(i); + let mut remove_clients = Vec::new(); + + for (id, client) in clients.iter_mut() { + if let Err(e) = Self::send_to_client(client, &msg).await { + remove_clients.push(*id); println!("{e:?}"); } } + for id in remove_clients { + clients.remove(&id); + } + Ok(()) } - pub async fn send_to_client( - clients: &mut tokio::sync::MutexGuard<'_, Vec>, - i: usize, - msg: &[u8], + pub async fn send_to( + self: &Arc, + id: &usize, + message: pulse_wire::terminal::TerminalServerMessage, ) -> tokio::io::Result<()> { - clients[i].write(&msg.len().to_le_bytes()).await?; - clients[i].write(msg).await?; - clients[i].flush().await?; + Self::send_to_client( + self.clients.lock().await.get_mut(id).ok_or_else(|| { + tokio::io::Error::new( + std::io::ErrorKind::Other, + format!("Client({id}) does not exist"), + ) + })?, + &message.to_com(), + ) + .await + } + + pub async fn send_to_client(client: &mut OwnedWriteHalf, msg: &[u8]) -> tokio::io::Result<()> { + client.write(&msg.len().to_le_bytes()).await?; + client.write(msg).await?; + client.flush().await?; Ok(()) } + + pub async fn log( + self: &Arc, + kind: LogKind, + name: &str, + message: &str, + ) -> tokio::io::Result<()> { + let log = EventLog { + kind, + name: name.to_string(), + message: message.to_string(), + }; + + self.logs.lock().await.push(log.clone()); + + self.broadcast(pulse_wire::terminal::TerminalServerMessage::AddLog(log)) + .await + } + + pub async fn info(self: &Arc, name: &str, message: &str) -> tokio::io::Result<()> { + self.log(LogKind::Info, name, message).await + } + + pub async fn warn(self: &Arc, name: &str, message: &str) -> tokio::io::Result<()> { + self.log(LogKind::Warn, name, message).await + } + + pub async fn error(self: &Arc, name: &str, message: &str) -> tokio::io::Result<()> { + self.log(LogKind::Err, name, message).await + } + + pub async fn debug(self: &Arc, name: &str, message: &str) -> tokio::io::Result<()> { + self.log(LogKind::Debug, name, message).await + } + + pub fn get_engine(&self) -> Arc { + self.engine + .upgrade() + .expect("Failed to upgrade engine(Weak) to Arc") + } }