cleans up run streams and removes unused code

Author: githubsands

Cargo.lock

@@ -60,7 +60,6 @@ fn main() {
                 .takes_value(true),
         )
         .get_matches();
-
     let config_path = PathBuf::from(matches.value_of("config").unwrap_or("/etc/config.yaml"));
     info!("config path given: {:?}", config_path);
     let file = PathBuf::from(config_path);

bin/orderbook-quoter-server/src/main.rs

@@ -16,3 +16,4 @@ config.workspace = true
 exchange.workspace = true
 market-objects.workspace = true
 quoter-errors.workspace = true
+anyhow = "1.0.99"

depth-driver/Cargo.toml

@@ -2,16 +2,11 @@ use std::cell::RefCell;
 use std::rc::Rc;
 
 use crossbeam_channel::Sender;
+use tracing::warn;
 
-use tokio::sync::watch::{
-    channel as watchChannel, Receiver as watchReceiver, Sender as watchSender,
-};
+use tokio::sync::watch::{channel as watchChannel, Receiver as watchReceiver};
+use tokio::task;
 use tokio_context::context::Context;
-use tracing::info;
-
-use futures::pin_mut;
-use futures::stream::FuturesOrdered;
-use futures::StreamExt;
 
 use config::ExchangeConfig;
 use exchange::exchange::Exchange;
