export functions :))

2 files changed, 136 insertions, 25 deletions

diff --git a/src/export_fun.rs b/src/export_fun.rs
index 64ac08b..9c2ab4a 100644
--- a/src/export_fun.rs
+++ b/src/export_fun.rs
@@ -1,14 +1,19 @@
 //! allow sub-programs to invoke arbitrary user-defined functions via a unix socket
 
 use crate::Session;
+use crate::run::Input;
+use crate::run::Output;
 use crate::run::get_command_kind;
 use std::env::current_exe;
 use std::ffi::OsStr;
 use std::fs;
+use std::fs::File;
 use std::io;
 use std::io::IoSliceMut;
 use std::io::Read;
 use std::io::Write;
+use std::os::fd::FromRawFd;
+use std::os::fd::OwnedFd;
 use std::os::unix::ffi::OsStrExt;
 use std::os::unix::fs::symlink;
 use std::os::unix::net::AncillaryData;
@@ -21,8 +26,9 @@ use std::process::exit;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::thread;
+use std::time::Duration;
 
-fn handle_server(session: Arc<Mutex<Session>>, stream: UnixStream) -> io::Result<()> {
+fn handle_server(session: Arc<Mutex<Session>>, mut stream: UnixStream) -> io::Result<()> {
     // TODO: figure out how to get a reasonable limit on CLI arg len
     let mut buf = [0u8; 8192];
     let mut iov = [IoSliceMut::new(&mut buf)];
@@ -64,18 +70,38 @@ fn handle_server(session: Arc<Mutex<Session>>, stream: UnixStream) -> io::Result
 
     let se = session.lock().unwrap();
     match get_command_kind(&se, cli_args[0].as_slice()) {
-        crate::run::CommandKind::Fun(ast) => todo!(),
+        crate::run::CommandKind::Fun(_) => (),
         crate::run::CommandKind::Path(_) | crate::run::CommandKind::Builtin(_) => {
             return Ok(());
         }
     }
+    drop(se);
 
-    todo!()
+    let stdin = File::from(unsafe { OwnedFd::from_raw_fd(fds[0]) });
+    let stdout = File::from(unsafe { OwnedFd::from_raw_fd(fds[1]) });
+
+    let res = crate::run::Executor::execute_fun(
+        session,
+        cli_args,
+        Input::File(stdin),
+        Output::File(stdout),
+    );
+
+    let exit_code = match res {
+        Ok(_) => 0,
+        Err(e) => match e {
+            crate::run::ExecError::UnknownVariable(_) => -3,
+            crate::run::ExecError::ExecError(x) => x,
+        },
+    };
+
+    let _ = stream.set_write_timeout(Some(Duration::from_secs(1)));
+    stream.write_all(&exit_code.to_le_bytes())?;
+
+    Ok(())
 }
 
 fn handle_client(mut stream: UnixStream) -> io::Result<()> {
-    println!("uhh hi");
-
     // give up all my file descriptors descriptors
     let mut ancillary_buffer = [0; 128];
     let mut ancillary = SocketAncillary::new(&mut ancillary_buffer[..]);
diff --git a/src/run/mod.rs b/src/run/mod.rs
index da62778..c7414a8 100644
--- a/src/run/mod.rs
+++ b/src/run/mod.rs
@@ -1,37 +1,101 @@
 use std::collections::HashMap;
+use std::fs::File;
 use std::io::{PipeReader, PipeWriter};
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 
-use crate::parse::{Ast, PostExpansion, PreExpansion};
+use crate::parse::{self, Ast, PostExpansion, PreExpansion};
 use crate::*;
 
 mod builtin;
 
 #[derive(Debug)]
-enum ExecError {
+pub enum ExecError {
     UnknownVariable(BString),
     ExecError(i32),
 }
 
 #[derive(Clone)]
-struct Executor {
+pub struct Executor {
     se: Arc<Mutex<Session>>,
     args: Option<Vec<BString>>,
 }
 
+pub enum Input {
+    Stdin,
+    Pipe(PipeReader),
+    File(File),
+}
+
+pub enum Output {
+    Stdout,
+    Pipe(PipeWriter),
+    File(File),
+}
+
+impl Into<Stdio> for Input {
+    fn into(self) -> Stdio {
+        match self {
+            Input::Stdin => Stdio::inherit(),
+            Input::Pipe(reader) => reader.into(),
+            Input::File(file) => file.into(),
+        }
+    }
+}
+
+impl Into<Stdio> for Output {
+    fn into(self) -> Stdio {
+        match self {
+            Output::Stdout => Stdio::inherit(),
+            Output::Pipe(writer) => writer.into(),
+            Output::File(file) => file.into(),
+        }
+    }
+}
+
+impl Read for Input {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            Input::Stdin => io::stdin().read(buf),
+            Input::Pipe(reader) => reader.read(buf),
+            Input::File(file) => file.read(buf),
+        }
+    }
+}
+
+impl Write for Output {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        match self {
+            Output::Stdout => io::stdout().write(buf),
+            Output::Pipe(writer) => writer.write(buf),
+            Output::File(file) => file.write(buf),
+        }
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        match self {
+            Output::Stdout => io::stdout().flush(),
+            Output::Pipe(writer) => writer.flush(),
+            Output::File(file) => file.flush(),
+        }
+    }
+}
+
 impl Executor {
     fn execute_pipeline(
         &mut self,
         pipes: parse::Pipes<parse::PostExpansion>,
-        mut stdin: Option<PipeReader>,
-        mut stdout: Option<PipeWriter>,
+        stdin: Input,
+        stdout: Output,
     ) -> Result<(), ExecError> {
         let mut children = Vec::new();
         let mut threads = Vec::new();
         let mut prev_reader = None;
         let mut spawn_error = false;
 
+        let mut stdin = Some(stdin);
+        let mut stdout = Some(stdout);
+
         let pipelen = pipes.cmds.len();
         for (i, cmd) in pipes.cmds.into_iter().enumerate() {
             let (reader, writer) = if i < pipelen - 1 {
@@ -52,14 +116,14 @@ impl Executor {
 
                     if let Some(r) = prev_reader.take() {
                         command.stdin(Stdio::from(r));
-                    } else if let Some(stdin) = stdin.take() {
-                        command.stdin(Stdio::from(stdin));
+                    } else {
+                        command.stdin(stdin.take().unwrap());
                     }
 
                     if let Some(w) = writer {
                         command.stdout(Stdio::from(w));
-                    } else if let Some(stdout) = stdout.take() {
-                        command.stdout(Stdio::from(stdout));
+                    } else {
+                        command.stdout(stdout.take().unwrap());
                     }
 
                     if let Some(sr) = self.se.lock().unwrap().socket_running.as_ref() {
@@ -107,18 +171,14 @@ impl Executor {
                 CommandKind::Builtin(builtin) => {
                     let mut input: Box<dyn io::Read + Send> = if let Some(r) = prev_reader.take() {
                         Box::new(r)
-                    } else if let Some(stdin) = stdin.take() {
-                        Box::new(stdin)
                     } else {
-                        Box::new(io::stdin())
+                        Box::new(stdin.take().unwrap())
                     };
 
                     let mut output: Box<dyn io::Write + Send> = if let Some(w) = writer {
                         Box::new(w)
-                    } else if let Some(w) = stdout.take() {
-                        Box::new(w)
                     } else {
-                        Box::new(io::stdout())
+                        Box::new(stdout.take().unwrap())
                     };
 
                     // currently only required for `re`, cannot happen in background thread
@@ -141,7 +201,12 @@ impl Executor {
                     this.args = Some(args);
 
                     let ast = ast.expand(&mut this)?;
-                    this.execute(ast, prev_reader, writer).unwrap();
+                    this.execute(
+                        ast,
+                        prev_reader.map(|r| Input::Pipe(r)).unwrap_or(Input::Stdin),
+                        writer.map(|w| Output::Pipe(w)).unwrap_or(Output::Stdout),
+                    )
+                    .unwrap();
                 }
             }
 
@@ -207,8 +272,8 @@ impl Executor {
     fn execute(
         &mut self,
         ast: Ast<parse::PostExpansion>,
-        stdin: Option<PipeReader>,
-        stdout: Option<PipeWriter>,
+        stdin: Input,
+        stdout: Output,
     ) -> Result<(), ExecError> {
         match ast {
             Ast::FunDecl(fd) => self.execute_fun_decl(fd),
@@ -216,6 +281,25 @@ impl Executor {
             Ast::Pipes(pipes) => self.execute_pipeline(pipes, stdin, stdout),
         }
     }
+
+    pub fn execute_fun(
+        session: Arc<Mutex<Session>>,
+        invoke: Vec<BString>,
+        stdin: Input,
+        stdout: Output,
+    ) -> Result<(), ExecError> {
+        let mut this = Self {
+            se: session,
+            args: None,
+        };
+
+        let cmd = parse::Command {
+            cmd: invoke[0].clone(),
+            args: invoke[1..].to_vec(),
+        };
+
+        this.execute_pipeline(parse::Pipes { cmds: vec![cmd] }, stdin, stdout)
+    }
 }
 
 impl parse::Expander for Executor {
@@ -255,7 +339,8 @@ impl parse::Expander for Executor {
         let (stdin, _) = io::pipe().unwrap();
         let (mut expansion, stdout) = io::pipe().unwrap();
         let mut this = self.clone();
-        let t = std::thread::spawn(move || this.execute(ast, Some(stdin), Some(stdout)));
+        let t =
+            std::thread::spawn(move || this.execute(ast, Input::Pipe(stdin), Output::Pipe(stdout)));
         let mut buf = Vec::new();
         expansion.read_to_end(&mut buf).unwrap();
         t.join().unwrap()?;
@@ -269,7 +354,7 @@ impl parse::Expander for Executor {
 fn exec(se: Arc<Mutex<Session>>, ast: Ast<PreExpansion>) -> Result<(), ExecError> {
     let mut exec = Executor { se, args: None };
     let ast = ast.expand(&mut exec)?;
-    exec.execute(ast, None, None)
+    exec.execute(ast, Input::Stdin, Output::Stdout)
 }
 
 pub fn run(se: Arc<Mutex<Session>>, cmd: Vec<u8>) {