Development of Parallel API

bastion/Cargo.toml

@@ -49,6 +49,10 @@ futures-timer = "3.0.0"
 fxhash = "0.2"
 lazy_static = "1.4"
 log = "0.4"
+ipc-channel = "0.14.0"
+serde = { version = "1.0", features = ["derive"] }
+
+
 # TODO: https://github.com/cogciprocate/qutex/pull/5
 # TODO: https://github.com/cogciprocate/qutex/pull/6
 bastion-qutex = { version = "0.2", features = ["async_await"] }

bastion/src/lib.rs

@@ -83,6 +83,8 @@ pub mod message;
 pub mod path;
 pub mod supervisor;
 
+pub mod parallel;
+
 ///
 /// Prelude of Bastion
 pub mod prelude {

bastion/src/parallel/callbacks.rs

@@ -0,0 +1,27 @@
+use std::fmt::{self, Debug, Formatter};
+use std::sync::{Arc, Mutex};
+use std::boxed::Box;
+use std::io;
+
+pub type CallbackFunc = dyn FnMut() -> io::Result<()> + Send + Sync + 'static;
+pub type SafeCallbackFunc = Arc<Mutex<Box<CallbackFunc>>>;
+
+#[derive(Default)]
+pub struct ProcessCallbacks {
+    pub before_start: Option<SafeCallbackFunc>,
+    pub before_restart: Option<SafeCallbackFunc>,
+    pub after_restart: Option<SafeCallbackFunc>,
+    pub after_stop: Option<SafeCallbackFunc>,
+}
+
+impl Debug for ProcessCallbacks {
+    fn fmt(&self, fmt: &mut Formatter) -> fmt::Result {
+        fmt.debug_struct("ProcessCallbacks")
+            .field("before_start", &self.before_start.is_some())
+            .field("before_restart", &self.before_start.is_some())
+            .field("after_restart", &self.before_start.is_some())
+            .field("after_stop", &self.before_start.is_some())
+            .finish()
+    }
+}
+

bastion/src/parallel/mod.rs

@@ -0,0 +1,3 @@
+pub mod callbacks;
+pub mod ops;
+pub mod process;

bastion/src/parallel/process.rs

@@ -0,0 +1,152 @@
+use std::sync::{Arc, Mutex};
+use std::collections::HashMap;
+use std::ffi::OsStr;
+use std::ffi::OsString;
+use std::process::Stdio;
+use std::io;
+use super::callbacks::*;
+
+#[derive(Debug)]
+pub struct ProcessData {
+    pub(crate) callbacks: ProcessCallbacks,
+    pub(crate) envs: HashMap<OsString, OsString>,
+}
+
+impl Default for ProcessData {
+    fn default() -> Self {
+        Self {
+            callbacks: ProcessCallbacks::default(),
+            envs: std::env::vars_os().collect(),
+        }
+    }
+}
+
+
+#[derive(Debug, Default)]
+pub struct Builder {
+    pub(crate) stdin: Option<Stdio>,
+    pub(crate) stdout: Option<Stdio>,
+    pub(crate) stderr: Option<Stdio>,
+    pub(crate) data: ProcessData,
+}
+
+impl Builder {
+    pub fn new() -> Self {
+        Self {
+            stdin: None,
+            stdout: None,
+            stderr: None,
+            data: ProcessData::default()
+        }
+    }
+
+    /// Process data which has given to spawned Process
+    pub(crate) fn data(&mut self, proc_data: ProcessData) -> &mut Self {
+        self.data = proc_data;
+        self
+    }
+
+    /// Set an environment variable in the spawned process. Equivalent to `Command::env`
+    pub fn env<K, V>(&mut self, key: K, val: V) -> &mut Self
+    where
+        K: AsRef<OsStr>,
+        V: AsRef<OsStr>,
+    {
+        self.data.envs
+            .insert(key.as_ref().to_owned(), val.as_ref().to_owned());
+        self
+    }
+
+    /// Set environment variables in the spawned process. Equivalent to `Command::envs`
+    pub fn envs<I, K, V>(&mut self, vars: I) -> &mut Self
+    where
+        I: IntoIterator<Item = (K, V)>,
+        K: AsRef<OsStr>,
+        V: AsRef<OsStr>,
+    {
+        self.data.envs.extend(
+            vars.into_iter()
+                .map(|(k, v)| (k.as_ref().to_owned(), v.as_ref().to_owned())),
+        );
+        self
+    }
+
+    ///
+    /// Removes an environment variable in the spawned process. Equivalent to `Command::env_remove`
+    pub fn env_remove<K: AsRef<OsStr>>(&mut self, key: K) -> &mut Self {
+        self.data.envs.remove(key.as_ref());
+        self
+    }
+
+    ///
+    /// Clears all environment variables in the spawned process. Equivalent to `Command::env_clear`
+    pub fn env_clear(&mut self) -> &mut Self {
+        self.data.envs.clear();
+        self
+    }
+
+    ///
+    /// Captures the `stdin` of the spawned process, allowing you to manually send data via `JoinHandle::stdin`
+    pub fn stdin<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.stdin = Some(cfg.into());
+        self
+    }
+
+    ///
+    /// Captures the `stdout` of the spawned process, allowing you to manually receive data via `JoinHandle::stdout`
+    pub fn stdout<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.stdout = Some(cfg.into());
+        self
+    }
+
+    ///
+    /// Captures the `stderr` of the spawned process, allowing you to manually receive data via `JoinHandle::stderr`
+    pub fn stderr<T: Into<Stdio>>(&mut self, cfg: T) -> &mut Self {
+        self.stderr = Some(cfg.into());
+        self
+    }
+
+    ///
+    /// Process before start
+    #[cfg(unix)]
+    pub fn before_start<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnMut() -> io::Result<()> + Send + Sync + 'static
+    {
+        self.data.callbacks.before_start = Some(Arc::new(Mutex::new(Box::new(f))));
+        self
+    }
+
+    ///
+    /// Process before restart
+    #[cfg(unix)]
+    pub fn before_restart<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnMut() -> io::Result<()> + Send + Sync + 'static
+    {
+        self.data.callbacks.before_restart = Some(Arc::new(Mutex::new(Box::new(f))));
+        self
+    }
+
+    ///
+    /// Process after restart
+    #[cfg(unix)]
+    pub fn after_restart<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnMut() -> io::Result<()> + Send + Sync + 'static
+    {
+        self.data.callbacks.after_restart = Some(Arc::new(Mutex::new(Box::new(f))));
+        self
+    }
+
+    ///
+    /// Process after stop
+    #[cfg(unix)]
+    pub fn after_stop<F>(&mut self, f: F) -> &mut Self
+    where
+        F: FnMut() -> io::Result<()> + Send + Sync + 'static
+    {
+        self.data.callbacks.after_stop = Some(Arc::new(Mutex::new(Box::new(f))));
+        self
+    }
+}