initial impl zombienet backchannel

Cargo.toml

@@ -88,6 +88,7 @@ members = [
 	"node/test/polkadot-simnet/common",
 	"node/test/polkadot-simnet/node",
 	"node/test/polkadot-simnet/test",
+	"node/zombienet-backchannel",
 	"parachain/test-parachains",
 	"parachain/test-parachains/adder",
 	"parachain/test-parachains/adder/collator",

node/zombienet-backchannel/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "backchannel-cli"
+version = "0.1.0"
+edition = "2021"
+
+[package]
+name = "zombienet-backchannel"
+description = "Zombienet backchannel to notify test runner and coordinate with malus actors."
+license = "GPL-3.0-only"
+version = "0.9.13"
+authors = ["Parity Technologies <[email protected]>"]
+edition = "2018"
+readme = "README.md"
+publish = false
+
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[dependencies]
+# futures-channel = "0.3"
+tokio = { version = "1.0.0", default-features = false, features = ["macros", "net", "rt-multi-thread"] }
+url = "2.0.0"
+tokio-tungstenite = "0.16"
+futures-util = "0.3.18"
+parity-scale-codec = { version = "2.3.1", features = ["derive"] }
+reqwest = "0.11"
+thiserror = "1.0.30"
+log = "0.4.13"

node/zombienet-backchannel/src/lib.rs

@@ -0,0 +1,57 @@
+/// Provides a backchannel to zombienet test runner.
+
+use std::env;
+
+use futures_util::{StreamExt, stream::SplitStream};
+use tokio::net::TcpStream;
+use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async};
+use parity_scale_codec as scale_codec;
+
+mod errors;
+use errors::BackchannelError;
+
+#[derive(Debug)]
+pub struct ZombienetBackchannel {
+    pub inner: SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>
+}
+
+const BACKCHANNEL_HOST: &str = "backchannel";
+const BACKCHANNEL_PORT: &str = "3000";
+
+impl ZombienetBackchannel {
+    pub async fn connect() -> Result<ZombienetBackchannel,BackchannelError> {
+        let backchannel_host = env::var("BACKCHANNEL_HOST").unwrap_or_else(|_| BACKCHANNEL_HOST.to_string());
+        let backchannel_port = env::var("BACKCHANNEL_PORT").unwrap_or_else(|_| BACKCHANNEL_PORT.to_string());
+
+        let ws_url = format!("ws://{}:{}/ws", backchannel_host, backchannel_port);
+        log::debug!("Connecting to : {}",&ws_url);
+        let (ws_stream, _) = connect_async(ws_url).await.map_err(|_| BackchannelError::CantConnectToWS)?;
+        let (_write, read) = ws_stream.split();
+
+        Ok( ZombienetBackchannel {
+            inner: read
+        })
+    }
+
+    pub async fn send(key: &'static str, val: impl scale_codec::Encode) -> Result<(),BackchannelError> {
+        let backchannel_host = env::var("BACKCHANNEL_HOST").unwrap_or_else(|_| BACKCHANNEL_HOST.to_string());
+        let backchannel_port = env::var("BACKCHANNEL_PORT").unwrap_or_else(|_| BACKCHANNEL_PORT.to_string());
+
+        let client = reqwest::Client::new();
+        let encoded = val.encode();
+        let body = String::from_utf8_lossy(&encoded);
+        let zombienet_endpoint = format!("http://{}:{}/{}",backchannel_host, backchannel_port, key);
+
+        log::debug!("Sending to : {}",&zombienet_endpoint);
+        let _res = client.post(zombienet_endpoint)
+            .body(body.to_string())
+            .send()
+            .await.map_err(|e| {
+                log::error!("Error sending new item: {}",e);
+                BackchannelError::SendItemFail
+            })?;
+
+        Ok(())
+    }
+}
+