node: Use JSON for control commands
This commit is contained in:
Alexis Sellier
parent
2649e9c6fe
commit
037ff39894
|
|
@ -5,15 +5,15 @@ use std::io::LineWriter;
|
|||
use std::os::unix::net::UnixListener;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::PathBuf;
|
||||
use std::str::FromStr;
|
||||
use std::{io, net};
|
||||
|
||||
use radicle::node::Handle;
|
||||
use serde_json as json;
|
||||
|
||||
use crate::identity::Id;
|
||||
use crate::node;
|
||||
use crate::node::FetchResult;
|
||||
use crate::node::NodeId;
|
||||
use crate::node::{Command, CommandName, CommandResult, FetchResult};
|
||||
use crate::runtime;
|
||||
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
|
|
@ -44,7 +44,7 @@ pub fn listen<H: Handle<Error = runtime::HandleError, FetchResult = FetchResult>
|
|||
handle.shutdown().ok();
|
||||
break;
|
||||
}
|
||||
writeln!(stream, "error: {e}").ok();
|
||||
CommandResult::error(e).to_writer(&mut stream).ok();
|
||||
|
||||
stream.flush().ok();
|
||||
stream.shutdown(net::Shutdown::Both).ok();
|
||||
|
|
@ -62,6 +62,8 @@ pub fn listen<H: Handle<Error = runtime::HandleError, FetchResult = FetchResult>
|
|||
enum CommandError {
|
||||
#[error("invalid command argument `{0}`, {1}")]
|
||||
InvalidCommandArg(String, Box<dyn std::error::Error>),
|
||||
#[error("invalid command arguments `{0:?}`")]
|
||||
InvalidCommandArgs(Vec<String>),
|
||||
#[error("unknown command `{0}`")]
|
||||
UnknownCommand(String),
|
||||
#[error("serialization failed: {0}")]
|
||||
|
|
@ -83,165 +85,110 @@ fn command<H: Handle<Error = runtime::HandleError, FetchResult = FetchResult>>(
|
|||
let mut line = String::new();
|
||||
|
||||
reader.read_line(&mut line)?;
|
||||
let input = line.trim_end();
|
||||
|
||||
let cmd = line.trim_end();
|
||||
log::debug!(target: "control", "Received `{input}` on control socket");
|
||||
let cmd: Command = json::from_str(input)?;
|
||||
|
||||
log::debug!(target: "control", "Received `{cmd}` on control socket");
|
||||
|
||||
// TODO: refactor to include helper
|
||||
match cmd.split_once(' ') {
|
||||
Some(("fetch", args)) => {
|
||||
if let Some((rid, node)) = args.split_once(' ') {
|
||||
let rid: Id = rid
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(rid.to_owned(), Box::new(e)))?;
|
||||
let node: NodeId = node
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?;
|
||||
|
||||
fetch(rid, node, LineWriter::new(stream), handle)?;
|
||||
}
|
||||
match cmd.name {
|
||||
CommandName::Fetch => {
|
||||
let (rid, nid): (Id, NodeId) = parse::args(cmd)?;
|
||||
fetch(rid, nid, LineWriter::new(stream), handle)?;
|
||||
}
|
||||
Some(("seeds", arg)) => {
|
||||
let rid: Id = arg
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(arg.to_owned(), Box::new(e)))?;
|
||||
CommandName::Seeds => {
|
||||
let rid: Id = parse::arg(cmd)?;
|
||||
let seeds = handle.seeds(rid)?;
|
||||
|
||||
for seed in handle.seeds(rid)? {
|
||||
writeln!(writer, "{seed}")?;
|
||||
}
|
||||
json::to_writer(writer, &seeds)?;
|
||||
}
|
||||
Some(("track-repo", arg)) => match arg.parse() {
|
||||
Ok(id) => match handle.track_repo(id) {
|
||||
CommandName::TrackRepo => {
|
||||
let rid: Id = parse::arg(cmd)?;
|
||||
|
||||
match handle.track_repo(rid) {
|
||||
Ok(updated) => {
|
||||
if updated {
|
||||
writeln!(writer, "{}", node::RESPONSE_OK)?;
|
||||
} else {
|
||||
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
|
||||
}
|
||||
CommandResult::Okay { updated }.to_writer(writer)?;
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
return Err(CommandError::InvalidCommandArg(
|
||||
arg.to_owned(),
|
||||
Box::new(err),
|
||||
));
|
||||
}
|
||||
},
|
||||
Some(("untrack-repo", arg)) => match arg.parse() {
|
||||
Ok(id) => match handle.untrack_repo(id) {
|
||||
}
|
||||
CommandName::UntrackRepo => {
|
||||
let rid: Id = parse::arg(cmd)?;
|
||||
|
||||
match handle.untrack_repo(rid) {
|
||||
Ok(updated) => {
|
||||
if updated {
|
||||
writeln!(writer, "{}", node::RESPONSE_OK)?;
|
||||
} else {
|
||||
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
|
||||
}
|
||||
CommandResult::Okay { updated }.to_writer(writer)?;
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
return Err(CommandError::InvalidCommandArg(
|
||||
arg.to_owned(),
|
||||
Box::new(err),
|
||||
));
|
||||
}
|
||||
},
|
||||
Some(("track-node", args)) => {
|
||||
let (peer, alias) = if let Some((peer, alias)) = args.split_once(' ') {
|
||||
(peer, Some(alias.to_owned()))
|
||||
} else {
|
||||
(args, None)
|
||||
}
|
||||
CommandName::TrackNode => {
|
||||
let (node, alias) = match cmd.args.as_slice() {
|
||||
[node] => (node.as_str(), None),
|
||||
[node, alias] => (node.as_str(), Some(alias.to_owned())),
|
||||
_ => return Err(CommandError::InvalidCommandArgs(cmd.args)),
|
||||
};
|
||||
match peer.parse() {
|
||||
Ok(id) => match handle.track_node(id, alias) {
|
||||
Ok(updated) => {
|
||||
if updated {
|
||||
writeln!(writer, "{}", node::RESPONSE_OK)?;
|
||||
} else {
|
||||
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
return Err(CommandError::InvalidCommandArg(
|
||||
args.to_owned(),
|
||||
Box::new(err),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(("untrack-node", arg)) => match arg.parse() {
|
||||
Ok(id) => match handle.untrack_node(id) {
|
||||
let nid = node
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(node.to_owned(), Box::new(e)))?;
|
||||
|
||||
match handle.track_node(nid, alias) {
|
||||
Ok(updated) => {
|
||||
if updated {
|
||||
writeln!(writer, "{}", node::RESPONSE_OK)?;
|
||||
} else {
|
||||
writeln!(writer, "{}", node::RESPONSE_NOOP)?;
|
||||
}
|
||||
CommandResult::Okay { updated }.to_writer(writer)?;
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
return Err(CommandError::InvalidCommandArg(
|
||||
arg.to_owned(),
|
||||
Box::new(err),
|
||||
));
|
||||
}
|
||||
},
|
||||
Some(("announce-refs", arg)) => match arg.parse() {
|
||||
Ok(id) => {
|
||||
if let Err(e) = handle.announce_refs(id) {
|
||||
}
|
||||
CommandName::UntrackNode => {
|
||||
let nid: NodeId = parse::arg(cmd)?;
|
||||
|
||||
match handle.untrack_node(nid) {
|
||||
Ok(updated) => {
|
||||
CommandResult::Okay { updated }.to_writer(writer)?;
|
||||
}
|
||||
Err(e) => {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
writeln!(writer, "{}", node::RESPONSE_OK)?;
|
||||
}
|
||||
Err(err) => {
|
||||
return Err(CommandError::InvalidCommandArg(
|
||||
arg.to_owned(),
|
||||
Box::new(err),
|
||||
));
|
||||
}
|
||||
},
|
||||
Some((cmd, _)) => return Err(CommandError::UnknownCommand(cmd.to_owned())),
|
||||
}
|
||||
CommandName::AnnounceRefs => {
|
||||
let rid: Id = parse::arg(cmd)?;
|
||||
|
||||
// Commands with no arguments.
|
||||
None => match cmd {
|
||||
"status" => {
|
||||
writeln!(writer, "{}", node::RESPONSE_OK).ok();
|
||||
if let Err(e) = handle.announce_refs(rid) {
|
||||
return Err(CommandError::Runtime(e));
|
||||
}
|
||||
"routing" => match handle.routing() {
|
||||
Ok(c) => {
|
||||
for (id, seed) in c.iter() {
|
||||
writeln!(writer, "{id} {seed}",)?;
|
||||
}
|
||||
CommandResult::ok().to_writer(writer).ok();
|
||||
}
|
||||
CommandName::Status => {
|
||||
CommandResult::ok().to_writer(writer).ok();
|
||||
}
|
||||
CommandName::Routing => match handle.routing() {
|
||||
Ok(c) => {
|
||||
for (id, seed) in c.iter() {
|
||||
writeln!(writer, "{id} {seed}")?;
|
||||
}
|
||||
Err(e) => return Err(CommandError::Runtime(e)),
|
||||
},
|
||||
"inventory" => match handle.inventory() {
|
||||
Ok(c) => {
|
||||
for id in c.iter() {
|
||||
writeln!(writer, "{id}")?;
|
||||
}
|
||||
}
|
||||
Err(e) => return Err(CommandError::Runtime(e)),
|
||||
},
|
||||
"shutdown" => {
|
||||
return Err(CommandError::Shutdown);
|
||||
}
|
||||
_ => {
|
||||
return Err(CommandError::UnknownCommand(line));
|
||||
}
|
||||
Err(e) => return Err(CommandError::Runtime(e)),
|
||||
},
|
||||
CommandName::Inventory => match handle.inventory() {
|
||||
Ok(c) => {
|
||||
for id in c.iter() {
|
||||
writeln!(writer, "{id}")?;
|
||||
}
|
||||
}
|
||||
Err(e) => return Err(CommandError::Runtime(e)),
|
||||
},
|
||||
CommandName::Shutdown => {
|
||||
return Err(CommandError::Shutdown);
|
||||
}
|
||||
_ => {
|
||||
return Err(CommandError::UnknownCommand(line));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -263,6 +210,45 @@ fn fetch<W: Write, H: Handle<Error = runtime::HandleError, FetchResult = FetchRe
|
|||
Ok(())
|
||||
}
|
||||
|
||||
mod parse {
|
||||
use super::*;
|
||||
|
||||
pub(super) fn arg<T: FromStr>(cmd: Command) -> Result<T, CommandError>
|
||||
where
|
||||
<T as FromStr>::Err: std::error::Error + 'static,
|
||||
{
|
||||
let [arg]: [String; 1] = cmd
|
||||
.args
|
||||
.clone()
|
||||
.try_into()
|
||||
.map_err(|_| CommandError::InvalidCommandArgs(cmd.args))?;
|
||||
|
||||
arg.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(arg, Box::new(e)))
|
||||
}
|
||||
|
||||
pub(super) fn args<S: FromStr, T: FromStr>(cmd: Command) -> Result<(S, T), CommandError>
|
||||
where
|
||||
<S as FromStr>::Err: std::error::Error + 'static,
|
||||
<T as FromStr>::Err: std::error::Error + 'static,
|
||||
{
|
||||
let [arg1, arg2]: [String; 2] = cmd
|
||||
.args
|
||||
.clone()
|
||||
.try_into()
|
||||
.map_err(|_| CommandError::InvalidCommandArgs(cmd.args))?;
|
||||
|
||||
let arg1 = arg1
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(arg1, Box::new(e)))?;
|
||||
let arg2 = arg2
|
||||
.parse()
|
||||
.map_err(|e| CommandError::InvalidCommandArg(arg2, Box::new(e)))?;
|
||||
|
||||
Ok((arg1, arg2))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::io::prelude::*;
|
||||
|
|
@ -290,15 +276,22 @@ mod tests {
|
|||
});
|
||||
|
||||
for proj in &projs {
|
||||
let mut buf = [0; 2];
|
||||
let mut stream = loop {
|
||||
let stream = loop {
|
||||
if let Ok(stream) = UnixStream::connect(&socket) {
|
||||
break stream;
|
||||
}
|
||||
};
|
||||
writeln!(&stream, "announce-refs {proj}").unwrap();
|
||||
stream.read_exact(&mut buf).unwrap();
|
||||
assert_eq!(&buf, &[b'o', b'k']);
|
||||
writeln!(
|
||||
&stream,
|
||||
"{}",
|
||||
json::to_string(&Command::new(CommandName::AnnounceRefs, [proj])).unwrap()
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let stream = BufReader::new(stream);
|
||||
let line = stream.lines().next().unwrap().unwrap();
|
||||
|
||||
assert_eq!(line, json::json!({ "status": "ok" }).to_string());
|
||||
}
|
||||
|
||||
for proj in &projs {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
use std::fmt;
|
||||
use std::io::{self, Write};
|
||||
use std::io;
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
|
@ -10,7 +10,7 @@ use thiserror::Error;
|
|||
|
||||
use crate::crypto::Signer;
|
||||
use crate::identity::Id;
|
||||
use crate::node::FetchResult;
|
||||
use crate::node::{Command, FetchResult};
|
||||
use crate::profile::Home;
|
||||
use crate::service;
|
||||
use crate::service::{CommandError, QueryState};
|
||||
|
|
@ -221,7 +221,7 @@ impl<G: Signer + EcSign + 'static> radicle::node::Handle for Handle<G> {
|
|||
// control thread gracefully. Since the control thread may have called this function,
|
||||
// the control socket may already be disconnected. Ignore errors.
|
||||
UnixStream::connect(self.home.socket())
|
||||
.and_then(|mut sock| sock.write_all(b"shutdown"))
|
||||
.and_then(|sock| Command::SHUTDOWN.to_writer(sock))
|
||||
.ok();
|
||||
|
||||
self.controller.shutdown().map_err(|_| Error::NotConnected)
|
||||
|
|
|
|||
|
|
@ -1,14 +1,14 @@
|
|||
mod features;
|
||||
|
||||
use std::io::{BufRead, BufReader, Write};
|
||||
use std::io::{BufRead, BufReader};
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::str::FromStr;
|
||||
use std::{io, net};
|
||||
use std::{fmt, io, net};
|
||||
|
||||
use amplify::WrapperMut;
|
||||
use crossbeam_channel as chan;
|
||||
use cyphernet::addr::{HostName, NetAddr};
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json as json;
|
||||
|
||||
|
|
@ -22,10 +22,58 @@ pub use features::Features;
|
|||
pub const DEFAULT_SOCKET_NAME: &str = "radicle.sock";
|
||||
/// Default radicle protocol port.
|
||||
pub const DEFAULT_PORT: u16 = 8776;
|
||||
/// Response on node socket indicating that a command was carried out successfully.
|
||||
pub const RESPONSE_OK: &str = "ok";
|
||||
/// Response on node socket indicating that a command had no effect.
|
||||
pub const RESPONSE_NOOP: &str = "noop";
|
||||
|
||||
/// Result of a command, on the node control socket.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(tag = "status")]
|
||||
pub enum CommandResult {
|
||||
/// Response on node socket indicating that a command was carried out successfully.
|
||||
#[serde(rename = "ok")]
|
||||
Okay {
|
||||
/// Whether the command had any effect.
|
||||
#[serde(default, skip_serializing_if = "crate::serde_ext::is_default")]
|
||||
updated: bool,
|
||||
},
|
||||
/// Response on node socket indicating that an error occured.
|
||||
Error {
|
||||
/// The reason for the error.
|
||||
reason: String,
|
||||
},
|
||||
}
|
||||
|
||||
impl CommandResult {
|
||||
/// Create an "updated" response.
|
||||
pub fn updated() -> Self {
|
||||
Self::Okay { updated: true }
|
||||
}
|
||||
|
||||
/// Create an "ok" response.
|
||||
pub fn ok() -> Self {
|
||||
Self::Okay { updated: false }
|
||||
}
|
||||
|
||||
/// Create an error result.
|
||||
pub fn error(err: impl std::error::Error) -> Self {
|
||||
Self::Error {
|
||||
reason: err.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Write this command result to a stream, including a terminating LF character.
|
||||
pub fn to_writer(&self, mut w: impl io::Write) -> io::Result<()> {
|
||||
json::to_writer(&mut w, self).map_err(|_| io::ErrorKind::InvalidInput)?;
|
||||
w.write_all(b"\n")
|
||||
}
|
||||
}
|
||||
|
||||
impl From<CommandResult> for Result<bool, Error> {
|
||||
fn from(value: CommandResult) -> Self {
|
||||
match value {
|
||||
CommandResult::Okay { updated } => Ok(updated),
|
||||
CommandResult::Error { reason } => Err(Error::Node(reason)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Peer public protocol address.
|
||||
#[derive(Wrapper, WrapperMut, Clone, Eq, PartialEq, Debug, From)]
|
||||
|
|
@ -54,6 +102,82 @@ impl From<net::SocketAddr> for Address {
|
|||
}
|
||||
}
|
||||
|
||||
/// Command name.
|
||||
#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "kebab-case")]
|
||||
pub enum CommandName {
|
||||
/// Announce repository references for given repository to peers.
|
||||
AnnounceRefs,
|
||||
/// Connect to node with the given address.
|
||||
Connect,
|
||||
/// Lookup seeds for the given repository in the routing table.
|
||||
Seeds,
|
||||
/// Fetch the given repository from the network.
|
||||
Fetch,
|
||||
/// Track the given repository.
|
||||
TrackRepo,
|
||||
/// Untrack the given repository.
|
||||
UntrackRepo,
|
||||
/// Track the given node.
|
||||
TrackNode,
|
||||
/// Untrack the given node.
|
||||
UntrackNode,
|
||||
/// Get the node's inventory.
|
||||
Inventory,
|
||||
/// Get the node's routing table.
|
||||
Routing,
|
||||
/// Get the node's status.
|
||||
Status,
|
||||
/// Shutdown the node.
|
||||
Shutdown,
|
||||
}
|
||||
|
||||
impl fmt::Display for CommandName {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
// SAFETY: The enum can always be converted to a value.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let val = json::to_value(self).unwrap();
|
||||
// SAFETY: The value is always a string.
|
||||
#[allow(clippy::unwrap_used)]
|
||||
let s = val.as_str().unwrap();
|
||||
|
||||
write!(f, "{s}")
|
||||
}
|
||||
}
|
||||
|
||||
/// Commands sent to the node via the control socket.
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct Command {
|
||||
/// Command name.
|
||||
#[serde(rename = "cmd")]
|
||||
pub name: CommandName,
|
||||
/// Command arguments.
|
||||
#[serde(rename = "args")]
|
||||
pub args: Vec<String>,
|
||||
}
|
||||
|
||||
impl Command {
|
||||
/// Shutdown command.
|
||||
pub const SHUTDOWN: Self = Self {
|
||||
name: CommandName::Shutdown,
|
||||
args: vec![],
|
||||
};
|
||||
|
||||
/// Create a new command.
|
||||
pub fn new<T: ToString>(name: CommandName, args: impl IntoIterator<Item = T>) -> Self {
|
||||
Self {
|
||||
name,
|
||||
args: args.into_iter().map(|a| a.to_string()).collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Write this command to a stream, including a terminating LF character.
|
||||
pub fn to_writer(&self, mut w: impl io::Write) -> io::Result<()> {
|
||||
json::to_writer(&mut w, self).map_err(|_| io::ErrorKind::InvalidInput)?;
|
||||
w.write_all(b"\n")
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
#[serde(tag = "status", rename_all = "kebab-case")]
|
||||
pub enum FetchResult {
|
||||
|
|
@ -85,20 +209,30 @@ impl<S: ToString> From<Result<Vec<RefUpdate>, S>> for FetchResult {
|
|||
}
|
||||
}
|
||||
|
||||
/// Error returned by [`Handle`] functions.
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
pub enum Error {
|
||||
#[error("failed to connect to node: {0}")]
|
||||
Connect(#[from] io::Error),
|
||||
#[error("received invalid response for `{cmd}` command: '{response}'")]
|
||||
InvalidResponse { cmd: &'static str, response: String },
|
||||
#[error("failed to call node: {0}")]
|
||||
Call(#[from] CallError),
|
||||
#[error("node: {0}")]
|
||||
Node(String),
|
||||
#[error("received empty response for `{cmd}` command")]
|
||||
EmptyResponse { cmd: CommandName },
|
||||
}
|
||||
|
||||
/// Error returned by [`Node::call`] iterator.
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
pub enum CallError {
|
||||
#[error("i/o: {0}")]
|
||||
Io(#[from] io::Error),
|
||||
#[error("received invalid json in response for `{cmd}` command: '{response}': {error}")]
|
||||
InvalidJson {
|
||||
cmd: &'static str,
|
||||
cmd: CommandName,
|
||||
response: String,
|
||||
error: json::Error,
|
||||
},
|
||||
#[error("received empty response for `{cmd}` command")]
|
||||
EmptyResponse { cmd: &'static str },
|
||||
}
|
||||
|
||||
/// A handle to send commands to the node or request information.
|
||||
|
|
@ -157,24 +291,24 @@ impl Node {
|
|||
}
|
||||
|
||||
/// Call a command on the node.
|
||||
pub fn call<A: ToString>(
|
||||
pub fn call<A: ToString, T: DeserializeOwned>(
|
||||
&self,
|
||||
cmd: &str,
|
||||
args: &[A],
|
||||
) -> Result<impl Iterator<Item = Result<String, io::Error>>, io::Error> {
|
||||
name: CommandName,
|
||||
args: impl IntoIterator<Item = A>,
|
||||
) -> Result<impl Iterator<Item = Result<T, CallError>>, io::Error> {
|
||||
let stream = UnixStream::connect(&self.socket)?;
|
||||
let args = args
|
||||
.iter()
|
||||
.map(ToString::to_string)
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ");
|
||||
Command::new(name, args).to_writer(&stream)?;
|
||||
|
||||
if args.is_empty() {
|
||||
writeln!(&stream, "{cmd}")?;
|
||||
} else {
|
||||
writeln!(&stream, "{cmd} {args}")?;
|
||||
}
|
||||
Ok(BufReader::new(stream).lines())
|
||||
Ok(BufReader::new(stream).lines().map(move |l| {
|
||||
let l = l?;
|
||||
let v = json::from_str(&l).map_err(|e| CallError::InvalidJson {
|
||||
cmd: name,
|
||||
response: l,
|
||||
error: e,
|
||||
})?;
|
||||
|
||||
Ok(v)
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -184,13 +318,13 @@ impl Handle for Node {
|
|||
type FetchResult = FetchResult;
|
||||
|
||||
fn is_running(&self) -> bool {
|
||||
let Ok(mut lines) = self.call::<&str>("status", &[]) else {
|
||||
let Ok(mut lines) = self.call::<&str, CommandResult>(CommandName::Status, []) else {
|
||||
return false;
|
||||
};
|
||||
let Some(Ok(line)) = lines.next() else {
|
||||
let Some(Ok(result)) = lines.next() else {
|
||||
return false;
|
||||
};
|
||||
line == RESPONSE_OK
|
||||
matches!(result, CommandResult::Okay { .. })
|
||||
}
|
||||
|
||||
fn connect(&mut self, _node: NodeId, _addr: Address) -> Result<(), Error> {
|
||||
|
|
@ -198,113 +332,73 @@ impl Handle for Node {
|
|||
}
|
||||
|
||||
fn seeds(&mut self, id: Id) -> Result<Vec<NodeId>, Error> {
|
||||
self.call("seeds", &[id.urn()])?
|
||||
.map(|line| {
|
||||
let line = line?;
|
||||
let node = NodeId::from_str(&line).map_err(|_| Error::InvalidResponse {
|
||||
cmd: "seeds",
|
||||
response: line,
|
||||
})?;
|
||||
Ok(node)
|
||||
})
|
||||
.collect()
|
||||
let seeds: Vec<NodeId> =
|
||||
self.call(CommandName::Seeds, [id.urn()])?
|
||||
.next()
|
||||
.ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::Seeds,
|
||||
})??;
|
||||
|
||||
Ok(seeds)
|
||||
}
|
||||
|
||||
fn fetch(&mut self, id: Id, from: NodeId) -> Result<Self::FetchResult, Error> {
|
||||
let result = self
|
||||
.call("fetch", &[id.urn(), from.to_human()])?
|
||||
.call(CommandName::Fetch, [id.urn(), from.to_human()])?
|
||||
.next()
|
||||
.ok_or(Error::EmptyResponse { cmd: "fetch" })??;
|
||||
let lookup = json::from_str(&result).map_err(|e| Error::InvalidJson {
|
||||
cmd: "fetch",
|
||||
response: result,
|
||||
error: e,
|
||||
})?;
|
||||
.ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::Fetch,
|
||||
})??;
|
||||
|
||||
Ok(lookup)
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
fn track_node(&mut self, id: NodeId, alias: Option<String>) -> Result<bool, Error> {
|
||||
let id = id.to_human();
|
||||
let mut line = if let Some(alias) = alias.as_deref() {
|
||||
self.call("track-node", &[id.as_str(), alias])
|
||||
let args = if let Some(alias) = alias.as_deref() {
|
||||
vec![id.as_str(), alias]
|
||||
} else {
|
||||
self.call("track-node", &[id.as_str()])
|
||||
}?;
|
||||
let line = line
|
||||
.next()
|
||||
.ok_or(Error::EmptyResponse { cmd: "track-node" })??;
|
||||
vec![id.as_str()]
|
||||
};
|
||||
|
||||
log::debug!("node: {}", line);
|
||||
let mut line = self.call(CommandName::TrackNode, args)?;
|
||||
let response: CommandResult = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::TrackNode,
|
||||
})??;
|
||||
|
||||
match line.as_str() {
|
||||
RESPONSE_OK => Ok(true),
|
||||
RESPONSE_NOOP => Ok(false),
|
||||
_ => Err(Error::InvalidResponse {
|
||||
cmd: "track-node",
|
||||
response: line,
|
||||
}),
|
||||
}
|
||||
response.into()
|
||||
}
|
||||
|
||||
fn track_repo(&mut self, id: Id) -> Result<bool, Error> {
|
||||
let mut line = self.call("track-repo", &[id.urn()])?;
|
||||
let line = line
|
||||
.next()
|
||||
.ok_or(Error::EmptyResponse { cmd: "track-repo" })??;
|
||||
let mut line = self.call(CommandName::TrackRepo, [id.urn()])?;
|
||||
let response: CommandResult = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::TrackRepo,
|
||||
})??;
|
||||
|
||||
log::debug!("node: {}", line);
|
||||
|
||||
match line.as_str() {
|
||||
RESPONSE_OK => Ok(true),
|
||||
RESPONSE_NOOP => Ok(false),
|
||||
_ => Err(Error::InvalidResponse {
|
||||
cmd: "track-repo",
|
||||
response: line,
|
||||
}),
|
||||
}
|
||||
response.into()
|
||||
}
|
||||
|
||||
fn untrack_node(&mut self, id: NodeId) -> Result<bool, Error> {
|
||||
let mut line = self.call("untrack-node", &[id])?;
|
||||
let line = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: "untrack-node",
|
||||
let mut line = self.call(CommandName::UntrackNode, [id])?;
|
||||
let response: CommandResult = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::UntrackNode,
|
||||
})??;
|
||||
|
||||
log::debug!("node: {}", line);
|
||||
|
||||
match line.as_str() {
|
||||
RESPONSE_OK => Ok(true),
|
||||
RESPONSE_NOOP => Ok(false),
|
||||
_ => Err(Error::InvalidResponse {
|
||||
cmd: "untrack-node",
|
||||
response: line,
|
||||
}),
|
||||
}
|
||||
response.into()
|
||||
}
|
||||
|
||||
fn untrack_repo(&mut self, id: Id) -> Result<bool, Error> {
|
||||
let mut line = self.call("untrack-repo", &[id.urn()])?;
|
||||
let line = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: "untrack-repo",
|
||||
let mut line = self.call(CommandName::UntrackRepo, [id.urn()])?;
|
||||
let response: CommandResult = line.next().ok_or(Error::EmptyResponse {
|
||||
cmd: CommandName::UntrackRepo,
|
||||
})??;
|
||||
|
||||
log::debug!("node: {}", line);
|
||||
|
||||
match line.as_str() {
|
||||
RESPONSE_OK => Ok(true),
|
||||
RESPONSE_NOOP => Ok(false),
|
||||
_ => Err(Error::InvalidResponse {
|
||||
cmd: "untrack-repo",
|
||||
response: line,
|
||||
}),
|
||||
}
|
||||
response.into()
|
||||
}
|
||||
|
||||
fn announce_refs(&mut self, id: Id) -> Result<(), Error> {
|
||||
for line in self.call("announce-refs", &[id.urn()])? {
|
||||
let line = line?;
|
||||
log::debug!("node: {}", line);
|
||||
for line in self.call(CommandName::AnnounceRefs, [id.urn()])? {
|
||||
line?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -325,3 +419,13 @@ impl Handle for Node {
|
|||
todo!();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_command_name_display() {
|
||||
assert_eq!(CommandName::TrackNode.to_string(), "track-node");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,3 +23,8 @@ pub mod string {
|
|||
.map_err(de::Error::custom)
|
||||
}
|
||||
}
|
||||
|
||||
/// Return true if the given value is the default for that type.
|
||||
pub fn is_default<T: Default + PartialEq>(t: &T) -> bool {
|
||||
t == &T::default()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue