feat: Introduce polling-based multiplexer

txpipe · Mar 27, 2023 · 217adce · 217adce
1 parent 381a46f
commit 217adce
Show file tree

Hide file tree

Showing 7 changed files with 560 additions and 136 deletions.
diff --git a/pallas-multiplexer/src/agents.rs b/pallas-multiplexer/src/agents.rs
@@ -3,6 +3,7 @@
 use crate::Payload;
 use pallas_codec::{minicbor, Fragment};
 use thiserror::Error;
+use tracing::error;
 
 #[derive(Debug, Error)]
 pub enum ChannelError {
@@ -41,7 +42,11 @@ where
     match maybe_msg {
         Ok(msg) => Decoding::Done(msg, decoder.position()),
         Err(err) if err.is_end_of_input() => Decoding::NotEnoughData,
-        Err(err) => Decoding::UnexpectedError(Box::new(err)),
+        Err(err) => {
+            error!(?err);
+            error!("{}", hex::encode(buffer));
+            Decoding::UnexpectedError(Box::new(err))
+        }
     }
 }
 

diff --git a/pallas-upstream/Cargo.toml b/pallas-upstream/Cargo.toml
@@ -11,7 +11,11 @@ readme = "README.md"
 authors = ["Santiago Carmuega <santiago@carmuega.me>"]
 
 [dependencies]
-gasket = { git = "https://github.com/construkts/gasket-rs" }
+byteorder = "1.4.3"
+# gasket = { git = "https://github.com/construkts/gasket-rs" }
+gasket = { path = "../../../construkts/gasket-rs" }
+hex = "0.4.3"
+mio = { version = "0.8.6", features = ["net", "os-poll"] }
 # gasket = { version = "0.1.0", path = "../../../construkts/gasket-rs" }
 pallas-codec = { version = "0.18.0", path = "../pallas-codec" }
 pallas-crypto = { version = "0.18.0", path = "../pallas-crypto" }

diff --git a/pallas-upstream/src/chainsync.rs b/pallas-upstream/src/chainsync.rs
@@ -47,16 +47,19 @@ impl Worker {
 
         match value {
             Intersection::Origin => {
+                info!("intersecting origin");
                 let point = self.client.intersect_origin().or_restart()?;
 
                 Ok(Some(point))
             }
             Intersection::Tip => {
+                info!("intersecting tip");
                 let point = self.client.intersect_tip().or_restart()?;
 
                 Ok(Some(point))
             }
             Intersection::Breadcrumbs(points) => {
+                info!("intersecting breadcrumbs");
                 let (point, _) = self.client.find_intersect(Vec::from(points)).or_restart()?;
 
                 Ok(point)

diff --git a/pallas-upstream/src/framework.rs b/pallas-upstream/src/framework.rs
@@ -96,8 +96,8 @@ impl Error {
         Error::Message(error.to_string())
     }
 
-    pub fn custom(error: Box<dyn std::error::Error>) -> Error {
-        Error::Custom(format!("{error}"))
+    pub fn custom(error: impl Into<Box<dyn std::error::Error>>) -> Error {
+        Error::Custom(format!("{}", error.into()))
     }
 }
 

diff --git a/pallas-upstream/src/lib.rs b/pallas-upstream/src/lib.rs
@@ -1,6 +1,7 @@
 pub(crate) mod blockfetch;
 pub(crate) mod chainsync;
 pub(crate) mod framework;
+pub(crate) mod newplexer;
 pub(crate) mod plexer;
 
 pub mod cursor;