Merge pull request #46 from butaneprotocol/more-sources

More sources

Author: SupernaviX

Cargo.lock

@@ -24,7 +24,7 @@ cron = "0.15"
 dashmap = "6"
 ed25519 = { version = "2.2", features = ["pkcs8", "pem"] }
 ed25519-dalek = { version = "2.1" }
-frost-ed25519 = { version = "2.0.0-rc.0", features = ["serialization"] }
+frost-ed25519 = { version = "2.1", features = ["serialization"] }
 futures = "0.3"
 hex = "0.4.3"
 kupon = { git = "https://github.com/SundaeSwap-finance/kupon", rev = "ccc0b37" }
@@ -47,8 +47,8 @@ serde = "1"
 serde_json = "1"
 temp-env = "0.3"
 tokio = { version = "1", features = ["full"] }
+tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots" ]}
 tokio-util = { version = "0.7", features = ["full"] }
-tokio-websockets = { version = "0.11", features = ["client", "rustls-webpki-roots", "ring", "rand"] }
 tonic = "0.12"
 tracing = "0.1.40"
 tracing-opentelemetry = "0.28"

Cargo.toml

@@ -185,12 +185,6 @@ binance:
     - token: BTC
       unit: USDT
       stream: btcusdt@ticker
-    - token: EUR
-      unit: USDT
-      stream: eurusdt@ticker
-    - token: PAXG
-      unit: USDT
-      stream: paxgusdt@ticker
     - token: POL
       unit: USDT
       stream: polusdt@ticker
@@ -231,6 +225,46 @@ coinbase:
     - token: USDT
       unit: USD
       product_id: USDT-USD
+crypto_com:
+  tokens:
+    - token: ADA
+      unit: USD
+      stream: ticker.ADA_USD
+    - token: BTC
+      unit: USD
+      stream: ticker.BTC_USD
+    - token: EUR
+      unit: USDT
+      stream: ticker.EUR_USDT
+    - token: PAXG
+      unit: USD
+      stream: ticker.PAXG_USD
+    - token: POL
+      unit: USD
+      stream: ticker.POL_USD
+    - token: SOL
+      unit: USD
+      stream: ticker.SOL_USD
+    - token: USDT
+      unit: USD
+      stream: ticker.USDT_USD
+kucoin:
+  tokens:
+    - token: ADA
+      unit: USDT
+      symbol: ADA-USDT
+    - token: BTC
+      unit: USDT
+      symbol: BTC-USDT
+    - token: PAXG
+      unit: USDT
+      symbol: PAXG-USDT
+    - token: POL
+      unit: USDT
+      symbol: POL-USDT
+    - token: SOL
+      unit: USDT
+      symbol: SOL-USDT
 maestro:
   tokens:
     - token: DJED

config.base.yaml

