fixup

Author: Frando

Cargo.lock

@@ -48,7 +48,7 @@ ed25519-dalek = { version = "2.0.0", features = ["serde", "rand_core"] }
 hex = "0.4.3"
 indexmap = "2.0"
 iroh-metrics = { version = "0.32", default-features = false }
-n0-future = "0.1.2"
+iroh-base = { version = "0.34", default-features = false, features = ["key"] }
 postcard = { version = "1", default-features = false, features = [
     "alloc",
     "use-std",
@@ -75,11 +75,7 @@ irpc = { git = "https://github.com/n0-computer/irpc", optional = true, default-f
     "message_spans",
 ] }
 irpc-derive = { git = "https://github.com/n0-computer/irpc", optional = true }
-# irpc = { path = "../irpc", optional = true, default-features = false, features = [
-#     "stream",
-#     "message_spans",
-# ] }
-# irpc-derive = { path = "../irpc/irpc-derive", optional = true }
+n0-future = { version = "0.1.2", optional = true }
 
 # rpc dependencies
 quinn = { package = "iroh-quinn", version = "0.13.0", optional = true }
@@ -130,12 +126,22 @@ net = [
     "dep:tokio",
     "dep:tokio-util",
     "dep:async-channel",
+    "dep:n0-future",
     "dep:futures-util",
     "dep:futures-concurrency",
     "dep:anyhow",
     "dep:thiserror",
 ]
-rpc = ["net", "dep:quinn", "irpc/rpc", "irpc/quinn_endpoint_setup"]
+rpc = [
+    "dep:irpc",
+    "dep:irpc-derive",
+    "dep:tokio",
+    "dep:quinn",
+    "dep:n0-future",
+    "dep:thiserror",
+    "irpc/rpc",
+    "irpc/quinn_endpoint_setup",
+]
 test-utils = ["dep:rand_chacha", "dep:humantime-serde"]
 simulator = [
     "test-utils",

Cargo.toml

@@ -9,11 +9,10 @@ use std::{
 };
 
 use bytes::Bytes;
-use futures_lite::{Stream, StreamExt};
-use iroh::NodeId;
-use irpc::{channel::spsc, Client, LocalSender};
+use iroh_base::NodeId;
+use irpc::{channel::spsc, Client};
 use irpc_derive::rpc_requests;
-use n0_future::TryStreamExt;
+use n0_future::{Stream, StreamExt, TryStreamExt};
 use serde::{Deserialize, Serialize};
 
 use crate::proto::{DeliveryScope, TopicId};
@@ -42,13 +41,13 @@ pub(crate) struct JoinRequest {
     pub bootstrap: BTreeSet<NodeId>,
 }
 
-///
+/// Errors returned from methods in [`GossipApi`]
 #[derive(Debug, thiserror::Error)]
 pub enum ApiError {
     /// RPC error
     #[error(transparent)]
     Rpc(#[from] irpc::Error),
-    ///
+    /// Received an unexpected event.
     #[error("received unexpected event")]
     UnexpectedEvent,
 }
@@ -72,8 +71,9 @@ pub struct GossipApi {
 }
 
 impl GossipApi {
+    #[cfg(feature = "net")]
     pub(crate) fn local(tx: tokio::sync::mpsc::Sender<RpcMessage>) -> Self {
-        let local = LocalSender::<RpcMessage, Service>::from(tx);
+        let local = irpc::LocalSender::<RpcMessage, Service>::from(tx);
         Self {
             client: local.into(),
         }
@@ -87,7 +87,7 @@ impl GossipApi {
     }
 
     /// Listen on a quinn endpoint for incoming RPC connections.
-    #[cfg(feature = "rpc")]
+    #[cfg(all(feature = "rpc", feature = "net"))]
     pub(crate) async fn listen(&self, endpoint: quinn::Endpoint) {
         use std::sync::Arc;
 

src/api.rs

@@ -13,6 +13,7 @@
 #[doc(inline)]
 pub use net::GOSSIP_ALPN as ALPN;
 
+#[cfg(any(feature = "net", feature = "rpc"))]
 pub mod api;
 pub mod metrics;
 #[cfg(feature = "net")]

src/lib.rs

@@ -850,7 +850,7 @@ struct TopicState {
     joined: bool,
     neighbors: BTreeSet<NodeId>,
     event_sender: broadcast::Sender<GossipEvent>,
-    /// Keys identifying [`GossipSender`]s.
+    /// Keys identifying command receivers in [`Actor::command_rx`].
     ///
     /// This represents the receiver side of gossip's publish public API.
     command_rx_keys: HashSet<stream_group::Key>,
@@ -982,7 +982,7 @@ async fn topic_subscriber_loop(
                    Err(broadcast::error::RecvError::Lagged(_)) => Event::Lagged,
                    Ok(event) => Event::Gossip(event)
                };
-               if let Err(_) = sender.send(event).await {
+               if sender.send(event).await.is_err() {
                    break;
                }
            }