[feat] Add New Binance Websocket Source

.gitignore

@@ -13,6 +13,4 @@ Cargo.lock
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
 
-
 .vscode/
-

Cargo.toml

@@ -4,9 +4,10 @@ edition = "2021"
 license = "MIT OR Apache-2.0"
 
 [workspace]
-members = ["price-adapter-raw", "price-adapter"]
+members = ["price-adapter-raw", "price-adapter", "bothan-binance"]
 resolver = "2"
 
 [workspace.dependencies]
 price-adapter = { version = "0.1.8", path = "price-adapter" }
 price-adapter-raw = { version = "0.1.8", path = "price-adapter-raw" }
+bothan-binance = { version = "0.1.8", path = "bothan-binance" }

bothan-binance/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "bothan-binance"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+reqwest = { version = "0.11.22", features = ["json"] }
+thiserror = "1.0.56"
+tracing = { version = "0.1.40", features = [] }
+tokio = { version = "1.0.3", features = ["full"] }
+tracing-subscriber = "0.3.17"
+serde = { version = "1.0.196", features = ["std", "derive", "alloc"] }
+serde_json = "1.0.108"
+itertools = "0.12.0"
+tokio-tungstenite = { version = "0.21.0", features = ["native-tls"] }
+futures-util = "0.3.29"
+tokio-util = "0.7.10"
+chrono = "0.4.31"
+rand = "0.8.4"
+dashmap = "5.5.3"
+derive_more = {  version = "1.0.0-beta.6", features = ["full"] }

bothan-binance/examples/dummy_example.rs

@@ -0,0 +1,17 @@
+use tracing_subscriber::fmt::init;
+
+use bothan_binance::BinanceService;
+
+#[tokio::main]
+async fn main() {
+    init();
+
+    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    if let Ok(mut service) = BinanceService::new(None).await {
+        loop {
+            let data = service.get_price_data(&["btcusdt", "ethusdt"]).await;
+            println!("{:?}", data);
+            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+        }
+    }
+}

bothan-binance/src/api.rs

@@ -0,0 +1,3 @@
+pub mod error;
+pub mod types;
+pub mod websocket;

bothan-binance/src/api/error.rs

