912a5ca protocol/service: create command module — heartwood

heartwood

rad:z3gqcJUoA1n9HaHKufZs5FCSGazv5

Radicle Heartwood Protocol & Stack

protocol/service: create command module

Fintan Halpenny committed 2 months ago

commit 912a5ca42b2c8a0b033d783954e2eba49018d96c
parent fa94638

2 files changed +88 -72

modified crates/radicle-protocol/src/service.rs

@@ -2,6 +2,9 @@

 #![allow(clippy::collapsible_match)]
 #![allow(clippy::collapsible_if)]
 #![warn(clippy::unwrap_used)]
 pub mod command;
 pub use command::{Command, CommandError, QueryState};
 pub mod filter;
 pub mod gossip;
 pub mod io;

@@ -33,7 +36,7 @@ use radicle::node::refs::Store as _;

 use radicle::node::routing::Store as _;
 use radicle::node::seed;
 use radicle::node::seed::Store as _;
 use radicle::node::{ConnectOptions, Penalty, Severity};
 use radicle::node::{Penalty, Severity};
 use radicle::storage::refs::SIGREFS_BRANCH;
 use radicle::storage::RepositoryError;
 use radicle_fetch::policy::SeedingPolicy;

@@ -51,9 +54,7 @@ use radicle::identity::RepoId;

 use radicle::node::events::Emitter;
 use radicle::node::routing;
 use radicle::node::routing::InsertResult;
 use radicle::node::{
     Address, Alias, Features, FetchResult, HostName, Seed, Seeds, SyncStatus, SyncedAt,
 };
 use radicle::node::{Address, Features, FetchResult, HostName, Seed, Seeds, SyncStatus, SyncedAt};
 use radicle::prelude::*;
 use radicle::storage;
 use radicle::storage::{refs::RefsAt, Namespaces, ReadStorage};

