New websocket handler with working async
This commit is contained in:
Jedrzej Stuczynski
Generated
+70
-37
@@ -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)" = "<none>"
|
||||
"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"
|
||||
|
||||
+4
-2
@@ -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"
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
+119
-64
@@ -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<Message>,
|
||||
tx: UnboundedSender<Message>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum WebSocketError {
|
||||
@@ -21,72 +34,114 @@ impl From<hex::FromHexError> for WebSocketError {
|
||||
}
|
||||
}
|
||||
|
||||
struct Server {
|
||||
out: Sender,
|
||||
input_tx: mpsc::UnboundedSender<InputMessage>,
|
||||
}
|
||||
|
||||
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<InputMessage, WebSocketError> {
|
||||
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<InputMessage>) {
|
||||
// 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<InputMessage, WebSocketError> {
|
||||
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<dyn std::error::Error>> {
|
||||
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(())
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user