@@ -0,0 +1,22 @@
+use tokio_tungstenite::tungstenite::{self, http::StatusCode};
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("unknown error")]
+    Unknown,
+
+    #[error("failed to connect with response code {0}")]
+    Connection(StatusCode),
+
+    #[error("failed to parse")]
+    Parse(#[from] serde_json::Error),
+
+    #[error("not connected")]
+    NotConnected(),
+
+    #[error("already connected")]
+    AlreadyConnected(),
+
+    #[error("tungstenite error")]
+    Tungstenite(#[from] tungstenite::Error),
+}

bothan-binance/src/api/types.rs

@@ -0,0 +1,62 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct SuccessResponse {
+    pub result: Option<String>,
+    pub id: u64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct ErrorResponse {
+    pub code: u16,
+    pub msg: String,
+    pub id: String,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "e")]
+pub enum Data {
+    #[serde(rename = "24hrMiniTicker")]
+    MiniTicker(MiniTickerInfo),
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct MiniTickerInfo {
+    #[serde(rename = "E")]
+    pub event_time: u64,
+
+    #[serde(rename = "s")]
+    pub symbol: String,
+
+    #[serde(rename = "c")]
+    pub close_price: String,
+
+    #[serde(rename = "o")]
+    pub open_price: String,
+
+    #[serde(rename = "h")]
+    pub high_price: String,
+
+    #[serde(rename = "l")]
+    pub low_price: String,
+
+    #[serde(rename = "v")]
+    pub base_volume: String,
+
+    #[serde(rename = "q")]
+    pub quote_volume: String,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct StreamResponse {
+    pub stream: String,
+    pub data: Data,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(untagged)]
+pub enum BinanceResponse {
+    Success(SuccessResponse),
+    Error(ErrorResponse),
+    Stream(StreamResponse),
+}

bothan-binance/src/api/websocket.rs

@@ -0,0 +1,106 @@
+use futures_util::stream::{SplitSink, SplitStream};
+use futures_util::{SinkExt, StreamExt};
+use serde_json::json;
+use tokio::net::TcpStream;
+use tokio_tungstenite::tungstenite::http::StatusCode;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream};
+use tracing::warn;
+
+use crate::api::error::Error;
+use crate::api::types::BinanceResponse;
+
+const DEFAULT_URL: &str = "wss://stream.binance.com:9443/stream";
+
+pub struct BinanceWebsocket {
+    url: String,
+    sender: Option<SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>>,
+    receiver: Option<SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>>,
+}
+
+impl BinanceWebsocket {
+    pub fn new(url: &str) -> Self {
+        Self {
+            url: url.to_string(),
+            sender: None,
+            receiver: None,
+        }
+    }
+
+    pub async fn connect(&mut self) -> Result<(), Error> {
+        let (socket, response) = connect_async(&self.url).await?;
+
+        let status = response.status();
+        if StatusCode::is_server_error(&status) || StatusCode::is_client_error(&status) {
+            return Err(Error::Connection(status));
+        }
+
+        let (sender, receiver) = socket.split();
+        self.sender = Some(sender);
+        self.receiver = Some(receiver);
+
+        Ok(())
+    }
+
+    pub async fn disconnect(&mut self) -> Result<(), Error> {
+        let mut sender = self.sender.take().ok_or(Error::NotConnected())?;
+        // Ignore result as we just want to send a close message
+        if sender.send(Message::Close(None)).await.is_err() {
+            warn!("unable to send close frame")
+        }
+
+        self.receiver = None;
+        Ok(())
+    }
+
+    pub async fn subscribe(&mut self, ids: &[&str]) -> Result<(), Error> {
+        let sender = self.sender.as_mut().ok_or(Error::NotConnected())?;
+
+        let stream_ids = ids
+            .iter()
+            .map(|id| format!("{}@miniTicker", id))
+            .collect::<Vec<_>>();
+
+        let payload = json!({
+            "method": "SUBSCRIBE",
+            "params": stream_ids,
+            "id": rand::random::<u32>()
+        });
+        let message = Message::Text(payload.to_string());
+        Ok(sender.send(message).await?)
+    }
+
+    pub async fn unsubscribe(&mut self, ids: &[&str]) -> Result<(), Error> {
+        let sender = self.sender.as_mut().ok_or(Error::NotConnected())?;
+        let stream_ids = ids
+            .iter()
+            .map(|id| format!("{}@miniTicker", id))
+            .collect::<Vec<_>>();
+        let payload = json!({
+            "method": "UNSUBSCRIBE",
+            "params": stream_ids,
+            "id": rand::random::<u32>()
+        });
+
+        let message = Message::Text(payload.to_string());
+
+        sender.send(message).await?;
+
+        Ok(())
+    }
+
+    pub async fn next(&mut self) -> Result<BinanceResponse, Error> {
+        let receiver = self.receiver.as_mut().ok_or(Error::NotConnected())?;
+        if let Some(Ok(msg)) = receiver.next().await {
+            Ok(serde_json::from_str::<BinanceResponse>(&msg.to_string())?)
+        } else {
+            Err(Error::Unknown)
+        }
+    }
+}
+
+impl Default for BinanceWebsocket {
+    fn default() -> Self {
+        Self::new(DEFAULT_URL)
+    }
+}

bothan-binance/src/cache.rs

@@ -0,0 +1,8 @@
+// TODO: move into own crate
+
+pub use dash_map::Cache;
+pub use error::Error;
+
+pub mod dash_map;
+pub mod error;
+mod types;

bothan-binance/src/cache/dash_map.rs

@@ -0,0 +1,108 @@
+use std::ops::Sub;
+use std::sync::Arc;
+
+use dashmap::{DashMap, DashSet};
+use tokio::select;
+use tokio::sync::mpsc::Sender;
+use tokio::time::{interval, Duration, Instant};
+use tokio_util::sync::CancellationToken;
+use tracing::info;
+
+use crate::cache::error::Error;
+use crate::cache::types::{StoredPriceData, DEFAULT_TIMEOUT};
+use crate::types::PriceData;
+
+pub struct Cache {
+    price_map: Arc<DashMap<String, StoredPriceData>>,
+    subscription_map: Arc<DashSet<String>>,
+    token: CancellationToken,
+}
+
+impl Drop for Cache {
+    fn drop(&mut self) {
+        self.token.cancel()
+    }
+}
+
+impl Cache {
+    pub fn new(sender: Sender<Vec<String>>) -> Self {
+        let price_map = Arc::new(DashMap::<String, StoredPriceData>::new());
+        let subscription_map = Arc::new(DashSet::<String>::new());
+        let token = CancellationToken::new();
+
+        let cloned_token = token.clone();
+        let cloned_price_map = price_map.clone();
+        let cloned_subscription_map = subscription_map.clone();
+
+        tokio::spawn(async move {
+            let mut interval = interval(Duration::from_secs(10));
+            loop {
+                select! {
+                    _ = interval.tick() => {
+                        let keys = cloned_price_map.iter().filter_map(|r| {
+                            let (k, v) = r.pair();
+                            if check_timeout(v.last_used) {
+                                cloned_price_map.remove(k);
+                                cloned_subscription_map.remove(k)
+                            } else {
+                                None
+                            }
+                        }).collect::<Vec<String>>();
+                        if !keys.is_empty() {
+                            info!("Removing unused keys: {:?}", keys);
+                            let _res = sender.send(keys).await;
+                        }
+                    }
+                    _ = cloned_token.cancelled() => {
+                        break
+                    }
+                }
+            }
+        });
+
+        Self {
+            price_map,
+            subscription_map,
+            token,
+        }
+    }
+
+    pub fn set_pending(&self, id: String) {
+        self.subscription_map.insert(id.to_ascii_lowercase());
+    }
+
+    pub fn set_data(&self, id: String, data: PriceData) {
+        if let Some((k, mut v)) = self.price_map.remove(&id) {
+            v.data = data;
+            self.price_map.insert(k, v);
+        } else {
+            self.price_map.insert(
+                id.to_ascii_lowercase(),
+                StoredPriceData {
+                    data,
+                    last_used: Instant::now(),
+                },
+            );
+        }
+    }
+
+    pub fn get(&self, id: &str) -> Result<PriceData, Error> {
+        if let Some(r) = self.price_map.get(&id.to_ascii_lowercase()) {
+            return Ok(r.data.clone());
+        }
+
+        if self.subscription_map.contains(&id.to_ascii_lowercase()) {
+            Err(Error::Invalid)
+        } else {
+            Err(Error::DoesNotExist)
+        }
+    }
+
+    pub fn keys(&self) -> Vec<String> {
+        self.price_map.iter().map(|r| r.key().clone()).collect()
+    }
+}
+
+fn check_timeout(last_used: Instant) -> bool {
+    Instant::now().sub(last_used) > Duration::from_secs(DEFAULT_TIMEOUT)
+}

bothan-binance/src/cache/error.rs

@@ -0,0 +1,4 @@
+pub enum Error {
+    DoesNotExist,
+    Invalid,
+}

bothan-binance/src/cache/types.rs

@@ -0,0 +1,10 @@
+use crate::types::PriceData;
+use tokio::time::Instant;
+
+pub const DEFAULT_TIMEOUT: u64 = 6000;
+
+#[derive(Debug, Clone)]
+pub struct StoredPriceData {
+    pub data: PriceData,
+    pub last_used: Instant,
+}