barter_data_sniper/streams/
consumer.rs

1use crate::{
2    error::DataError,
3    event::MarketEvent,
4    exchange::StreamSelector,
5    instrument::InstrumentData,
6    streams::{
7        reconnect,
8        reconnect::stream::{
9            init_reconnecting_stream, ReconnectingStream, ReconnectionBackoffPolicy,
10        },
11    },
12    subscription::{Subscription, SubscriptionKind},
13    Identifier, MarketStream,
14};
15use barter_instrument_copy::exchange::ExchangeId;
16use futures::Stream;
17use serde::{Deserialize, Serialize};
18use tracing::info;
19
20/// Default [`ReconnectionBackoffPolicy`] for a [`reconnecting`](`ReconnectingStream`) [`MarketStream`].
21pub const STREAM_RECONNECTION_POLICY: ReconnectionBackoffPolicy = ReconnectionBackoffPolicy {
22    backoff_ms_initial: 125,
23    backoff_multiplier: 2,
24    backoff_ms_max: 60000,
25};
26
27/// Convenient type alias for a [`MarketEvent`] [`Result`] consumed via a
28/// [`reconnecting`](`ReconnectingStream`) [`MarketStream`].
29pub type MarketStreamResult<InstrumentKey, Kind> =
30    reconnect::Event<ExchangeId, Result<MarketEvent<InstrumentKey, Kind>, DataError>>;
31
32/// Convenient type alias for a [`MarketEvent`] consumed via a
33/// [`reconnecting`](`ReconnectingStream`) [`MarketStream`].
34pub type MarketStreamEvent<InstrumentKey, Kind> =
35    reconnect::Event<ExchangeId, MarketEvent<InstrumentKey, Kind>>;
36
37/// Initialises a [`reconnecting`](`ReconnectingStream`) [`MarketStream`] using a collection of
38/// [`Subscription`]s.
39///
40/// The provided [`ReconnectionBackoffPolicy`] dictates how the exponential backoff scales
41/// between reconnections.
42pub async fn init_market_stream<Exchange, Instrument, Kind>(
43    policy: ReconnectionBackoffPolicy,
44    subscriptions: Vec<Subscription<Exchange, Instrument, Kind>>,
45) -> Result<impl Stream<Item = MarketStreamResult<Instrument::Key, Kind::Event>>, DataError>
46where
47    Exchange: StreamSelector<Instrument, Kind>,
48    Instrument: InstrumentData,
49    Kind: SubscriptionKind,
50    Subscription<Exchange, Instrument, Kind>:
51        Identifier<Exchange::Channel> + Identifier<Exchange::Market>,
52{
53    // Determine ExchangeId associated with these Subscriptions
54    let exchange = Exchange::ID;
55
56    // Determine StreamKey for use in logging
57    let stream_key = subscriptions
58        .first()
59        .map(|sub| StreamKey {
60            exchange,
61            kind: sub.kind.as_str(),
62        })
63        .ok_or(DataError::SubscriptionsEmpty)?;
64
65    info!(
66        %exchange,
67        ?subscriptions,
68        ?policy,
69        ?stream_key,
70        "MarketStream with auto reconnect running"
71    );
72
73    Ok(init_reconnecting_stream(move || {
74        let subscriptions = subscriptions.clone();
75        async move { Exchange::Stream::init::<Exchange::SnapFetcher>(&subscriptions).await }
76    })
77    .await?
78    .with_reconnect_backoff(policy, stream_key)
79    .with_termination_on_error(|error| error.is_terminal(), stream_key)
80    .with_reconnection_events(exchange))
81}
82
83#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
84pub struct StreamKey<Kind = &'static str> {
85    pub exchange: ExchangeId,
86    pub kind: Kind,
87}
88
89impl std::fmt::Debug for StreamKey {
90    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
91        write!(f, "market_stream-{}-{}", self.exchange, self.kind)
92    }
93}
barter_data_sniper/streams/consumer.rs

barter_data_sniper/streams/
consumer.rs
