refactor: extract RPC definitions into here

Author: dignifiedquire

Cargo.lock

@@ -50,14 +50,21 @@ redb = { version = "2.0.0" }
 redb_v1  = { package = "redb", version = "1.5.1" }
 self_cell = "1.0.3"
 serde = { version = "1.0.164", features = ["derive"] }
-strum = { version = "0.25", features = ["derive"] }
+strum = { version = "0.26", features = ["derive"] }
 tempfile = { version = "3.4" }
 thiserror = "1"
 tokio = { version = "1", features = ["sync", "rt", "time", "macros"] }
 tokio-stream = { version = "0.1", optional = true, features = ["sync"]}
 tokio-util = { version = "0.7.12", optional = true, features = ["codec", "io-util", "io", "rt"] }
 tracing = "0.1"
 
+# rpc
+nested_enum_utils = { version = "0.1.0", optional = true }
+quic-rpc = { version = "0.13", optional = true }
+quic-rpc-derive = { version = "0.13", optional = true }
+serde-error = { version = "0.1.3", optional = true }
+portable-atomic = { version = "1.9.0", optional = true }
+
 [dev-dependencies]
 iroh-test = "0.27.0"
 rand_chacha = "0.3.1"
@@ -67,10 +74,18 @@ tempfile = "3.4"
 test-strategy = "0.3.1"
 
 [features]
-default = ["net", "metrics", "engine"]
+default = ["net", "metrics", "engine", "rpc"]
 net = ["dep:iroh-net", "tokio/io-util", "dep:tokio-stream", "dep:tokio-util"]
 metrics = ["iroh-metrics/metrics"]
 engine = ["net", "dep:iroh-gossip", "dep:iroh-blobs", "dep:iroh-router"]
+rpc = [
+  "engine",
+  "dep:nested_enum_utils",
+  "dep:quic-rpc",
+  "dep:quic-rpc-derive",
+  "dep:serde-error",
+  "dep:portable-atomic",
+]
 
 [package.metadata.docs.rs]
 all-features = true

Cargo.toml

@@ -11,7 +11,9 @@ use std::{
 
 use anyhow::{bail, Context, Result};
 use futures_lite::{Stream, StreamExt};
-use iroh_blobs::{downloader::Downloader, store::EntryStatus, Hash};
+use iroh_blobs::{
+    downloader::Downloader, store::EntryStatus, util::local_pool::LocalPoolHandle, Hash,
+};
 use iroh_gossip::net::Gossip;
 use iroh_net::{key::PublicKey, Endpoint, NodeAddr};
 use serde::{Deserialize, Serialize};
@@ -52,6 +54,7 @@ pub struct Engine {
     actor_handle: Arc<AbortOnDropHandle<()>>,
     #[debug("ContentStatusCallback")]
     content_status_cb: ContentStatusCallback,
+    local_pool_handle: LocalPoolHandle,
 }
 
 impl Engine {
@@ -66,6 +69,7 @@ impl Engine {
         bao_store: B,
         downloader: Downloader,
         default_author_storage: DefaultAuthorStorage,
+        local_pool_handle: LocalPoolHandle,
     ) -> anyhow::Result<Self> {
         let (live_actor_tx, to_live_actor_recv) = mpsc::channel(ACTOR_CHANNEL_CAP);
         let me = endpoint.node_id().fmt_short();
@@ -111,6 +115,7 @@ impl Engine {
             actor_handle: Arc::new(AbortOnDropHandle::new(actor_handle)),
             content_status_cb,
             default_author: Arc::new(default_author),
+            local_pool_handle,
         })
     }
 
@@ -205,6 +210,10 @@ impl Engine {
         reply_rx.await?;
         Ok(())
     }
+
+    pub(crate) fn local_pool_handle(&self) -> &LocalPoolHandle {
+        &self.local_pool_handle
+    }
 }
 
 /// Converts an [`EntryStatus`] into a ['ContentStatus'].

src/engine.rs

@@ -48,6 +48,9 @@ mod ticket;
 #[cfg(feature = "engine")]
 #[cfg_attr(iroh_docsrs, doc(cfg(feature = "engine")))]
 pub mod engine;
+#[cfg(feature = "engine")]
+#[cfg_attr(iroh_docsrs, doc(cfg(feature = "engine")))]
+pub mod rpc;
 
 pub mod actor;
 pub mod store;

src/lib.rs

@@ -0,0 +1,56 @@
+//! Quic RPC implemenation for docs.
+
+use crate::engine::Engine;
+
+pub mod client;
+pub mod proto;
+
+mod docs_handle_request;
+
+type RpcError = serde_error::Error;
+type RpcResult<T> = std::result::Result<T, RpcError>;
+
+impl Engine {
+    /// Handle a docs request from the RPC server.
+    pub async fn handle_rpc_request<S: quic_rpc::Service, C: quic_rpc::ServiceEndpoint<S>>(
+        &self,
+        msg: crate::rpc::proto::Request,
+        chan: quic_rpc::server::RpcChannel<crate::rpc::proto::RpcService, C, S>,
+    ) -> Result<(), quic_rpc::server::RpcServerError<C>> {
+        use crate::rpc::proto::Request::*;
+
+        let this = self.clone();
+        match msg {
+            Open(msg) => chan.rpc(msg, this, Self::doc_open).await,
+            Close(msg) => chan.rpc(msg, this, Self::doc_close).await,
+            Status(msg) => chan.rpc(msg, this, Self::doc_status).await,
+            List(msg) => chan.server_streaming(msg, this, Self::doc_list).await,
+            Create(msg) => chan.rpc(msg, this, Self::doc_create).await,
+            Drop(msg) => chan.rpc(msg, this, Self::doc_drop).await,
+            Import(msg) => chan.rpc(msg, this, Self::doc_import).await,
+            Set(msg) => chan.rpc(msg, this, Self::doc_set).await,
+            ImportFile(msg) => {
+                chan.server_streaming(msg, this, Self::doc_import_file)
+                    .await
+            }
+            ExportFile(msg) => {
+                chan.server_streaming(msg, this, Self::doc_export_file)
+                    .await
+            }
+            Del(msg) => chan.rpc(msg, this, Self::doc_del).await,
+            SetHash(msg) => chan.rpc(msg, this, Self::doc_set_hash).await,
+            Get(msg) => chan.server_streaming(msg, this, Self::doc_get_many).await,
+            GetExact(msg) => chan.rpc(msg, this, Self::doc_get_exact).await,
+            StartSync(msg) => chan.rpc(msg, this, Self::doc_start_sync).await,
+            Leave(msg) => chan.rpc(msg, this, Self::doc_leave).await,
+            Share(msg) => chan.rpc(msg, this, Self::doc_share).await,
+            Subscribe(msg) => {
+                chan.try_server_streaming(msg, this, Self::doc_subscribe)
+                    .await
+            }
+            SetDownloadPolicy(msg) => chan.rpc(msg, this, Self::doc_set_download_policy).await,
+            GetDownloadPolicy(msg) => chan.rpc(msg, this, Self::doc_get_download_policy).await,
+            GetSyncPeers(msg) => chan.rpc(msg, this, Self::doc_get_sync_peers).await,
+        }
+    }
+}