Merge pull request #74 from mum-rs/yeet-ipc

4 files changed, 64 insertions, 54 deletions

diff --git a/mumd/Cargo.toml b/mumd/Cargo.toml
index 9080bdf..a853622 100644
--- a/mumd/Cargo.toml
+++ b/mumd/Cargo.toml
@@ -28,7 +28,6 @@ dasp_ring_buffer = "0.11"
 futures-util = { version = "0.3", features = ["sink"]}
 futures-channel = "0.3"
 hound = "3.4"
-ipc-channel = "0.15"
 log = "0.4"
 mumble-protocol = "0.4.1"
 native-tls = "0.2"
@@ -37,10 +36,11 @@ opus = "0.2"
 serde = { version = "1.0", features = ["derive"] }
 strum = "0.20"
 strum_macros = "0.20"
-tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync", "net", "time"] }
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync", "net", "time", "fs"] }
 tokio-stream = "0.1.0"
 tokio-native-tls = "0.3"
 tokio-util = { version = "0.6", features = ["codec", "net"] }
+bincode = "1.3.2"
 
 libnotify = { version = "1.0", optional = true }
 
diff --git a/mumd/src/client.rs b/mumd/src/client.rs
index 84c1ea1..cdae7eb 100644
--- a/mumd/src/client.rs
+++ b/mumd/src/client.rs
@@ -2,16 +2,15 @@ use crate::command;
 use crate::network::{tcp, udp, ConnectionInfo};
 use crate::state::State;
 
-use ipc_channel::ipc::IpcSender;
 use mumble_protocol::{Serverbound, control::ControlPacket, crypt::ClientCryptState};
 use mumlib::command::{Command, CommandResponse};
 use std::sync::Arc;
