crowtty: implement script execution (WIP)

platforms/melpomene/melpo.toml

@@ -35,6 +35,9 @@ enabled = true
 # message = "hello\r\n"
 # interval =  { secs = 1, nanos = 0 }
 
+[services.sermux_shell]
+enabled = true
+
 [platform]
 # sleep_cap = { secs = 0, nanos = 100_000_000 } # 100ms
 

source/kernel/src/daemons/shells.rs

@@ -27,32 +27,54 @@ use futures::FutureExt;
 use input_mgr::RingLine;
 use key_event::KeyEvent;
 use profont::PROFONT_12_POINT;
+use serde::{Deserialize, Serialize};
 
 use crate::forth::Forth;
 
 /// Settings for the [sermux_shell] daemon
-#[derive(Debug)]
+#[derive(Debug, Serialize, Deserialize)]
 #[non_exhaustive]
 pub struct SermuxShellSettings {
+    /// Should the shell be enabled?
+    ///
+    /// Defaults to false
+    #[serde(default)]
+    pub enabled: bool,
     /// Sermux port to serve the shell on
     ///
     /// Defaults to [WellKnown::ForthShell0]
+    #[serde(default = "SermuxShellSettings::default_port")]
     pub port: u16,
     /// Number of bytes used for the sermux buffer
     ///
     /// Defaults to 256
+    #[serde(default = "SermuxShellSettings::default_capacity")]
     pub capacity: usize,
     /// Forth parameters for the shell
     ///
     /// Uses the default value of [Params]
+    #[serde(default)]
     pub forth_settings: Params,
 }
 
+impl SermuxShellSettings {
+    const DEFAULT_PORT: u16 = WellKnown::ForthShell0 as u16;
+    const DEFAULT_CAPACITY: usize = 256;
+
+    const fn default_port() -> u16 {
+        Self::DEFAULT_PORT
+    }
+    const fn default_capacity() -> usize {
+        Self::DEFAULT_CAPACITY
+    }
+}
+
 impl Default for SermuxShellSettings {
     fn default() -> Self {
         Self {
-            port: WellKnown::ForthShell0.into(),
-            capacity: 256,
+            enabled: false,
+            port: Self::DEFAULT_PORT,
+            capacity: Self::DEFAULT_CAPACITY,
             forth_settings: Default::default(),
         }
     }
@@ -65,6 +87,7 @@ pub async fn sermux_shell(k: &'static Kernel, settings: SermuxShellSettings) {
         port,
         capacity,
         forth_settings,
+        ..
     } = settings;
     let port = PortHandle::open(k, port, capacity).await.unwrap();
     let (task, tid_io) = Forth::new(k, forth_settings)

source/kernel/src/lib.rs

@@ -154,6 +154,7 @@ pub struct KernelServiceSettings {
     pub spawnulator: SpawnulatorSettings,
     pub sermux_loopback: daemons::sermux::LoopbackSettings,
     pub sermux_hello: daemons::sermux::HelloSettings,
+    pub sermux_shell: daemons::shells::SermuxShellSettings,
     #[cfg(feature = "serial-trace")]
     pub sermux_trace: serial_trace::SerialTraceSettings,
 }
@@ -336,6 +337,11 @@ impl Kernel {
                 self.initialize(daemons::sermux::hello(self, settings.sermux_hello))
                     .expect("failed to spawn SerMux hello world daemon");
             }
+
+            if settings.sermux_shell.enabled {
+                self.initialize(daemons::shells::sermux_shell(self, settings.sermux_shell))
+                    .expect("failed to spawn SerMux hello world daemon");
+            }
         } else {
             let deps = [
                 #[cfg(feature = "serial-trace")]

tools/crowtty/src/connection.rs

@@ -51,6 +51,7 @@ pub enum Connect {
         #[arg(default_value_t = Self::DEFAULT_BAUD_RATE)]
         baud: u32,
     },
+    Exec,
 }
 
 impl Write for Connection {
@@ -127,6 +128,7 @@ impl Connect {
                 let port = serial::new(path, baud).timeout(Self::READ_TIMEOUT).open()?;
                 Ok(Connection::Serial(port))
             }
+            Self::Exec => unreachable!("exec should not go up to here"),
         }
     }
 }
@@ -136,6 +138,7 @@ impl fmt::Display for Connect {
         match self {
             Self::Tcp { ip, port } => write!(f, "{ip}:{port}"),
             Self::Serial { path, baud } => write!(f, "{} (@ {baud})", path.display()),
+            Self::Exec => write!(f, "Exec"),
         }
     }
 }

tools/crowtty/src/main.rs

@@ -1,3 +1,8 @@
+use std::{
+    io::{stdin, Read},
+    net::{Ipv4Addr, TcpStream},
+};
+
 use clap::Parser;
 use connection::Connect;
 use miette::{Context, IntoDiagnostic};
@@ -49,6 +54,13 @@ fn main() -> miette::Result<()> {
         verbose,
         trace_filter,
     } = Args::parse();