@@ -239,74 +240,6 @@ pub trait Store:

 impl Store for radicle::node::Database {}
 /// 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.
 pub enum Command {
     /// Announce repository references for given repository and namespaces to peers.
     AnnounceRefs(RepoId, HashSet<PublicKey>, chan::Sender<RefsAt>),
     /// Announce local repositories to peers.
     AnnounceInventory,
     /// Add repository to local inventory.
     AddInventory(RepoId, chan::Sender<bool>),
     /// Connect to node with the given address.
     Connect(NodeId, Address, ConnectOptions),
     /// Disconnect from node.
     Disconnect(NodeId),
     /// Get the node configuration.
     Config(chan::Sender<Config>),
     /// Get the node's listen addresses.
     ListenAddrs(chan::Sender<Vec<std::net::SocketAddr>>),
     /// Lookup seeds for the given repository in the routing table, and report
     /// sync status for given namespaces.
     Seeds(RepoId, HashSet<PublicKey>, chan::Sender<Seeds>),
     /// Fetch the given repository from the network.
     Fetch(RepoId, NodeId, time::Duration, chan::Sender<FetchResult>),
     /// Seed the given repository.
     Seed(RepoId, Scope, chan::Sender<bool>),
     /// Unseed the given repository.
     Unseed(RepoId, chan::Sender<bool>),
     /// Follow the given node.
     Follow(NodeId, Option<Alias>, chan::Sender<bool>),
     /// Unfollow the given node.
     Unfollow(NodeId, chan::Sender<bool>),
     /// Query the internal service state.
     QueryState(Arc<QueryState>, chan::Sender<Result<(), CommandError>>),
 }
 impl fmt::Debug for Command {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::AnnounceRefs(id, _, _) => write!(f, "AnnounceRefs({id})"),
             Self::AnnounceInventory => write!(f, "AnnounceInventory"),
             Self::AddInventory(rid, _) => write!(f, "AddInventory({rid})"),
             Self::Connect(id, addr, opts) => write!(f, "Connect({id}, {addr}, {opts:?})"),
             Self::Disconnect(id) => write!(f, "Disconnect({id})"),
             Self::Config(_) => write!(f, "Config"),
             Self::ListenAddrs(_) => write!(f, "ListenAddrs"),
             Self::Seeds(id, _, _) => write!(f, "Seeds({id})"),
             Self::Fetch(id, node, _, _) => write!(f, "Fetch({id}, {node})"),
             Self::Seed(id, scope, _) => write!(f, "Seed({id}, {scope})"),
             Self::Unseed(id, _) => write!(f, "Unseed({id})"),
             Self::Follow(id, _, _) => write!(f, "Follow({id})"),
             Self::Unfollow(id, _) => write!(f, "Unfollow({id})"),
             Self::QueryState { .. } => write!(f, "QueryState(..)"),
         }
     }
 }
 /// Command-related errors.
 #[derive(thiserror::Error, Debug)]
 pub enum CommandError {
     #[error(transparent)]
     Storage(#[from] storage::Error),
     #[error(transparent)]
     Routing(#[from] routing::Error),
     #[error(transparent)]
     Policy(#[from] policy::Error),
 }
 /// Fetch state for an ongoing fetch.
 #[derive(Debug)]
 pub struct FetchState {

added crates/radicle-protocol/src/service/command.rs

@@ -0,0 +1,83 @@

 use std::{collections::HashSet, fmt, sync::Arc, time};
 use crossbeam_channel as chan;
 use radicle::crypto::PublicKey;
 use radicle::node::policy::config as policy;
 use radicle::node::policy::Scope;
 use radicle::node::routing;
 use radicle::node::FetchResult;
 use radicle::node::Seeds;
 use radicle::node::{Address, Alias, Config, ConnectOptions};
 use radicle::storage;
 use radicle::storage::refs::RefsAt;
 use radicle_core::{NodeId, RepoId};
 use super::ServiceState;
 /// 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.
 pub enum Command {
     /// Announce repository references for given repository and namespaces to peers.
     AnnounceRefs(RepoId, HashSet<PublicKey>, chan::Sender<RefsAt>),
     /// Announce local repositories to peers.
     AnnounceInventory,
     /// Add repository to local inventory.
     AddInventory(RepoId, chan::Sender<bool>),
     /// Connect to node with the given address.
     Connect(NodeId, Address, ConnectOptions),
     /// Disconnect from node.
     Disconnect(NodeId),
     /// Get the node configuration.
     Config(chan::Sender<Config>),
     /// Get the node's listen addresses.
     ListenAddrs(chan::Sender<Vec<std::net::SocketAddr>>),
     /// Lookup seeds for the given repository in the routing table, and report
     /// sync status for given namespaces.
     Seeds(RepoId, HashSet<PublicKey>, chan::Sender<Seeds>),
     /// Fetch the given repository from the network.
     Fetch(RepoId, NodeId, time::Duration, chan::Sender<FetchResult>),
     /// Seed the given repository.
     Seed(RepoId, Scope, chan::Sender<bool>),
     /// Unseed the given repository.
     Unseed(RepoId, chan::Sender<bool>),
     /// Follow the given node.
     Follow(NodeId, Option<Alias>, chan::Sender<bool>),
     /// Unfollow the given node.
     Unfollow(NodeId, chan::Sender<bool>),
     /// Query the internal service state.
     QueryState(Arc<QueryState>, chan::Sender<Result<(), CommandError>>),
 }
 impl fmt::Debug for Command {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::AnnounceRefs(id, _, _) => write!(f, "AnnounceRefs({id})"),
             Self::AnnounceInventory => write!(f, "AnnounceInventory"),
             Self::AddInventory(rid, _) => write!(f, "AddInventory({rid})"),
             Self::Connect(id, addr, opts) => write!(f, "Connect({id}, {addr}, {opts:?})"),
             Self::Disconnect(id) => write!(f, "Disconnect({id})"),
             Self::Config(_) => write!(f, "Config"),
             Self::ListenAddrs(_) => write!(f, "ListenAddrs"),
             Self::Seeds(id, _, _) => write!(f, "Seeds({id})"),
             Self::Fetch(id, node, _, _) => write!(f, "Fetch({id}, {node})"),
             Self::Seed(id, scope, _) => write!(f, "Seed({id}, {scope})"),
             Self::Unseed(id, _) => write!(f, "Unseed({id})"),
             Self::Follow(id, _, _) => write!(f, "Follow({id})"),
             Self::Unfollow(id, _) => write!(f, "Unfollow({id})"),
             Self::QueryState { .. } => write!(f, "QueryState(..)"),
         }
     }
 }
 /// Command-related errors.
 #[derive(thiserror::Error, Debug)]
 pub enum CommandError {
     #[error(transparent)]
     Storage(#[from] storage::Error),
     #[error(transparent)]
     Routing(#[from] routing::Error),
     #[error(transparent)]
     Policy(#[from] policy::Error),
 }