much progress has been made

Author: Frando

Cargo.lock

@@ -76,6 +76,8 @@ tracing = "0.1"
 # rpc
 irpc = { version = "0.5.0" }
 n0-future = "0.1.3"
+serde-error = "0.1.3"
+quinn = { package = "iroh-quinn", version = "0.14.0" }
 
 [dev-dependencies]
 rand_chacha = "0.3.1"
@@ -92,7 +94,7 @@ testdir = "0.7"
 data-encoding = "2.6.0"
 
 [features]
-default = ["net", "metrics", "engine", "test-utils"]
+default = ["net", "metrics", "engine", "test-utils", "rpc"]
 net = ["dep:iroh", "tokio/io-util", "dep:tokio-stream", "dep:tokio-util"]
 metrics = ["iroh-metrics/metrics", "iroh/metrics"]
 engine = ["net", "dep:iroh-gossip"]

Cargo.toml

@@ -12,20 +12,24 @@ use anyhow::{anyhow, Context, Result};
 use bytes::Bytes;
 use futures_util::FutureExt;
 use iroh_blobs::Hash;
+use irpc::channel::mpsc;
 use serde::{Deserialize, Serialize};
 use tokio::{sync::oneshot, task::JoinSet};
 use tracing::{debug, error, error_span, trace, warn};
 
 use crate::{
     metrics::Metrics,
     ranger::Message,
+    rpc2::{
+        protocol::{AuthorListResponse, ListResponse},
+        RpcError, RpcResult,
+    },
     store::{
         fs::{ContentHashesIterator, StoreInstance},
         DownloadPolicy, ImportNamespaceOutcome, Query, Store,
     },
-    Author, AuthorHeads, AuthorId, Capability, CapabilityKind, ContentStatus,
-    ContentStatusCallback, Event, NamespaceId, NamespaceSecret, PeerIdBytes, Replica, ReplicaInfo,
-    SignedEntry, SyncOutcome,
+    Author, AuthorHeads, AuthorId, Capability, ContentStatus, ContentStatusCallback, Event,
+    NamespaceId, NamespaceSecret, PeerIdBytes, Replica, ReplicaInfo, SignedEntry, SyncOutcome,
 };
 
 const ACTION_CAP: usize = 1024;
