Merge 127a9fd into 73ca073

Author: Frando

Cargo.lock

@@ -42,7 +42,7 @@ self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
 iroh-base = "0.90"
 reflink-copy = "0.1.24"
-irpc = { version = "0.5.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream", "derive"], default-features = false }
+irpc = { version = "0.5.0", features = ["rpc", "quinn_endpoint_setup", "spans", "stream", "derive"], default-features = false }
 iroh-metrics = { version = "0.35" }
 
 [dev-dependencies]
@@ -69,4 +69,3 @@ default = ["hide-proto-docs"]
 iroh = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
 iroh-base = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
 irpc = { git = "https://github.com/n0-computer/irpc.git", branch = "main" }
-irpc-derive = { git = "https://github.com/n0-computer/irpc.git", branch = "main" }

Cargo.toml

@@ -12,11 +12,11 @@
 //!
 //! You can also [`connect`](Store::connect) to a remote store that is listening
 //! to rpc requests.
-use std::{io, net::SocketAddr, ops::Deref, sync::Arc};
+use std::{io, net::SocketAddr, ops::Deref};
 
 use bao_tree::io::EncodeError;
 use iroh::Endpoint;
-use irpc::rpc::{listen, Handler};
+use irpc::rpc::{listen, RemoteService};
 use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
 use proto::{Request, ShutdownRequest, SyncDbRequest};
@@ -32,7 +32,7 @@ pub mod remote;
 pub mod tags;
 pub use crate::{store::util::Tag, util::temp_tag::TempTag};
 
-pub(crate) type ApiClient = irpc::Client<proto::Command, proto::Request, proto::StoreService>;
+pub(crate) type ApiClient = irpc::Client<proto::Request>;
 
 #[common_fields({
     backtrace: Option<Backtrace>,
@@ -281,42 +281,8 @@ impl Store {
 
     /// Listen on a quinn endpoint for incoming rpc connections.
     pub async fn listen(self, endpoint: quinn::Endpoint) {
-        let local = self.client.local().unwrap().clone();
-        let handler: Handler<Request> = Arc::new(move |req, rx, tx| {
-            let local = local.clone();
-            Box::pin({
-                match req {
-                    Request::SetTag(msg) => local.send((msg, tx)),
-                    Request::CreateTag(msg) => local.send((msg, tx)),
-                    Request::DeleteTags(msg) => local.send((msg, tx)),
-                    Request::RenameTag(msg) => local.send((msg, tx)),
-                    Request::ListTags(msg) => local.send((msg, tx)),
-
-                    Request::ListTempTags(msg) => local.send((msg, tx)),
-                    Request::CreateTempTag(msg) => local.send((msg, tx)),
-
-                    Request::BlobStatus(msg) => local.send((msg, tx)),
-
-                    Request::ImportBytes(msg) => local.send((msg, tx)),
-                    Request::ImportByteStream(msg) => local.send((msg, tx, rx)),
-                    Request::ImportBao(msg) => local.send((msg, tx, rx)),
-                    Request::ImportPath(msg) => local.send((msg, tx)),
-                    Request::ListBlobs(msg) => local.send((msg, tx)),
-                    Request::DeleteBlobs(msg) => local.send((msg, tx)),
-                    Request::Batch(msg) => local.send((msg, tx, rx)),
-
-                    Request::ExportBao(msg) => local.send((msg, tx)),
-                    Request::ExportRanges(msg) => local.send((msg, tx)),
-                    Request::ExportPath(msg) => local.send((msg, tx)),
-
-                    Request::Observe(msg) => local.send((msg, tx)),
-
-                    Request::ClearProtected(msg) => local.send((msg, tx)),
-                    Request::SyncDb(msg) => local.send((msg, tx)),
-                    Request::Shutdown(msg) => local.send((msg, tx)),
-                }
-            })
-        });
+        let local = self.client.as_local().unwrap().clone();
+        let handler = Request::remote_handler(local);
         listen::<Request>(endpoint, handler).await
     }
 

src/api.rs

@@ -29,15 +29,10 @@ use crate::{
 
 #[derive(Debug, Clone)]
 pub struct Downloader {
-    client: irpc::Client<SwarmMsg, SwarmProtocol, DownloaderService>,
+    client: irpc::Client<SwarmProtocol>,
 }
 
-#[derive(Debug, Clone)]
-pub struct DownloaderService;
-
-impl irpc::Service for DownloaderService {}
-
-#[rpc_requests(DownloaderService, message = SwarmMsg, alias = "Msg")]
+#[rpc_requests(message = SwarmMsg, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 enum SwarmProtocol {
     #[rpc(tx = mpsc::Sender<DownloadProgessItem>)]

src/api/downloader.rs

@@ -86,11 +86,7 @@ impl HashSpecific for CreateTagMsg {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct StoreService;
-impl irpc::Service for StoreService {}
-
-#[rpc_requests(StoreService, message = Command, alias = "Msg")]
+#[rpc_requests(message = Command, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 pub enum Request {
     #[rpc(tx = mpsc::Sender<super::Result<Hash>>)]

src/api/proto.rs

@@ -1259,7 +1259,7 @@ impl AsRef<Store> for FsStore {
 
 impl FsStore {
     fn new(
-        sender: irpc::LocalSender<proto::Command, proto::StoreService>,
+        sender: irpc::LocalSender<proto::Request>,
         db: tokio::sync::mpsc::Sender<InternalCommand>,
     ) -> Self {
         Self {

src/store/fs.rs

@@ -39,8 +39,7 @@ use crate::{
         blobs::{AddProgressItem, ImportMode},
         proto::{
             HashSpecific, ImportByteStreamMsg, ImportByteStreamRequest, ImportByteStreamUpdate,
-            ImportBytesMsg, ImportBytesRequest, ImportPathMsg, ImportPathRequest, Scope,
-            StoreService,
+            ImportBytesMsg, ImportBytesRequest, ImportPathMsg, ImportPathRequest, Request, Scope,
         },
     },
     store::{
@@ -136,12 +135,12 @@ impl std::fmt::Debug for ImportEntry {
     }
 }
 
-impl Channels<StoreService> for ImportEntry {
+impl Channels<Request> for ImportEntry {
     type Tx = mpsc::Sender<AddProgressItem>;
     type Rx = NoReceiver;
 }
 
-pub type ImportEntryMsg = WithChannels<ImportEntry, StoreService>;
+pub type ImportEntryMsg = WithChannels<ImportEntry, Request>;
 
 impl HashSpecific for ImportEntryMsg {
     fn hash(&self) -> Hash {