From 175d257e34ca3cf40a00cf0c4a13d357ff4cb3a8 Mon Sep 17 00:00:00 2001 From: Zomatree Date: Sun, 5 Sep 2021 22:01:58 +0100 Subject: [PATCH] fix msgpack encoding --- Cargo.lock | 6 ++---- Cargo.toml | 2 +- src/notifications/websocket.rs | 20 +++++++++++++------- 3 files changed, 16 insertions(+), 12 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 54a0f2a1..da27c482 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2758,8 +2758,7 @@ dependencies = [ [[package]] name = "rmp" version = "0.8.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f55e5fa1446c4d5dd1f5daeed2a4fe193071771a2636274d0d7a3b082aa7ad6" +source = "git+https://github.com/3Hren/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814" dependencies = [ "byteorder", "num-traits", @@ -2768,8 +2767,7 @@ dependencies = [ [[package]] name = "rmp-serde" version = "0.15.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "723ecff9ad04f4ad92fe1c8ca6c20d2196d9286e9c60727c4cb5511629260e9d" +source = "git+https://github.com/3Hren/msgpack-rust?rev=5bf2c24203ad422233cf35b7b7bfad9f7e811814#5bf2c24203ad422233cf35b7b7bfad9f7e811814" dependencies = [ "byteorder", "rmp", diff --git a/Cargo.toml b/Cargo.toml index e1e6c522..b3c02b41 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,7 +36,7 @@ base64 = "0.13.0" serde_json = "1.0.57" serde = { version = "1.0.115", features = ["derive"] } validator = { version = "0.11", features = ["derive"] } -rmp-serde = "0.15.5" +rmp-serde = { git = "https://github.com/3Hren/msgpack-rust", rev = "5bf2c24203ad422233cf35b7b7bfad9f7e811814" } # async futures = "0.3.8" diff --git a/src/notifications/websocket.rs b/src/notifications/websocket.rs index e6ae3506..a9a02d60 100644 --- a/src/notifications/websocket.rs +++ b/src/notifications/websocket.rs @@ -30,7 +30,7 @@ use super::{ }; type Tx = UnboundedSender; -type PeerMap = Arc>>; +type PeerMap = Arc>>; lazy_static! { static ref CONNECTIONS: PeerMap = Arc::new(Mutex::new(HashMap::new())); @@ -48,7 +48,7 @@ pub async fn launch_server() { } } -#[derive(Debug)] +#[derive(Debug, Clone)] enum MSGFormat { JSON, MSGPACK @@ -92,7 +92,7 @@ async fn accept(stream: TcpStream) { let (write, read) = ws_stream.split(); let (tx, rx) = unbounded(); - CONNECTIONS.lock().unwrap().insert(addr, tx.clone()); + CONNECTIONS.lock().unwrap().insert(addr, (tx.clone(), msg_format.clone())); let send = |notification: ClientboundNotification| { let res = match msg_format { @@ -100,7 +100,7 @@ async fn accept(stream: TcpStream) { Ok(s) => Message::Text(s), Err(_) => return } - MSGFormat::MSGPACK => match rmp_serde::to_vec(¬ification) { + MSGFormat::MSGPACK => match rmp_serde::to_vec_named(¬ification) { Ok(v) => Message::Binary(v), Err(_) => return } @@ -358,12 +358,18 @@ pub fn publish(ids: Vec, notification: ClientboundNotification) { } } - let msg = Message::Text(serde_json::to_string(¬ification).unwrap()); + let json_msg = Message::Text(serde_json::to_string(¬ification.clone()).unwrap()); + let msgpack_msg = Message::Binary(rmp_serde::to_vec_named(¬ification).unwrap()); let connections = CONNECTIONS.lock().unwrap(); for target in targets { - if let Some(conn) = connections.get(&target) { - if let Err(_) = conn.unbounded_send(msg.clone()) { + if let Some((conn, msg_format)) = connections.get(&target) { + let msg = match msg_format { + MSGFormat::JSON => json_msg.clone(), + MSGFormat::MSGPACK => msgpack_msg.clone() + }; + + if let Err(_) = conn.unbounded_send(msg) { debug!("Failed unbounded_send."); } }