rustrade_data/subscriber/
mod.rs

1use self::{
2    mapper::{SubscriptionMapper, WebSocketSubMapper},
3    validator::SubscriptionValidator,
4};
5use crate::{
6    Identifier,
7    exchange::Connector,
8    instrument::InstrumentData,
9    subscription::{Map, Subscription, SubscriptionKind, SubscriptionMeta},
10};
11use futures::SinkExt;
12use rustrade_integration::{
13    error::SocketError,
14    protocol::websocket::{WebSocket, WsMessage, connect},
15};
16use serde::{Deserialize, Serialize};
17use std::{fmt::Debug, future::Future};
18use tracing::debug;
19
20/// [`SubscriptionMapper`] implementations defining how to map a
21/// collection of Barter [`Subscription`]s into exchange specific [`SubscriptionMeta`].
22pub mod mapper;
23
24/// [`SubscriptionValidator`] implementations defining how to
25/// validate actioned [`Subscription`]s were successful.
26pub mod validator;
27
28// One connection shared by every stream a subscriber and its clones open; documented in its own
29// file.
30#[cfg(any(feature = "alpaca", feature = "lse", feature = "massive"))]
31pub(crate) mod shared;
32
33/// The market stream of every provider whose streams share one connection.
34#[cfg(any(feature = "alpaca", feature = "lse", feature = "massive"))]
35pub mod shared_stream;
36
37/// Defines how to connect to a socket and subscribe to market data streams.
38///
39/// Subscribers may carry state such as authentication credentials.
40/// The trait requires `Clone` to support reconnection (subscribers are cloned
41/// into the reconnect closure).
42pub trait Subscriber: Clone + Send + Sync {
43    type SubMapper: SubscriptionMapper;
44
45    /// What a successful subscribe hands the stream to read from.
46    ///
47    /// A [`WebSocket`] of its own for almost every exchange, and the only transport the standard
48    /// [`MarketStream`](crate::MarketStream) initialisation accepts. A subscriber whose streams
49    /// share one connection instead hands out a per-stream view of it, and pairs with a
50    /// [`MarketStream`](crate::MarketStream) of its own that knows how to read that view.
51    ///
52    /// `Send`, because a stream's transport is moved into the task that drives it.
53    type Transport: Send;
54
55    fn subscribe<Exchange, Instrument, Kind>(
56        &self,
57        subscriptions: &[Subscription<Exchange, Instrument, Kind>],
58    ) -> impl Future<Output = Result<Subscribed<Instrument::Key, Self::Transport>, SocketError>> + Send
59    where
60        Exchange: Connector + Send + Sync,
61        Kind: SubscriptionKind + Send + Sync,
62        Instrument: InstrumentData,
63        Subscription<Exchange, Instrument, Kind>:
64            Identifier<Exchange::Channel> + Identifier<Exchange::Market>;
65}
66
67/// The outcome of a successful [`Subscriber::subscribe`].
68#[derive(Debug)]
69pub struct Subscribed<InstrumentKey, Transport = WebSocket> {
70    /// The connection, or view of one, that the subscribed events arrive on.
71    pub transport: Transport,
72    /// Each confirmed subscription's identifier, mapped to its instrument key.
73    pub map: Map<InstrumentKey>,
74    /// Frames that arrived during validation without being a confirmation, in arrival order —
75    /// typically the first events of an already-confirmed subscription. They are the stream's first
76    /// input and must be processed before anything read from `transport`.
77    ///
78    /// Raw WebSocket frames whatever the transport: a shared connection relays the frames it reads,
79    /// so every stream parses its input the same way.
80    pub buffered_websocket_events: Vec<WsMessage>,
81}
82
83/// Standard [`Subscriber`] for [`WebSocket`]s suitable for most exchanges.
84///
85/// This is a stateless subscriber for unauthenticated market data streams.
86#[derive(
87    Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Debug, Default, Deserialize, Serialize,
88)]
89pub struct WebSocketSubscriber;
90
91impl Subscriber for WebSocketSubscriber {
92    type SubMapper = WebSocketSubMapper;
93    type Transport = WebSocket;
94
95    async fn subscribe<Exchange, Instrument, Kind>(
96        &self,
97        subscriptions: &[Subscription<Exchange, Instrument, Kind>],
98    ) -> Result<Subscribed<Instrument::Key, Self::Transport>, SocketError>
99    where
100        Exchange: Connector + Send + Sync,
101        Kind: SubscriptionKind + Send + Sync,
102        Instrument: InstrumentData,
103        Subscription<Exchange, Instrument, Kind>:
104            Identifier<Exchange::Channel> + Identifier<Exchange::Market>,
105    {
106        // Define variables for logging ergonomics
107        let exchange = Exchange::ID;
108        let url = Exchange::url()?;
109        debug!(%exchange, %url, ?subscriptions, "subscribing to WebSocket");
110
111        // Connect to exchange
112        let mut websocket = connect(url).await?;
113        debug!(%exchange, ?subscriptions, "connected to WebSocket");
114
115        // Map &[Subscription<Exchange, Kind>] to SubscriptionMeta
116        let SubscriptionMeta {
117            instrument_map,
118            ws_subscriptions,
119        } = Self::SubMapper::map::<Exchange, Instrument, Kind>(subscriptions);
120
121        // Send Subscriptions over WebSocket
122        for subscription in ws_subscriptions {
123            debug!(%exchange, payload = ?subscription, "sending exchange subscription");
124            websocket
125                .send(subscription)
126                .await
127                .map_err(|error| SocketError::WebSocket(Box::new(error)))?;
128        }
129
130        // Validate Subscription responses
131        let (map, buffered_websocket_events) = Exchange::SubValidator::validate::<
132            Exchange,
133            Instrument::Key,
134            Kind,
135        >(instrument_map, &mut websocket)
136        .await?;
137
138        debug!(%exchange, "successfully initialised WebSocket stream with confirmed Subscriptions");
139        Ok(Subscribed {
140            transport: websocket,
141            map,
142            buffered_websocket_events,
143        })
144    }
145}
rustrade_data/subscriber/mod.rs

rustrade_data/subscriber/
mod.rs
