add naive worker and coordinator; output filename not corrected

Author: adwinwhite

playground-worker/Cargo.lock

@@ -0,0 +1,11 @@
+[package]
+name = "playground_worker"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+worker_message = { path = "../worker-message" }
+bincode = "1.3"
+log = "0.4"

playground-worker/Cargo.toml

@@ -0,0 +1,149 @@
+use bincode::deserialize_from;
+use std::collections::HashMap;
+use std::fs;
+use std::io::{Read, Write};
+use std::os::unix::net::{UnixListener, UnixStream};
+use std::path::Path;
+use std::process::Child;
+use std::process::Command;
+use std::process::Stdio;
+use std::str::from_utf8;
+use std::{io, thread};
+use worker_message::{ExecuteOutput, Pid, Request, Response};
+
+pub fn listen(request_socket: &str, stdio_socket: &str) {
+    let request_listener = UnixListener::bind(request_socket).unwrap();
+
+    let stdio_listener = UnixListener::bind(stdio_socket).unwrap();
+
+    let (request_stream, _) = request_listener.accept().unwrap();
+    let (stdio_stream, _) = stdio_listener.accept().unwrap();
+    handle_socket(request_stream, stdio_stream);
+}
+
+fn handle_socket(request_stream: UnixStream, stdio_stream: UnixStream) {
+    let mut children = HashMap::<Pid, Child>::new();
+    loop {
+        let req: Request = deserialize_from(&request_stream).unwrap();
+        match req {
+            Request::WriteFile { path, content } => {
+                let path = Path::new(&path);
+                let parent_dir = path.parent().unwrap();
+                fs::create_dir_all(parent_dir).unwrap();
+                let result = fs::write(path, content);
+                let response = Response::WriteResult(result.map_err(|e| e.to_string()));
+                bincode::serialize_into(&request_stream, &response).unwrap();
+            }
+            Request::ReadFile { path } => {
+                let result = fs::read(path);
+                let response = Response::ReadResult(result.map_err(|e| e.to_string()));
+                bincode::serialize_into(&request_stream, &response).unwrap();
+            }
+            Request::ExecuteCommand {
+                cmd,
+                args,
+                envs,
+                cwd,
+            } => {
+                let result = Command::new(cmd)
+                    .args(args)
+                    .envs(envs)
+                    .current_dir(cwd)
+                    .output();
+                let response = Response::ExecuteResult(
+                    result
+                        .map(|output| ExecuteOutput {
+                            status: output.status.code(),
+                            stderr: output.stderr,
+                            stdout: output.stdout,
+                        })
+                        .map_err(|e| e.to_string()),
+                );
+                bincode::serialize_into(&request_stream, &response).unwrap();
+            }
+            Request::StreamCommand {
+                cmd,
+                args,
+                envs,
+                cwd,
+            } => {
+                let result = Command::new(cmd)
+                    .args(args)
+                    .envs(envs)
+                    .current_dir(cwd)
+                    .stdin(Stdio::piped())
+                    .stdout(Stdio::piped())
+                    .stderr(Stdio::piped())
+                    .spawn();
+                let result = match result {
+                    Ok(mut child) => {
+                        let pid = child.id();
+                        stream_stdio(&stdio_stream, &mut child);
+                        children.insert(pid, child);
+                        Ok(pid)
+                    }
+                    Err(err) => Err(err.to_string()),
+                };
+                let response = Response::StreamResult(result);
+                bincode::serialize_into(&request_stream, &response).unwrap();
+            }
+            Request::KillProcess { pid } => {
+                let result = match children.remove(&pid) {
+                    Some(mut child) => match child.kill() {
+                        Ok(()) => Ok(()),
+                        Err(err) => {
+                            if err.kind() == io::ErrorKind::InvalidInput {
+                                Ok(())
+                            } else {
+                                Err(err.to_string())
+                            }
+                        }
+                    },
+                    None => Ok(()),
+                };
+                let response = Response::KillResult(result);
+                bincode::serialize_into(&request_stream, &response).unwrap();
+            }
+        }
+    }
+}
+
+fn stream_stdio(stream: &UnixStream, child: &mut Child) {
+    let mut stdin = child.stdin.take().unwrap();
+    let mut stdout = child.stdout.take().unwrap();
+    let mut stderr = child.stderr.take().unwrap();
+
+    let mut stream_in = stream.try_clone().unwrap();
+    let mut stream_out = stream.try_clone().unwrap();
+    let mut stream_err = stream.try_clone().unwrap();
+    thread::spawn(move || {
+        let mut buffer = [0; 1024];
+        loop {
+            let n = stream_in.read(&mut buffer).unwrap();
+            if n != 0 {
+                stdin.write_all(&buffer[..n]).unwrap();
+                println!("To stdin {} - {}", n, from_utf8(&buffer[..n]).unwrap());
+            }
+        }
+    });
+    thread::spawn(move || {
+        let mut buffer = [0; 1024];
+        loop {
+            let n = stdout.read(&mut buffer).unwrap();
+            if n != 0 {
+                stream_out.write_all(&buffer[..n]).unwrap();
+                println!("From stdout {} - {}", n, from_utf8(&buffer[..n]).unwrap());
+            }
+        }
+    });
+    thread::spawn(move || {
+        let mut buffer = [0; 1024];
+        loop {
+            let n = stderr.read(&mut buffer).unwrap();
+            if n != 0 {
+                stream_err.write_all(&buffer[..n]).unwrap();
+                println!("From stderr {} - {}", n, from_utf8(&buffer[..n]).unwrap());
+            }
+        }
+    });
+}

playground-worker/src/lib.rs

@@ -0,0 +1,9 @@
+use playground_worker::listen;
+
+
+fn main() {
+    let request_socket_path = "/tmp/request_sock";
+    let stdio_socket_path = "/tmp/stdio_sock";
+    listen(request_socket_path, stdio_socket_path);
+}
+

playground-worker/src/main.rs

@@ -30,3 +30,6 @@ tokio = { version = "1.9", features = ["macros", "time", "process", "rt-multi-th
 tower-http = { version = "0.4", features = ["cors", "fs", "set-header", "trace"] }
 tracing = "0.1.37"
 tracing-subscriber = "0.3.16"
+worker_message = { path = "../worker-message" }
+playground_worker = { path = "../playground-worker" }
+bincode = "1.3"