refactor: Use Rayon for plexer stage (#5)

Cargo.toml

@@ -14,7 +14,6 @@ authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 [dependencies]
 # pallas = "0.14.0"
 pallas = { path = "../pallas/pallas" }
-# pallas = { git = "https://github.com/txpipe/pallas.git" }
 hex = "0.4.3"
 net2 = "0.2.37"
 bech32 = "0.8.1"
@@ -35,3 +34,5 @@ thiserror = "1.0.30"
 lazy_static = "1.4.0"
 rayon = "1.5.3"
 rocksdb = "0.19.0"
+tracing = "0.1.37"
+tracing-subscriber = "0.3.16"

src/lib.rs

@@ -5,3 +5,6 @@ pub mod model;
 pub mod prelude;
 pub mod storage;
 pub mod upstream;
+
+#[cfg(test)]
+mod tests;

src/prelude.rs

@@ -1,5 +1,6 @@
 pub use super::model::*;
 pub use super::storage::Cursor;
+pub use super::upstream::prelude::*;
 
 use std::fmt::Display;
 

src/tests/plexing.rs

@@ -0,0 +1,43 @@
+use pallas::network::miniprotocols::Point;
+use tracing::info;
+
+use crate::upstream::plexer;
+use crate::upstream::prelude::*;
+
+#[test]
+fn connect_to_real_relay() {
+    tracing::subscriber::set_global_default(
+        tracing_subscriber::FmtSubscriber::builder()
+            .with_max_level(tracing::Level::TRACE)
+            .finish(),
+    )
+    .unwrap();
+
+    let known_point = Point::Specific(
+        3866155,
+        hex::decode("9a5446c4178c708706f8218ee05cec7674396e5f044f911eacc1ad1147cc353e").unwrap(),
+    );
+
+    let mut input = MuxInputPort::default();
+    let mut output = DemuxOutputPort::default();
+
+    let client_channel = protocol_channel(2, &mut input, &mut output);
+
+    let worker = plexer::Worker::new(
+        "preview-node.world.dev.cardano.org:30002".into(),
+        2,
+        input,
+        output,
+    );
+
+    let tether = gasket::runtime::spawn_stage(worker, gasket::runtime::Policy::default(), None);
+
+    let mut client = OuroborosClient::new(client_channel);
+    let (point, _) = client.find_intersect(vec![known_point.clone()]).unwrap();
+
+    assert_eq!(point.unwrap(), known_point);
+
+    info!("dismissing");
+    tether.dismiss_stage().unwrap();
+    tether.join_stage();
+}

src/upstream/mod.rs

@@ -1,2 +1,3 @@
 pub mod chainsync;
 pub mod plexer;
+pub mod prelude;