From fc30dc26687862458a378901d3273e38a1b28c83 Mon Sep 17 00:00:00 2001 From: Alexis Sellier Date: Thu, 1 Sep 2022 15:10:53 +0200 Subject: [PATCH] node: Add client control socket Signed-off-by: Alexis Sellier --- node/src/client.rs | 13 +++++++ node/src/client/handle.rs | 77 +++++++++++++++++++++++++++++++++++++ node/src/client/socket.rs | 80 +++++++++++++++++++++++++++++++++++++++ node/src/protocol.rs | 8 ++++ node/src/rad.rs | 1 + 5 files changed, 179 insertions(+) create mode 100644 node/src/client/handle.rs create mode 100644 node/src/client/socket.rs diff --git a/node/src/client.rs b/node/src/client.rs index 2c5cc0d9..83f33942 100644 --- a/node/src/client.rs +++ b/node/src/client.rs @@ -10,6 +10,9 @@ use crate::crypto::Signer; use crate::protocol; use crate::storage::git::Storage; +pub mod handle; +pub mod socket; + /// Client configuration. #[derive(Debug, Clone)] pub struct Config { @@ -101,6 +104,16 @@ impl Client { Ok(()) } + + /// Create a new handle to communicate with the client. + pub fn handle(&self) -> handle::Handle { + handle::Handle { + waker: self.reactor.waker(), + commands: self.handle.clone(), + shutdown: self.shutdown.clone(), + listening: self.listening.clone(), + } + } } pub struct Events {} diff --git a/node/src/client/handle.rs b/node/src/client/handle.rs new file mode 100644 index 00000000..9d7e4d82 --- /dev/null +++ b/node/src/client/handle.rs @@ -0,0 +1,77 @@ +use std::net; + +use crossbeam_channel as chan; +use nakamoto_net::Reactor; +use thiserror::Error; + +use crate::identity::ProjId; +use crate::protocol; +use crate::protocol::CommandError; + +/// An error resulting from a handle method. +#[derive(Error, Debug)] +pub enum Error { + /// The command channel is no longer connected. + #[error("command channel is not connected")] + NotConnected, + /// The command returned an error. + #[error("command failed: {0}")] + Command(#[from] CommandError), + /// The operation timed out. + #[error("the operation timed out")] + Timeout, + /// An I/O error occured. + #[error(transparent)] + Io(#[from] std::io::Error), +} + +impl From for Error { + fn from(_: chan::RecvError) -> Self { + Self::NotConnected + } +} + +impl From for Error { + fn from(err: chan::RecvTimeoutError) -> Self { + match err { + chan::RecvTimeoutError::Timeout => Self::Timeout, + chan::RecvTimeoutError::Disconnected => Self::NotConnected, + } + } +} + +impl From> for Error { + fn from(_: chan::SendError) -> Self { + Self::NotConnected + } +} + +pub struct Handle { + pub(crate) commands: chan::Sender, + pub(crate) waker: R::Waker, + pub(crate) shutdown: chan::Sender<()>, + pub(crate) listening: chan::Receiver, +} + +impl Handle { + /// Notify the client that a project has been updated. + pub fn updated(&self, id: ProjId) -> Result<(), Error> { + self.command(protocol::Command::AnnounceInventory(id)) + } + + /// Send a command to the command channel, and wake up the event loop. + pub fn command(&self, cmd: protocol::Command) -> Result<(), Error> { + self.commands.send(cmd)?; + R::wake(&self.waker)?; + + Ok(()) + } + + /// Ask the client to shutdown. + pub fn shutdown(self) -> Result<(), Error> { + self.shutdown.send(())?; + R::wake(&self.waker)?; + + Ok(()) + } +} diff --git a/node/src/client/socket.rs b/node/src/client/socket.rs new file mode 100644 index 00000000..a8bd9406 --- /dev/null +++ b/node/src/client/socket.rs @@ -0,0 +1,80 @@ +use std::io::prelude::*; +use std::io::BufReader; +use std::os::unix::net::UnixListener; +use std::os::unix::net::UnixStream; +use std::path::Path; +use std::str::FromStr; +use std::{fs, io, net}; + +use nakamoto_net::Reactor; + +use crate::client; +use crate::client::handle::Handle; +use crate::identity::ProjId; + +/// Default name for control socket file. +pub const DEFAULT_NAME: &str = "radicle.sock"; + +#[derive(thiserror::Error, Debug)] +pub enum Error { + #[error("failed to bind control socket listener: {0}")] + Bind(io::Error), +} + +/// Listen for commands on the control socket, and process them. +pub fn listen, R: Reactor>(path: P, handle: Handle) -> Result<(), Error> { + // Remove the socket file on startup before rebinding. + fs::remove_file(&path).ok(); + + let listener = UnixListener::bind(path).map_err(Error::Bind)?; + for incoming in listener.incoming() { + match incoming { + Ok(mut stream) => { + if let Err(e) = drain(&stream, &handle) { + log::error!("Received {} on control socket", e); + + write!(stream, "error: {}", e).ok(); + + stream.flush().ok(); + stream.shutdown(net::Shutdown::Both).ok(); + } + } + Err(e) => log::error!("Failed to open control socket stream: {}", e), + } + } + + Ok(()) +} + +#[derive(thiserror::Error, Debug)] +enum DrainError { + #[error("invalid command argument `{0}`")] + InvalidCommandArg(String), + #[error("unknown command `{0}`")] + UnknownCommand(String), + #[error("invalid command")] + InvalidCommand, + #[error("client error: {0}")] + Client(#[from] client::handle::Error), +} + +fn drain(stream: &UnixStream, handle: &Handle) -> Result<(), DrainError> { + let mut reader = BufReader::new(stream); + + for line in reader.by_ref().lines().flatten() { + match line.split_once(' ') { + Some(("update", arg)) => { + if let Ok(id) = ProjId::from_str(arg) { + if let Err(e) = handle.updated(id) { + return Err(DrainError::Client(e)); + } + } else { + return Err(DrainError::InvalidCommandArg(arg.to_owned())); + } + } + Some((cmd, _)) => return Err(DrainError::UnknownCommand(cmd.to_owned())), + None => return Err(DrainError::InvalidCommand), + } + } + Ok(()) +} diff --git a/node/src/protocol.rs b/node/src/protocol.rs index 00478b00..707342e4 100644 --- a/node/src/protocol.rs +++ b/node/src/protocol.rs @@ -54,8 +54,13 @@ pub enum Event {} pub enum Command { Connect(net::SocketAddr), Fetch(ProjId, net::SocketAddr), + AnnounceInventory(ProjId), } +/// Command-related errors. +#[derive(thiserror::Error, Debug)] +pub enum CommandError {} + #[derive(Debug)] pub struct Protocol { /// Peers currently or recently connected. @@ -330,6 +335,9 @@ where }) .unwrap(); } + Command::AnnounceInventory(_proj) => { + todo!() + } } } diff --git a/node/src/rad.rs b/node/src/rad.rs index 662a6b2d..237bee27 100644 --- a/node/src/rad.rs +++ b/node/src/rad.rs @@ -107,6 +107,7 @@ pub fn init( git::set_upstream( repo, REMOTE_NAME, + // FIXME: Should use default branch here. "master", &format!("refs/remotes/{user_id}/heads/master"), )?;