Merge branch 'main' into update-htx

Cargo.toml

@@ -10,10 +10,12 @@ members = [
     "bothan-api/server",
     "bothan-api/server-cli",
     "bothan-binance",
+    "bothan-coinbase",
     "bothan-coingecko",
     "bothan-core",
     "bothan-htx",
     "bothan-kraken",
+    "bothan-okx",
 ]
 exclude = ["bothan-api", "bothan-api-proxy"]
 resolver = "2"
@@ -46,10 +48,10 @@ bothan-api = { path = "bothan-api/server" }
 
 bothan-binance = { path = "bothan-binance" }
 #bothan-bybit = { path = "bothan-bybit" }
-#bothan-coinbase = { path = "bothan-coinbase" }
+bothan-coinbase = { path = "bothan-coinbase" }
 bothan-coingecko = { path = "bothan-coingecko" }
 #bothan-coinmarketcap = { path = "bothan-coinmarketcap" }
 #bothan-cryptocompare = { path = "bothan-cryptocompare" }
 bothan-htx = { path = "bothan-htx" }
 bothan-kraken = { path = "bothan-kraken" }
-#bothan-okx = { path = "bothan-okx" }
+bothan-okx = { path = "bothan-okx" }

bothan-binance/src/api/error.rs

@@ -5,8 +5,8 @@ pub enum ConnectionError {
     #[error("failed to connect to endpoint {0}")]
     ConnectionFailure(#[from] tungstenite::Error),
 
-    #[error("received unsuccessful HTTP response: {0}")]
-    UnsuccessfulHttpResponse(tungstenite::http::StatusCode),
+    #[error("received unsuccessful WebSocket response: {0}")]
+    UnsuccessfulWebSocketResponse(tungstenite::http::StatusCode),
 }
 
 #[derive(Debug, thiserror::Error)]

bothan-binance/src/api/msgs.rs

@@ -1,26 +1,26 @@
 use serde::{Deserialize, Serialize};
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct SuccessResponse {
     pub result: Option<String>,
     pub id: u64,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct ErrorResponse {
     pub code: u16,
     pub msg: String,
     pub id: u64,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(tag = "e")]
 pub enum Data {
     #[serde(rename = "24hrMiniTicker")]
     MiniTicker(MiniTickerInfo),
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct MiniTickerInfo {
     #[serde(rename = "E")]
     pub event_time: i64,
@@ -47,13 +47,13 @@ pub struct MiniTickerInfo {
     pub quote_volume: String,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct StreamResponse {
     pub stream: String,
     pub data: Data,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum BinanceResponse {
     Success(SuccessResponse),

bothan-binance/src/api/websocket.rs

@@ -41,7 +41,7 @@ impl BinanceWebSocketConnector {
         // Check the HTTP response status.
         let status = resp.status();
         if status.as_u16() >= 400 {
-            return Err(ConnectionError::UnsuccessfulHttpResponse(status));
+            return Err(ConnectionError::UnsuccessfulWebSocketResponse(status));
         }
 
         // Return the WebSocket connection.

bothan-bybit/src/api/types/ticker.rs

@@ -67,15 +67,15 @@ pub struct SpotTicker {
 }
 
 /// Represents different types of tickers.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(untagged, rename_all = "camelCase")]
 pub enum Tickers {
     /// A list of spot tickers.
     Spot(Vec<SpotTicker>),
 }
 
 /// Represents the response containing tickers.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct TickersResponse {
     /// The category of the tickers.
     #[serde(default)]

bothan-coinbase/Cargo.toml

@@ -11,11 +11,15 @@ futures-util = "0.3.29"
 async-trait = { workspace = true }
 bothan-core = { workspace = true }
 chrono = { workspace = true }
+rand = { workspace = true }
+rust_decimal = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
-rand = { workspace = true }
-serde = { workspace = true }
-serde_json = { workspace = true }
+
+[dev-dependencies]
+ws-mock = { git = "https://github.com/bandprotocol/ws-mock.git", branch = "master" }

bothan-coinbase/examples/coinbase_basic.rs

@@ -1,20 +1,39 @@
+use std::time::Duration;
+
+use tokio::time::sleep;
 use tracing_subscriber::fmt::init;
 
-use bothan_coinbase::service::builder::CoinbaseServiceBuilder;
-use bothan_core::service::Service;
+use bothan_coinbase::{CoinbaseWorkerBuilder, CoinbaseWorkerBuilderOpts};
+use bothan_core::registry::Registry;
+use bothan_core::store::SharedStore;
+use bothan_core::worker::{AssetWorker, AssetWorkerBuilder};
 
 #[tokio::main]
 async fn main() {
     init();
+    let path = std::env::current_dir().unwrap();
+    let registry = Registry::default().validate().unwrap();
+    let store = SharedStore::new(registry, path.as_path()).await.unwrap();
+
+    let worker_store = store.create_worker_store(CoinbaseWorkerBuilder::worker_name());
+    let opts = CoinbaseWorkerBuilderOpts::default();
+
+    let worker = CoinbaseWorkerBuilder::new(worker_store, opts)
+        .build()
+        .await
+        .unwrap();
+
+    worker
+        .set_query_ids(vec!["BTC-USD".to_string(), "ETH-USD".to_string()])
+        .await
+        .unwrap();
+
+    sleep(Duration::from_secs(2)).await;
 
-    let service = CoinbaseServiceBuilder::default().build().await;
-    if let Ok(mut service) = service {
-        loop {
-            let data = service
-                .get_price_data(&["BTC-USD", "ETH-USD", "USDT-USD"])
-                .await;
-            println!("{:?}", data);
-            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
-        }
+    loop {
+        let btc_data = worker.get_asset("BTC-USD").await;
+        let eth_data = worker.get_asset("ETH-USD").await;
+        println!("{:?} {:?}", btc_data, eth_data);
+        sleep(Duration::from_secs(5)).await;
     }
 }

bothan-coinbase/src/api.rs

@@ -1,3 +1,7 @@
+pub use error::{ConnectionError, MessageError, SendError};
+pub use types::channels::ticker::Ticker;
+pub use websocket::{CoinbaseWebSocketConnection, CoinbaseWebSocketConnector};
+
 pub mod error;
 pub mod types;
 pub mod websocket;

bothan-coinbase/src/api/error.rs

@@ -1,25 +1,32 @@
-use tokio_tungstenite::tungstenite::{self, http::StatusCode};
+use tokio_tungstenite::tungstenite;
 
-/// Represents the various errors that can occur.
 #[derive(Debug, thiserror::Error)]
-pub enum Error {
-    /// Connection failure with a response code.
-    #[error("failed to connect with response code {0}")]
-    ConnectionFailure(StatusCode),
+pub enum ConnectionError {
+    #[error("failed to connect to endpoint {0}")]
+    ConnectionFailure(#[from] tungstenite::Error),
 
-    /// Error parsing JSON.
-    #[error("failed to parse")]
-    Parse(#[from] serde_json::Error),
+    #[error("received unsuccessful WebSocket response: {0}")]
+    UnsuccessfulWebSocketResponse(tungstenite::http::StatusCode),
+}
 
-    /// Error from the Tungstenite library.
-    #[error("tungstenite error")]
-    Tungstenite(#[from] tungstenite::Error),
+#[derive(Debug, thiserror::Error)]
+pub enum MessageError {
+    #[error("failed to parse message")]
+    Parse(#[from] serde_json::Error),
 
-    /// Channel closed unexpectedly.
     #[error("channel closed")]
     ChannelClosed,
 
-    /// Received an unsupported message.
     #[error("unsupported message")]
     UnsupportedMessage,
 }
+
+#[derive(Debug, thiserror::Error)]
+pub enum SendError {
+    #[error(transparent)]
+    Tungstenite(#[from] tungstenite::Error),
+
+    /// Indicates a failure to parse a message.
+    #[error("failed to parse")]
+    Parse(#[from] serde_json::Error),
+}

bothan-coinbase/src/api/types.rs

@@ -10,7 +10,7 @@ pub mod request;
 pub const DEFAULT_URL: &str = "wss://ws-feed.exchange.coinbase.com";
 
 /// Represents the possible responses from the Coinbase feed.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case", tag = "type")]
 pub enum CoinbaseResponse {
     /// A ticker update.

bothan-coinbase/src/api/types/channels/ticker.rs

@@ -1,7 +1,7 @@
 use serde::{Deserialize, Serialize};
 
 /// Represents a ticker with various market data.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct Ticker {
     /// The sequence number of the ticker.
     pub sequence: usize,

bothan-coinbase/src/api/types/request.rs

@@ -25,7 +25,7 @@ pub struct Request {
 }
 
 /// Represents a subscription channel.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct SubscriptionChannel {
     /// The name of the channel.
     pub name: String,
@@ -34,7 +34,7 @@ pub struct SubscriptionChannel {
 }
 
 /// Represents the current subscriptions.
-#[derive(Clone, Debug, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub struct Subscriptions {
     /// The list of subscribed channels.
     pub channels: Vec<SubscriptionChannel>,