chore: update Iceoryx and tungstenite.

nkaz001 · nkaz001 · commit 0e3f799c31a9 · 2025-01-10T07:24:13.000-05:00
diff --git a/collector/Cargo.toml b/collector/Cargo.toml
@@ -11,7 +11,7 @@ serde = { version = "1.0.196", features = ["derive"] }
 tokio = { version = "1.35.1", features = ["full"] }
 serde_json = "1.0.113"
 futures-util = "0.3.30"
-tokio-tungstenite = { version = "0.24.0", features = ["native-tls"] }
+tokio-tungstenite = { version = "0.26.1", features = ["native-tls"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = [] }
 anyhow = "1.0.79"
diff --git a/collector/src/binance/http.rs b/collector/src/binance/http.rs
@@ -10,7 +10,7 @@ use futures_util::{SinkExt, StreamExt};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 use tokio_tungstenite::{
     connect_async,
-    tungstenite::{client::IntoClientRequest, Message},
+    tungstenite::{client::IntoClientRequest, Bytes, Message, Utf8Bytes},
 };
 use tracing::{error, warn};
 
@@ -52,7 +52,7 @@ pub async fn fetch_depth_snapshot(symbol: &str) -> Result<String, reqwest::Error
 
 pub async fn connect(
     url: &str,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) -> Result<(), anyhow::Error> {
     let request = url.into_client_request()?;
     let (ws_stream, _) = connect_async(request).await?;
@@ -61,7 +61,7 @@ pub async fn connect(
 
     tokio::spawn(async move {
         while rx.recv().await.is_some() {
-            if write.send(Message::Pong(Vec::new())).await.is_err() {
+            if write.send(Message::Pong(Bytes::default())).await.is_err() {
                 return;
             }
         }
@@ -102,7 +102,7 @@ pub async fn connect(
 pub async fn keep_connection(
     streams: Vec<String>,
     symbol_list: Vec<String>,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) {
     let mut error_count = 0;
     loop {
diff --git a/collector/src/binance/mod.rs b/collector/src/binance/mod.rs
@@ -5,6 +5,7 @@ use std::collections::HashMap;
 use chrono::{DateTime, Utc};
 pub use http::{fetch_depth_snapshot, keep_connection};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
+use tokio_tungstenite::tungstenite::Utf8Bytes;
 use tracing::{error, warn};
 
 use crate::{error::ConnectorError, throttler::Throttler};
@@ -13,10 +14,10 @@ fn handle(
     prev_u_map: &mut HashMap<String, i64>,
     writer_tx: &UnboundedSender<(DateTime<Utc>, String, String)>,
     recv_time: DateTime<Utc>,
-    data: String,
+    data: Utf8Bytes,
     throttler: &Throttler,
 ) -> Result<(), ConnectorError> {
-    let j: serde_json::Value = serde_json::from_str(&data)?;
+    let j: serde_json::Value = serde_json::from_str(data.as_str())?;
     if let Some(j_data) = j.get("data") {
         if let Some(j_symbol) = j_data
             .as_object()
@@ -69,7 +70,7 @@ fn handle(
                     *prev_u_map.entry(symbol.to_string()).or_insert(0) = u;
                 }
             }
-            let _ = writer_tx.send((recv_time, symbol.to_string(), data));
+            let _ = writer_tx.send((recv_time, symbol.to_string(), data.to_string()));
         }
     }
     Ok(())
diff --git a/collector/src/binancefuturescm/http.rs b/collector/src/binancefuturescm/http.rs
@@ -10,7 +10,7 @@ use futures_util::{SinkExt, StreamExt};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 use tokio_tungstenite::{
     connect_async,
-    tungstenite::{client::IntoClientRequest, Message},
+    tungstenite::{client::IntoClientRequest, Bytes, Message, Utf8Bytes},
 };
 use tracing::{error, warn};
 
@@ -53,7 +53,7 @@ pub async fn fetch_depth_snapshot(symbol: &str) -> Result<String, reqwest::Error
 
 pub async fn connect(
     url: &str,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) -> Result<(), anyhow::Error> {
     let request = url.into_client_request()?;
     let (ws_stream, _) = connect_async(request).await?;
@@ -62,7 +62,7 @@ pub async fn connect(
 
     tokio::spawn(async move {
         while rx.recv().await.is_some() {
-            if write.send(Message::Pong(Vec::new())).await.is_err() {
+            if write.send(Message::Pong(Bytes::default())).await.is_err() {
                 return;
             }
         }
@@ -103,7 +103,7 @@ pub async fn connect(
 pub async fn keep_connection(
     streams: Vec<String>,
     symbol_list: Vec<String>,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) {
     let mut error_count = 0;
     loop {
diff --git a/collector/src/binancefuturescm/mod.rs b/collector/src/binancefuturescm/mod.rs
@@ -5,6 +5,7 @@ use std::collections::HashMap;
 use chrono::{DateTime, Utc};
 pub use http::{fetch_depth_snapshot, keep_connection};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
+use tokio_tungstenite::tungstenite::Utf8Bytes;
 use tracing::{error, warn};
 
 use crate::{error::ConnectorError, throttler::Throttler};
@@ -13,10 +14,10 @@ fn handle(
     prev_u_map: &mut HashMap<String, i64>,
     writer_tx: &UnboundedSender<(DateTime<Utc>, String, String)>,
     recv_time: DateTime<Utc>,
-    data: String,
+    data: Utf8Bytes,
     throttler: &Throttler,
 ) -> Result<(), ConnectorError> {
-    let j: serde_json::Value = serde_json::from_str(&data)?;
+    let j: serde_json::Value = serde_json::from_str(data.as_str())?;
     if let Some(j_data) = j.get("data") {
         if let Some(j_symbol) = j_data
             .as_object()
@@ -70,7 +71,7 @@ fn handle(
                 }
                 *prev_u_map.entry(symbol.to_string()).or_insert(0) = u;
             }
-            let _ = writer_tx.send((recv_time, symbol.to_string(), data));
+            let _ = writer_tx.send((recv_time, symbol.to_string(), data.to_string()));
         }
     }
     Ok(())
diff --git a/collector/src/binancefuturesum/http.rs b/collector/src/binancefuturesum/http.rs
@@ -10,7 +10,7 @@ use futures_util::{SinkExt, StreamExt};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 use tokio_tungstenite::{
     connect_async,
-    tungstenite::{client::IntoClientRequest, Message},
+    tungstenite::{client::IntoClientRequest, Bytes, Message, Utf8Bytes},
 };
 use tracing::{error, warn};
 
@@ -53,7 +53,7 @@ pub async fn fetch_depth_snapshot(symbol: &str) -> Result<String, reqwest::Error
 
 pub async fn connect(
     url: &str,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) -> Result<(), anyhow::Error> {
     let request = url.into_client_request()?;
     let (ws_stream, _) = connect_async(request).await?;
@@ -62,7 +62,7 @@ pub async fn connect(
 
     tokio::spawn(async move {
         while rx.recv().await.is_some() {
-            if write.send(Message::Pong(Vec::new())).await.is_err() {
+            if write.send(Message::Pong(Bytes::default())).await.is_err() {
                 return;
             }
         }
@@ -103,7 +103,7 @@ pub async fn connect(
 pub async fn keep_connection(
     streams: Vec<String>,
     symbol_list: Vec<String>,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) {
     let mut error_count = 0;
     loop {
diff --git a/collector/src/binancefuturesum/mod.rs b/collector/src/binancefuturesum/mod.rs
@@ -5,6 +5,7 @@ use std::collections::HashMap;
 use chrono::{DateTime, Utc};
 pub use http::{fetch_depth_snapshot, keep_connection};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
+use tokio_tungstenite::tungstenite::Utf8Bytes;
 use tracing::{error, warn};
 
 use crate::{error::ConnectorError, throttler::Throttler};
@@ -13,10 +14,10 @@ fn handle(
     prev_u_map: &mut HashMap<String, i64>,
     writer_tx: &UnboundedSender<(DateTime<Utc>, String, String)>,
     recv_time: DateTime<Utc>,
-    data: String,
+    data: Utf8Bytes,
     throttler: &Throttler,
 ) -> Result<(), ConnectorError> {
-    let j: serde_json::Value = serde_json::from_str(&data)?;
+    let j: serde_json::Value = serde_json::from_str(data.as_str())?;
     if let Some(j_data) = j.get("data") {
         if let Some(j_symbol) = j_data
             .as_object()
@@ -70,7 +71,7 @@ fn handle(
                 }
                 *prev_u_map.entry(symbol.to_string()).or_insert(0) = u;
             }
-            let _ = writer_tx.send((recv_time, symbol.to_string(), data));
+            let _ = writer_tx.send((recv_time, symbol.to_string(), data.to_string()));
         }
     }
     Ok(())
diff --git a/collector/src/bybit/http.rs b/collector/src/bybit/http.rs
@@ -13,29 +13,32 @@ use tokio::{
 };
 use tokio_tungstenite::{
     connect_async,
-    tungstenite::{client::IntoClientRequest, Message},
+    tungstenite::{client::IntoClientRequest, Bytes, Message, Utf8Bytes},
 };
 use tracing::{error, warn};
 
 pub async fn connect(
     url: &str,
     topics: Vec<String>,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) -> Result<(), anyhow::Error> {
     let request = url.into_client_request()?;
     let (ws_stream, _) = connect_async(request).await?;
     let (mut write, mut read) = ws_stream.split();
     let (tx, mut rx) = unbounded_channel::<()>();
 
     write
-        .send(Message::Text(format!(
-            r#"{{"req_id": "subscribe", "op": "subscribe", "args": [{}]}}"#,
-            topics
-                .iter()
-                .map(|s| format!("\"{s}\""))
-                .collect::<Vec<_>>()
-                .join(",")
-        )))
+        .send(Message::Text(
+            format!(
+                r#"{{"req_id": "subscribe", "op": "subscribe", "args": [{}]}}"#,
+                topics
+                    .iter()
+                    .map(|s| format!("\"{s}\""))
+                    .collect::<Vec<_>>()
+                    .join(",")
+            )
+            .into(),
+        ))
         .await?;
 
     tokio::spawn(async move {
@@ -45,7 +48,7 @@ pub async fn connect(
                 result = rx.recv() => {
                     match result {
                         Some(_) => {
-                            if write.send(Message::Pong(Vec::new())).await.is_err() {
+                            if write.send(Message::Pong(Bytes::default())).await.is_err() {
                                 return;
                             }
                         }
@@ -56,7 +59,7 @@ pub async fn connect(
                 }
                 _ = ping_interval.tick() => {
                     if write.send(
-                        Message::Text(r#"{"req_id": "ping", "op": "ping"}"#.to_string())
+                        Message::Text(r#"{"req_id": "ping", "op": "ping"}"#.into())
                     ).await.is_err() {
                         return;
                     }
@@ -100,7 +103,7 @@ pub async fn connect(
 pub async fn keep_connection(
     topics: Vec<String>,
     symbol_list: Vec<String>,
-    ws_tx: UnboundedSender<(DateTime<Utc>, String)>,
+    ws_tx: UnboundedSender<(DateTime<Utc>, Utf8Bytes)>,
 ) {
     let mut error_count = 0;
     loop {
diff --git a/collector/src/bybit/mod.rs b/collector/src/bybit/mod.rs
@@ -1,5 +1,6 @@
 use chrono::{DateTime, Utc};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
+use tokio_tungstenite::tungstenite::Utf8Bytes;
 use tracing::error;
 
 use self::http::keep_connection;
@@ -10,13 +11,13 @@ mod http;
 fn handle(
     writer_tx: &UnboundedSender<(DateTime<Utc>, String, String)>,
     recv_time: DateTime<Utc>,
-    data: String,
+    data: Utf8Bytes,
 ) -> Result<(), ConnectorError> {
-    let j: serde_json::Value = serde_json::from_str(&data)?;
+    let j: serde_json::Value = serde_json::from_str(data.as_str())?;
     if let Some(j_topic) = j.get("topic") {
         let topic = j_topic.as_str().ok_or(ConnectorError::FormatError)?;
         let symbol = topic.split(".").last().ok_or(ConnectorError::FormatError)?;
-        let _ = writer_tx.send((recv_time, symbol.to_string(), data));
+        let _ = writer_tx.send((recv_time, symbol.to_string(), data.to_string()));
     } else if let Some(j_success) = j.get("success") {
         let success = j_success.as_bool().ok_or(ConnectorError::FormatError)?;
         if !success {
diff --git a/connector/Cargo.toml b/connector/Cargo.toml
@@ -18,12 +18,12 @@ chrono = { version = "0.4.33" }
 serde = { version = "1.0.196", features = ["derive"] }
 serde_json = { version = "1.0.113" }
 tokio = { version = "1.35.1", features = ["full"] }
-tokio-tungstenite = { version = "0.24.0", features = ["rustls-tls-native-roots"] }
+tokio-tungstenite = { version = "0.26.1", features = ["rustls-tls-native-roots"] }
 reqwest = { version = "0.12.3", default-features = false, features = ["json", "rustls-tls-native-roots"] }
 futures-util = { version = "0.3.30" }
 sha2 = { version = "0.11.0-pre.3" }
 hmac = { version = "0.13.0-pre.3" }
-iceoryx2 = { version = "0.4.1", features = ["logger_tracing"] }
+iceoryx2 = { version = "0.5.0", features = ["logger_tracing"] }
 toml = "0.8.19"
 tracing-subscriber = "0.3.18"
 clap = { version = "4.5.15", features = ["derive"] }
diff --git a/connector/src/binancefutures/market_data_stream.rs b/connector/src/binancefutures/market_data_stream.rs
@@ -275,7 +275,7 @@ impl MarketDataStream {
                                 "{symbol}@depth@0ms"
                             ],
                             "id": "{id}"
-                        }}"#))).await?;
+                        }}"#).into())).await?;
                     }
                     Err(RecvError::Closed) => {
                         return Ok(());
diff --git a/connector/src/bybit/private_stream.rs b/connector/src/bybit/private_stream.rs
@@ -14,7 +14,7 @@ use tokio::{
 };
 use tokio_tungstenite::{
     connect_async,
-    tungstenite::{client::IntoClientRequest, Message},
+    tungstenite::{client::IntoClientRequest, Bytes, Message},
     MaybeTlsStream,
     WebSocketStream,
 };
@@ -90,7 +90,7 @@ impl PrivateStream {
                             ],
                         };
                         let s = serde_json::to_string(&op).unwrap();
-                        write.send(Message::Text(s)).await?;
+                        write.send(Message::Text(s.into())).await?;
                     } else {
                         return Err(BybitError::AuthError {
                             msg: resp.ret_msg.unwrap(),
@@ -258,7 +258,7 @@ impl PrivateStream {
             args: vec![self.api_key.clone(), expires.to_string(), signature],
         };
         let s = serde_json::to_string(&op).unwrap();
-        write.send(Message::Text(s)).await?;
+        write.send(Message::Text(s.into())).await?;
 
         loop {
             select! {
@@ -269,7 +269,7 @@ impl PrivateStream {
                         args: vec![]
                     };
                     let s = serde_json::to_string(&op).unwrap();
-                    write.send(Message::Text(s)).await?;
+                    write.send(Message::Text(s.into())).await?;
                 }
                 msg = self.symbol_rx.recv() => {
                     match msg {
@@ -337,7 +337,7 @@ impl PrivateStream {
                             }
                         }
                         Some(Ok(Message::Ping(_))) => {
-                            write.send(Message::Pong(Vec::new())).await?;
+                            write.send(Message::Pong(Bytes::default())).await?;
                         }
                         Some(Ok(Message::Close(close_frame))) => {
                             return Err(BybitError::ConnectionAbort(
diff --git a/connector/src/bybit/public_stream.rs b/connector/src/bybit/public_stream.rs
diff --git a/connector/src/bybit/trade_stream.rs b/connector/src/bybit/trade_stream.rs
diff --git a/connector/src/main.rs b/connector/src/main.rs
diff --git a/hftbacktest/Cargo.toml b/hftbacktest/Cargo.toml
diff --git a/hftbacktest/src/live/ipc/iceoryx.rs b/hftbacktest/src/live/ipc/iceoryx.rs

Original file line number	Diff line number	Diff line change
`@@ -275,7 +275,7 @@ impl MarketDataStream {`
`275`	`275`	`"{symbol}@depth@0ms"`
`276`	`276`	`],`
`277`	`277`	`"id": "{id}"`
`278`		`- }}"#))).await?;`
	`278`	`+ }}"#).into())).await?;`
`279`	`279`	`}`
`280`	`280`	`Err(RecvError::Closed) => {`
`281`	`281`	`return Ok(());`