rustrade_data/streams/
mod.rs

1use self::builder::{StreamBuilder, multi::MultiStreamBuilder};
2use crate::subscription::SubscriptionKind;
3use fnv::FnvHashMap;
4use futures::Stream;
5use rustrade_instrument::exchange::ExchangeId;
6use rustrade_integration::channel::UnboundedRx;
7
8/// Bridges a blocking source — a Parquet artifact, a compressed archive — into a bounded,
9/// back-pressured `Stream`, off the async runtime's workers.
10pub mod blocking;
11
12/// Defines the [`StreamBuilder`] and [`MultiStreamBuilder`] APIs for ergonomically initialising
13/// [`MarketStream`](super::MarketStream) [`Streams`].
14pub mod builder;
15
16/// Central consumer loop functionality used by the [`StreamBuilder`] to
17/// drive a re-connecting [`MarketStream`](super::MarketStream).
18pub mod consumer;
19
20/// Lazily merge N time-sorted market streams into one, for replaying multi-instrument historical
21/// data as a single time-ordered feed.
22pub mod merge;
23
24/// Defines a [`ReconnectingStream`](reconnect::stream::ReconnectingStream) and associated logic
25/// for generating an auto reconnecting `Stream`.
26pub mod reconnect;
27
28/// Ergonomic collection of exchange market event receivers.
29#[derive(Debug)]
30pub struct Streams<T> {
31    pub streams: FnvHashMap<ExchangeId, UnboundedRx<T>>,
32}
33
34impl<T> Streams<T> {
35    /// Construct a [`StreamBuilder`] for configuring new market event [`Streams`].
36    pub fn builder<InstrumentKey, Kind>() -> StreamBuilder<InstrumentKey, Kind>
37    where
38        Kind: SubscriptionKind,
39    {
40        StreamBuilder::<InstrumentKey, Kind>::new()
41    }
42
43    /// Construct a [`MultiStreamBuilder`] for configuring new
44    /// [`MarketEvent<T>`](crate::event::MarketEvent) [`Streams`].
45    pub fn builder_multi() -> MultiStreamBuilder<T> {
46        MultiStreamBuilder::<T>::new()
47    }
48
49    /// Remove an exchange market event [`Stream`] from the [`Streams`] `HashMap`.
50    pub fn select(&mut self, exchange: ExchangeId) -> Option<impl Stream<Item = T> + '_> {
51        self.streams.remove(&exchange).map(UnboundedRx::into_stream)
52    }
53
54    /// Select and merge every exchange `Stream` using
55    /// [`select_all`](futures_util::stream::select_all::select_all).
56    pub fn select_all(self) -> impl Stream<Item = T> {
57        let all = self.streams.into_values().map(UnboundedRx::into_stream);
58        futures_util::stream::select_all::select_all(all)
59    }
60}
rustrade_data/streams/mod.rs

rustrade_data/streams/
mod.rs
