Command processing

pulse-ui/src/state.rs

@@ -83,7 +83,15 @@ impl<'a, T> Drop for StateGuard<'a, T> {
 }
 
 impl Context {
+    pub async fn event<E: Any + Send + Sync>(&self, event: E) {
+        self.tx.send(Box::new(event)).await.unwrap()
+    }
+
     pub async fn close(&self) {
-        self.tx.send(Box::new(Close)).await.unwrap()
+        self.event(Close).await
+    }
+
+    pub async fn refresh(&self) {
+        self.event(Refresh).await
     }
 }

src/engine/terminal.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use pulse_wire::PulseWire;
+use pulse_wire::{PulseWire, terminal::TerminalClientMessage};
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::{
@@ -40,37 +40,52 @@ impl TerminalServer {
 
             self.clients.lock().await.push(writer);
 
+            let s = self.clone();
+
             tokio::spawn(async move {
-                if let Err(err) = Self::handle_client(reader).await {
+                if let Err(err) = s.handle_client(reader).await {
                     eprintln!("Terminal connection error: {err}");
                 }
             });
         }
     }
 
-    async fn handle_client(mut reader: OwnedReadHalf) -> tokio::io::Result<()> {
+    async fn handle_client(&self, mut reader: OwnedReadHalf) -> tokio::io::Result<()> {
-        let input = tokio::spawn(async move {
+        loop {
-            loop {
+            let mut len_buf = [0u8; size_of::<usize>()];
-                let mut len_buf = [0u8; size_of::<usize>()];
+            let size = reader.read_exact(&mut len_buf).await?;
-                let size = reader.read_exact(&mut len_buf).await?;
 
-                let len = usize::from_le_bytes(len_buf);
+            let len = usize::from_le_bytes(len_buf);
 
-                if size == 0 || len == 0 {
+            if size == 0 || len == 0 {
-                    break;
+                break;
-                }
-
-                let mut buffer = vec![0u8; len];
-
-                reader.read_exact(&mut buffer).await?;
-
-                println!("Received {} bytes", len);
             }
 
-            Ok::<(), tokio::io::Error>(())
+            let mut buffer = vec![0u8; len];
-        });
 
-        let _ = tokio::try_join!(input)?;
+            reader.read_exact(&mut buffer).await?;
+
+            match TerminalClientMessage::from_com(&mut buffer) {
+                TerminalClientMessage::ExecuteCommand(command) => {
+                    let command = command.as_str();
+
+                    let (command, _args) = if let Some((command, args)) = command.split_once(" ") {
+                        (command, args.split(" ").collect())
+                    } else {
+                        (command, Vec::new())
+                    };
+
+                    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?;
+                }
+            }
+        }
 
         Ok(())
     }

src/terminal/command.rs

@@ -6,24 +6,12 @@ impl PulseTradeApp {
             return;
         }
 
-        let (command, _args) = if let Some((command, args)) = command.split_once(" ") {
+        match command.trim() {
-            (command, args.split(" ").collect())
-        } else {
-            (command, Vec::new())
-        };
-
-        match command {
             "exit" | "quit" | "leave" => {
                 ctx.close().await;
             }
 
-            _ => {
+            _ => {}
-                self.logs.lock().await.push(pulse_wire::terminal::EventLog {
-                    kind: pulse_wire::terminal::LogKind::Err,
-                    name: "cmd".to_string(),
-                    message: format!("Command '{}' not found", command),
-                });
-            }
         }
     }
 }

src/terminal/main.rs

@@ -43,9 +43,7 @@ impl App for PulseTradeApp {
     async fn update(&mut self, ctx: &pulse_ui::state::Context, event: Box<dyn Any + Send + Sync>) {
         if let Some(Refresh) = event.downcast_ref() {
             return;
-        }
+        } else if let Some(event) = event.downcast_ref() {
-
-        if let Some(event) = event.downcast_ref() {
             if self.command.value.lock().await.handle_event(event) {
                 return;
             } else if let crossterm::event::Event::Key(key) = event {