Struct DynamicStreams 

pub struct DynamicStreams<InstrumentKey> {
    pub trades: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, PublicTrade>>>,
    pub l1s: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookL1>>>,
    pub l2s: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookEvent>>>,
    pub liquidations: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Liquidation>>>,
    pub candles: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Candle>>>,
}

Fields

trades: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, PublicTrade>>>

l1s: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookL1>>>

l2s: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookEvent>>>

liquidations: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Liquidation>>>

candles: VecMap<ExchangeId, UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Candle>>>

Implementations

impl<InstrumentKey> DynamicStreams<InstrumentKey>

pub async fn init<SubBatchIter, SubIter, Sub, Instrument>( subscription_batches: SubBatchIter, ) -> Result<Self, DataError>

where SubBatchIter: IntoIterator<Item = SubIter>, SubIter: IntoIterator<Item = Sub>, Sub: Into<Subscription<ExchangeId, Instrument, SubKind>>, Instrument: DynamicInstrument + InstrumentData<Key = InstrumentKey>, InstrumentKey: Debug + Clone + PartialEq + Send + Sync + 'static,

Initialise a set of Streams by providing one or more Subscription batches, for venues that need no subscriber from the caller.

Equivalent to init_with with no DynamicSubscribers: a subscription to a venue that needs one fails with DataError::SubscriberRequired.

Examples

Please see rustrade-data-rs/examples/dynamic_multi_stream_multi_exchange.rs for a comprehensive example of how to use this market data stream initialiser.

pub async fn init_with<SubBatchIter, SubIter, Sub, Instrument>( subscribers: &DynamicSubscribers, subscription_batches: SubBatchIter, ) -> Result<Self, DataError>

where SubBatchIter: IntoIterator<Item = SubIter>, SubIter: IntoIterator<Item = Sub>, Sub: Into<Subscription<ExchangeId, Instrument, SubKind>>, Instrument: DynamicInstrument + InstrumentData<Key = InstrumentKey>, InstrumentKey: Debug + Clone + PartialEq + Send + Sync + 'static,

Initialise a set of Streams by providing one or more Subscription batches, and the subscribers for venues that cannot be served without one.

Each batch (ie/ impl Iterator<Item = Subscription>) will initialise at-least-one Stream under the hood. If the batch contains more-than-one ExchangeId and/or SubKind, it will be further split under the hood for compile-time reasons.

Streams sharing a connection

A venue supplied through subscribers receives a clone of that subscriber for every group, whichever batch it came from. Where clones share a connection, every group on that venue shares it: across datasets and kinds on London Strategic Edge, across kinds on each Alpaca feed, and across kinds on each Massive cluster.

Errors

Nothing is connected until every group has been routed, so these fail the call with no connection opened:

• a subscription the support matrix refuses (see exchange_supports_instrument_kind_sub_kind);
• DataError::Unsupported for a pair the matrix accepts but no connector here serves;
• DataError::SubscriberRequired for a venue subscribers has no subscriber for;
• DataError::FeatureDisabled for a venue whose cargo feature this build lacks.

Every group is then initialised concurrently. If any fails, the call waits for the rest to finish initialising, stops every stream that succeeded, and returns the first error — so a failed call leaves no stream running: each is dropped before the call returns.

A dropped London Strategic Edge, Alpaca or Massive stream asks its connection to release its share, and the connection does so asynchronously, closing the socket once no stream remains. A retry through the same subscriber, or a clone of it, reuses that connection whatever state it is in. A retry through a subscriber built separately for the same account may still be refused — TOO_MANY_CONNECTIONS on London Strategic Edge, connection limit exceeded on Alpaca — until the socket has closed. Massive refuses nothing: it closes the older socket instead, ending every stream still on it.

Examples

Please see rustrade-data-rs/examples/dynamic_multi_stream_multi_exchange.rs for a comprehensive example of how to use this market data stream initialiser.

pub fn select_trades( &mut self, exchange: ExchangeId, ) -> Option<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, PublicTrade>>>

Remove an exchange PublicTrade Stream from the DynamicStreams collection.

Note that calling this method will permanently remove this Stream from Self.

pub fn select_all_trades( &mut self, ) -> SelectAll<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, PublicTrade>>>

Select and merge every exchange PublicTrade Stream using SelectAll.

pub fn select_l1s( &mut self, exchange: ExchangeId, ) -> Option<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookL1>>>

Remove an exchange OrderBookL1 Stream from the DynamicStreams collection.

Note that calling this method will permanently remove this Stream from Self.

pub fn select_all_l1s( &mut self, ) -> SelectAll<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookL1>>>

Select and merge every exchange OrderBookL1 Stream using SelectAll.

pub fn select_l2s( &mut self, exchange: ExchangeId, ) -> Option<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookEvent>>>

Remove an exchange OrderBookEvent Stream from the DynamicStreams collection.

Note that calling this method will permanently remove this Stream from Self.

pub fn select_all_l2s( &mut self, ) -> SelectAll<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, OrderBookEvent>>>

Select and merge every exchange OrderBookEvent Stream using SelectAll.

pub fn select_liquidations( &mut self, exchange: ExchangeId, ) -> Option<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Liquidation>>>

Remove an exchange Liquidation Stream from the DynamicStreams collection.

Note that calling this method will permanently remove this Stream from Self.

pub fn select_all_liquidations( &mut self, ) -> SelectAll<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Liquidation>>>

Select and merge every exchange Liquidation Stream using SelectAll.

pub fn select_candles( &mut self, exchange: ExchangeId, ) -> Option<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Candle>>>

Remove an exchange Candle Stream from the DynamicStreams collection.

Note that calling this method will permanently remove this Stream from Self.

Mixed intervals

DynamicStreams keys candle streams by ExchangeId only, so subscribing to multiple intervals on one exchange (e.g. Candles { Min1 } + Candles { Hour1 } on BinanceSpot) merges them into this single per-exchange stream. Each distinct interval is a separate upstream WebSocket subscription (mind venue per-connection stream limits at high symbol × interval fan-out). Because Candle carries no interval field, a consumer mixing intervals must recover it from close_time spacing or the originating subscription; if per-interval routing matters, use the typed Streams builder with one StreamSelector per interval.

pub fn select_all_candles( &mut self, ) -> SelectAll<UnboundedReceiverStream<MarketStreamResult<InstrumentKey, Candle>>>

Select and merge every exchange Candle Stream using SelectAll.

See select_candles for how multiple intervals on one exchange are merged — the merged Candles carry no interval to disambiguate them.

pub fn select_all<Output>(self) -> impl Stream<Item = Output>

where InstrumentKey: Send + 'static, Output: 'static, MarketStreamResult<InstrumentKey, PublicTrade>: Into<Output>, MarketStreamResult<InstrumentKey, OrderBookL1>: Into<Output>, MarketStreamResult<InstrumentKey, OrderBookEvent>: Into<Output>, MarketStreamResult<InstrumentKey, Liquidation>: Into<Output>, MarketStreamResult<InstrumentKey, Candle>: Into<Output>,

Select and merge every exchange Stream for every data type using select_all

Note that using MarketStreamResult<Instrument, DataKind> as the Output is suitable for most use cases.

Trait Implementations

impl<InstrumentKey: Debug> Debug for DynamicStreams<InstrumentKey>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.