starting to work on converting integration test (Connect) with new sync paradigm

Author: orthecreedence

src/error.rs

@@ -61,3 +61,9 @@ impl PartialEq for Error {
 /// Wraps `std::result::Result` around our `Error` enum
 pub type Result<T> = std::result::Result<T, Error>;
 
+impl From<Error> for String {
+    fn from(err: Error) -> Self {
+        format!("{:?}", err)
+    }
+}
+

src/message.rs

@@ -163,6 +163,16 @@ pub enum MessageWrapped {
     Init(PeerPubkey),
 }
 
+impl MessageWrapped {
+    /// Grab this wrapped message's sender pubkey
+    pub fn pubkey_sender(&self) -> &PeerPubkey {
+        match self {
+            Self::Sealed(sealed) => sealed.pubkey_sender(),
+            Self::Init(pubkey) => pubkey,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

tests/common/mod.rs

@@ -11,11 +11,12 @@ use futures::{
 use serde::{Serialize, de::DeserializeOwned};
 use std::{
     collections::HashMap,
+    fmt::Debug,
     ops::Deref,
 };
 use tp2p::{
-    ConnectionInfo, Sync,
     action::Action,
+    sync::{ConnectionInfo, Sync},
 };
 
 pub fn serialize<T: Serialize>(obj: &T) -> Result<Vec<u8>, String> {
@@ -30,10 +31,10 @@ pub fn deserialize<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, String> {
     Ok(obj)
 }
 
-pub async fn action_dispatcher<U, T, S>(sync: &mut Sync, tx: &Sender<PeerEvent>, action: Action<U, T, S>) -> Result<(), String>
-    where U: Clone + Debug + serde::Serialize + serde::de::DeserializeOwned,
-          T: Clone + Debug + serde::Serialize + serde::de::DeserializeOwned,
-          S: Clone + Debug + serde::Serialize + serde::de::DeserializeOwned,
+pub async fn action_dispatcher<U, T, S>(sync: &mut Sync<T>, tx: &Sender<PeerEvent>, action: Action<U, T, S>) -> Result<(), String>
+    where U: Debug,
+          T: Debug + Eq + std::hash::Hash,
+          S: Debug,
 {
     if let Action::Sync(sync_action) = action {
         sync.apply_sync_actions(vec![sync_action]);

tests/connect.rs

@@ -1,37 +1,83 @@
-/*
 mod common;
 
 use async_std::{
     channel,
     task,
 };
+use chrono::{DateTime, Utc};
 use common::{
     action_dispatcher,
     deserialize,
     PeerEvent,
     PeerManager,
 };
 use tp2p::{
-    Sync,
+    event::Event,
     message::MessageWrapped,
     peer::{Keypair, PeeringConfig, ConfirmationMode, SubscriptionConfig},
+    sync::Sync,
 };
 
+fn now() -> DateTime<Utc> {
+    Utc::now()
+}
+
 macro_rules! main_loop {
-    ($sync:expr, $peer:expr, $tx:expr, $rx:expr, $name:expr) => {
+    ($sync:expr, $peer:expr, $tx:expr, $rx:expr, $name:expr, $U:ty, $T:ty, $S:ty) => {
         loop {
             match $rx.try_recv() {
                 Ok(PeerEvent::Connect(from)) => {
                     println!(concat!($name, ": new connection! {:?}"), from);
                 }
                 Ok(PeerEvent::Recv(from, message_bytes)) => {
                     let msg: MessageWrapped = deserialize(message_bytes.as_ref())?;
-                    //println!(concat!($name, ": recv: {:?} -- {:?}"), from, msg);  // DEBUG: remove
-                    let actions = $sync.process_incoming_message(&msg, from)
-                        .map_err(|err| format!("process: {:?}", err))?;
+                    let actions = match &msg {
+                        MessageWrapped::Init(pubkey) => {
+                            $sync.process_init_message(&msg, &from, &now())?
+                        }
+                        MessageWrapped::Sealed(sealed) => {
+                            let message_opened = $sync.unwrap_incoming_message(sealed)?;
+                            match message_opened.body() {
+                                Event::Hello => {
+                                    $sync.process_event_hello(&message_opened, sealed.pubkey_sender(), &from, &now())?
+                                }
+                                Event::PeerInit { .. } => {
+                                    $sync.process_event_peer_init(&message_opened, sealed.pubkey_sender(), &from, &now())?
+                                }
+                                Event::PeerConfirm { .. } => {
+                                    $sync.process_event_peer_confirm(&message_opened, sealed.pubkey_sender(), &now())?
+                                }
+                                Event::Ping => {
+                                    $sync.process_event_ping(&message_opened, sealed.pubkey_sender(), &now())?
+                                }
+                                Event::Pong => {
+                                    $sync.process_event_pong(&message_opened, sealed.pubkey_sender(), &now())?
+                                }
+                                Event::QueryMessagesByID { ids } => {
+                                    let messages = {
+                                        drop(ids);
+                                        vec![]
+                                    };
+                                    $sync.process_event_query_messages_by_id(&message_opened, sealed.pubkey_sender(), &messages)?
+                                }
+                                Event::QueryMessagesByDepth { topic, depth } => {
+                                    let messages = {
+                                        drop(topic);
+                                        drop(depth);
+                                        vec![]
+                                    };
+                                    $sync.process_event_query_messages_by_depth(&message_opened, sealed.pubkey_sender(), &messages)?
+                                }
+                                Event::Subscribe(..) => {
+                                }
+                                _ => panic!("oi"),
+                            }
+                        }
+                    };
+
                     for action in actions {
                         println!(concat!($name, ": action -- {:?}"), action);
-                        action_dispatcher(&mut $sync, &$tx, action).await?;
+                        action_dispatcher::<$U, $T, $S>(&mut $sync, &$tx, action).await?;
                     }
                 }
                 Err(channel::TryRecvError::Closed) => {
@@ -46,7 +92,6 @@ macro_rules! main_loop {
 
 #[async_std::test]
 async fn peer_connect() -> Result<(), String> {
-    env_logger::init();
     let peer1_task = task::spawn(async move {
         let keypair = Keypair::new_random();
         let peering_config = PeeringConfig::new(
@@ -61,7 +106,7 @@ async fn peer_connect() -> Result<(), String> {
         let peer_task = task::spawn(async move {
             peer.start("127.0.0.1", 50020).await.expect("error running peer");
         });
-        main_loop! { sync , peer, tx, rx, "peer1" }
+        main_loop! { sync , peer, tx, rx, "peer1", (), (), () }
 
         peer_task.await;
         let res: Result<(), String> = Ok(());
@@ -84,12 +129,12 @@ async fn peer_connect() -> Result<(), String> {
             peer.start("127.0.0.1", 50021).await.expect("error running peer");
         });
 
-        let actions = sync.init_comm("127.0.0.1:50020").expect("peer_init failed");
+        let actions = sync.init_comm::<(), ()>("127.0.0.1:50020", &now()).expect("peer_init failed");
         for action in actions {
             println!("peer2: action -- {:?}", action);
             action_dispatcher(&mut sync, &tx, action).await?;
         }
-        main_loop! { sync , peer, tx, rx, "peer2" }
+        main_loop! { sync , peer, tx, rx, "peer2", (), (), () }
 
         peer_task.await;
         let res: Result<(), String> = Ok(());
@@ -101,5 +146,4 @@ async fn peer_connect() -> Result<(), String> {
 
     Ok(())
 }
-*/