From d484c05e56194346944b295968c66ccc0e543534 Mon Sep 17 00:00:00 2001 From: Eskil Queseth Date: Wed, 24 Mar 2021 21:38:51 +0100 Subject: remove ipc-channel dependency --- mumd/Cargo.toml | 4 +-- mumd/src/client.rs | 5 ++-- mumd/src/command.rs | 3 +-- mumd/src/main.rs | 73 +++++++++++++++++++++++++---------------------------- 4 files changed, 39 insertions(+), 46 deletions(-) (limited to 'mumd') diff --git a/mumd/Cargo.toml b/mumd/Cargo.toml index 9080bdf..f271e69 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", "io-util"] } 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>>, + oneshot::Sender>>, )>, ) { 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>, mut command_receiver: mpsc::UnboundedReceiver<( Command, - IpcSender>>, + oneshot::Sender>>, )>, tcp_event_register_sender: mpsc::UnboundedSender<(TcpEvent, TcpEventCallback)>, ping_request_sender: mpsc::UnboundedSender, diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 7c3745c..d4151e3 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -5,13 +5,14 @@ mod network; mod notify; mod state; -use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender}; +use futures_util::StreamExt; +//use ipc_channel::ipc::{self, IpcOneShotServer, IpcSender}; 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, sync::{mpsc, oneshot}}; +use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec}; +use bytes::{BufMut, BytesMut}; #[tokio::main] async fn main() { @@ -24,7 +25,7 @@ async fn main() { notify::init(); // check if another instance is live - let (tx_client, rx_client) = + /*let (tx_client, rx_client) = ipc::channel::>>().unwrap(); if let Ok(server_name) = fs::read_to_string(mumlib::SOCKET_PATH) { if let Ok(tx0) = IpcSender::connect(server_name) { @@ -41,50 +42,44 @@ async fn main() { } } } - } + }*/ - let (command_sender, command_receiver) = mpsc::unbounded_channel::<( - Command, - IpcSender>>, - )>(); + 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>>, + oneshot::Sender>>, )>, ) { + let socket = UnixListener::bind("/tmp/mumd").unwrap(); + loop { - // create listener - let (server, server_name): ( - IpcOneShotServer<( - Command, - IpcSender>>, - )>, - 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 (reader, writer) = incoming.into_split(); + let reader = FramedRead::new(reader, LengthDelimitedCodec::new()); + let writer = FramedWrite::new(writer, LengthDelimitedCodec::new()); + + reader.filter_map(|buf| async { + buf.ok() + }) + .map(|buf| bincode::deserialize::(&buf).unwrap()) + .filter_map(|command| async { + let (tx, rx) = oneshot::channel(); - // receive command and response channel - let (_, conn): ( - _, - ( - Command, - IpcSender>>, - ), - ) = server.accept().unwrap(); - debug!("Sending command {:?} to command handler", conn.0); - command_sender.send(conn).unwrap(); + command_sender.send((command, tx)).unwrap(); + + let response = rx.await.unwrap(); + let mut serialized = BytesMut::new(); + bincode::serialize_into((&mut serialized).writer(), &response).unwrap(); + Some(Ok(serialized.freeze())) + }).forward(writer).await.unwrap(); + } } -} +} \ No newline at end of file -- cgit v1.2.1