-use tokio::{join, sync::{mpsc, watch, Mutex}};
+use tokio::{join, sync::{Mutex, mpsc, oneshot, watch}};
 
 pub async fn handle(
     command_receiver: mpsc::UnboundedReceiver<(
         Command,
-        IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
+        oneshot::Sender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
 ) {
     let (connection_info_sender, connection_info_receiver) =
diff --git a/mumd/src/command.rs b/mumd/src/command.rs
index 653d1fa..3e462b1 100644
--- a/mumd/src/command.rs
+++ b/mumd/src/command.rs
@@ -5,7 +5,6 @@ use crate::network::{
 };
 use crate::state::{ExecutionContext, State};
 
-use ipc_channel::ipc::IpcSender;
 use log::*;
 use mumble_protocol::{Serverbound, control::ControlPacket};
 use mumlib::command::{Command, CommandResponse};
@@ -16,7 +15,7 @@ pub async fn handle(
     state: Arc<Mutex<State>>,
     mut command_receiver: mpsc::UnboundedReceiver<(
         Command,
-        IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
+        oneshot::Sender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
     tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>,
     ping_request_sender: mpsc::UnboundedSender<PingRequest>,
diff --git a/mumd/src/main.rs b/mumd/src/main.rs
index 7c3745c..26e8d49 100644
--- a/mumd/src/main.rs
+++ b/mumd/src/main.rs
@@ -5,13 +5,13 @@ mod network;
 mod notify;
 mod state;
 
-use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender};
+use futures_util::{SinkExt, StreamExt};
 use log::*;
 use mumlib::command::{Command, CommandResponse};
 use mumlib::setup_logger;
-use std::fs;
-use tokio::{join, sync::mpsc};
-use tokio::task::spawn_blocking;
+use tokio::{join, net::{UnixListener, UnixStream}, sync::{mpsc, oneshot}};
+use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
+use bytes::{BufMut, BytesMut};
 
 #[tokio::main]
 async fn main() {
@@ -24,67 +24,79 @@ async fn main() {
     notify::init();
 
     // check if another instance is live
-    let (tx_client, rx_client) =
-        ipc::channel::<mumlib::error::Result<Option<CommandResponse>>>().unwrap();
-    if let Ok(server_name) = fs::read_to_string(mumlib::SOCKET_PATH) {
-        if let Ok(tx0) = IpcSender::connect(server_name) {
-            if tx0.send((Command::Ping, tx_client)).is_ok() {
-                match rx_client.recv() {
-                    Ok(Ok(Some(CommandResponse::Pong))) => {
+    let connection = UnixStream::connect(mumlib::SOCKET_PATH).await;
+    match connection {
+        Ok(stream) => {
+            let (reader, writer) = stream.into_split();
+            let mut reader = FramedRead::new(reader, LengthDelimitedCodec::new());
+            let mut writer = FramedWrite::new(writer, LengthDelimitedCodec::new());
+            let mut command = BytesMut::new();
+            bincode::serialize_into((&mut command).writer(), &Command::Ping).unwrap();
+            if let Ok(()) = writer.send(command.freeze()).await {
+                if let Some(Ok(buf)) = reader.next().await {
+                    if let Ok(Ok::<Option<CommandResponse>, mumlib::error::Error>(Some(CommandResponse::Pong))) = bincode::deserialize(&buf) {
                         error!("Another instance of mumd is already running");
                         return;
-                    },
-                    resp => {
-                        warn!("Ping with weird response. Continuing...");
-                        debug!("Response was {:?}", resp);
                     }
                 }
             }
+            debug!("a dead socket was found, removing");
+            tokio::fs::remove_file(mumlib::SOCKET_PATH).await.unwrap();
+        }
+        Err(e) => {
+            if matches!(e.kind(), std::io::ErrorKind::ConnectionRefused) {
+                debug!("a dead socket was found, removing");
+                tokio::fs::remove_file(mumlib::SOCKET_PATH).await.unwrap();
+            }
         }
     }
 
-    let (command_sender, command_receiver) = mpsc::unbounded_channel::<(
-        Command,
-        IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
-    )>();
+    let (command_sender, command_receiver) = mpsc::unbounded_channel();
 
-    let (_, e) = join!(
+    join!(
         client::handle(command_receiver),
-        spawn_blocking(move || {
-            // IpcSender is blocking
-            receive_oneshot_commands(command_sender);
-        }),
+        receive_commands(command_sender),
     );
-    e.unwrap();
 }
 
-fn receive_oneshot_commands(
+async fn receive_commands(
     command_sender: mpsc::UnboundedSender<(
         Command,
-        IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
+        oneshot::Sender<mumlib::error::Result<Option<CommandResponse>>>,
     )>,
 ) {
+    let socket = UnixListener::bind(mumlib::SOCKET_PATH).unwrap();
+
     loop {
-        // create listener
-        let (server, server_name): (
-            IpcOneShotServer<(
-                Command,
-                IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
-            )>,
-            String,
-        ) = IpcOneShotServer::new().unwrap();
-        fs::write(mumlib::SOCKET_PATH, &server_name).unwrap();
-        debug!("Listening to {}", server_name);
+        if let Ok((incoming, _)) = socket.accept().await {
+            let sender = command_sender.clone();
+            tokio::spawn(async move {
+                let (reader, writer) = incoming.into_split();
+                let mut reader = FramedRead::new(reader, LengthDelimitedCodec::new());
+                let mut writer = FramedWrite::new(writer, LengthDelimitedCodec::new());
+                
+                while let Some(next) = reader.next().await {
+                    let buf = match next {
+                        Ok(buf) => buf,
+                        Err(_) => continue,
+                    };
+
+                    let command = match bincode::deserialize::<Command>(&buf) {
+                        Ok(e) => e,
+                        Err(_) => continue,
+                    };
 
-        // receive command and response channel
-        let (_, conn): (
-            _,
-            (
-                Command,
-                IpcSender<mumlib::error::Result<Option<CommandResponse>>>,
-            ),
-        ) = server.accept().unwrap();
-        debug!("Sending command {:?} to command handler", conn.0);
-        command_sender.send(conn).unwrap();
+                    let (tx, rx) = oneshot::channel();
+
+                    sender.send((command, tx)).unwrap();
+
+                    let response = rx.await.unwrap();
+                    let mut serialized = BytesMut::new();
+                    bincode::serialize_into((&mut serialized).writer(), &response).unwrap();
+
+                    let _ = writer.send(serialized.freeze()).await;
+                }
+            });
+        }
     }
 }