+
+    if let Connect::Exec = connect {
+        let mut cmd = Vec::new();
+        stdin().read_to_end(&mut cmd).into_diagnostic()?;
+        return libcrowtty::Exec::new().settings(settings).run(cmd);
+    }
+
     let conn = connect
         .connect()
         .into_diagnostic()

tools/libcrowtty/src/lib.rs

@@ -6,7 +6,7 @@ use std::{
     collections::HashMap,
     fmt,
     io::{ErrorKind, Read, Write},
-    net::TcpListener,
+    net::{Ipv4Addr, TcpListener},
     sync::mpsc::{channel, Receiver, Sender},
     thread::{sleep, spawn, JoinHandle},
     time::{Duration, Instant},
@@ -216,6 +216,93 @@ impl Crowtty {
             manager.workers.insert(i, handle);
         }
 
+        {
+            let i = WellKnown::ForthShell0.into();
+            let (inp_send, inp_recv) = channel();
+            let (out_send, out_recv) = channel();
+
+            let socket =
+                std::net::TcpListener::bind(dbg!(format!("127.0.0.1:{}", tcp_port_base + i)))
+                    .unwrap();
+
+            let work = TcpWorker {
+                out: out_recv,
+                inp: inp_send,
+                socket,
+                port: i,
+            };
+            let tag = tag.port(i);
+            let thread_hdl = spawn(move || {
+                let mux = " MUX".if_supports_color(Stream::Stdout, |s| s.cyan());
+                let dmux = "DMUX".if_supports_color(Stream::Stdout, |s| s.bright_purple());
+                let err = "ERR!".if_supports_color(Stream::Stdout, |err| err.red());
+                'incoming: for skt in work.socket.incoming() {
+                    let mut skt = match skt {
+                        Ok(skt) => skt,
+                        Err(e) => {
+                            panic!(
+                                "{tag} CONN failed to accept host connection to port {} (:{}): {e}",
+                                tcp_port_base + work.port,
+                                work.port
+                            );
+                        }
+                    };
+
+                    println!(
+                        "{tag} CONN host connected to port {} (:{})",
+                        tcp_port_base + work.port,
+                        work.port
+                    );
+
+                    skt.set_read_timeout(Some(Duration::from_millis(10))).ok();
+                    // skt.set_nonblocking(true).ok();
+                    // skt.set_nodelay(true).ok();
+
+                    // let mut last = Instant::now();
+
+                    let mut input = Vec::new();
+
+                    'inner: loop {
+                        let mut buf = [0u8; 128];
+                        match skt.read(&mut buf) {
+                            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
+                            Err(_) => {
+                                skt.shutdown(std::net::Shutdown::Both).ok();
+                                continue 'incoming;
+                            }
+                            Ok(0) => {
+                                break 'inner;
+                            }
+                            Ok(n) => {
+                                tag.if_verbose(format_args!("{mux} {n}B <- :{}", work.port));
+                                input.extend_from_slice(&buf[..n]);
+                            }
+                        }
+                    }
+
+                    work.inp.send(input).ok();
+
+                    if let Ok(msg) = work.out.recv_timeout(Duration::from_secs(1)) {
+                        match skt.write_all(&msg) {
+                            Ok(_) => {}
+                            Err(e) => {
+                                println!("{tag} {dmux} {err} write error: {e}");
+                            }
+                        }
+                        continue 'incoming;
+                    } else {
+                        continue 'incoming;
+                    }
+                }
+            });
+            let handle = WorkerHandle {
+                out: out_send,
+                inp: inp_recv,
+                _thread_hdl: thread_hdl,
+            };
+
+            manager.workers.insert(i, handle);
+        }
         // spawn tracing listener
         let trace_port = WellKnown::BinaryTracing as u16;
         let trace_handle = {
@@ -399,3 +486,42 @@ struct TcpWorker {
     port: u16,
     socket: TcpListener,
 }
+
+pub struct Exec {
+    settings: Settings,
+}
+
+impl Exec {
+    pub fn new() -> Self {
+        Self {
+            settings: Settings::default(),
+        }
+    }
+
+    pub fn settings(self, settings: Settings) -> Self {
+        Self { settings }
+    }
+
+    pub fn run(self, cmd: Vec<u8>) -> Result<(), miette::Error> {
+        let port = self.settings.tcp_port_base + WellKnown::ForthShell0 as u16;
+        let mut stream =
+            std::net::TcpStream::connect((Ipv4Addr::LOCALHOST, port)).into_diagnostic()?;
+        eprintln!("[stderr] connected to crowtty on {:?}", stream.peer_addr());
+
+        stream.write_all(&cmd).into_diagnostic()?;
+        stream.shutdown(std::net::Shutdown::Write).unwrap();
+
+        let mut response = Vec::new();
+        stream.read_to_end(&mut response).into_diagnostic()?;
+
+        println!("{}", String::from_utf8_lossy(&response));
+
+        Ok(())
+    }
+}
+
+impl Default for Exec {
+    fn default() -> Self {
+        Self::new()
+    }
+}