@@ -60,12 +64,12 @@ enum Action {
     #[display("ListAuthors")]
     ListAuthors {
         #[debug("reply")]
-        reply: async_channel::Sender<Result<AuthorId>>,
+        reply: mpsc::Sender<RpcResult<AuthorListResponse>>,
     },
     #[display("ListReplicas")]
     ListReplicas {
         #[debug("reply")]
-        reply: async_channel::Sender<Result<(NamespaceId, CapabilityKind)>>,
+        reply: mpsc::Sender<RpcResult<ListResponse>>,
     },
     #[display("ContentHashes")]
     ContentHashes {
@@ -165,7 +169,7 @@ enum ReplicaAction {
     },
     GetMany {
         query: Query,
-        reply: async_channel::Sender<Result<SignedEntry>>,
+        reply: mpsc::Sender<RpcResult<SignedEntry>>,
     },
     DropReplica {
         reply: oneshot::Sender<Result<()>>,
@@ -290,6 +294,7 @@ impl SyncHandle {
     }
 
     pub async fn open(&self, namespace: NamespaceId, opts: OpenOpts) -> Result<()> {
+        tracing::debug!("SyncHandle::open called");
         let (reply, rx) = oneshot::channel();
         let action = ReplicaAction::Open { reply, opts };
         self.send_replica(namespace, action).await?;
@@ -443,7 +448,7 @@ impl SyncHandle {
         &self,
         namespace: NamespaceId,
         query: Query,
-        reply: async_channel::Sender<Result<SignedEntry>>,
+        reply: mpsc::Sender<RpcResult<SignedEntry>>,
     ) -> Result<()> {
         let action = ReplicaAction::GetMany { query, reply };
         self.send_replica(namespace, action).await?;
@@ -497,14 +502,14 @@ impl SyncHandle {
         Ok(store)
     }
 
-    pub async fn list_authors(&self, reply: async_channel::Sender<Result<AuthorId>>) -> Result<()> {
+    pub async fn list_authors(
+        &self,
+        reply: mpsc::Sender<RpcResult<AuthorListResponse>>,
+    ) -> Result<()> {
         self.send(Action::ListAuthors { reply }).await
     }
 
-    pub async fn list_replicas(
-        &self,
-        reply: async_channel::Sender<Result<(NamespaceId, CapabilityKind)>>,
-    ) -> Result<()> {
+    pub async fn list_replicas(&self, reply: mpsc::Sender<RpcResult<ListResponse>>) -> Result<()> {
         self.send(Action::ListReplicas { reply }).await
     }
 
@@ -696,15 +701,18 @@ impl Actor {
                 let iter = self
                     .store
                     .list_authors()
-                    .map(|a| a.map(|a| a.map(|a| a.id())));
+                    .map(|a| a.map(|a| a.map(|a| AuthorListResponse { author_id: a.id() })));
                 self.tasks
-                    .spawn_local(iter_to_channel_async(reply, iter).map(|_| ()));
+                    .spawn_local(iter_to_irpc(reply, iter).map(|_| ()));
                 Ok(())
             }
             Action::ListReplicas { reply } => {
                 let iter = self.store.list_namespaces();
+                let iter = iter.map(|inner| {
+                    inner.map(|res| res.map(|(id, capability)| ListResponse { id, capability }))
+                });
                 self.tasks
-                    .spawn_local(iter_to_channel_async(reply, iter).map(|_| ()));
+                    .spawn_local(iter_to_irpc(reply, iter).map(|_| ()));
                 Ok(())
             }
             Action::ContentHashes { reply } => {
@@ -838,7 +846,7 @@ impl Actor {
                     .ensure_open(&namespace)
                     .and_then(|_| self.store.get_many(namespace, query));
                 self.tasks
-                    .spawn_local(iter_to_channel_async(reply, iter).map(|_| ()));
+                    .spawn_local(iter_to_irpc(reply, iter).map(|_| ()));
                 Ok(())
             }
             ReplicaAction::DropReplica { reply } => send_reply_with(reply, self, |this| {
@@ -984,6 +992,7 @@ impl OpenReplicas {
             }
             hash_map::Entry::Occupied(mut e) => {
                 let state = e.get_mut();
+                tracing::debug!("STATE {state:?}");
                 state.handles = state.handles.wrapping_sub(1);
                 if state.handles == 0 {
                     let _ = e.remove_entry();
@@ -1001,14 +1010,18 @@ impl OpenReplicas {
     }
 }
 
-async fn iter_to_channel_async<T: Send + 'static>(
-    channel: async_channel::Sender<Result<T>>,
+async fn iter_to_irpc<T: irpc::RpcMessage>(
+    channel: mpsc::Sender<RpcResult<T>>,
     iter: Result<impl Iterator<Item = Result<T>>>,
 ) -> Result<(), SendReplyError> {
     match iter {
-        Err(err) => channel.send(Err(err)).await.map_err(send_reply_error)?,
+        Err(err) => channel
+            .send(Err(RpcError::new(&*err)))
+            .await
+            .map_err(send_reply_error)?,
         Ok(iter) => {
             for item in iter {
+                let item = item.map_err(|err| RpcError::new(&*err));
                 channel.send(item).await.map_err(send_reply_error)?;
             }
         }

src/actor.rs

@@ -52,6 +52,8 @@ pub mod actor;
 pub mod store;
 pub mod sync;
 
+pub mod rpc2;
+
 mod heads;
 mod keys;
 mod ranger;

src/lib.rs

@@ -9,6 +9,7 @@ use iroh_gossip::net::Gossip;
 
 use crate::{
     engine::{DefaultAuthorStorage, Engine},
+    rpc2::api::DocsApi,
     store::Store,
 };
 
@@ -32,8 +33,8 @@ impl ProtocolHandler for Docs {
 #[derive(Debug, Clone)]
 pub struct Docs {
     engine: Arc<Engine>,
-    // #[cfg(feature = "rpc")]
-    // pub(crate) rpc_handler: Arc<std::sync::OnceLock<crate::rpc::RpcHandler>>,
+    api: DocsApi, // #[cfg(feature = "rpc")]
+                  // pub(crate) rpc_handler: Arc<std::sync::OnceLock<crate::rpc::RpcHandler>>,
 }
 
 impl Docs {
@@ -47,6 +48,11 @@ impl Docs {
     pub fn persistent(path: PathBuf) -> Builder {
         Builder { path: Some(path) }
     }
+
+    /// Returns the API for this docs instance.
+    pub fn api(&self) -> &DocsApi {
+        &self.api
+    }
 }
 
 impl Docs {
@@ -63,10 +69,12 @@ impl Docs {
     ///
     /// Note that usually you would use the [`Builder`] to create a new docs protocol.
     pub fn new(engine: Engine) -> Self {
+        let engine = Arc::new(engine);
+        let api = DocsApi::spawn(engine.clone());
         Self {
-            engine: Arc::new(engine),
-            // #[cfg(feature = "rpc")]
-            // rpc_handler: Default::default(),
+            engine,
+            api, // #[cfg(feature = "rpc")]
+                 // rpc_handler: Default::default(),
         }
     }