refactor(iroh)!: Remove gossip rpc types (#2834)

rklaehn · web-flow · commit a55529b52e19 · 2024-10-30T10:56:03.000Z
## Description Remove gossip rpc types. They live in iroh-gossip now under a feature flag. ## Breaking Changes  ## Notes & open questions note: updates to quic-rpc 0.13 ## Change checklist - [ ] Self-review. - [ ] Documentation updates following the [style guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text), if relevant. - [ ] Tests if relevant. - [ ] All breaking changes documented.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/iroh-cli/src/commands/gossip.rs b/iroh-cli/src/commands/gossip.rs
@@ -7,10 +7,8 @@ use bao_tree::blake3;
 use clap::{ArgGroup, Subcommand};
 use futures_lite::StreamExt;
 use futures_util::SinkExt;
-use iroh::{
-    client::{gossip::SubscribeOpts, Iroh},
-    net::NodeId,
-};
+use iroh::{client::Iroh, net::NodeId};
+use iroh_gossip::rpc::client::SubscribeOpts;
 use tokio::io::AsyncBufReadExt;
 
 /// Commands to manage gossiping.
diff --git a/iroh/Cargo.toml b/iroh/Cargo.toml
@@ -39,8 +39,8 @@ iroh-docs = { version = "0.27.0" }
 iroh-gossip = "0.27.0"
 parking_lot = "0.12.1"
 postcard = { version = "1", default-features = false, features = ["alloc", "use-std", "experimental-derive"] }
-quic-rpc = { version = "0.12", default-features = false, features = ["flume-transport", "quinn-transport"] }
-quic-rpc-derive = { version = "0.12" }
+quic-rpc = { version = "0.13", default-features = false, features = ["flume-transport", "quinn-transport"] }
+quic-rpc-derive = { version = "0.13" }
 quinn = { package = "iroh-quinn", version = "0.11" }
 rand = "0.8"
 serde = { version = "1", features = ["derive"] }
diff --git a/iroh/src/client.rs b/iroh/src/client.rs
@@ -6,6 +6,7 @@ use std::collections::BTreeMap;
 
 use anyhow::Result;
 use futures_lite::{Stream, StreamExt};
+use quic_rpc::client::BoxedServiceConnection;
 use ref_cast::RefCast;
 
 use crate::rpc_protocol::node::{CounterStats, ShutdownRequest, StatsRequest, StatusRequest};
@@ -20,18 +21,16 @@ pub use self::{docs::Doc, net::NodeStatus};
 pub mod authors;
 pub mod blobs;
 pub mod docs;
-pub mod gossip;
 pub mod net;
 pub mod tags;
 
 /// Iroh rpc connection - boxed so that we can have a concrete type.
-pub(crate) type RpcConnection = quic_rpc::transport::boxed::Connection<RpcService>;
+pub(crate) type RpcConnection = BoxedServiceConnection<RpcService>;
 
 // Keep this type exposed, otherwise every occurrence of `RpcClient` in the API
 // will show up as `RpcClient<RpcService, Connection<RpcService>>` in the docs.
 /// Iroh rpc client - boxed so that we can have a concrete type.
-pub type RpcClient =
-    quic_rpc::RpcClient<RpcService, quic_rpc::transport::boxed::Connection<RpcService>>;
+pub type RpcClient = quic_rpc::RpcClient<RpcService>;
 
 /// An iroh client.
 ///
@@ -81,8 +80,9 @@ impl Iroh {
     }
 
     /// Returns the gossip client.
-    pub fn gossip(&self) -> &gossip::Client {
-        gossip::Client::ref_cast(&self.rpc)
+    pub fn gossip(&self) -> iroh_gossip::RpcClient<RpcService> {
+        let channel = self.rpc.clone().map::<iroh_gossip::RpcService>();
+        iroh_gossip::RpcClient::new(channel)
     }
 
     /// Returns the net client.
diff --git a/iroh/src/client/gossip.rs b/iroh/src/client/gossip.rs
diff --git a/iroh/src/client/quic.rs b/iroh/src/client/quic.rs
@@ -11,10 +11,7 @@ use anyhow::{bail, Context};
 use quic_rpc::transport::{boxed::Connection as BoxedConnection, quinn::QuinnConnection};
 
 use super::{Iroh, RpcClient};
-use crate::{
-    node::RpcStatus,
-    rpc_protocol::{node::StatusRequest, RpcService},
-};
+use crate::{node::RpcStatus, rpc_protocol::node::StatusRequest};
 
 /// ALPN used by irohs RPC mechanism.
 // TODO: Change to "/iroh-rpc/1"
@@ -46,7 +43,7 @@ pub(crate) async fn connect_raw(addr: SocketAddr) -> anyhow::Result<RpcClient> {
     let endpoint = create_quinn_client(bind_addr, vec![RPC_ALPN.to_vec()], false)?;
 
     let server_name = "localhost".to_string();
-    let connection = QuinnConnection::<RpcService>::new(endpoint, addr, server_name);
+    let connection = QuinnConnection::new(endpoint, addr, server_name);
     let connection = BoxedConnection::new(connection);
     let client = RpcClient::new(connection);
     // Do a status request to check if the server is running.
diff --git a/iroh/src/lib.rs b/iroh/src/lib.rs
@@ -49,7 +49,7 @@
 //!   manage and share content-addressed blobs of data
 //! - [tags](crate::client::tags):
 //!   tags to tell iroh what data is important
-//! - [gossip](crate::client::gossip):
+//! - [gossip](iroh_gossip::RpcClient):
 //!   exchange data with other nodes via a gossip protocol
 //!
 //! - [authors](crate::client::authors):
diff --git a/iroh/src/node/builder.rs b/iroh/src/node/builder.rs
@@ -691,7 +691,8 @@ where
         .await?;
 
         // Initialize the internal RPC connection.
-        let (internal_rpc, controller) = quic_rpc::transport::flume::connection::<RpcService>(32);
+        let (internal_rpc, controller) =
+            quic_rpc::transport::flume::service_connection::<RpcService>(32);
         let internal_rpc = quic_rpc::transport::boxed::ServerEndpoint::new(internal_rpc);
         // box the controller. Boxing has a special case for the flume channel that avoids allocations,
         // so this has zero overhead.
@@ -951,7 +952,10 @@ pub const DEFAULT_RPC_ADDR: SocketAddr =
 fn make_rpc_endpoint(
     secret_key: &SecretKey,
     mut rpc_addr: SocketAddr,
-) -> Result<(QuinnServerEndpoint<RpcService>, u16)> {
+) -> Result<(
+    QuinnServerEndpoint<crate::rpc_protocol::Request, crate::rpc_protocol::Response>,
+    u16,
+)> {
     let mut transport_config = quinn::TransportConfig::default();
     transport_config
         .max_concurrent_bidi_streams(MAX_RPC_STREAMS.into())
@@ -982,7 +986,7 @@ fn make_rpc_endpoint(
     };
 
     let actual_rpc_port = rpc_quinn_endpoint.local_addr()?.port();
-    let rpc_endpoint = QuinnServerEndpoint::<RpcService>::new(rpc_quinn_endpoint)?;
+    let rpc_endpoint = QuinnServerEndpoint::new(rpc_quinn_endpoint)?;
 
     Ok((rpc_endpoint, actual_rpc_port))
 }
diff --git a/iroh/src/node/rpc.rs b/iroh/src/node/rpc.rs
@@ -60,7 +60,6 @@ use crate::{
             ExportFileRequest, ExportFileResponse, ImportFileRequest, ImportFileResponse,
             Request as DocsRequest, SetHashRequest,
         },
-        gossip,
         net::{
             self, AddAddrRequest, AddrRequest, IdRequest, NodeWatchRequest, RelayRequest,
             RemoteInfoRequest, RemoteInfoResponse, RemoteInfosIterRequest, RemoteInfosIterResponse,
@@ -245,31 +244,15 @@ impl<D: BaoStore> Handler<D> {
 
     async fn handle_gossip_request(
         self,
-        msg: gossip::Request,
+        msg: iroh_gossip::RpcRequest,
         chan: RpcChannel<RpcService, IrohServerEndpoint>,
     ) -> Result<(), RpcServerError<IrohServerEndpoint>> {
-        use gossip::Request::*;
-        match msg {
-            Subscribe(msg) => {
-                chan.bidi_streaming(msg, self, |handler, req, updates| {
-                    let stream = handler
-                        .router
-                        .get_protocol::<Gossip>(GOSSIP_ALPN)
-                        .expect("missing gossip")
-                        .join_with_stream(
-                            req.topic,
-                            iroh_gossip::net::JoinOptions {
-                                bootstrap: req.bootstrap,
-                                subscription_capacity: req.subscription_capacity,
-                            },
-                            Box::pin(updates),
-                        );
-                    futures_util::TryStreamExt::map_err(stream, |e| RpcError::new(&*e))
-                })
-                .await
-            }
-            Update(_msg) => Err(RpcServerError::UnexpectedUpdateMessage),
-        }
+        let gossip = self
+            .router
+            .get_protocol::<Gossip>(GOSSIP_ALPN)
+            .expect("missing gossip");
+        let chan = chan.map::<iroh_gossip::RpcService>();
+        gossip.handle_rpc_request(msg, chan).await
     }
 
     async fn handle_authors_request(
diff --git a/iroh/src/rpc_protocol.rs b/iroh/src/rpc_protocol.rs
@@ -20,7 +20,6 @@ use serde::{Deserialize, Serialize};
 pub mod authors;
 pub mod blobs;
 pub mod docs;
-pub mod gossip;
 pub mod net;
 pub mod node;
 pub mod tags;
@@ -40,7 +39,7 @@ pub enum Request {
     Docs(docs::Request),
     Tags(tags::Request),
     Authors(authors::Request),
-    Gossip(gossip::Request),
+    Gossip(iroh_gossip::RpcRequest),
 }
 
 /// The response enum, listing all possible responses.
@@ -54,7 +53,7 @@ pub enum Response {
     Tags(tags::Response),
     Docs(docs::Response),
     Authors(authors::Response),
-    Gossip(gossip::Response),
+    Gossip(iroh_gossip::RpcResponse),
 }
 
 impl quic_rpc::Service for RpcService {
diff --git a/iroh/src/rpc_protocol/gossip.rs b/iroh/src/rpc_protocol/gossip.rs
