feat: add protobuf integration (#198)

* Added protobuf integration

* feat: protobuf impl refactor

* fix: pr suggestions

---------

Co-authored-by: Rok Černič <rok.cernic@gmail.com>

Author: levnikolaevich

Cargo.toml

@@ -71,3 +71,4 @@ rust_decimal_macros = { version = "1.29.1" }
 bytes = { version = "1.5.0" }
 spin_sleep = { version = "1.3.0 "}
 criterion = { version = "0.5.1" }
+prost = { version = "0.12.4" }

barter-data/Cargo.toml

@@ -53,3 +53,4 @@ derive_more = { workspace = true }
 itertools = { workspace = true }
 vecmap-rs = { workspace = true }
 fnv = { workspace = true }
+prost = { workspace = true }

barter-data/src/exchange/binance/futures/mod.rs

@@ -1,11 +1,15 @@
 use self::liquidation::BinanceLiquidation;
 use super::{Binance, ExchangeServer};
 use crate::{
-    ExchangeWsStream, NoInitialSnapshots,
+    NoInitialSnapshots,
     exchange::{
         StreamSelector,
-        binance::futures::l2::{
-            BinanceFuturesUsdOrderBooksL2SnapshotFetcher, BinanceFuturesUsdOrderBooksL2Transformer,
+        binance::{
+            BinanceWsStream,
+            futures::l2::{
+                BinanceFuturesUsdOrderBooksL2SnapshotFetcher,
+                BinanceFuturesUsdOrderBooksL2Transformer,
+            },
         },
     },
     instrument::InstrumentData,
@@ -46,15 +50,15 @@ where
     Instrument: InstrumentData,
 {
     type SnapFetcher = BinanceFuturesUsdOrderBooksL2SnapshotFetcher;
-    type Stream = ExchangeWsStream<BinanceFuturesUsdOrderBooksL2Transformer<Instrument::Key>>;
+    type Stream = BinanceWsStream<BinanceFuturesUsdOrderBooksL2Transformer<Instrument::Key>>;
 }
 
 impl<Instrument> StreamSelector<Instrument, Liquidations> for BinanceFuturesUsd
 where
     Instrument: InstrumentData,
 {
     type SnapFetcher = NoInitialSnapshots;
-    type Stream = ExchangeWsStream<
+    type Stream = BinanceWsStream<
         StatelessTransformer<Self, Instrument::Key, Liquidations, BinanceLiquidation>,
     >;
 }

barter-data/src/exchange/binance/mod.rs

@@ -11,7 +11,10 @@ use crate::{
     transformer::stateless::StatelessTransformer,
 };
 use barter_instrument::exchange::ExchangeId;
-use barter_integration::{error::SocketError, protocol::websocket::WsMessage};
+use barter_integration::{
+    error::SocketError,
+    protocol::websocket::{WebSocketSerdeParser, WsMessage},
+};
 use std::{fmt::Debug, marker::PhantomData};
 use url::Url;
 
@@ -44,6 +47,9 @@ pub mod subscription;
 /// [`BinanceFuturesUsd`](futures::BinanceFuturesUsd).
 pub mod trade;
 
+/// Convenient type alias for a Binance [`ExchangeWsStream`] using [`WebSocketSerdeParser`].
+pub type BinanceWsStream<Transformer> = ExchangeWsStream<WebSocketSerdeParser, Transformer>;
+
 /// Generic [`Binance<Server>`](Binance) exchange.
 ///
 /// ### Notes
@@ -106,7 +112,7 @@ where
 {
     type SnapFetcher = NoInitialSnapshots;
     type Stream =
-        ExchangeWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BinanceTrade>>;
+        BinanceWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BinanceTrade>>;
 }
 
 impl<Instrument, Server> StreamSelector<Instrument, OrderBooksL1> for Binance<Server>
@@ -115,7 +121,7 @@ where
     Server: ExchangeServer + Debug + Send + Sync,
 {
     type SnapFetcher = NoInitialSnapshots;
-    type Stream = ExchangeWsStream<
+    type Stream = BinanceWsStream<
         StatelessTransformer<Self, Instrument::Key, OrderBooksL1, BinanceOrderBookL1>,
     >;
 }

barter-data/src/exchange/binance/spot/mod.rs

@@ -1,10 +1,12 @@
 use super::{Binance, ExchangeServer};
 use crate::{
-    ExchangeWsStream,
     exchange::{
         StreamSelector,
-        binance::spot::l2::{
-            BinanceSpotOrderBooksL2SnapshotFetcher, BinanceSpotOrderBooksL2Transformer,
+        binance::{
+            BinanceWsStream,
+            spot::l2::{
+                BinanceSpotOrderBooksL2SnapshotFetcher, BinanceSpotOrderBooksL2Transformer,
+            },
         },
     },
     instrument::InstrumentData,
@@ -41,7 +43,7 @@ where
     Instrument: InstrumentData,
 {
     type SnapFetcher = BinanceSpotOrderBooksL2SnapshotFetcher;
-    type Stream = ExchangeWsStream<BinanceSpotOrderBooksL2Transformer<Instrument::Key>>;
+    type Stream = BinanceWsStream<BinanceSpotOrderBooksL2Transformer<Instrument::Key>>;
 }
 
 impl Display for BinanceSpot {

barter-data/src/exchange/bitfinex/mod.rs

@@ -32,7 +32,10 @@ use crate::{
     transformer::stateless::StatelessTransformer,
 };
 use barter_instrument::exchange::ExchangeId;
-use barter_integration::{error::SocketError, protocol::websocket::WsMessage};
+use barter_integration::{
+    error::SocketError,
+    protocol::websocket::{WebSocketSerdeParser, WsMessage},
+};
 use barter_macro::{DeExchange, SerExchange};
 use derive_more::Display;
 use serde_json::json;
@@ -64,6 +67,9 @@ pub mod validator;
 /// See docs: <https://docs.bitfinex.com/docs/ws-general>
 pub const BASE_URL_BITFINEX: &str = "wss://api-pub.bitfinex.com/ws/2";
 
+/// Convenient type alias for a Bitfinex [`ExchangeWsStream`] using [`WebSocketSerdeParser`](barter_integration::protocol::websocket::WebSocketSerdeParser).
+pub type BitfinexWsStream<Transformer> = ExchangeWsStream<WebSocketSerdeParser, Transformer>;
+
 /// [`Bitfinex`] exchange.
 ///
 /// See docs: <https://docs.bitfinex.com/docs/ws-general>
@@ -117,7 +123,7 @@ where
     Instrument: InstrumentData,
 {
     type SnapFetcher = NoInitialSnapshots;
-    type Stream = ExchangeWsStream<
+    type Stream = BitfinexWsStream<
         StatelessTransformer<Self, Instrument::Key, PublicTrades, BitfinexMessage>,
     >;
 }

barter-data/src/exchange/bitfinex/validator.rs

@@ -11,7 +11,7 @@ use barter_integration::{
     error::SocketError,
     protocol::{
         StreamParser,
-        websocket::{WebSocket, WebSocketParser, WsMessage},
+        websocket::{WebSocket, WebSocketSerdeParser, WsMessage},
     },
     subscription::SubscriptionId,
 };
@@ -36,7 +36,7 @@ pub struct BitfinexWebSocketSubValidator;
 
 #[async_trait]
 impl SubscriptionValidator for BitfinexWebSocketSubValidator {
-    type Parser = WebSocketParser;
+    type Parser = WebSocketSerdeParser;
 
     async fn validate<Exchange, Instrument, Kind>(
         mut instrument_map: Map<Instrument>,
@@ -82,7 +82,7 @@ impl SubscriptionValidator for BitfinexWebSocketSubValidator {
                         None => break Err(SocketError::Subscribe("WebSocket stream terminated unexpectedly".to_string()))
                     };
 
-                    match Self::Parser::parse::<BitfinexPlatformEvent>(response) {
+                    match <WebSocketSerdeParser as StreamParser<BitfinexPlatformEvent>>::parse(response) {
                         Some(Ok(response)) => match response.validate() {
                             // Bitfinex server is online
                             Ok(BitfinexPlatformEvent::PlatformStatus(status)) => {

barter-data/src/exchange/bitmex/mod.rs

@@ -14,7 +14,10 @@ use crate::{
     transformer::stateless::StatelessTransformer,
 };
 use barter_instrument::exchange::ExchangeId;
-use barter_integration::{error::SocketError, protocol::websocket::WsMessage};
+use barter_integration::{
+    error::SocketError,
+    protocol::websocket::{WebSocketSerdeParser, WsMessage},
+};
 use derive_more::Display;
 use serde::de::{Error, Unexpected};
 use std::fmt::Debug;
@@ -38,6 +41,9 @@ pub mod subscription;
 /// Public trade types for [`Bitmex`].
 pub mod trade;
 
+/// Convenient type alias for a Bitmex [`ExchangeWsStream`] using [`WebSocketSerdeParser`](barter_integration::protocol::websocket::WebSocketSerdeParser).
+pub type BitmexWsStream<Transformer> = ExchangeWsStream<WebSocketSerdeParser, Transformer>;
+
 /// [`Bitmex`] server base url.
 ///
 /// See docs: <https://www.bitmex.com/app/wsAPI>
@@ -84,7 +90,7 @@ where
 {
     type SnapFetcher = NoInitialSnapshots;
     type Stream =
-        ExchangeWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BitmexTrade>>;
+        BitmexWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BitmexTrade>>;
 }
 
 impl<'de> serde::Deserialize<'de> for Bitmex {

barter-data/src/exchange/bybit/mod.rs

@@ -15,7 +15,10 @@ use crate::{
     transformer::stateless::StatelessTransformer,
 };
 use barter_instrument::exchange::ExchangeId;
-use barter_integration::{error::SocketError, protocol::websocket::WsMessage};
+use barter_integration::{
+    error::SocketError,
+    protocol::websocket::{WebSocketSerdeParser, WsMessage},
+};
 use book::{BybitOrderBookMessage, l2::BybitOrderBooksL2Transformer};
 use serde::de::{Error, Unexpected};
 use std::{fmt::Debug, marker::PhantomData, time::Duration};
@@ -56,6 +59,9 @@ pub mod trade;
 /// [`BybitFuturesUsd`](futures::BybitPerpetualsUsd).
 pub mod book;
 
+/// Convenient type alias for a Bybit [`ExchangeWsStream`] using [`WebSocketSerdeParser`](barter_integration::protocol::websocket::WebSocketSerdeParser).
+pub type BybitWsStream<Transformer> = ExchangeWsStream<WebSocketSerdeParser, Transformer>;
+
 /// Generic [`Bybit<Server>`](Bybit) exchange.
 ///
 /// ### Notes
@@ -122,7 +128,7 @@ where
 {
     type SnapFetcher = NoInitialSnapshots;
     type Stream =
-        ExchangeWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BybitTrade>>;
+        BybitWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, BybitTrade>>;
 }
 
 impl<Instrument, Server> StreamSelector<Instrument, OrderBooksL1> for Bybit<Server>
@@ -131,7 +137,7 @@ where
     Server: ExchangeServer + Debug + Send + Sync,
 {
     type SnapFetcher = NoInitialSnapshots;
-    type Stream = ExchangeWsStream<
+    type Stream = BybitWsStream<
         StatelessTransformer<Self, Instrument::Key, OrderBooksL1, BybitOrderBookMessage>,
     >;
 }
@@ -142,7 +148,7 @@ where
     Server: ExchangeServer + Debug + Send + Sync,
 {
     type SnapFetcher = NoInitialSnapshots;
-    type Stream = ExchangeWsStream<BybitOrderBooksL2Transformer<Instrument::Key>>;
+    type Stream = BybitWsStream<BybitOrderBooksL2Transformer<Instrument::Key>>;
 }
 
 impl<'de, Server> serde::Deserialize<'de> for Bybit<Server>

barter-data/src/exchange/coinbase/mod.rs

@@ -11,7 +11,10 @@ use crate::{
     transformer::stateless::StatelessTransformer,
 };
 use barter_instrument::exchange::ExchangeId;
-use barter_integration::{error::SocketError, protocol::websocket::WsMessage};
+use barter_integration::{
+    error::SocketError,
+    protocol::websocket::{WebSocketSerdeParser, WsMessage},
+};
 use barter_macro::{DeExchange, SerExchange};
 use derive_more::Display;
 use serde_json::json;
@@ -37,6 +40,9 @@ pub mod trade;
 /// See docs: <https://docs.cloud.coinbase.com/exchange/docs/websocket-overview>
 pub const BASE_URL_COINBASE: &str = "wss://ws-feed.exchange.coinbase.com";
 
+/// Convenient type alias for a Coinbase [`ExchangeWsStream`] using [`WebSocketSerdeParser`](barter_integration::protocol::websocket::WebSocketSerdeParser).
+pub type CoinbaseWsStream<Transformer> = ExchangeWsStream<WebSocketSerdeParser, Transformer>;
+
 /// [`Coinbase`] exchange.
 ///
 /// See docs: <https://docs.cloud.coinbase.com/exchange/docs/websocket-overview>
@@ -91,5 +97,5 @@ where
 {
     type SnapFetcher = NoInitialSnapshots;
     type Stream =
-        ExchangeWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, CoinbaseTrade>>;
+        CoinbaseWsStream<StatelessTransformer<Self, Instrument::Key, PublicTrades, CoinbaseTrade>>;
 }