diff --git a/radicle-node/src/client/handle.rs b/radicle-node/src/client/handle.rs index 82838a05..4480a325 100644 --- a/radicle-node/src/client/handle.rs +++ b/radicle-node/src/client/handle.rs @@ -1,4 +1,5 @@ use std::net; +use std::sync::Arc; use crossbeam_channel as chan; use nakamoto_net::Waker; @@ -6,7 +7,8 @@ use thiserror::Error; use crate::identity::Id; use crate::service; -use crate::service::{CommandError, FetchLookup}; +use crate::service::{CommandError, FetchLookup, QueryState}; +use crate::service::{NodeId, Session}; /// An error resulting from a handle method. #[derive(Error, Debug)] @@ -88,6 +90,45 @@ impl traits::Handle for Handle { Ok(()) } + fn routing(&self) -> Result)>, Error> { + let (sender, receiver) = chan::unbounded(); + let query: Arc = Arc::new(move |state| { + for (id, nodes) in state.routing().iter() { + if sender.send((*id, nodes.iter().cloned().collect())).is_err() { + break; + } + } + Ok(()) + }); + let (err_sender, err_receiver) = chan::bounded(1); + self.command(service::Command::QueryState(query, err_sender))?; + err_receiver.recv()??; + + Ok(receiver) + } + + fn sessions(&self) -> Result, Error> { + // TODO: This can be implemented once we have real peer sessions. + todo!() + } + + fn inventory(&self) -> Result, Error> { + let (sender, receiver) = chan::unbounded(); + let query: Arc = Arc::new(move |state| { + for id in state.inventory()?.iter() { + if sender.send(*id).is_err() { + break; + } + } + Ok(()) + }); + let (err_sender, err_receiver) = chan::bounded(1); + self.command(service::Command::QueryState(query, err_sender))?; + err_receiver.recv()??; + + Ok(receiver) + } + /// Ask the client to shutdown. fn shutdown(self) -> Result<(), Error> { self.shutdown.send(())?; @@ -114,5 +155,11 @@ pub mod traits { fn command(&self, cmd: service::Command) -> Result<(), Error>; /// Ask the client to shutdown. fn shutdown(self) -> Result<(), Error>; + /// Query the routing table. + fn routing(&self) -> Result)>, Error>; + /// Query the peer session state. + fn sessions(&self) -> Result, Error>; + /// Query the inventory. + fn inventory(&self) -> Result, Error>; } } diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index f9856831..388e123d 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -52,8 +52,6 @@ enum DrainError { InvalidCommandArg(String), #[error("unknown command `{0}`")] UnknownCommand(String), - #[error("invalid command")] - InvalidCommand, #[error("client error: {0}")] Client(#[from] client::handle::Error), #[error("i/o error: {0}")] @@ -101,7 +99,38 @@ fn drain(stream: &UnixStream, handle: &H) -> Result<(), DrainError> { } } Some((cmd, _)) => return Err(DrainError::UnknownCommand(cmd.to_owned())), - None => return Err(DrainError::InvalidCommand), + + // Commands with no arguments. + None => match line.as_str() { + "routing" => match handle.routing() { + Ok(c) => { + let mut writer = LineWriter::new(stream); + + for (id, seeds) in c.iter() { + let seeds = seeds + .into_iter() + .map(String::from) + .collect::>() + .join(" "); + writeln!(writer, "{id} {seeds}",)?; + } + } + Err(e) => return Err(DrainError::Client(e)), + }, + "inventory" => match handle.inventory() { + Ok(c) => { + let mut writer = LineWriter::new(stream); + + for id in c.iter() { + writeln!(writer, "{id}")?; + } + } + Err(e) => return Err(DrainError::Client(e)), + }, + _ => { + return Err(DrainError::UnknownCommand(line)); + } + }, } } Ok(()) diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 49247dac..a94635ce 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -6,6 +6,7 @@ pub mod reactor; use std::collections::BTreeMap; use std::ops::{Deref, DerefMut}; +use std::sync::Arc; use std::{fmt, net, net::IpAddr}; use crossbeam_channel as chan; @@ -30,12 +31,13 @@ use crate::identity::{Doc, Id}; use crate::service::config::ProjectTracking; use crate::service::message::{Address, Announcement, AnnouncementMessage}; use crate::service::message::{NodeAnnouncement, RefsAnnouncement}; -use crate::service::peer::{Session, SessionError, SessionState}; +use crate::service::peer::{SessionError, SessionState}; use crate::storage; use crate::storage::{Inventory, ReadRepository, RefUpdate, WriteRepository, WriteStorage}; pub use crate::service::config::{Config, Network}; pub use crate::service::message::{Envelope, Message}; +pub use crate::service::peer::Session; use self::gossip::Gossip; use self::message::{InventoryAnnouncement, NodeFeatures}; @@ -109,19 +111,44 @@ pub enum FetchResult { }, } +/// Function used to query internal service state. +pub type QueryState = dyn Fn(&dyn ServiceState) -> Result<(), CommandError> + Send + Sync; + /// Commands sent to the service by the operator. -#[derive(Debug)] pub enum Command { + /// Announce repository references for given project id to peers. AnnounceRefs(Id), + /// Connect to node with the given address. Connect(net::SocketAddr), + /// Fetch the given project from the network. Fetch(Id, chan::Sender), + /// Track the given project. Track(Id, chan::Sender), + /// Untrack the given project. Untrack(Id, chan::Sender), + /// Query the internal service state. + QueryState(Arc, chan::Sender>), +} + +impl fmt::Debug for Command { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::AnnounceRefs(id) => write!(f, "AnnounceRefs({})", id), + Self::Connect(addr) => write!(f, "Connect({})", addr), + Self::Fetch(id, _) => write!(f, "Fetch({})", id), + Self::Track(id, _) => write!(f, "Track({})", id), + Self::Untrack(id, _) => write!(f, "Untrack({})", id), + Self::QueryState { .. } => write!(f, "QueryState(..)"), + } + } } /// Command-related errors. #[derive(thiserror::Error, Debug)] -pub enum CommandError {} +pub enum CommandError { + #[error(transparent)] + Storage(#[from] storage::Error), +} #[derive(Debug)] pub struct Service { @@ -161,6 +188,21 @@ pub struct Service { start_time: LocalTime, } +impl Service +where + G: crypto::Signer, +{ + /// Get the local node id. + pub fn node_id(&self) -> NodeId { + *self.signer.public_key() + } + + /// Get the local service time. + pub fn local_time(&self) -> LocalTime { + self.clock.local_time() + } +} + impl Service where A: address_book::Store, @@ -201,10 +243,6 @@ where } } - pub fn node_id(&self) -> NodeId { - *self.signer.public_key() - } - pub fn seeds(&self, id: &Id) -> Box + '_> { if let Some(peers) = self.routing.get(id) { Box::new( @@ -255,16 +293,6 @@ where todo!() } - /// Get the connected peers. - pub fn sessions(&self) -> &Sessions { - &self.sessions - } - - /// Get the current inventory. - pub fn inventory(&self) -> Result { - self.storage.inventory() - } - /// Get the storage instance. pub fn storage(&self) -> &S { &self.storage @@ -275,36 +303,11 @@ where &mut self.storage } - /// Get a project from storage, using the local node's key. - pub fn get(&self, proj: Id) -> Result>, storage::Error> { - self.storage.get(&self.node_id(), proj) - } - /// Get the local signer. pub fn signer(&self) -> &G { &self.signer } - /// Get the clock. - pub fn clock(&self) -> &RefClock { - &self.clock - } - - /// Get the local service time. - pub fn local_time(&self) -> LocalTime { - self.clock.local_time() - } - - /// Get service configuration. - pub fn config(&self) -> &Config { - &self.config - } - - /// Get reference to routing table. - pub fn routing(&self) -> &Routing { - &self.routing - } - /// Get I/O reactor. pub fn reactor(&mut self) -> &mut Reactor { &mut self.reactor @@ -463,6 +466,9 @@ where self.reactor.broadcast(ann, peers); } + Command::QueryState(query, sender) => { + sender.send(query(self)).ok(); + } } } @@ -789,6 +795,52 @@ where } } +/// Gives read access to the service state. +pub trait ServiceState { + /// Get the connected peers. + fn sessions(&self) -> &Sessions; + /// Get the current inventory. + fn inventory(&self) -> Result; + /// Get a project from storage, using the local node's key. + fn get(&self, proj: Id) -> Result>, storage::Error>; + /// Get the clock. + fn clock(&self) -> &RefClock; + /// Get service configuration. + fn config(&self) -> &Config; + /// Get reference to routing table. + fn routing(&self) -> &Routing; +} + +impl ServiceState for Service +where + G: Signer, + S: ReadStorage, +{ + fn sessions(&self) -> &Sessions { + &self.sessions + } + + fn inventory(&self) -> Result { + self.storage.inventory() + } + + fn get(&self, proj: Id) -> Result>, storage::Error> { + self.storage.get(&self.node_id(), proj) + } + + fn clock(&self) -> &RefClock { + &self.clock + } + + fn config(&self) -> &Config { + &self.config + } + + fn routing(&self) -> &Routing { + &self.routing + } +} + #[derive(Debug, Clone)] pub enum DisconnectReason { User, diff --git a/radicle-node/src/service/peer.rs b/radicle-node/src/service/peer.rs index fcc79525..b7db8c80 100644 --- a/radicle-node/src/service/peer.rs +++ b/radicle-node/src/service/peer.rs @@ -1,7 +1,7 @@ use crate::service::message::*; use crate::service::*; -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] #[allow(clippy::large_enum_variant)] pub enum SessionState { /// Initial peer state. For outgoing peers this @@ -38,7 +38,7 @@ pub enum SessionError { } /// A peer session. Each connected peer will have one session. -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct Session { /// Peer address. pub addr: net::SocketAddr, diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index 1d3f599b..a02dd9f3 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -1,5 +1,7 @@ use std::sync::{Arc, Mutex}; +use crossbeam_channel as chan; + use crate::client::handle::traits; use crate::client::handle::Error; use crate::identity::Id; @@ -34,6 +36,18 @@ impl traits::Handle for Handle { Ok(()) } + fn routing(&self) -> Result)>, Error> { + unimplemented!(); + } + + fn sessions(&self) -> Result, Error> { + unimplemented!(); + } + + fn inventory(&self) -> Result, Error> { + unimplemented!(); + } + fn shutdown(self) -> Result<(), Error> { Ok(()) } diff --git a/radicle-node/src/test/tests.rs b/radicle-node/src/test/tests.rs index 7e9d7bc3..e3e5291c 100644 --- a/radicle-node/src/test/tests.rs +++ b/radicle-node/src/test/tests.rs @@ -11,6 +11,7 @@ use crate::service::filter::Filter; use crate::service::message::*; use crate::service::peer::*; use crate::service::reactor::Io; +use crate::service::ServiceState as _; use crate::service::*; use crate::storage::git::Storage; use crate::storage::ReadStorage;