wip

Cargo.toml

@@ -30,6 +30,9 @@ tokio = { version = "1", features = ["full"] }
 toml = { version = "0.8", features = ["indexmap", "preserve_order"] }
 xx = { version = "2", features = ["fslock", "hash"] }
 chrono = "0.4.38"
+uuid = { version = "1.11.0", features = ["v4", "fast-rng"] }
+rmp-serde = "1.3.0"
+strum = { version = "0.26.3", features = ["derive"] }
 
 [target.'cfg(unix)'.dependencies]
 exec = "0.3"

src/cli/daemon/mod.rs

@@ -29,6 +29,7 @@ impl Daemon {
 pub fn kill_or_stop(existing_pid: u32, force: bool) -> Result<bool> {
     if let Some(process) = procs::get_process(existing_pid) {
         if force {
+            debug!("Killing existing pitchfork daemon with pid {existing_pid}");
             if sysinfo::Process::kill_with(process, sysinfo::Signal::Term).is_none() {
                 sysinfo::Process::kill(process);
             }

src/cli/daemon/run.rs

@@ -21,6 +21,6 @@ impl Run {
             }
         }
 
-        Supervisor::new(pid_file).start().await
+        Supervisor::new(pid_file).await?.start().await
     }
 }

src/cli/daemon/start.rs

@@ -1,6 +1,6 @@
+use duct::cmd;
 use crate::cli::daemon::kill_or_stop;
 use crate::state_file::StateFile;
-use crate::supervisor::Supervisor;
 use crate::{env, Result};
 
 /// Starts the internal pitchfork daemon in the background
@@ -21,9 +21,10 @@ impl Start {
             }
         }
 
-        if let Ok(fork::Fork::Child) = fork::daemon(false, false) {
-            Supervisor::new(pid_file).start().await?;
-        }
+        cmd!(&*env::BIN_PATH, "daemon", "run")
+            .stdout_null()
+            .stderr_null()
+            .start()?;
 
         Ok(())
     }

src/cli/run.rs

@@ -1,9 +1,7 @@
-use crate::{env, Result};
+use crate::{ipc, Result};
 use eyre::bail;
-use interprocess::local_socket::traits::tokio::Stream;
-use interprocess::local_socket::{GenericFilePath, ToFsName};
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
-use tokio::try_join;
+use tokio::io::AsyncWriteExt;
+use crate::ipc::client::IpcClient;
 
 /// Runs a one-off daemon
 #[derive(Debug, clap::Args)]
@@ -25,17 +23,8 @@ impl Run {
         }
         dbg!(&self);
 
-        let conn = interprocess::local_socket::tokio::Stream::connect(
-            env::IPC_SOCK_PATH.clone().to_fs_name::<GenericFilePath>()?,
-        )
-        .await?;
-        let (recv, mut send) = conn.split();
-        let mut read = tokio::io::BufReader::new(recv);
-        let mut buffer = String::with_capacity(1024);
-        let send = send.write_all(b"Hello from client!\n");
-        let recv = read.read_line(&mut buffer);
-        try_join!(recv, send)?;
-        println!("Received: {}", buffer.trim());
+        let _ipc = IpcClient::connect().await?;
+        // ipc.send.write_all(b"Hello from client!\n").await?;
         Ok(())
     }
 }

src/env.rs

@@ -2,7 +2,7 @@ use once_cell::sync::Lazy;
 pub use std::env::*;
 use std::path::PathBuf;
 
-pub static BIN_PATH: Lazy<PathBuf> = Lazy::new(|| current_exe().unwrap());
+pub static BIN_PATH: Lazy<PathBuf> = Lazy::new(|| current_exe().unwrap().canonicalize().unwrap());
 
 pub static HOME_DIR: Lazy<PathBuf> = Lazy::new(|| dirs::home_dir().unwrap_or_default());
 pub static PITCHFORK_STATE_DIR: Lazy<PathBuf> = Lazy::new(|| {
@@ -28,7 +28,9 @@ pub static PITCHFORK_LOG_FILE: Lazy<PathBuf> = Lazy::new(|| {
 });
 pub static PITCHFORK_EXEC: Lazy<bool> = Lazy::new(|| var_true("PITCHFORK_EXEC"));
 
-pub static IPC_SOCK_PATH: Lazy<PathBuf> = Lazy::new(|| PITCHFORK_STATE_DIR.join("pitchfork.sock"));
+pub static IPC_SOCK_DIR: Lazy<PathBuf> = Lazy::new(|| PITCHFORK_STATE_DIR.join("sock"));
+pub static IPC_SOCK_MAIN: Lazy<PathBuf> = Lazy::new(|| IPC_SOCK_DIR.join("main.sock"));
+pub static IPC_JSON: Lazy<bool> = Lazy::new(|| var_true("IPC_JSON"));
 
 fn var_path(name: &str) -> Option<PathBuf> {
     var(name).map(PathBuf::from).ok()

src/ipc/client.rs

@@ -1 +1,46 @@
+use tokio::sync::Mutex;
+use crate::ipc::{fs_name, IpcMessage};
+use crate::{env, ipc, Result};
+use interprocess::local_socket::tokio::{RecvHalf, SendHalf};
+use interprocess::local_socket::traits::tokio::Stream;
+use interprocess::local_socket::{GenericFilePath, ToFsName};
+use tokio::io::{AsyncWriteExt, BufReader};
+use uuid::Uuid;
 
+pub struct IpcClient {
+    id: String,
+    recv: BufReader<RecvHalf>,
+    send: Mutex<SendHalf>,
+}
+
+impl IpcClient {
+    pub async fn connect() -> Result<Self> {
+        // ensure nobody else can connect to the IPC server at the same time
+        let _fslock = xx::fslock::get(&*env::IPC_SOCK_MAIN, false)?;
+        let conn =
+            interprocess::local_socket::tokio::Stream::connect(fs_name(&env::IPC_SOCK_MAIN)?)
+                .await?;
+        debug!("Connected to IPC main");
+        let (recv, send) = conn.split();
+        let recv = BufReader::new(recv);
+        let id = Uuid::new_v4().to_string();
+        let client = IpcClient { id, recv, send: Mutex::new(send) };
+        client.send(IpcMessage::Connect(client.id.clone())).await?;
+        Ok(client)
+    }
+
+    pub async fn send(&self, msg: IpcMessage) -> Result<()> {
+        let mut msg = if *env::IPC_JSON {
+            serde_json::to_vec(&msg)?
+        } else {
+            rmp_serde::to_vec(&msg)?
+        };
+        // if msg.contains(&b'\n') {
+        //     panic!("IPC message contains newline");
+        // }
+        msg.push(0);
+        let mut send = self.send.lock().await;
+        send.write_all(&msg).await?;
+        Ok(())
+    }
+}

src/ipc/mod.rs

@@ -1,2 +1,16 @@
+use std::path::Path;
+use interprocess::local_socket::{GenericFilePath, Name, ToFsName};
+
 pub(crate) mod client;
 pub(crate) mod server;
+
+#[derive(Debug, serde::Serialize, serde::Deserialize, strum::Display)]
+pub enum IpcMessage {
+    Connect(String),
+    Response(String),
+}
+
+pub fn fs_name(path: &Path) -> eyre::Result<Name> {
+    let fs_name = path.to_fs_name::<GenericFilePath>()?;
+    Ok(fs_name)
+}