rustrade_data/streams/builder/
mod.rs

1use super::Streams;
2use crate::{
3    Identifier,
4    error::DataError,
5    exchange::StreamSelector,
6    instrument::InstrumentData,
7    streams::{
8        consumer::{MarketStreamResult, STREAM_RECONNECTION_POLICY, init_market_stream},
9        reconnect::stream::ReconnectingStream,
10    },
11    subscription::{Subscription, SubscriptionKind},
12};
13use rustrade_instrument::exchange::ExchangeId;
14use rustrade_integration::{Validator, channel::Channel};
15use std::{
16    collections::HashMap,
17    fmt::{Debug, Display},
18    future::Future,
19    pin::Pin,
20};
21
22/// Defines the [`MultiStreamBuilder`](multi::MultiStreamBuilder) API for ergonomically
23/// initialising a common [`Streams<Output>`](Streams) from multiple
24/// [`StreamBuilder<SubscriptionKind>`](StreamBuilder)s.
25pub mod multi;
26
27/// Defines the [`DynamicStreams`](dynamic::DynamicStreams) API for initialising an arbitrary number
28/// of `MarketStream`s from the [`ExchangeId`] and [`SubKind`](crate::subscription::SubKind) enums, rather than concrete
29/// types.
30pub mod dynamic;
31
32/// Communicative type alias representing the [`Future`] result of a [`Subscription`] validation
33/// call generated whilst executing [`StreamBuilder::subscribe`].
34pub type SubscribeFuture = Pin<Box<dyn Future<Output = Result<(), DataError>>>>;
35
36/// Builder to configure and initialise a [`Streams<MarketEvent<SubscriptionKind::Event>`](Streams) instance
37/// for a specific [`SubscriptionKind`].
38#[derive(Default)]
39pub struct StreamBuilder<InstrumentKey, Kind>
40where
41    Kind: SubscriptionKind,
42{
43    pub channels: HashMap<ExchangeId, Channel<MarketStreamResult<InstrumentKey, Kind::Event>>>,
44    pub futures: Vec<SubscribeFuture>,
45}
46
47impl<InstrumentKey, Kind> Debug for StreamBuilder<InstrumentKey, Kind>
48where
49    InstrumentKey: Debug,
50    Kind: SubscriptionKind,
51{
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        f.debug_struct("StreamBuilder<InstrumentKey, SubscriptionKind>")
54            .field("channels", &self.channels)
55            .field("num_futures", &self.futures.len())
56            .finish()
57    }
58}
59
60impl<InstrumentKey, Kind> StreamBuilder<InstrumentKey, Kind>
61where
62    Kind: SubscriptionKind,
63{
64    /// Construct a new [`Self`].
65    pub fn new() -> Self {
66        Self {
67            channels: HashMap::new(),
68            futures: Vec::new(),
69        }
70    }
71
72    /// Add a collection of [`Subscription`]s to the [`StreamBuilder`] that will be actioned on
73    /// a connection the `subscriber` provides.
74    ///
75    /// For almost every exchange that is a distinct
76    /// [`WebSocket`](rustrade_integration::protocol::websocket::WebSocket) per call. A subscriber
77    /// whose clones share one connection — for a provider that allows a key a single connection —
78    /// instead serves each call over a view of it; see
79    /// [`Subscriber::Transport`](crate::subscriber::Subscriber::Transport).
80    ///
81    /// The `subscriber` handles the connection and authentication.
82    /// For unauthenticated exchanges, use [`WebSocketSubscriber`](crate::subscriber::WebSocketSubscriber).
83    /// For authenticated exchanges like Alpaca, use the exchange-specific subscriber with credentials.
84    ///
85    /// Note that [`Subscription`]s are not actioned until the
86    /// [`init()`](StreamBuilder::init()) method is invoked.
87    pub fn subscribe<SubIter, Sub, Exchange, Instrument>(
88        mut self,
89        subscriber: Exchange::Subscriber,
90        subscriptions: SubIter,
91    ) -> Self
92    where
93        SubIter: IntoIterator<Item = Sub>,
94        Sub: Into<Subscription<Exchange, Instrument, Kind>>,
95        Exchange: StreamSelector<Instrument, Kind> + Ord + Send + Sync + 'static,
96        Instrument: InstrumentData<Key = InstrumentKey> + Ord + Display + 'static,
97        Instrument::Key: Debug + Clone + Send + 'static,
98        Kind: Ord + Display + Send + Sync + 'static,
99        Kind::Event: Clone + Send,
100        Subscription<Exchange, Instrument, Kind>:
101            Identifier<Exchange::Channel> + Identifier<Exchange::Market>,
102    {
103        // Construct Vec<Subscriptions> from input SubIter
104        let subscriptions = subscriptions.into_iter().map(Sub::into).collect::<Vec<_>>();
105
106        // Acquire channel Sender to send Market<Kind::Event> from consumer loop to user
107        // '--> Add ExchangeChannel Entry if this Exchange <--> SubscriptionKind combination is new
108        let exchange_tx = self.channels.entry(Exchange::ID).or_default().tx.clone();
109
110        // Add Future that once awaited will yield the Result<(), SocketError> of subscribing
111        self.futures.push(Box::pin(async move {
112            // Validate Subscriptions
113            let mut subscriptions = subscriptions
114                .into_iter()
115                .map(Subscription::validate)
116                .collect::<Result<Vec<_>, _>>()?;
117
118            // Remove duplicate Subscriptions
119            subscriptions.sort();
120            subscriptions.dedup();
121
122            // Initialise a MarketEvent `ReconnectingStream`
123            let stream =
124                init_market_stream(STREAM_RECONNECTION_POLICY, subscriber, subscriptions).await?;
125
126            // Forward MarketEvents to ExchangeTx
127            tokio::spawn(stream.forward_to(exchange_tx));
128
129            Ok(())
130        }));
131
132        self
133    }
134
135    /// Spawn a [`MarketStreamResult<SubscriptionKind::Event>`](MarketStreamResult) consumer loop
136    /// for each collection of [`Subscription`]s added to [`StreamBuilder`] via the
137    /// [`subscribe()`](StreamBuilder::subscribe()) method.
138    ///
139    /// Each consumer loop distributes consumed [`MarketStreamResult`] to
140    /// the [`Streams`] `HashMap` returned by this method.
141    pub async fn init(
142        self,
143    ) -> Result<Streams<MarketStreamResult<InstrumentKey, Kind::Event>>, DataError> {
144        // Await Stream initialisation perpetual and ensure success
145        futures::future::try_join_all(self.futures).await?;
146
147        // Construct Streams using each ExchangeChannel receiver
148        Ok(Streams {
149            streams: self
150                .channels
151                .into_iter()
152                .map(|(exchange, channel)| (exchange, channel.rx))
153                .collect(),
154        })
155    }
156}
rustrade_data/streams/builder/mod.rs

rustrade_data/streams/builder/
mod.rs