@@ -20,30 +15,31 @@ use quoter_errors::{ErrorHotPath, ErrorInitialState};
 
 pub struct DepthDriver {
     exchanges: Vec<Rc<RefCell<Exchange>>>,
-    orderbook_snapshot_trigger: watchReceiver<()>,
-    exchange_snapshot_trigger: watchSender<()>,
+    // todo: implement syncs
+    // orderbook_snapshot_sync: watchReceiver<()>,
+    // exchange_snapshot_sync: watchSender<()>,
 }
 
 impl DepthDriver {
     pub fn new(
         exchange_configs: &Vec<ExchangeConfig>,
         depths_producer: Sender<DepthUpdate>,
-        orderbook_snapshot_trigger: watchReceiver<()>,
+        _: watchReceiver<()>,
     ) -> Result<DepthDriver, ErrorInitialState> {
         let mut exchanges: Vec<Rc<RefCell<Exchange>>> = Vec::new();
-        let (snapshot_trigger, inner_snapshot_consumer) = watchChannel(());
+        let (_, inner_snapshot_consumer) = watchChannel(());
         for exchange_config in exchange_configs {
             let exchange = Exchange::new(
                 exchange_config,
                 depths_producer.clone(),
                 inner_snapshot_consumer.clone(),
             );
-            exchanges.push(Rc::new(RefCell::new(exchange)));
+            exchanges.push(Rc::new(RefCell::new(exchange?)));
         }
         Ok(DepthDriver {
             exchanges,
-            orderbook_snapshot_trigger,
-            exchange_snapshot_trigger: snapshot_trigger,
+            // orderbook_snapshot_trigger,
+            // exchange_snapshot_trigger: snapshot_trigger,
         })
     }
 
@@ -65,6 +61,7 @@ impl DepthDriver {
         Ok(())
     }
 
+    // todo: this can be done on multiplie threads?
     pub async fn build_orderbook(&mut self) -> Result<(), ErrorInitialState> {
         for exchange in self.exchanges.iter_mut() {
             exchange.as_ref().borrow_mut().run_snapshot().await?;
@@ -79,50 +76,47 @@ impl DepthDriver {
         Ok(())
     }
 
-    // run_streams is the main driver the entire program that continously polls websocket depths from
-    // N exchanges and also handles orderbook rebuilds
-    pub async fn run_streams(&mut self, ctx: &mut Context) -> Result<(), ErrorHotPath> {
-        info!("running streams");
-        let inner_trigger = &mut self.orderbook_snapshot_trigger;
-        let mut exchanges: Vec<*mut Exchange> = vec![];
-        // let mut streams = FuturesOrdered::new();
-        for exchange in &mut self.exchanges {
-            exchanges.push(exchange.as_ref().as_ptr())
-        }
-
-        /*
-        unsafe {
-            for stream in exchanges {
-                streams.push_back(stream.as_mut().unwrap().stream_depths())
-            }
-        }
-        */
-
-        // pin_mut!(streams);
-
-        loop {
-            info!("waiting for future");
-            tokio::select! {
-                    // TODO: This may not rebuild orderbook correctly but the trigger currently is not
-                    // implemented within the orderbook. This future handles orderbook rebuilds.
-                        // _ = inner_trigger.changed()=> {
-                            // send snapshot trigger to our N exchange streams
-                            // this sets a streams run function to grab a http depth load from multiplie exchanges, and buffers
-                            // the websocket depths.
-                            // let _ = self.exchange_snapshot_trigger.send(());
-                        _ = unsafe {exchanges[0].as_mut().unwrap().stream_depths() } => {}
-                        _ = unsafe {exchanges[1].as_mut().unwrap().stream_depths() } => {}
-                        _ = ctx.done() => {
-                            return Ok(());
+    // todo: think about implementing context for graceful shutdowns
+    pub async fn run_streams(&mut self, _: &mut Context) -> Result<(), ErrorHotPath> {
+        let local = task::LocalSet::new();
+        local
+            .run_until(async move {
+                for (idx, exchange) in self.exchanges.iter_mut().enumerate() {
+                    let exchange = exchange.clone();
+                    tokio::task::spawn_local(async move {
+                        loop {
+                            match exchange.as_ref().borrow_mut().stream_depths().await {
+                                Ok(_) => continue,
+                                Err(err) => {
+                                    warn!(
+                                        "received error {} for exchange when streaming depths{}",
+                                        err, idx
+                                    );
+                                    // todo: don't return but reconnect - we don't want to end the
+                                    //       entire system if one websocket fails
+                                    return Err(ErrorHotPath::OrderBookDealSendFail);
+                                }
+                            }
                         }
-            }
-        }
+                        // even though this code is unreachable we need to infer
+                        // the return type
+                        Ok::<(), ErrorHotPath>(())
+                    });
+                }
+            })
+            .await;
+        Ok(())
     }
 
-    pub async fn close_exchanges(&mut self) {
+    // todo: update the error here
+    pub async fn close_exchanges(&mut self) -> Result<(), ErrorHotPath> {
         for exchange in &mut self.exchanges {
             let mut exchange = exchange.borrow_mut();
-            exchange.close().await
+            exchange
+                .close()
+                .await
+                .map_err(|_| ErrorHotPath::OrderBookDealSendFail)?;
         }
+        Ok(())
     }
 }

depth-driver/src/lib.rs

@@ -89,7 +89,7 @@ impl Exchange {
 
     pub async fn stream_depths(&mut self) -> Result<(), ErrorHotPath> {
         info!("streaming depths");
-        self.inner.run().await;
+        self.inner.run().await?;
         Ok(())
     }
 

exchange/src/exchange.rs

@@ -1,5 +1,4 @@
 use std::error::Error;
-use std::fmt;
 use std::{pin::Pin, task::Poll};
 
 use serde_json::from_str;
@@ -135,9 +134,9 @@ impl ExchangeStream {
                 Ok(mut depths) => {
                     while let Some(depth) = depths.next() {
                         self.buffer.push(depth);
-                        // we must keep processing  snapshot depths and depths from the websocket
+                        // we must keep processing snapshot depths and depths from the websocket
                         // but this time the websocket depths are stored in their own buffer
-                        // to be sequenced
+                        // to be sequenced aftr snapshot depths are processed
                         self.next().await;
                     }
                     success = true;
@@ -174,9 +173,6 @@ impl ExchangeStream {
                             Ok(mut depths) => {
                             while let Some(depth) = depths.next() {
                                 self.buffer.push(depth);
-                                // we must keep processing snapshot depths and depths from the websocket
-                                // but this time the websocket depths are stored in their own buffer
-                                // to be sequenced
                                 self.next().await;
                             }
                                 break
@@ -249,7 +245,7 @@ impl ExchangeStream {
             impl Iterator<Item = DepthUpdate>,
             impl Iterator<Item = DepthUpdate>,
         ),
-        Box<dyn Error + Sync + Send + 'static>,
+        orderbook_snapshot_error,
     > {
         let req_builder = self
             .http_client
@@ -328,6 +324,8 @@ impl ExchangeStream {
 
 #[derive(Debug)]
 pub enum WSStreamState {
+    // todo: collapse these errors into 1 Faulty state or do something differently
+    // -- this is our async state not a error
     WSError(tokio_tungstenite::tungstenite::Error),
     SenderError,
     FailedStream,
@@ -459,7 +457,6 @@ impl Stream for ExchangeStream {
                 }
             }
         }
-        info!("streaming -- here 1");
         return Poll::Ready(Some(WSStreamState::WaitingForDepth));
     }
 }