Add websocket tool

Cargo.toml

@@ -5,6 +5,7 @@ members = [
   "shared",
   "extractor",
   "tools/logger",
+  "tools/websocket",
   "tools/metrics",
   "tools/addr-connectivity-rs",
 ]

shared/Cargo.toml

@@ -6,9 +6,10 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-prost = "0.10"
+prost = "0.10.4"
 bitcoin = "0.29.0"
 base32 = "0.4.0" # for encoding Tor/Onion addresses
+serde = { version = "1.0.203", features = ["derive"] }
 
 [build-dependencies]
 prost-build = "0.10"

shared/build.rs

@@ -6,10 +6,14 @@ use prost_build;
 
 fn main() {
     // Generate Rust types for the protobuf's
-    if let Err(e) = prost_build::compile_protos(
-        &["../protobuf/proto-types/wrapper.proto"],
-        &["../protobuf/proto-types/"],
-    ) {
+    if let Err(e) = prost_build::Config::new()
+        .compile_well_known_types()
+        .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]")
+        .compile_protos(
+            &["../protobuf/proto-types/wrapper.proto"],
+            &["../protobuf/proto-types/"],
+        )
+    {
         println!("Error while compiling protos: {}", e);
         panic!("Failed to code-gen the Rust structs from the Protobuf definitions");
     }

shared/src/lib.rs

@@ -1,6 +1,7 @@
 #![cfg_attr(feature = "strict", deny(warnings))]
 
 pub extern crate bitcoin;
+pub extern crate prost;
 
 pub mod addrman;
 pub mod ctypes;

tools/addr-connectivity-rs/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2021"
 [dependencies]
 shared = { path = "../../shared" }
 
-prost = "0.10"
 nng = "1.0.1"
 crossbeam = { version = "0.8.2", features = ["crossbeam-channel"] }
 rand = "0.8.5"

tools/addr-connectivity-rs/src/main.rs

@@ -15,6 +15,7 @@ use shared::net_msg::message::Msg;
 use shared::net_msg::Message as NetMessage;
 use shared::primitive::address::Address as AddressType;
 use shared::primitive::Address;
+use shared::prost::Message as ProstMessage;
 use shared::util;
 use shared::wrapper;
 use shared::wrapper::wrapper::Wrap;
@@ -24,7 +25,6 @@ use crossbeam::channel::{unbounded, Receiver, Sender};
 use nng::options::protocol::pubsub::Subscribe;
 use nng::options::Options;
 use nng::{Protocol, Socket};
-use prost::Message as ProstMessage;
 use rand::Rng;
 
 mod metrics;

tools/logger/Cargo.toml

@@ -6,7 +6,6 @@ edition = "2021"
 [dependencies]
 shared = { path = "../../shared" }
 
-prost = "0.10"
 nng = "1.0.1"
 
 [features]

tools/logger/src/main.rs

@@ -4,7 +4,7 @@ use nng::options::protocol::pubsub::Subscribe;
 use nng::options::Options;
 use nng::{Protocol, Socket};
 
-use prost::Message;
+use shared::prost::Message;
 use shared::wrapper;
 use shared::wrapper::wrapper::Wrap;
 

tools/metrics/Cargo.toml

@@ -12,7 +12,6 @@ lazy_static = "1.4.0"
 log = "0.4.14"
 simple_logger = "1.9.0"
 
-prost = "0.10"
 nng = "1.0.1"
 
 [features]

tools/metrics/src/main.rs

@@ -4,13 +4,12 @@ use nng::options::protocol::pubsub::Subscribe;
 use nng::options::Options;
 use nng::{Protocol, Socket};
 
-use prost::Message;
-
 use shared::addrman::addrman_event;
 use shared::mempool::mempool_event;
 use shared::net_conn::connection_event;
 use shared::net_msg;
 use shared::net_msg::{message::Msg, reject::RejectReason};
+use shared::prost::Message;
 use shared::util;
 use shared::validation::validation_event;
 use shared::wrapper;

tools/websocket/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "websocket"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+shared = { path = "../../shared" }
+nng = "1.0.1"
+tungstenite = "0.21.0"
+async-std = { version = "1.12.0", features = ["attributes"] }
+async-broadcast = "0.7.0"
+serde_json = "1.0.117"
+
+[features]
+# Treat warnings as a build error.
+strict = []

tools/websocket/src/main.rs

@@ -0,0 +1,103 @@
+#![cfg_attr(feature = "strict", deny(warnings))]
+
+use async_broadcast::broadcast;
+use async_std::task;
+use nng::options::protocol::pubsub::Subscribe;
+use nng::options::Options;
+use nng::{Protocol, Socket};
+use shared::prost::Message;
+use shared::wrapper;
+use shared::wrapper::wrapper::Wrap;
+use std::net::TcpListener;
+use tungstenite::accept;
+
+const ADDRESS: &'static str = "tcp://127.0.0.1:8883";
+const WS_ADDRESS: &'static str = "127.0.0.1:47482";
+
+#[async_std::main]
+async fn main() {
+    let (mut sender, broadcast_receiver) = broadcast(128);
+    sender.set_overflow(true);
+    let inactive_broadcast_receiver = broadcast_receiver.deactivate();
+
+    // nano message receive task
+    task::spawn(async move {
+        let sub = Socket::new(Protocol::Sub0).unwrap();
+        sub.dial(ADDRESS).unwrap();
+
+        let all_topics = vec![];
+        sub.set_opt::<Subscribe>(all_topics).unwrap();
+        loop {
+            let msg = sub.recv().unwrap();
+            let unwrapped = wrapper::Wrapper::decode(msg.as_slice()).unwrap().wrap;
+            if let Some(event) = unwrapped {
+                if let Err(e) = sender.broadcast(event).await {
+                    println!("Could not send msg event into broadcast channel: {}", e);
+                }
+            }
+        }
+    });
+
+    println!("Starting websocket server on {}", WS_ADDRESS);
+    let server = match TcpListener::bind(WS_ADDRESS) {
+        Ok(s) => s,
+        Err(e) => {
+            println!("Could not start websocket server on {}: {}", WS_ADDRESS, e);
+            return;
+        }
+    };
+
+    for stream in server.incoming() {
+        match stream {
+            Ok(stream) => {
+                let mut r = inactive_broadcast_receiver.clone().activate();
+                task::spawn(async move {
+                    match accept(stream) {
+                        Ok(mut websocket) => {
+                            println!(
+                                "Accepted new websocket connection: connections={}",
+                                r.receiver_count()
+                            );
+
+                            loop {
+                                match r.recv().await {
+                                    Ok(msg) => {
+                                        match serde_json::to_string::<Wrap>(&msg.clone().into()) {
+                                            Ok(msg) => {
+                                                if let Err(e) =
+                                                    websocket.send(tungstenite::Message::Text(msg))
+                                                {
+                                                    println!("Could not send msg to websocket: {}. Connection probably closed.", e);
+                                                    // Try our best to close and flush the websocket. If we can't,
+                                                    // we can't..
+                                                    let _ = websocket.close(None);
+                                                    let _ = websocket.flush();
+                                                    break;
+                                                }
+                                            }
+                                            Err(e) => {
+                                                println!(
+                                                    "Could not serialize the message to JSON: {}",
+                                                    e
+                                                )
+                                            }
+                                        }
+                                    }
+                                    Err(e) => {
+                                        println!("Could not receive msg: {}", e);
+                                    }
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            println!("Failed to open websocket on incoming connection: {}", e);
+                        }
+                    }
+                });
+            }
+            Err(e) => {
+                println!("Failed to accept incomming TCP connection: {}", e);
+            }
+        }
+    }
+}