node: Start implementing fetch control

Signed-off-by: Alexis Sellier <self@cloudhead.io>

Cargo.lock

@@ -111,6 +111,16 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "771fe0050b883fcc3ea2359b1a96bcfbc090b7116eae7c3c512c7a083fdf23d3"
 
+[[package]]
+name = "bstr"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba3569f383e8f1598449f1a423e72e99569137b47740b1da11ef19af3d5c3223"
+dependencies = [
+ "memchr",
+ "serde",
+]
+
 [[package]]
 name = "bumpalo"
 version = "3.11.0"
@@ -356,6 +366,19 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "git-url"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9fcc88c679f3dc55e1e4fe0324142abec612cafaafae1bf6a951c324c9d96f8"
+dependencies = [
+ "bstr",
+ "home",
+ "quick-error",
+ "serde",
+ "url",
+]
+
 [[package]]
 name = "git2"
 version = "0.13.25"
@@ -392,6 +415,15 @@ version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
 
+[[package]]
+name = "home"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2456aef2e6b6a9784192ae780c0f15bc57df0e918585282325e8c8ac27737654"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "iana-time-zone"
 version = "0.1.48"
@@ -526,6 +558,12 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "memchr"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
+
 [[package]]
 name = "multibase"
 version = "0.9.1"
@@ -689,6 +727,12 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "quick-error"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
+
 [[package]]
 name = "quickcheck"
 version = "1.0.3"
