feat: route protocol events via global notification channel

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<String>,
+    pub msg: Option<Vec<u8>>,
+    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<u128>,
+    pub count: u128,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum GlobalEvent {
+    Floodsub(FloodsubEvent),
+    Ping(PingEvent),
+}

crates/identity/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod events;
 pub mod keys;
 pub mod multiaddr;
 pub mod peer;

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());

crates/node/src/node.rs

@@ -32,6 +32,7 @@ pub struct Node {
     pub key_pair: RsaKeyPair,
     pub mpsc_tx: Sender<Vec<u8>>,
     pub handlers: Arc<Mutex<HashMap<String, ProtocolHanldler>>>,
+    pub global_event_tx: Sender<Vec<u8>>,
 
     // Protocols
     pub floodsub: Arc<Mutex<Option<Arc<FloodSub>>>>,
@@ -44,12 +45,14 @@ impl Node {
         key_pair: RsaKeyPair,
         handlers: Arc<Mutex<HashMap<String, ProtocolHanldler>>>,
         local_peer_info: PeerInfo,
+        global_event_tx: Sender<Vec<u8>>,
     ) -> 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();
@@ -139,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();
@@ -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");
             }
         }
 

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" }

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.
+It is not suitable for adversarial or large-scale production environments without significant extension.
+
+
+global_event_tx: UNSUBSCRIBE + SUBSCRIBE + RECV_MSG

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};
@@ -25,12 +28,13 @@ use crate::{
 
 pub type LastSeenCache = HashMap<MessageKey, u64>;
 pub type FloodsubPayload = (Option<String>, Option<Vec<String>>, Option<Vec<u8>>);
+pub type SubscrbedTopicApi = HashMap<String, Sender<(Vec<u8>, Vec<u8>)>>;
 
 #[derive(Debug, Clone)]
 pub struct FloosubStore {
     peer_topics: HashMap<String, Vec<String>>,
     peers: HashMap<String, Sender<Vec<u8>>>,
-    subscribed_topic_api: HashMap<String, Sender<Vec<u8>>>,
+    subscribed_topic_api: SubscrbedTopicApi,
 }
 
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
@@ -45,6 +49,7 @@ pub struct FloodSub {
     pub floodsub_mpsc_tx: Sender<Vec<u8>>,
     last_seen_cache: Arc<Mutex<LastSeenCache>>,
     pub floodsub_store: Arc<Mutex<FloosubStore>>,
+    pub global_event_tx: Sender<Vec<u8>>,
 }
 
 #[async_trait]
@@ -102,7 +107,7 @@ impl IProtocolHandler for FloodSub {
 }
 
 impl FloodSub {
-    pub async fn new(local_peer: &PeerInfo) -> Result<Arc<Self>> {
+    pub async fn new(local_peer: &PeerInfo, global_event_tx: Sender<Vec<u8>>) -> Result<Arc<Self>> {
         let (floodsub_mpsc_tx, floodsub_mpsc_rx) = mpsc::channel::<Vec<u8>>(300);
         let last_seen_cache = Arc::new(Mutex::new(HashMap::new()));
         let local_peer_info = local_peer.clone();
@@ -116,6 +121,7 @@ impl FloodSub {
                 peers: HashMap::new(),
                 subscribed_topic_api: HashMap::new(),
             })),
+            global_event_tx,
         });
 
         tokio::spawn(async move {
@@ -161,7 +167,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 +224,26 @@ impl FloodSub {
             return Ok(());
         }
 
-        debug!("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(bincode::serialize(&floosub_event).unwrap())
+            .await
+            .unwrap();
+        // ---------------------------------------------------------
 
-        let (topic_mpsc_tx, topic_mpsc_rx) = mpsc::channel::<Vec<u8>>(100);
+        let (topic_mpsc_tx, topic_mpsc_rx) = mpsc::channel::<(Vec<u8>, Vec<u8>)>(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 +291,19 @@ impl FloodSub {
                 return Ok(());
             }
 
-            debug!("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(bincode::serialize(&floosub_event).unwrap())
+                .await
+                .unwrap();
+            // ---------------------------------------------------------
 
             {
                 let mut store = self.floodsub_store.lock().await;
@@ -355,7 +385,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();
 

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;
 
@@ -8,19 +9,22 @@ use crate::pubsub::FloodsubPayload;
 pub struct SubscriptionAPI {
     pub topic_id: String,
     floodsub_mpsc_tx: Sender<Vec<u8>>,
-    topic_mpsc_rx: Receiver<Vec<u8>>,
+    topic_mpsc_rx: Receiver<(Vec<u8>, Vec<u8>)>,
+    global_event_tx: Sender<Vec<u8>>,
 }
 
 impl SubscriptionAPI {
     pub fn new(
         topic_id: String,
         floodsub_mpsc_tx: Sender<Vec<u8>>,
-        topic_mpsc_rx: Receiver<Vec<u8>>,
+        topic_mpsc_rx: Receiver<(Vec<u8>, Vec<u8>)>,
+        global_event_tx: Sender<Vec<u8>>,
     ) -> Self {
         SubscriptionAPI {
             topic_id,
             floodsub_mpsc_tx,
             topic_mpsc_rx,
+            global_event_tx,
         }
     }
 
@@ -48,20 +52,32 @@ impl SubscriptionAPI {
         Ok(())
     }
 
-    pub async fn recv(&mut self) -> Option<Vec<u8>> {
+    pub async fn recv(&mut self) -> Option<(Vec<u8>, Vec<u8>)> {
         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) => {
-                    let msg = String::from_utf8_lossy(&payload).to_string();
-                    println!("[{}]: {}", topic, msg);
+                Some((payload, source_peer)) => {
+                    // 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 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(bincode::serialize(&floosub_event).unwrap())
+                        .await
+                        .unwrap();
                 }
                 None => {
                     debug!("[{}]: Receiver loop ended", topic);

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<u128>,
-    pub remote_peer_id: String,
-}
-
 pub struct Ping {
     pub count: Arc<Mutex<u32>>,
+    global_event_tx: Sender<Vec<u8>>,
 }
 
 impl Ping {
-    pub fn new(count: Option<u32>) -> Self {
+    pub fn new(count: Option<u32>, global_event_tx: Sender<Vec<u8>>) -> 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();