Merge pull request #73 from Zomatree/msgpack_fix

fix msgpack encoding

Cargo.lock

@@ -2785,8 +2785,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",
@@ -2795,8 +2794,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",

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"

src/notifications/websocket.rs

@@ -30,7 +30,7 @@ use super::{
 };
 
 type Tx = UnboundedSender<Message>;
-type PeerMap = Arc<Mutex<HashMap<SocketAddr, Tx>>>;
+type PeerMap = Arc<Mutex<HashMap<SocketAddr, (Tx, MSGFormat)>>>;
 
 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,
@@ -108,15 +108,15 @@ 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 {
             MSGFormat::JSON => match serde_json::to_string(&notification) {
                 Ok(s) => Message::Text(s),
-                Err(_) => return,
-            },
-            MSGFormat::MSGPACK => match rmp_serde::to_vec(&notification) {
+                Err(_) => return
+            }
+            MSGFormat::MSGPACK => match rmp_serde::to_vec_named(&notification) {
                 Ok(v) => Message::Binary(v),
                 Err(_) => return,
             },
@@ -370,12 +370,18 @@ pub fn publish(ids: Vec<String>, notification: ClientboundNotification) {
         }
     }
 
-    let msg = Message::Text(serde_json::to_string(&notification).unwrap());
+    let json_msg = Message::Text(serde_json::to_string(&notification).unwrap());
+    let msgpack_msg = Message::Binary(rmp_serde::to_vec_named(&notification).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.");
             }
         }