Merge pull request #14 from n0-computer/Frando/gossip

feat: add support for iroh-gossip

Author: b5

Cargo.lock

@@ -28,6 +28,7 @@ tracing = "0.1.41"
 rand = "0.8"
 derive_more = { version = "2.0.1", features = ["from"] }
 strum = { version = "0.27.1", features = ["derive"] }
+iroh-gossip = { version = "0.34.1", default-features = false }
 
 [features]
 bin = ["iroh-blobs/rpc"]

Cargo.toml

@@ -6,6 +6,7 @@ use iroh::{
     Endpoint, NodeAddr, NodeId,
 };
 use iroh_blobs::{ticket::BlobTicket, BlobFormat, Hash};
+use iroh_gossip::proto::TopicId;
 use n0_future::{task::AbortOnDropHandle, SinkExt, StreamExt};
 use rand::Rng;
 use rcan::Rcan;
@@ -192,6 +193,36 @@ impl Client {
         let res = r.await??;
         Ok(res)
     }
+
+    /// Create a gossip topic.
+    pub async fn put_gossip_topic(
+        &mut self,
+        topic: TopicId,
+        label: String,
+        bootstrap: Vec<NodeId>,
+    ) -> Result<()> {
+        let (s, r) = oneshot::channel();
+        self.sender
+            .send(ActorMessage::PutTopic {
+                topic,
+                label,
+                bootstrap,
+                s,
+            })
+            .await?;
+        let res = r.await??;
+        Ok(res)
+    }
+
+    /// Delete a gossip topic.
+    pub async fn delete_gossip_topic(&mut self, topic: TopicId) -> Result<()> {
+        let (s, r) = oneshot::channel();
+        self.sender
+            .send(ActorMessage::DeleteTopic { topic, s })
+            .await?;
+        let res = r.await??;
+        Ok(res)
+    }
 }
 
 struct Actor {
@@ -237,6 +268,16 @@ enum ActorMessage {
         name: String,
         s: oneshot::Sender<anyhow::Result<Hash>>,
     },
+    PutTopic {
+        topic: iroh_gossip::proto::TopicId,
+        label: String,
+        bootstrap: Vec<NodeId>,
+        s: oneshot::Sender<anyhow::Result<()>>,
+    },
+    DeleteTopic {
+        topic: iroh_gossip::proto::TopicId,
+        s: oneshot::Sender<anyhow::Result<()>>,
+    },
 }
 
 impl Actor {
@@ -378,6 +419,61 @@ impl Actor {
                 };
                 s.send(response).ok();
             }
+            ActorMessage::PutTopic {
+                topic,
+                label,
+                bootstrap,
+                s,
+            } => {
+                if let Err(err) = self
+                    .writer
+                    .send(ServerMessage::PutTopic {
+                        topic: *topic.as_bytes(),
+                        label,
+                        bootstrap,
+                    })
+                    .await
+                {
+                    s.send(Err(err.into())).ok();
+                    return;
+                }
+                let response = match self.reader.next().await {
+                    Some(Ok(msg)) => match msg {
+                        ClientMessage::PutTopicResponse(None) => Ok(()),
+                        ClientMessage::PutTopicResponse(Some(err)) => {
+                            Err(anyhow!("put topic failed: {}", err))
+                        }
+                        _ => Err(anyhow!("unexpected message from server: {:?}", msg)),
+                    },
+                    Some(Err(err)) => Err(anyhow!("failed to receive response: {:?}", err)),
+                    None => Err(anyhow!("connection closed")),
+                };
+                s.send(response).ok();
+            }
+            ActorMessage::DeleteTopic { topic, s } => {
+                if let Err(err) = self
+                    .writer
+                    .send(ServerMessage::DeleteTopic {
+                        topic: *topic.as_bytes(),
+                    })
+                    .await
+                {
+                    s.send(Err(err.into())).ok();
+                    return;
+                }
+                let response = match self.reader.next().await {
+                    Some(Ok(msg)) => match msg {
+                        ClientMessage::DeleteTopicResponse(None) => Ok(()),
+                        ClientMessage::DeleteTopicResponse(Some(err)) => {
+                            Err(anyhow!("delete topic failed: {}", err))
+                        }
+                        _ => Err(anyhow!("unexpected message from server: {:?}", msg)),
+                    },
+                    Some(Err(err)) => Err(anyhow!("failed to receive response: {:?}", err)),
+                    None => Err(anyhow!("connection closed")),
+                };
+                s.send(response).ok();
+            }
         }
     }
 

src/client.rs

@@ -1,3 +1,4 @@
+use iroh::NodeId;
 use iroh_blobs::{ticket::BlobTicket, Hash};
 use rcan::Rcan;
 use serde::{Deserialize, Serialize};
@@ -7,6 +8,8 @@ use crate::caps::Caps;
 
 pub const ALPN: &[u8] = b"/iroh/n0des/1";
 
+pub type ProtoTopicId = [u8; 32];
+
 /// Messages sent from the client to the server
 #[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -15,6 +18,14 @@ pub enum ServerMessage {
     Auth(Rcan<Caps>),
     /// Request that the node fetches the given blob.
     PutBlob { ticket: BlobTicket, name: String },
+    /// Request that the node joins the given tossip topic
+    PutTopic {
+        topic: ProtoTopicId,
+        label: String,
+        bootstrap: Vec<NodeId>,
+    },
+    /// Request that the node joins the given tossip topic
+    DeleteTopic { topic: ProtoTopicId },
     /// Request the name of a blob held by the node
     GetTag { name: String },
     /// Request to store the given metrics data
@@ -33,6 +44,10 @@ pub enum ClientMessage {
     AuthResponse(Option<String>),
     /// If set, this means it was an error.
     PutBlobResponse(Option<String>),
+    /// If set, this means it was an error.
+    PutTopicResponse(Option<String>),
+    /// If set, this means it was an error.
+    DeleteTopicResponse(Option<String>),
     /// Simple pong response
     Pong {
         req: [u8; 32],