fix: re-add tests, adjust format to send out ready and echo

topos-protocol · Mar 22, 2024 · 3ce420a · 3ce420a
1 parent 54bd18b
commit 3ce420a
Show file tree

Hide file tree

Showing 6 changed files with 1,536 additions and 1,508 deletions.
diff --git a/crates/topos-p2p/src/behaviour/gossip.rs b/crates/topos-p2p/src/behaviour/gossip.rs
@@ -45,9 +45,31 @@ impl Behaviour {
         topic: &'static str,
         message: Vec<u8>,
     ) -> Result<MessageId, PublishError> {
+        println!("Publishing {} {}", message.len(), topic);
         debug!("Publishing {} {}", message.len(), topic);
-        self.gossipsub.publish(IdentTopic::new(topic), message)
-        // let mut messag_id = MessageId::new(&[0]);
+        match topic {
+            TOPOS_GOSSIP => {
+                if let Ok(msg_id) = self.gossipsub.publish(IdentTopic::new(topic), message) {
+                    debug!("Published on topos_gossip: {:?}", msg_id);
+                    return Ok(msg_id);
+                }
+            }
+            TOPOS_ECHO | TOPOS_READY => {
+                let batch = Batch {
+                    messages: vec![message],
+                };
+                if let Ok(msg_id) = self
+                    .gossipsub
+                    .publish(IdentTopic::new(topic), batch.encode_to_vec())
+                {
+                    debug!("Published on topos_gossip: {:?}", msg_id);
+                    return Ok(msg_id);
+                }
+            }
+            _ => {}
+        }
+
+        let messag_id = MessageId::new(&[0]);
         // match topic {
         //     TOPOS_GOSSIP => {
         //         if let Ok(msg_id) = self.gossipsub.publish(IdentTopic::new(topic), message) {
@@ -56,10 +78,10 @@ impl Behaviour {
         //         }
         //     }
         //     TOPOS_ECHO | TOPOS_READY => self.pending.entry(topic).or_default().push_back(message),
-        //     _ => return Err("Invalid topic"),
+        //     _ => (),
         // }
-        //
-        // Ok(messag_id)
+
+        Ok(messag_id)
     }
 
     pub fn subscribe(&mut self) -> Result<(), P2PError> {
@@ -204,7 +226,7 @@ impl NetworkBehaviour for Behaviour {
                     let batch = Batch {
                         messages: queue.drain(0..num_of_message).collect(),
                     };
-
+                    println!("Publishing {} {}", batch.messages.len(), topic);
                     debug!("Publishing {} {}", batch.messages.len(), topic);
                     let msg = batch.encode_to_vec();
                     P2P_GOSSIP_BATCH_SIZE.observe(batch.messages.len() as f64);

diff --git a/crates/topos-p2p/src/runtime/handle_command.rs b/crates/topos-p2p/src/runtime/handle_command.rs
@@ -61,19 +61,22 @@ impl Runtime {
                 }
             }
 
-            //TODO: Return `MessageId` to the caller here
             Command::Gossip {
                 topic,
                 data: message,
                 sender,
-            } => match self.swarm.behaviour_mut().gossipsub.publish(topic, message) {
-                Ok(message_id) => {
-                    debug!("Published message to {topic}");
-                    P2P_MESSAGE_SENT_ON_GOSSIPSUB_TOTAL.inc();
-                    let _ = sender.send(message_id);
+            } => {
+                println!("Send to GossipSub: {topic}");
+                match self.swarm.behaviour_mut().gossipsub.publish(topic, message) {
+                    Ok(message_id) => {
+                        println!("Published message to {topic}");
+                        debug!("Published message to {topic}");
+                        P2P_MESSAGE_SENT_ON_GOSSIPSUB_TOTAL.inc();
+                        let _ = sender.send(message_id);
+                    }
+                    Err(err) => error!("Failed to publish message to {topic}: {err}"),
                 }
-                Err(err) => error!("Failed to publish message to {topic}: {err}"),
-            },
+            }
         }
     }
 }