[feat] Add Telemetry for Binance

Cargo.toml

@@ -33,6 +33,7 @@ futures = "0.3.30"
 humantime-serde = "1.1.1"
 itertools = "0.13.0"
 mockito = "1.4.0"
+opentelemetry = { version = "0.26.0", features = ["metrics"] }
 prost = "0.13.1"
 protoc-gen-prost = "0.4.0"
 protoc-gen-tonic = "0.4.1"

bothan-binance/Cargo.toml

@@ -9,6 +9,7 @@ repository.workspace = true
 async-trait = { workspace = true }
 bothan-core = { workspace = true }
 humantime-serde = { workspace = true }
+opentelemetry = { workspace = true }
 rand = { workspace = true }
 rust_decimal = { workspace = true }
 serde = { workspace = true }

bothan-binance/src/api/msgs.rs

@@ -3,14 +3,14 @@ use serde::{Deserialize, Serialize};
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct SuccessResponse {
     pub result: Option<String>,
-    pub id: u64,
+    pub id: i64,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct ErrorResponse {
-    pub code: u16,
+    pub code: i16,
     pub msg: String,
-    pub id: u64,
+    pub id: i64,
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]

bothan-binance/src/api/websocket.rs

@@ -70,19 +70,20 @@ impl BinanceWebSocketConnection {
     /// ```
     pub async fn subscribe_mini_ticker_stream<K: AsRef<str>>(
         &mut self,
-        ids: &[K],
+        id: i64,
+        tickers: &[K],
     ) -> Result<(), SendError> {
         // Format the stream IDs for subscription.
-        let stream_ids = ids
+        let tickers = tickers
             .iter()
             .map(|id| format!("{}@miniTicker", id.as_ref()))
             .collect::<Vec<_>>();
 
         // Create the subscription payload.
         let payload = json!({
             "method": "SUBSCRIBE",
-            "params": stream_ids,
-            "id": rand::random::<u32>()
+            "params": tickers,
+            "id": id,
         });
 
         // Send the subscription message.
@@ -101,10 +102,11 @@ impl BinanceWebSocketConnection {
     /// ```
     pub async fn unsubscribe_mini_ticker_stream<K: AsRef<str>>(
         &mut self,
-        ids: &[K],
+        id: i64,
+        tickers: &[K],
     ) -> Result<(), SendError> {
         // Format the stream IDs for unsubscription.
-        let stream_ids = ids
+        let stream_ids = tickers
             .iter()
             .map(|id| format!("{}@miniTicker", id.as_ref()))
             .collect::<Vec<_>>();
@@ -113,7 +115,7 @@ impl BinanceWebSocketConnection {
         let payload = json!({
             "method": "UNSUBSCRIBE",
             "params": stream_ids,
-            "id": rand::random::<u32>()
+            "id":id,
         });
 
         // Send the unsubscription message.

bothan-binance/src/worker.rs

@@ -49,7 +49,7 @@ impl AssetWorker for BinanceWorker {
     async fn set_query_ids(&self, ids: Vec<String>) -> Result<(), SetQueryIDError> {
         let (to_sub, to_unsub) = self
             .store
-            .set_query_ids(ids)
+            .compute_query_id_differences(ids)
             .await
             .map_err(|e| SetQueryIDError::new(e.to_string()))?;