@@ -740,9 +784,11 @@ dependencies = [
  "bs58",
  "chrono",
  "colored",
+ "crossbeam-channel",
  "ed25519-consensus",
  "fastrand",
  "git-ref-format",
+ "git-url",
  "git2",
  "log",
  "multibase",

node/src/client.rs

@@ -105,7 +105,7 @@ impl<R: Reactor> Client<R> {
     }
 
     /// Create a new handle to communicate with the client.
-    pub fn handle(&self) -> handle::Handle<R> {
+    pub fn handle(&self) -> handle::Handle<R::Waker> {
         handle::Handle {
             waker: self.reactor.waker(),
             commands: self.handle.clone(),

node/src/client/handle.rs

@@ -1,12 +1,12 @@
 use std::net;
 
 use crossbeam_channel as chan;
-use nakamoto_net::Reactor;
+use nakamoto_net::Waker;
 use thiserror::Error;
 
 use crate::identity::ProjId;
 use crate::protocol;
-use crate::protocol::CommandError;
+use crate::protocol::{CommandError, FetchLookup};
 
 /// An error resulting from a handle method.
 #[derive(Error, Debug)]
@@ -46,17 +46,19 @@ impl<T> From<chan::SendError<T>> for Error {
     }
 }
 
-pub struct Handle<R: Reactor> {
+pub struct Handle<W: Waker> {
     pub(crate) commands: chan::Sender<protocol::Command>,
-    pub(crate) waker: R::Waker,
     pub(crate) shutdown: chan::Sender<()>,
     pub(crate) listening: chan::Receiver<net::SocketAddr>,
+    pub(crate) waker: W,
 }
 
-impl<R: Reactor> traits::Handle for Handle<R> {
-    /// Retrieve or update the project from network.
-    fn fetch(&self, _id: ProjId) -> Result<(), Error> {
-        todo!()
+impl<W: Waker> traits::Handle for Handle<W> {
+    /// Retrieve or update the given project from the network.
+    fn fetch(&self, id: ProjId) -> Result<FetchLookup, Error> {
+        let (sender, receiver) = chan::bounded(1);
+        self.commands.send(protocol::Command::Fetch(id, sender))?;
+        receiver.recv().map_err(Error::from)
     }
 
     /// Start tracking the given project. Doesn't do anything if the project is already tracked.
@@ -81,7 +83,7 @@ impl<R: Reactor> traits::Handle for Handle<R> {
     /// Send a command to the command channel, and wake up the event loop.
     fn command(&self, cmd: protocol::Command) -> Result<(), Error> {
         self.commands.send(cmd)?;
-        R::wake(&self.waker)?;
+        self.waker.wake()?;
 
         Ok(())
     }
@@ -89,7 +91,7 @@ impl<R: Reactor> traits::Handle for Handle<R> {
     /// Ask the client to shutdown.
     fn shutdown(self) -> Result<(), Error> {
         self.shutdown.send(())?;
-        R::wake(&self.waker)?;
+        self.waker.wake()?;
 
         Ok(())
     }
@@ -100,7 +102,7 @@ pub mod traits {
 
     pub trait Handle {
         /// Retrieve or update the project from network.
-        fn fetch(&self, id: ProjId) -> Result<(), Error>;
+        fn fetch(&self, id: ProjId) -> Result<FetchLookup, Error>;
         /// Start tracking the given project. Doesn't do anything if the project is already
         /// tracked.
         fn track(&self, id: ProjId) -> Result<bool, Error>;

node/src/control.rs

@@ -1,6 +1,7 @@
 //! Client control socket implementation.
 use std::io::prelude::*;
 use std::io::BufReader;
+use std::io::LineWriter;
 use std::os::unix::net::UnixListener;
 use std::os::unix::net::UnixStream;
 use std::path::Path;
@@ -8,6 +9,9 @@ use std::{fs, io, net};
 
 use crate::client;
 use crate::client::handle::traits::Handle;
+use crate::identity::ProjId;
+use crate::protocol::FetchLookup;
+use crate::protocol::FetchResult;
 
 /// Default name for control socket file.
 pub const DEFAULT_SOCKET_NAME: &str = "radicle.sock";
@@ -55,6 +59,8 @@ enum DrainError {
     InvalidCommand,
     #[error("client error: {0}")]
     Client(#[from] client::handle::Error),
+    #[error("i/o error: {0}")]
+    Io(#[from] io::Error),
 }
 
 fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
@@ -65,9 +71,7 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
         match line.split_once(' ') {
             Some(("fetch", arg)) => {
                 if let Ok(id) = arg.parse() {
-                    if let Err(e) = handle.fetch(id) {
-                        return Err(DrainError::Client(e));
-                    }
+                    fetch(id, LineWriter::new(stream), handle)?;
                 } else {
                     return Err(DrainError::InvalidCommandArg(arg.to_owned()));
                 }
@@ -106,6 +110,47 @@ fn drain<H: Handle>(stream: &UnixStream, handle: &H) -> Result<(), DrainError> {
     Ok(())
 }
 
+fn fetch<W: Write, H: Handle>(id: ProjId, mut writer: W, handle: &H) -> Result<(), DrainError> {
+    match handle.fetch(id.clone()) {
+        Err(e) => {
+            return Err(DrainError::Client(e));
+        }
+        Ok(FetchLookup::Found { providers, results }) => {
+            let providers = Vec::from(providers);
+
+            writeln!(
+                writer,
+                "ok: found {} providers for {} ({:?})",
+                providers.len(),
+                &id,
+                &providers,
+            )?;
+
+            for result in results.iter() {
+                match result {
+                    FetchResult::Fetched { from } => {
+                        writeln!(writer, "ok: {} fetched from {}", &id, from)?;
+                    }
+                    FetchResult::Error { from, error } => {
+                        writeln!(
+                            writer,
+                            "error: {} failed to fetch from {}: {}",
+                            &id, from, error
+                        )?;
+                    }
+                }
+            }
+        }
+        Ok(FetchLookup::NotFound) => {
+            writeln!(writer, "error: {} was not found", &id)?;
+        }
+        Ok(FetchLookup::Error(err)) => {
+            writeln!(writer, "error: {}", err)?;
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use std::io::prelude::*;

node/src/test/handle.rs

@@ -4,6 +4,7 @@ use crate::client::handle::traits;
 use crate::client::handle::Error;
 use crate::identity::ProjId;
 use crate::protocol;
+use crate::protocol::FetchLookup;
 
 #[derive(Default, Clone)]
 pub struct Handle {
@@ -11,8 +12,8 @@ pub struct Handle {
 }
 
 impl traits::Handle for Handle {
-    fn fetch(&self, _id: ProjId) -> Result<(), Error> {
-        Ok(())
+    fn fetch(&self, _id: ProjId) -> Result<FetchLookup, Error> {
+        Ok(FetchLookup::NotFound)
     }
 
     fn track(&self, _id: ProjId) -> Result<bool, Error> {