@@ -40,7 +40,9 @@ struct RawOracleConfig {
     pub binance: BinanceConfig,
     pub bybit: ByBitConfig,
     pub coinbase: CoinbaseConfig,
+    pub crypto_com: CryptoComConfig,
     pub fxratesapi: FxRatesApiConfig,
+    pub kucoin: KucoinConfig,
     pub maestro: MaestroConfig,
     pub okx: OkxConfig,
     pub sundaeswap: SundaeSwapConfig,
@@ -72,7 +74,9 @@ pub struct OracleConfig {
     pub bybit: ByBitConfig,
     pub binance: BinanceConfig,
     pub coinbase: CoinbaseConfig,
+    pub crypto_com: CryptoComConfig,
     pub fxratesapi: FxRatesApiConfig,
+    pub kucoin: KucoinConfig,
     pub maestro: MaestroConfig,
     pub okx: OkxConfig,
     pub sundaeswap: SundaeSwapConfig,
@@ -177,7 +181,9 @@ impl TryFrom<RawOracleConfig> for OracleConfig {
             binance: raw.binance,
             bybit: raw.bybit,
             coinbase: raw.coinbase,
+            crypto_com: raw.crypto_com,
             fxratesapi: raw.fxratesapi,
+            kucoin: raw.kucoin,
             maestro: raw.maestro,
             okx: raw.okx,
             sundaeswap: raw.sundaeswap,
@@ -305,13 +311,37 @@ pub struct CoinbaseTokenConfig {
     pub product_id: String,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct CryptoComConfig {
+    pub tokens: Vec<CryptoComTokenConfig>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct CryptoComTokenConfig {
+    pub token: String,
+    pub unit: String,
+    pub stream: String,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct FxRatesApiConfig {
     pub cron: String,
     pub currencies: Vec<String>,
     pub base: String,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct KucoinConfig {
+    pub tokens: Vec<KucoinTokenConfig>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct KucoinTokenConfig {
+    pub token: String,
+    pub unit: String,
+    pub symbol: String,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct MaestroConfig {
     pub tokens: Vec<MaestroTokenConfig>,

src/config.rs

@@ -24,7 +24,9 @@ use crate::{
         binance::BinanceSource,
         bybit::ByBitSource,
         coinbase::CoinbaseSource,
+        crypto_com::CryptoComSource,
         fxratesapi::FxRatesApiSource,
+        kucoin::KucoinSource,
         maestro::MaestroSource,
         minswap::MinswapSource,
         okx::OkxSource,
@@ -71,6 +73,8 @@ impl PriceAggregator {
             SourceAdapter::new(BinanceSource::new(&config), &config),
             SourceAdapter::new(ByBitSource::new(&config), &config),
             SourceAdapter::new(CoinbaseSource::new(&config), &config),
+            SourceAdapter::new(CryptoComSource::new(&config), &config),
+            SourceAdapter::new(KucoinSource::new(&config), &config),
             SourceAdapter::new(MinswapSource::new(&config)?, &config),
             SourceAdapter::new(OkxSource::new(&config)?, &config),
             SourceAdapter::new(SpectrumSource::new(&config)?, &config),

src/price_aggregator.rs

@@ -1,11 +1,11 @@
 use std::{collections::BTreeMap, str::FromStr, time::Duration};
 
 use anyhow::{anyhow, Result};
-use futures::{future::BoxFuture, FutureExt, SinkExt, StreamExt};
+use futures::{future::BoxFuture, FutureExt, StreamExt};
 use rust_decimal::Decimal;
 use serde::Deserialize;
 use tokio::time::timeout;
-use tokio_websockets::{ClientBuilder, Message};
+use tokio_tungstenite::connect_async;
 use tracing::{trace, warn};
 
 use crate::{
@@ -53,9 +53,8 @@ impl BinanceSource {
             .map(|k| k.as_str())
             .collect::<Vec<&str>>()
             .join("/");
-        let url = format!("{BASE_URL}?streams={streams}");
-        let uri = url.try_into()?;
-        let (mut stream, _) = ClientBuilder::from_uri(uri).connect().await?;
+        let uri = format!("{BASE_URL}?streams={streams}");
+        let (mut stream, _) = connect_async(uri).await?;
         trace!("Connected to binance!");
 
         let connection_timeout = Duration::from_secs(60);
@@ -67,18 +66,12 @@ impl BinanceSource {
                     continue;
                 }
             };
-            if let Some(contents) = message.as_text() {
-                if let Err(err) = self.process_binance_message(contents, sink) {
-                    warn!("Unexpected error updating binance data: {:?}", err);
-                }
-            } else if message.is_ping() {
-                let data = message.into_payload();
-                trace!("Ping received from binance: {:?}", data);
-                if let Err(err) = stream.send(Message::pong(data)).await {
-                    warn!("Unexpected error replying to binance ping: {}", err);
-                }
-            } else {
-                warn!("Unexpected response from binance: {:?}", message);
+            if !message.is_text() {
+                continue;
+            }
+            let contents = message.into_text()?;
+            if let Err(err) = self.process_binance_message(&contents, sink) {
+                warn!("Unexpected error updating binance data: {:?}", err);
             }
         }
 

src/sources/binance.rs

@@ -1,6 +1,6 @@
 use std::str::FromStr;
 
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Result};
 use dashmap::DashMap;
 use futures::{future::BoxFuture, FutureExt, SinkExt, StreamExt};
 use rust_decimal::Decimal;
@@ -9,7 +9,7 @@ use tokio::{
     select,
     time::{sleep, timeout, Duration},
 };
-use tokio_websockets::{ClientBuilder, Message};
+use tokio_tungstenite::{connect_async, tungstenite::Message};
 
 use crate::config::OracleConfig;
 
@@ -68,8 +68,7 @@ impl ByBitSource {
     }
 
     async fn query_impl(&self, sink: &PriceSink) -> Result<()> {
-        let uri = URL.try_into()?;
-        let (mut stream, _) = ClientBuilder::from_uri(uri).connect().await?;
+        let (mut stream, _) = connect_async(URL).await?;
 
         let subscribe_timeout = Duration::from_secs(60);
         timeout(
@@ -107,53 +106,49 @@ impl ByBitSource {
         let consumer = async move {
             while let Some(result) = stream.next().await {
                 let message = result?;
-                match message.as_text() {
-                    Some(content) => {
-                        let response: ByBitResponse = serde_json::from_str(content)?;
-                        let data = match response {
-                            ByBitResponse::StatusResponse { success, ret_msg } => {
-                                if !success {
-                                    return Err(anyhow!("Error subscribing: {}", ret_msg));
-                                }
-                                continue;
-                            }
-                            ByBitResponse::TickerResponse { data } => data,
-                        };
-                        let Some(mut info) = self.stream_info.get_mut(&data.symbol) else {
-                            continue;
-                        };
-                        let Some(value) = data
-                            .mark_price
-                            .and_then(|x| Decimal::from_str(&x).ok())
-                            .or(info.last_value)
-                        else {
-                            continue;
-                        };
-                        info.last_value = Some(value);
-
-                        let Some(volume) = data
-                            .volume_24h
-                            .and_then(|x| Decimal::from_str(&x).ok())
-                            .or(info.last_volume)
-                        else {
-                            continue;
-                        };
-                        info.last_volume = Some(volume);
-
-                        let price_info = PriceInfo {
-                            token: info.token.clone(),
-                            unit: info.unit.clone(),
-                            value,
-                            reliability: volume,
-                        };
-                        sink.send(price_info)?;
-                    }
-                    None => {
-                        return Err(anyhow!("Unexpected response {:?}", message));
+                if !message.is_text() {
+                    continue;
+                }
+                let response: ByBitResponse = serde_json::from_str(&message.into_text()?)?;
+                let data = match response {
+                    ByBitResponse::StatusResponse { success, ret_msg } => {
+                        if !success {
+                            bail!("Error subscribing: {}", ret_msg);
+                        }
+                        continue;
                     }
+                    ByBitResponse::TickerResponse { data } => data,
+                };
+                let Some(mut info) = self.stream_info.get_mut(&data.symbol) else {
+                    continue;
+                };
+                let Some(value) = data
+                    .mark_price
+                    .and_then(|x| Decimal::from_str(&x).ok())
+                    .or(info.last_value)
+                else {
+                    continue;
+                };
+                info.last_value = Some(value);
+
+                let Some(volume) = data
+                    .volume_24h
+                    .and_then(|x| Decimal::from_str(&x).ok())
+                    .or(info.last_volume)
+                else {
+                    continue;
+                };
+                info.last_volume = Some(volume);
+
+                let price_info = PriceInfo {
+                    token: info.token.clone(),
+                    unit: info.unit.clone(),
+                    value,
+                    reliability: volume,
                 };
+                sink.send(price_info)?;
             }
-            Err(anyhow!("ByBit stream has closed"))
+            bail!("ByBit stream has closed")
         };
 
         select! {

src/sources/bybit.rs

@@ -1,11 +1,11 @@
 use std::{collections::BTreeMap, str::FromStr, time::Duration};
 
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, Context, Result};
 use futures::{future::BoxFuture, FutureExt, SinkExt, StreamExt};
 use rust_decimal::Decimal;
 use serde::{Deserialize, Serialize};
 use tokio::time::timeout;
-use tokio_websockets::{ClientBuilder, Message};
+use tokio_tungstenite::{connect_async, tungstenite::Message};
 use tracing::{trace, warn};
 
 use crate::config::{CoinbaseTokenConfig, OracleConfig};
@@ -46,10 +46,8 @@ impl CoinbaseSource {
     async fn query_impl(&self, sink: &PriceSink) -> Result<()> {
         trace!("Connecting to coinbase");
         let connection_timeout = Duration::from_secs(60);
-        let uri = URL.try_into()?;
 
-        let (mut stream, _) =
-            timeout(connection_timeout, ClientBuilder::from_uri(uri).connect()).await??;
+        let (mut stream, _) = timeout(connection_timeout, connect_async(URL)).await??;
 
         let request = CoinbaseRequest::Subscribe {
             product_ids: self.products.keys().cloned().collect(),
@@ -143,9 +141,7 @@ impl TryFrom<Message> for CoinbaseResponse {
     type Error = anyhow::Error;
 
     fn try_from(value: Message) -> Result<Self> {
-        match value.as_text() {
-            Some(msg) => Ok(serde_json::from_str(msg)?),
-            None => Err(anyhow::anyhow!("Unexpected response: {:?}", value)),
-        }
+        let text = value.into_text().context("unexpected response")?;
+        Ok(serde_json::from_str(&text)?)
     }
 }