From 21f4e7e5657cea6ad2e2d6f10c23fb742f306763 Mon Sep 17 00:00:00 2001 From: lla-dane Date: Mon, 25 May 2026 22:42:46 +0530 Subject: [PATCH 1/3] integrate global_event_tx with Floodsub API --- crates/node/src/inner.rs | 1 + crates/node/src/node.rs | 9 ++++--- crates/protocols/floodsub/README.md | 5 +++- crates/protocols/floodsub/src/pubsub.rs | 24 +++++++++++++------ crates/protocols/floodsub/src/subscription.rs | 21 +++++++++++----- examples/floodsub/src/main.rs | 20 ++++++++++++++++ 6 files changed, 63 insertions(+), 17 deletions(-) diff --git a/crates/node/src/inner.rs b/crates/node/src/inner.rs index 79ef374..4697ba9 100644 --- a/crates/node/src/inner.rs +++ b/crates/node/src/inner.rs @@ -106,6 +106,7 @@ impl NodeInner { keypair.clone(), handlers.clone(), local_peer_info.clone(), + global_event_tx.clone(), )); info!("Node listening on: {}", listen_addr.to_string()); diff --git a/crates/node/src/node.rs b/crates/node/src/node.rs index 845903f..ad6b68e 100644 --- a/crates/node/src/node.rs +++ b/crates/node/src/node.rs @@ -32,6 +32,7 @@ pub struct Node { pub key_pair: RsaKeyPair, pub mpsc_tx: Sender>, pub handlers: Arc>>, + pub global_event_tx: Sender>, // Protocols pub floodsub: Arc>>>, @@ -44,12 +45,14 @@ impl Node { key_pair: RsaKeyPair, handlers: Arc>>, local_peer_info: PeerInfo, + global_event_tx: Sender>, ) -> Self { Node { local_peer_info, key_pair, mpsc_tx, handlers, + global_event_tx, floodsub: Arc::new(Mutex::new(None)), ping: Arc::new(Mutex::new(None)), @@ -120,7 +123,9 @@ impl Node { warn!("Floodsub already running!!") } None => { - let floodsub = FloodSub::new(local_peer).await.unwrap(); + let floodsub = FloodSub::new(local_peer, self.global_event_tx.clone()) + .await + .unwrap(); self.set_stream_handler(FLOODSUB, Box::new(floodsub.clone())) .await .unwrap(); @@ -189,7 +194,6 @@ impl INodeFloodsubAPI for Node { match self.floodsub_is_running().await { false => warn!("Floodsub not running!!"), true => { - warn!("FLOODSUB publishing"); let floodsub_guard = self.floodsub.lock().await; let floodsub = floodsub_guard.as_ref().unwrap(); @@ -201,7 +205,6 @@ impl INodeFloodsubAPI for Node { ); floodsub.floodsub_mpsc_tx.send(frame).await.unwrap(); - warn!("FLOODSUB published"); } } diff --git a/crates/protocols/floodsub/README.md b/crates/protocols/floodsub/README.md index 1a29afa..0669c4d 100644 --- a/crates/protocols/floodsub/README.md +++ b/crates/protocols/floodsub/README.md @@ -76,4 +76,7 @@ This FloodSub implementation is suitable for: - small trusted networks - prototyping higher-level pubsub APIs -It is not suitable for adversarial or large-scale production environments without significant extension. \ No newline at end of file +It is not suitable for adversarial or large-scale production environments without significant extension. + + +global_event_tx: UNSUBSCRIBE + SUBSCRIBE + RECV_MSG \ No newline at end of file diff --git a/crates/protocols/floodsub/src/pubsub.rs b/crates/protocols/floodsub/src/pubsub.rs index 8f29d89..d7fbbd6 100644 --- a/crates/protocols/floodsub/src/pubsub.rs +++ b/crates/protocols/floodsub/src/pubsub.rs @@ -30,7 +30,7 @@ pub type FloodsubPayload = (Option, Option>, Option> pub struct FloosubStore { peer_topics: HashMap>, peers: HashMap>>, - subscribed_topic_api: HashMap>>, + subscribed_topic_api: HashMap, Vec)>>, } #[derive(Debug, Clone, Hash, PartialEq, Eq)] @@ -45,6 +45,7 @@ pub struct FloodSub { pub floodsub_mpsc_tx: Sender>, last_seen_cache: Arc>, pub floodsub_store: Arc>, + pub global_event_tx: Sender>, } #[async_trait] @@ -102,7 +103,7 @@ impl IProtocolHandler for FloodSub { } impl FloodSub { - pub async fn new(local_peer: &PeerInfo) -> Result> { + pub async fn new(local_peer: &PeerInfo, global_event_tx: Sender>) -> Result> { let (floodsub_mpsc_tx, floodsub_mpsc_rx) = mpsc::channel::>(300); let last_seen_cache = Arc::new(Mutex::new(HashMap::new())); let local_peer_info = local_peer.clone(); @@ -116,6 +117,7 @@ impl FloodSub { peers: HashMap::new(), subscribed_topic_api: HashMap::new(), })), + global_event_tx, }); tokio::spawn(async move { @@ -161,7 +163,6 @@ impl FloodSub { topic_ids: opt_vec_pld.unwrap(), }; - warn!("About to publish: floodsub"); self.publish(self.local_peer_info.clone().peer_id, pub_msg).await.unwrap(); } } @@ -219,13 +220,18 @@ impl FloodSub { return Ok(()); } - debug!("Subscribing to topic: {}", topic_id); + let subscribe_msg = format!("Subscribing to topic: {topic_id}"); + self.global_event_tx + .send(subscribe_msg.as_bytes().to_vec()) + .await + .unwrap(); - let (topic_mpsc_tx, topic_mpsc_rx) = mpsc::channel::>(100); + let (topic_mpsc_tx, topic_mpsc_rx) = mpsc::channel::<(Vec, Vec)>(100); let mut sub_api = SubscriptionAPI::new( topic_id.clone(), self.floodsub_mpsc_tx.clone(), topic_mpsc_rx, + self.global_event_tx.clone(), ); { @@ -273,7 +279,11 @@ impl FloodSub { return Ok(()); } - debug!("Unsubscribing from topic: {}", topic_id); + let unsubscribe_msg = format!("Unsubscribing from topic: {topic_id}"); + self.global_event_tx + .send(unsubscribe_msg.as_bytes().to_vec()) + .await + .unwrap(); { let mut store = self.floodsub_store.lock().await; @@ -355,7 +365,7 @@ impl FloodSub { }; topic_mpsc_tx - .send(pubsub_msg.data().to_vec()) + .send((pubsub_msg.data().to_vec(), pubsub_msg.from().to_vec())) .await .unwrap(); diff --git a/crates/protocols/floodsub/src/subscription.rs b/crates/protocols/floodsub/src/subscription.rs index f4244af..2018a68 100644 --- a/crates/protocols/floodsub/src/subscription.rs +++ b/crates/protocols/floodsub/src/subscription.rs @@ -8,19 +8,22 @@ use crate::pubsub::FloodsubPayload; pub struct SubscriptionAPI { pub topic_id: String, floodsub_mpsc_tx: Sender>, - topic_mpsc_rx: Receiver>, + topic_mpsc_rx: Receiver<(Vec, Vec)>, + global_event_tx: Sender>, } impl SubscriptionAPI { pub fn new( topic_id: String, floodsub_mpsc_tx: Sender>, - topic_mpsc_rx: Receiver>, + topic_mpsc_rx: Receiver<(Vec, Vec)>, + global_event_tx: Sender>, ) -> Self { SubscriptionAPI { topic_id, floodsub_mpsc_tx, topic_mpsc_rx, + global_event_tx, } } @@ -48,20 +51,26 @@ impl SubscriptionAPI { Ok(()) } - pub async fn recv(&mut self) -> Option> { + pub async fn recv(&mut self) -> Option<(Vec, Vec)> { self.topic_mpsc_rx.recv().await } pub async fn receive_loop(&mut self) { let topic = self.topic_id.clone(); - debug!("[{}]: Receiver loop started", topic); // TODO: send global event from here loop { match self.recv().await { - Some(payload) => { + Some((payload, source_peer)) => { let msg = String::from_utf8_lossy(&payload).to_string(); - println!("[{}]: {}", topic, msg); + let source_peer_id = String::from_utf8_lossy(&source_peer).to_string(); + let fsub_msg = format!("[{topic}] - [{source_peer_id}]: {msg}"); + + self.global_event_tx + .send(fsub_msg.as_bytes().to_vec()) + .await + .unwrap(); + } None => { debug!("[{}]: Receiver loop ended", topic); diff --git a/examples/floodsub/src/main.rs b/examples/floodsub/src/main.rs index f794e9e..12c24bb 100644 --- a/examples/floodsub/src/main.rs +++ b/examples/floodsub/src/main.rs @@ -1,8 +1,11 @@ mod cli; +use std::time::Duration; + use anyhow::Result; use identity::multiaddr::Multiaddr; use node::{inner::NodeInner, protocol::InnerProtocolOpt}; +use tokio::sync::mpsc::Receiver; use tracing::info; use tracing_subscriber::EnvFilter; @@ -27,8 +30,25 @@ async fn main() -> Result<()> { .await .unwrap(); + tokio::spawn(async move { + global_notification_receiver(_global_rx).await.unwrap(); + }); + + tokio::time::sleep(Duration::from_millis(200)).await; + info!("Run in new terminal: \ncargo run --bin floodsub --release"); cli_loop(host_mpsc_tx).await.unwrap(); Ok(()) } + +async fn global_notification_receiver(mut global_event_rx: Receiver>) -> Result<()> { + info!("Global notification receiver initiated"); + + loop { + let notification = global_event_rx.recv().await.unwrap(); + let msg = String::from_utf8_lossy(¬ification).to_string(); + + info!("{}", msg); + } +} From f811dfd96062dd76d3b442c05029a2f919793f8e Mon Sep 17 00:00:00 2001 From: lla-dane Date: Tue, 26 May 2026 00:20:01 +0530 Subject: [PATCH 2/3] create global_event + integrate with Ping API --- crates/identity/src/events.rs | 29 ++++++++++++ crates/identity/src/lib.rs | 1 + crates/node/src/node.rs | 2 +- crates/protocols/floodsub/Cargo.toml | 1 + crates/protocols/floodsub/src/pubsub.rs | 29 ++++++++++-- crates/protocols/floodsub/src/subscription.rs | 17 +++++-- crates/protocols/ping/src/handler.rs | 47 ++++++++++++------- examples/floodsub/src/main.rs | 23 +++++++-- 8 files changed, 119 insertions(+), 30 deletions(-) create mode 100644 crates/identity/src/events.rs diff --git a/crates/identity/src/events.rs b/crates/identity/src/events.rs new file mode 100644 index 0000000..3ed20e2 --- /dev/null +++ b/crates/identity/src/events.rs @@ -0,0 +1,29 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct FloodsubEvent { + pub msg_type: FloodsubMsgType, + pub source: Option, + pub msg: Option>, + pub topic: String, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub enum FloodsubMsgType { + Publish, + Subscribe, + Unsubscribe, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct PingEvent { + pub remote: String, + pub rtts: Vec, + pub count: u128, +} + +#[derive(Debug, Serialize, Deserialize)] +pub enum GlobalEvent { + Floodsub(FloodsubEvent), + Ping(PingEvent), +} diff --git a/crates/identity/src/lib.rs b/crates/identity/src/lib.rs index e6985b6..7510080 100644 --- a/crates/identity/src/lib.rs +++ b/crates/identity/src/lib.rs @@ -2,3 +2,4 @@ pub mod keys; pub mod multiaddr; pub mod peer; pub mod traits; +pub mod events; \ No newline at end of file diff --git a/crates/node/src/node.rs b/crates/node/src/node.rs index ad6b68e..3cf6b23 100644 --- a/crates/node/src/node.rs +++ b/crates/node/src/node.rs @@ -144,7 +144,7 @@ impl Node { warn!("Ping already running!!"); } None => { - let ping = Arc::new(Ping::new(None)); + let ping = Arc::new(Ping::new(None, self.global_event_tx.clone())); self.set_stream_handler(PING, Box::new(ping.clone())) .await .unwrap(); diff --git a/crates/protocols/floodsub/Cargo.toml b/crates/protocols/floodsub/Cargo.toml index ee5919b..469931d 100644 --- a/crates/protocols/floodsub/Cargo.toml +++ b/crates/protocols/floodsub/Cargo.toml @@ -9,6 +9,7 @@ anyhow = { workspace = true } async-trait = { workspace = true } tracing = { workspace = true } prost = { workspace = true } +bincode = {workspace = true} schema = { path = "../../schema" } identity = { path = "../../identity" } diff --git a/crates/protocols/floodsub/src/pubsub.rs b/crates/protocols/floodsub/src/pubsub.rs index d7fbbd6..0f28bb2 100644 --- a/crates/protocols/floodsub/src/pubsub.rs +++ b/crates/protocols/floodsub/src/pubsub.rs @@ -5,7 +5,10 @@ use std::{ use anyhow::Result; use async_trait::async_trait; -use identity::traits::muxer::IMuxedStream; +use identity::{ + events::{FloodsubEvent, FloodsubMsgType, GlobalEvent}, + traits::muxer::IMuxedStream, +}; use identity::{peer::PeerInfo, traits::core::IProtocolHandler}; use prost::Message as ProstMessage; use schema::floodsub::{rpc::SubOpts, Message, Rpc}; @@ -220,11 +223,19 @@ impl FloodSub { return Ok(()); } - let subscribe_msg = format!("Subscribing to topic: {topic_id}"); + // Send out a global event, for subscribing to a new topic + let floosub_event = GlobalEvent::Floodsub(FloodsubEvent { + msg_type: FloodsubMsgType::Subscribe, + source: None, + msg: None, + topic: topic_id.clone(), + }); + self.global_event_tx - .send(subscribe_msg.as_bytes().to_vec()) + .send(bincode::serialize(&floosub_event).unwrap()) .await .unwrap(); + // --------------------------------------------------------- let (topic_mpsc_tx, topic_mpsc_rx) = mpsc::channel::<(Vec, Vec)>(100); let mut sub_api = SubscriptionAPI::new( @@ -279,11 +290,19 @@ impl FloodSub { return Ok(()); } - let unsubscribe_msg = format!("Unsubscribing from topic: {topic_id}"); + // Send out a global event, for unsubscribing to a new topic + let floosub_event = GlobalEvent::Floodsub(FloodsubEvent { + msg_type: FloodsubMsgType::Unsubscribe, + source: None, + msg: None, + topic: topic_id.clone(), + }); + self.global_event_tx - .send(unsubscribe_msg.as_bytes().to_vec()) + .send(bincode::serialize(&floosub_event).unwrap()) .await .unwrap(); + // --------------------------------------------------------- { let mut store = self.floodsub_store.lock().await; diff --git a/crates/protocols/floodsub/src/subscription.rs b/crates/protocols/floodsub/src/subscription.rs index 2018a68..d7446e2 100644 --- a/crates/protocols/floodsub/src/subscription.rs +++ b/crates/protocols/floodsub/src/subscription.rs @@ -1,4 +1,5 @@ use anyhow::Result; +use identity::events::{FloodsubEvent, FloodsubMsgType, GlobalEvent}; use tokio::sync::mpsc::{Receiver, Sender}; use tracing::debug; @@ -62,15 +63,21 @@ impl SubscriptionAPI { loop { match self.recv().await { Some((payload, source_peer)) => { - let msg = String::from_utf8_lossy(&payload).to_string(); + // let msg = String::from_utf8_lossy(&payload).to_string(); let source_peer_id = String::from_utf8_lossy(&source_peer).to_string(); - let fsub_msg = format!("[{topic}] - [{source_peer_id}]: {msg}"); - + // let fsub_msg = format!("[{topic}] - [{source_peer_id}]: {msg}"); + + let floosub_event = GlobalEvent::Floodsub(FloodsubEvent { + msg_type: FloodsubMsgType::Publish, + source: Some(source_peer_id), + msg: Some(payload), + topic: topic.clone(), + }); + self.global_event_tx - .send(fsub_msg.as_bytes().to_vec()) + .send(bincode::serialize(&floosub_event).unwrap()) .await .unwrap(); - } None => { debug!("[{}]: Receiver loop ended", topic); diff --git a/crates/protocols/ping/src/handler.rs b/crates/protocols/ping/src/handler.rs index 9236e97..bdce5e9 100644 --- a/crates/protocols/ping/src/handler.rs +++ b/crates/protocols/ping/src/handler.rs @@ -5,26 +5,28 @@ use std::{ use anyhow::{Error, Result}; use async_trait::async_trait; -use identity::traits::{core::IProtocolHandler, muxer::IMuxedStream}; -use tokio::{sync::Mutex, time::timeout}; +use identity::{ + events::{GlobalEvent, PingEvent}, + traits::{core::IProtocolHandler, muxer::IMuxedStream}, +}; +use tokio::{ + sync::{mpsc::Sender, Mutex}, + time::timeout, +}; use tracing::{debug, error, warn}; const PING_LENGTH: usize = 32; -pub struct PingEvent { - pub is_initiator: bool, - pub rtts: Vec, - pub remote_peer_id: String, -} - pub struct Ping { pub count: Arc>, + global_event_tx: Sender>, } impl Ping { - pub fn new(count: Option) -> Self { + pub fn new(count: Option, global_event_tx: Sender>) -> Self { Ping { count: Arc::new(Mutex::new(count.unwrap_or(5))), + global_event_tx, } } @@ -94,7 +96,16 @@ impl Ping { match self.ping(stream).await { Err(e) => error!("Error in ping sequence: {}, {}", e, peer_id), Ok(rtt) => { - debug!("Ping rtt exchange: RTT = [{}]μs", rtt) + let ping_event = GlobalEvent::Ping(PingEvent { + remote: stream.get_peer_id(), + rtts: vec![rtt], + count: 1, + }); + + self.global_event_tx + .send(bincode::serialize(&ping_event).unwrap()) + .await + .unwrap(); } } tokio::time::sleep(Duration::from_millis(1000)).await; @@ -144,12 +155,16 @@ impl Ping { } } - debug!( - "Ping completed with {} RTT samples from {}:\n{:?}μs", - rtts.len(), - stream.get_peer_id(), - rtts - ); + let ping_event = GlobalEvent::Ping(PingEvent { + remote: stream.get_peer_id(), + count: rtts.len() as u128, + rtts: rtts.clone(), + }); + + self.global_event_tx + .send(bincode::serialize(&ping_event).unwrap()) + .await + .unwrap(); } false => { let count_buf = stream.read().await.unwrap(); diff --git a/examples/floodsub/src/main.rs b/examples/floodsub/src/main.rs index 12c24bb..421c4ea 100644 --- a/examples/floodsub/src/main.rs +++ b/examples/floodsub/src/main.rs @@ -3,7 +3,10 @@ mod cli; use std::time::Duration; use anyhow::Result; -use identity::multiaddr::Multiaddr; +use identity::{ + events::{FloodsubMsgType, GlobalEvent}, + multiaddr::Multiaddr, +}; use node::{inner::NodeInner, protocol::InnerProtocolOpt}; use tokio::sync::mpsc::Receiver; use tracing::info; @@ -47,8 +50,22 @@ async fn global_notification_receiver(mut global_event_rx: Receiver>) -> loop { let notification = global_event_rx.recv().await.unwrap(); - let msg = String::from_utf8_lossy(¬ification).to_string(); + let decoded = bincode::deserialize::(¬ification).unwrap(); + match decoded { + GlobalEvent::Floodsub(event) => { + let original = event.clone(); - info!("{}", msg); + match event.msg_type { + FloodsubMsgType::Publish => { + let msg = String::from_utf8_lossy(&event.msg.unwrap()).to_string(); + let topic = event.topic; + let source = event.source.unwrap(); + info!("FloodsubEvent: {topic} - {source}: {msg}"); + } + _ => info!("{:?}", original), + } + } + GlobalEvent::Ping(event) => info!("{:?}", event), + } } } From f721fd32960a5cc2dc6bb197e8ef1aeb374bbbe6 Mon Sep 17 00:00:00 2001 From: lla-dane Date: Tue, 26 May 2026 00:22:06 +0530 Subject: [PATCH 3/3] fix linter errors --- crates/identity/src/lib.rs | 2 +- crates/protocols/floodsub/src/pubsub.rs | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/identity/src/lib.rs b/crates/identity/src/lib.rs index 7510080..6b6fe94 100644 --- a/crates/identity/src/lib.rs +++ b/crates/identity/src/lib.rs @@ -1,5 +1,5 @@ +pub mod events; pub mod keys; pub mod multiaddr; pub mod peer; pub mod traits; -pub mod events; \ No newline at end of file diff --git a/crates/protocols/floodsub/src/pubsub.rs b/crates/protocols/floodsub/src/pubsub.rs index 0f28bb2..d5cd5a3 100644 --- a/crates/protocols/floodsub/src/pubsub.rs +++ b/crates/protocols/floodsub/src/pubsub.rs @@ -28,12 +28,13 @@ use crate::{ pub type LastSeenCache = HashMap; pub type FloodsubPayload = (Option, Option>, Option>); +pub type SubscrbedTopicApi = HashMap, Vec)>>; #[derive(Debug, Clone)] pub struct FloosubStore { peer_topics: HashMap>, peers: HashMap>>, - subscribed_topic_api: HashMap, Vec)>>, + subscribed_topic_api: SubscrbedTopicApi, } #[derive(Debug, Clone, Hash, PartialEq, Eq)]