From 3ddde4c8ef2c38327377a1642f66614cd658c0b5 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Tue, 17 Dec 2019 12:17:49 +0000 Subject: [PATCH] New websocket handler with working async --- Cargo.lock | 107 ++++++++++++++-------- Cargo.toml | 6 +- src/commands/websocket.rs | 2 +- src/sockets/ws.rs | 183 +++++++++++++++++++++++++------------- 4 files changed, 194 insertions(+), 104 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1db7bd75cb..eb8d66ab7c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -789,6 +789,14 @@ dependencies = [ "autocfg 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "input_buffer" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "iovec" version = "0.1.4" @@ -821,11 +829,6 @@ name = "lazy_static" version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -[[package]] -name = "lazycell" -version = "1.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" - [[package]] name = "libc" version = "0.2.66" @@ -921,17 +924,6 @@ dependencies = [ "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)", ] -[[package]] -name = "mio-extras" -version = "2.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -dependencies = [ - "lazycell 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)", - "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", - "mio 0.6.21 (registry+https://github.com/rust-lang/crates.io-index)", - "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", -] - [[package]] name = "mio-named-pipes" version = "0.1.6" @@ -1046,7 +1038,8 @@ dependencies = [ "sfw-provider-requests 0.1.0", "sphinx 0.1.0", "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", - "ws 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)", + "tokio-tungstenite 0.10.0 (git+https://github.com/dbcfd/tokio-tungstenite?branch=tokio2)", + "tungstenite 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)", ] [[package]] @@ -1128,6 +1121,24 @@ name = "percent-encoding" version = "2.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" +[[package]] +name = "pin-project" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "pin-project-internal 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "pin-project-internal" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "pin-project-lite" version = "0.1.1" @@ -1857,6 +1868,18 @@ dependencies = [ "tokio-executor 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "tokio-tungstenite" +version = "0.10.0" +source = "git+https://github.com/dbcfd/tokio-tungstenite?branch=tokio2#6dc2018cbfe8fe7ddd75ff977343086503135b38" +dependencies = [ + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-project 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)", + "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)", + "tungstenite 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "try-lock" version = "0.2.2" @@ -1870,6 +1893,25 @@ dependencies = [ "cfg-if 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "tungstenite" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "base64 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)", + "byteorder 1.3.2 (registry+https://github.com/rust-lang/crates.io-index)", + "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", + "http 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)", + "httparse 1.3.4 (registry+https://github.com/rust-lang/crates.io-index)", + "input_buffer 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", + "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", + "native-tls 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)", + "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", + "sha-1 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)", + "url 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)", + "utf-8 0.7.5 (registry+https://github.com/rust-lang/crates.io-index)", +] + [[package]] name = "typenum" version = "1.11.2" @@ -1929,6 +1971,11 @@ dependencies = [ "percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "utf-8" +version = "0.7.5" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "uuid" version = "0.7.4" @@ -2009,23 +2056,6 @@ dependencies = [ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)", ] -[[package]] -name = "ws" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -dependencies = [ - "byteorder 1.3.2 (registry+https://github.com/rust-lang/crates.io-index)", - "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)", - "httparse 1.3.4 (registry+https://github.com/rust-lang/crates.io-index)", - "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", - "mio 0.6.21 (registry+https://github.com/rust-lang/crates.io-index)", - "mio-extras 2.0.6 (registry+https://github.com/rust-lang/crates.io-index)", - "rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)", - "sha-1 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)", - "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", - "url 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)", -] - [[package]] name = "ws2_32-sys" version = "0.2.1" @@ -2128,12 +2158,12 @@ dependencies = [ "checksum idna 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "38f09e0f0b1fb55fdee1f17470ad800da77af5186a1a76c026b679358b7e844e" "checksum idna 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "02e2673c30ee86b5b96a9cb52ad15718aa1f966f5ab9ad54a8b95d5ca33120a9" "checksum indexmap 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "712d7b3ea5827fcb9d4fda14bf4da5f136f0db2ae9c8f4bd4e2d1c6fde4e6db2" +"checksum input_buffer 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8e1b822cc844905551931d6f81608ed5f50a79c1078a4e2b4d42dbc7c1eedfbf" "checksum iovec 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e" "checksum itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "501266b7edd0174f8530248f87f99c88fbe60ca4ef3dd486835b8d8d53136f7f" "checksum kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d" "checksum keystream 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "c33070833c9ee02266356de0c43f723152bd38bd96ddf52c82b3af10c9138b28" "checksum lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" -"checksum lazycell 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b294d6fa9ee409a054354afc4352b0b9ef7ca222c69b8812cbea9e7d2bf3783f" "checksum libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)" = "d515b1f41455adea1313a4a2ac8a8a477634fbae63cc6100e3aebb207ce61558" "checksum lioness 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "4ae926706ba42c425c9457121178330d75e273df2e82e28b758faf3de3a9acb9" "checksum lock_api 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)" = "e57b3997725d2b60dbec1297f6c2e2957cc383db1cebd6be812163f969c7d586" @@ -2146,7 +2176,6 @@ dependencies = [ "checksum mime_guess 2.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1a0ed03949aef72dbdf3116a383d7b38b4768e6f960528cd6a6044aa9ed68599" "checksum miniz_oxide 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)" = "6f3f74f726ae935c3f514300cc6773a0c9492abc5e972d42ba0c0ebb88757625" "checksum mio 0.6.21 (registry+https://github.com/rust-lang/crates.io-index)" = "302dec22bcf6bae6dfb69c647187f4b4d0fb6f535521f7bc022430ce8e12008f" -"checksum mio-extras 2.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "52403fe290012ce777c4626790c8951324a2b9e3316b3143779c72b029742f19" "checksum mio-named-pipes 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "f5e374eff525ce1c5b7687c4cef63943e7686524a387933ad27ca7ec43779cb3" "checksum mio-uds 0.6.7 (registry+https://github.com/rust-lang/crates.io-index)" = "966257a94e196b11bb43aca423754d87429960a768de9414f3691d6957abf125" "checksum miow 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "8c1f2f3b1cf331de6896aabf6e9d55dca90356cc9960cca7eaaf408a355ae919" @@ -2164,6 +2193,8 @@ dependencies = [ "checksum pem 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "a1581760c757a756a41f0ee3ff01256227bdf64cb752839779b95ffb01c59793" "checksum percent-encoding 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "31010dd2e1ac33d5b46a5b413495239882813e0369f8ed8a5e266f173602f831" "checksum percent-encoding 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e" +"checksum pin-project 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "94b90146c7216e4cb534069fb91366de4ea0ea353105ee45ed297e2d1619e469" +"checksum pin-project-internal 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "44ca92f893f0656d3cba8158dd0f2b99b94de256a4a54e870bd6922fcc6c8355" "checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991" "checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587" "checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677" @@ -2239,8 +2270,10 @@ dependencies = [ "checksum tokio-tcp 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "1d14b10654be682ac43efee27401d792507e30fd8d26389e1da3b185de2e4119" "checksum tokio-threadpool 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)" = "f0c32ffea4827978e9aa392d2f743d973c1dfa3730a2ed3f22ce1e6984da848c" "checksum tokio-timer 0.2.12 (registry+https://github.com/rust-lang/crates.io-index)" = "1739638e364e558128461fc1ad84d997702c8e31c2e6b18fb99842268199e827" +"checksum tokio-tungstenite 0.10.0 (git+https://github.com/dbcfd/tokio-tungstenite?branch=tokio2)" = "" "checksum try-lock 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "e604eb7b43c06650e854be16a2a03155743d3752dd1c943f6829e26b7a36e382" "checksum try_from 0.3.2 (registry+https://github.com/rust-lang/crates.io-index)" = "283d3b89e1368717881a9d51dad843cc435380d8109c9e47d38780a324698d8b" +"checksum tungstenite 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)" = "8a0c2bd5aeb7dcd2bb32e472c8872759308495e5eccc942e929a513cd8d36110" "checksum typenum 1.11.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6d2783fe2d6b8c1101136184eb41be8b1ad379e4657050b8aaff0c79ee7575f9" "checksum unicase 2.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "50f37be617794602aabbeee0be4f259dc1778fabe05e2d67ee8f79326d5cb4f6" "checksum unicode-bidi 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)" = "49f2bd0c6468a8230e1db229cff8029217cf623c767ea5d60bfbd42729ea54d5" @@ -2249,6 +2282,7 @@ dependencies = [ "checksum unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c" "checksum url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "dd4e7c0d531266369519a4aa4f399d748bd37043b00bde1e4ff1f60a120b355a" "checksum url 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "75b414f6c464c879d7f9babf951f23bc3743fb7313c081b2e6ca719067ea9d61" +"checksum utf-8 0.7.5 (registry+https://github.com/rust-lang/crates.io-index)" = "05e42f7c18b8f902290b009cde6d651262f956c98bc51bca4cd1d511c9cd85c7" "checksum uuid 0.7.4 (registry+https://github.com/rust-lang/crates.io-index)" = "90dbc611eb48397705a6b0f6e917da23ae517e4d127123d2cf7674206627d32a" "checksum vcpkg 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)" = "3fc439f2794e98976c88a2a2dafce96b930fe8010b0a256b3c2199a773933168" "checksum vec_map 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "05c78687fb1a80548ae3250346c3db86a80a7cdd77bda190189f2d0a0987c81a" @@ -2262,5 +2296,4 @@ dependencies = [ "checksum winapi-i686-pc-windows-gnu 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" "checksum winapi-x86_64-pc-windows-gnu 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" "checksum winreg 0.6.2 (registry+https://github.com/rust-lang/crates.io-index)" = "b2986deb581c4fe11b621998a5e53361efe6b48a151178d0cd9eeffa4dc6acc9" -"checksum ws 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c51a2c47b5798ccc774ffb93ff536aec7c4275d722fd9c740c83cdd1af1f2d94" "checksum ws2_32-sys 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "d59cefebd0c892fa2dd6de581e937301d8552cb44489cdff035c6187cb63fa5e" diff --git a/Cargo.toml b/Cargo.toml index 2cef8f55f0..0ca4ee4fd7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,8 +26,10 @@ serde_json = "1.0.44" sphinx = { path = "../sphinx" } sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" } tokio = { version = "0.2", features = ["full"] } -ws = "0.9.1" +tungstenite = "0.9.2" + +# putting this explicitly below everything and most likely, the next time we look into it, it will already have a proper release +tokio-tungstenite = { git = "https://github.com/dbcfd/tokio-tungstenite", branch = "tokio2" } [dev-dependencies] mockito = "0.22.0" -serde_json = "1.0.44" diff --git a/src/commands/websocket.rs b/src/commands/websocket.rs index c5ecfabd81..7d9889019c 100644 --- a/src/commands/websocket.rs +++ b/src/commands/websocket.rs @@ -19,5 +19,5 @@ pub fn execute(matches: &ArgMatches) { .next() .expect("Failed to extract the socket address from the iterator"); - // ws::start(socket_address); + ws::start(socket_address); } diff --git a/src/sockets/ws.rs b/src/sockets/ws.rs index 85f82d46e8..832f5d5da9 100644 --- a/src/sockets/ws.rs +++ b/src/sockets/ws.rs @@ -1,11 +1,24 @@ use crate::clients::InputMessage; use futures::channel::mpsc; +use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; +use futures::stream::Stream; +use futures::Future; +use futures::{SinkExt, StreamExt}; use hex::FromHexError; use serde::{Deserialize, Serialize}; use serde_json::Value; use sphinx::route::Destination; use std::net::SocketAddr; -use ws::{listen, CloseCode, Handler, Message, Sender}; +use tokio::runtime::Runtime; +use tokio_tungstenite::accept_async; +use tungstenite::protocol::Message; +use std::time::Duration; + +struct Connection { + address: SocketAddr, + rx: UnboundedReceiver, + tx: UnboundedSender, +} #[derive(Debug)] enum WebSocketError { @@ -21,72 +34,114 @@ impl From for WebSocketError { } } -struct Server { - out: Sender, - input_tx: mpsc::UnboundedSender, -} - -impl Handler for Server { - fn on_message(&mut self, msg: Message) -> ws::Result<()> { - let parsed_msg = Server::parse_message(msg.clone()); - - println!("msg: {:?}", parsed_msg); - // Echo the message back - // self.out.send(msg) - Ok(()) - } - - fn on_close(&mut self, code: CloseCode, reason: &str) { - match code { - CloseCode::Normal => println!("The client is done with the connection."), - CloseCode::Away => println!("The client is leaving the site."), - _ => { - println!("The client encountered an error: {}", reason); - } - } - } -} - -impl Server { - // Proves we can call Rust methods from the websocket listener. Re-route it to wherever JS puts - // the `send_message` functionality. - fn parse_message(msg: Message) -> Result { - let text_msg = match msg { - Message::Text(msg) => msg, - Message::Binary(_) => panic!("binary messages are not supported!"), - }; - - let raw_msg: ClientMessageJSON = serde_json::from_str(&text_msg).unwrap(); - let address_vec = hex::decode(raw_msg.recipient_address)?; - - if address_vec.len() != 32 { - return Err(WebSocketError::InvalidDestinationLength); - } - - let mut address = [0; 32]; - address.copy_from_slice(&address_vec); - - let dummy_surb = [0; 16]; - - Ok(InputMessage( - Destination::new(address, dummy_surb), - raw_msg.message.into_bytes(), - )) - } -} - -pub fn start(socket_address: SocketAddr, input_tx: mpsc::UnboundedSender) { - // listen("127.0.0.1:3012", |out| { - // async move |msg| { - // out.send(msg) - // } - // }); - - // listen(socket_address, |out| Server { out, input_tx }).unwrap() -} #[derive(Serialize, Deserialize, Debug)] struct ClientMessageJSON { message: String, recipient_address: String, } + +fn dummy_response() -> Message { + Message::Text("foomp".to_string()) +} + +async fn handle_connection(conn: Connection) { + let mut conn = conn; + while let Some(msg) = conn.rx.next().await { + println!("Received a message from {}: {}", conn.address, msg); + // TODO: currently only hardcoded sends are supported + let parsed_message = parse_message(msg); + println!("parsed: {:?}", parsed_message); + println!("test async pre wait"); + + tokio::time::delay_for(Duration::from_secs(2)).await; + + println!("test async post wait"); + conn + .tx + .unbounded_send(dummy_response()) + .expect("Failed to forward message"); + } +} + +// Proves we can call Rust methods from the websocket listener. Re-route it to wherever JS puts +// the `send_message` functionality. +fn parse_message(msg: Message) -> Result { + let text_msg = match msg { + Message::Text(msg) => msg, + Message::Binary(_) => panic!("binary messages are not supported!"), + Message::Close(_) => panic!("todo: handle close!"), + _ => panic!("Other types of messages are also unsupported!"), + }; + + let raw_msg: ClientMessageJSON = serde_json::from_str(&text_msg).unwrap(); + let address_vec = hex::decode(raw_msg.recipient_address)?; + + if address_vec.len() != 32 { + return Err(WebSocketError::InvalidDestinationLength); + } + + let mut address = [0; 32]; + address.copy_from_slice(&address_vec); + + let dummy_surb = [0; 16]; + + Ok(InputMessage( + Destination::new(address, dummy_surb), + raw_msg.message.into_bytes(), + )) +} + +async fn accept_connection(stream: tokio::net::TcpStream) { + let address = stream + .peer_addr() + .expect("connected streams should have a peer address"); + println!("Peer address: {}", address); + + let mut ws_stream = tokio_tungstenite::accept_async(stream) + .await + .expect("Error during the websocket handshake occurred"); + + println!("New WebSocket connection: {}", address); + + // Create a channel for our stream, which other sockets will use to + // send us messages. Then register our address with the stream to send + // data to us. + let (msg_tx, msg_rx) = futures::channel::mpsc::unbounded(); + let (response_tx, mut response_rx) = futures::channel::mpsc::unbounded(); + let conn = Connection { + address, + rx: msg_rx, + tx: response_tx, + }; + tokio::spawn(handle_connection(conn)); + + while let Some(message) = ws_stream.next().await { + let message = message.expect("Failed to get request"); + msg_tx + .unbounded_send(message) + .expect("Failed to forward request"); + if let Some(resp) = response_rx.next().await { + ws_stream.send(resp).await.expect("Failed to send response"); + } + } +} + + + +pub fn start(address: SocketAddr) -> Result<(), Box> { + let mut rt = Runtime::new()?; + + rt.block_on(async { + let mut listener = tokio::net::TcpListener::bind(address).await?; + + while let Ok((stream, _)) = listener.accept().await { + // TODO: should it rather be rt.spawn? + tokio::spawn(accept_connection(stream)); + } + + eprintln!("The websocket went kaput..."); + Ok(()) + }) +} +