digdigdig3_station/
subscription.rs

1use digdigdig3::core::traits::Credentials;
2use digdigdig3::core::types::{AccountType, ExchangeId, SymbolInfo};
3use digdigdig3::core::websocket::KlineInterval;
4
5use crate::data::{
6    AggTradePoint, BalanceUpdatePoint, BarPoint, BasisPoint, BlockTradePoint, CompositeIndexPoint,
7    FootprintPoint, FundingRatePoint, FundingSettlementPoint, HistoricalVolatilityPoint,
8    IndexPriceKlinePoint, IndexPricePoint, InsuranceFundPoint, LiquidationPoint,
9    LongShortRatioPoint, MarkPriceKlinePoint, MarkPricePoint,
10    MarketWarningPoint, ObDeltaPoint, ObSnapshotPoint, OpenInterestPoint, OptionGreeksPoint,
11    OrderUpdatePoint, OrderbookL3Point, PositionUpdatePoint, PredictedFundingPoint,
12    PremiumIndexKlinePoint, RiskLimitPoint,
13    SettlementEventPoint, TickerPoint, TradePoint, VolatilityIndexPoint,
14};
15use crate::series::{Kind, SeriesKey};
16
17/// User-facing stream class to request in a `SubscriptionSet`.
18///
19/// `Kline` carries a typed `KlineInterval` (e.g. `KlineInterval::new("1m")`).
20/// All other variants are parameterless.
21#[derive(Debug, Clone, PartialEq, Eq, Hash)]
22pub enum Stream {
23    Ticker,
24    Trade,
25    Orderbook,
26    OrderbookDelta,
27    Kline(KlineInterval),
28    MarkPrice,
29    FundingRate,
30    OpenInterest,
31    Liquidation,
32    AggTrade,
33    // --- extended stream types ---
34    BlockTrade,
35    IndexPrice,
36    CompositeIndex,
37    OptionGreeks,
38    VolatilityIndex,
39    HistoricalVolatility,
40    LongShortRatio,
41    Basis,
42    InsuranceFund,
43    OrderbookL3,
44    SettlementEvent,
45    MarketWarning,
46    RiskLimit,
47    PredictedFunding,
48    FundingSettlement,
49    MarkPriceKline(KlineInterval),
50    IndexPriceKline(KlineInterval),
51    PremiumIndexKline(KlineInterval),
52    // --- derived bar aggregators ---
53    /// Range bar: close when |trade.price − bar_open| ≥ range.
54    /// `range` = price × 1e8 (fixed-point integer, same unit as `Kind::RangeBar`).
55    RangeBar(u64),
56    /// Tick bar: close every `n` trades.
57    TickBar(u32),
58    /// Volume bar: close when cumulative volume ≥ threshold.
59    /// `threshold` = volume × 1e8 (fixed-point integer, same unit as `Kind::VolumeBar`).
60    VolumeBar(u64),
61    /// Footprint bar: time-bucketed OHLCV with per-price buy/sell breakdown.
62    Footprint(KlineInterval),
63    // --- private (auth-required) streams ---
64    /// Order lifecycle events (create/fill/cancel/expire).  Requires credentials.
65    OrderUpdate,
66    /// Account balance changes.  Requires credentials.
67    BalanceUpdate,
68    /// Futures position changes.  Requires credentials.
69    PositionUpdate,
70}
71
72impl Stream {
73    pub(crate) fn to_kind(&self) -> Kind {
74        match self {
75            Stream::Trade => Kind::Trade,
76            Stream::AggTrade => Kind::AggTrade,
77            Stream::Kline(iv) => Kind::Kline(iv.clone()),
78            Stream::Ticker => Kind::Ticker,
79            Stream::Orderbook => Kind::Orderbook,
80            Stream::OrderbookDelta => Kind::OrderbookDelta,
81            Stream::MarkPrice => Kind::MarkPrice,
82            Stream::FundingRate => Kind::FundingRate,
83            Stream::OpenInterest => Kind::OpenInterest,
84            Stream::Liquidation => Kind::Liquidation,
85            Stream::BlockTrade => Kind::BlockTrade,
86            Stream::IndexPrice => Kind::IndexPrice,
87            Stream::CompositeIndex => Kind::CompositeIndex,
88            Stream::OptionGreeks => Kind::OptionGreeks,
89            Stream::VolatilityIndex => Kind::VolatilityIndex,
90            Stream::HistoricalVolatility => Kind::HistoricalVolatility,
91            Stream::LongShortRatio => Kind::LongShortRatio,
92            Stream::Basis => Kind::Basis,
93            Stream::InsuranceFund => Kind::InsuranceFund,
94            Stream::OrderbookL3 => Kind::OrderbookL3,
95            Stream::SettlementEvent => Kind::SettlementEvent,
96            Stream::MarketWarning => Kind::MarketWarning,
97            Stream::RiskLimit => Kind::RiskLimit,
98            Stream::PredictedFunding => Kind::PredictedFunding,
99            Stream::FundingSettlement => Kind::FundingSettlement,
100            Stream::MarkPriceKline(iv) => Kind::MarkPriceKline(iv.clone()),
101            Stream::IndexPriceKline(iv) => Kind::IndexPriceKline(iv.clone()),
102            Stream::PremiumIndexKline(iv) => Kind::PremiumIndexKline(iv.clone()),
103            Stream::RangeBar(r) => Kind::RangeBar(*r),
104            Stream::TickBar(n) => Kind::TickBar(*n),
105            Stream::VolumeBar(v) => Kind::VolumeBar(*v),
106            Stream::Footprint(iv) => Kind::Footprint(iv.clone()),
107            Stream::OrderUpdate => Kind::OrderUpdate,
108            Stream::BalanceUpdate => Kind::BalanceUpdate,
109            Stream::PositionUpdate => Kind::PositionUpdate,
110        }
111    }
112
113    /// True if this stream requires authentication credentials.
114    pub fn is_private(&self) -> bool {
115        matches!(self, Stream::OrderUpdate | Stream::BalanceUpdate | Stream::PositionUpdate)
116    }
117}
118
119#[derive(Debug, Clone)]
120pub(crate) struct Entry {
121    pub(crate) exchange: ExchangeId,
122    pub(crate) symbol: String,
123    pub(crate) account_type: AccountType,
124    pub(crate) streams: Vec<Stream>,
125    /// If true, `symbol` is the raw exchange-native string and must be
126    /// passed through to the WS connector verbatim — no `SymbolNormalizer`
127    /// translation. Set by `SubscriptionSet::add_raw`. Used for exotic
128    /// instrument IDs that don't fit the canonical BASE-QUOTE shape
129    /// (Deribit options like `BTC-23MAY26-86000-C`, exchange-specific
130    /// suffixes, etc.).
131    pub(crate) is_raw: bool,
132    /// Credentials for private (auth-required) streams.  `None` for public
133    /// streams.  Set by [`SubscriptionSet::add_authenticated`].
134    pub(crate) credentials: Option<Credentials>,
135}
136
137/// Declarative subscription request — built up fluently, consumed by
138/// [`crate::Station::subscribe`].
139#[derive(Debug, Default, Clone)]
140pub struct SubscriptionSet {
141    pub(crate) entries: Vec<Entry>,
142}
143
144impl SubscriptionSet {
145    pub fn new() -> Self { Self::default() }
146
147    /// Add a subscription. `symbol` is canonical (e.g. `"BTC-USDT"`,
148    /// `"BTCUSDT"`, `"BTC/USDT"`) — it is parsed into a canonical
149    /// `Symbol` and translated to the exchange-native form via
150    /// `SymbolNormalizer`. Use [`Self::add_raw`] for instrument IDs that
151    /// don't fit the canonical BASE-QUOTE shape (Deribit options,
152    /// exchange-specific futures suffixes, etc.).
153    pub fn add(
154        mut self,
155        exchange: ExchangeId,
156        symbol: impl Into<String>,
157        account_type: AccountType,
158        streams: impl IntoIterator<Item = Stream>,
159    ) -> Self {
160        self.entries.push(Entry {
161            exchange,
162            symbol: symbol.into(),
163            account_type,
164            streams: streams.into_iter().collect(),
165            is_raw: false,
166            credentials: None,
167        });
168        self
169    }
170
171    /// Add a subscription with a raw exchange-native symbol. `symbol` is
172    /// passed through to the connector verbatim — no `SymbolNormalizer`
173    /// translation. Use for instrument IDs that don't fit the canonical
174    /// BASE-QUOTE shape:
175    /// - Deribit options: `"BTC-23MAY26-86000-C"`
176    /// - Futures with date suffix: `"BTCUSDT_240329"`
177    /// - Index symbols: `".DEFI"`, `"BTCUSD-PERP"`
178    ///
179    /// The caller is responsible for using the exact wire format the
180    /// exchange expects — `Event.symbol` on the handle will mirror the
181    /// raw string back.
182    pub fn add_raw(
183        mut self,
184        exchange: ExchangeId,
185        symbol: impl Into<String>,
186        account_type: AccountType,
187        streams: impl IntoIterator<Item = Stream>,
188    ) -> Self {
189        self.entries.push(Entry {
190            exchange,
191            symbol: symbol.into(),
192            account_type,
193            streams: streams.into_iter().collect(),
194            is_raw: true,
195            credentials: None,
196        });
197        self
198    }
199
200    /// Add authenticated (private) streams for `(exchange, account_type)`.
201    ///
202    /// Credentials are forwarded to the WS connector so it can open an
203    /// authenticated channel.  Multiple `add_authenticated` calls for the
204    /// same `(exchange, account_type)` pair will create separate entries;
205    /// Station's `acquire_or_spawn` reuses an already-connected authenticated
206    /// WS if one is present in the hub for that pair.
207    ///
208    /// `symbol` is ignored for account-wide private streams (`BalanceUpdate`).
209    /// Pass an empty string (`""`) or any placeholder — `Event::symbol()` for
210    /// those variants returns `""` or the asset identifier.
211    pub fn add_authenticated(
212        mut self,
213        exchange: ExchangeId,
214        account_type: AccountType,
215        credentials: Credentials,
216        streams: impl IntoIterator<Item = Stream>,
217    ) -> Self {
218        self.entries.push(Entry {
219            exchange,
220            symbol: String::new(),
221            account_type,
222            streams: streams.into_iter().collect(),
223            is_raw: true,
224            credentials: Some(credentials),
225        });
226        self
227    }
228
229    pub fn len(&self) -> usize { self.entries.len() }
230    pub fn is_empty(&self) -> bool { self.entries.is_empty() }
231}
232
233/// Events forwarded to consumers. One variant per market-data class.
234#[derive(Debug, Clone)]
235pub enum Event {
236    Trade {
237        exchange: ExchangeId,
238        symbol: String,
239        point: TradePoint,
240    },
241    AggTrade {
242        exchange: ExchangeId,
243        symbol: String,
244        point: AggTradePoint,
245    },
246    Bar {
247        exchange: ExchangeId,
248        symbol: String,
249        timeframe: KlineInterval,
250        point: BarPoint,
251    },
252    Ticker {
253        exchange: ExchangeId,
254        symbol: String,
255        point: TickerPoint,
256    },
257    OrderbookSnapshot {
258        exchange: ExchangeId,
259        symbol: String,
260        point: ObSnapshotPoint,
261    },
262    OrderbookDelta {
263        exchange: ExchangeId,
264        symbol: String,
265        point: ObDeltaPoint,
266    },
267    MarkPrice {
268        exchange: ExchangeId,
269        symbol: String,
270        point: MarkPricePoint,
271    },
272    FundingRate {
273        exchange: ExchangeId,
274        symbol: String,
275        point: FundingRatePoint,
276    },
277    OpenInterest {
278        exchange: ExchangeId,
279        symbol: String,
280        point: OpenInterestPoint,
281    },
282    Liquidation {
283        exchange: ExchangeId,
284        symbol: String,
285        point: LiquidationPoint,
286    },
287    // --- extended stream types ---
288    BlockTrade {
289        exchange: ExchangeId,
290        symbol: String,
291        point: BlockTradePoint,
292    },
293    IndexPrice {
294        exchange: ExchangeId,
295        symbol: String,
296        point: IndexPricePoint,
297    },
298    CompositeIndex {
299        exchange: ExchangeId,
300        symbol: String,
301        point: CompositeIndexPoint,
302    },
303    OptionGreeks {
304        exchange: ExchangeId,
305        symbol: String,
306        point: OptionGreeksPoint,
307    },
308    VolatilityIndex {
309        exchange: ExchangeId,
310        symbol: String,
311        point: VolatilityIndexPoint,
312    },
313    HistoricalVolatility {
314        exchange: ExchangeId,
315        symbol: String,
316        point: HistoricalVolatilityPoint,
317    },
318    LongShortRatio {
319        exchange: ExchangeId,
320        symbol: String,
321        point: LongShortRatioPoint,
322    },
323    Basis {
324        exchange: ExchangeId,
325        symbol: String,
326        point: BasisPoint,
327    },
328    InsuranceFund {
329        exchange: ExchangeId,
330        symbol: String,
331        point: InsuranceFundPoint,
332    },
333    OrderbookL3 {
334        exchange: ExchangeId,
335        symbol: String,
336        point: OrderbookL3Point,
337    },
338    SettlementEvent {
339        exchange: ExchangeId,
340        symbol: String,
341        point: SettlementEventPoint,
342    },
343    MarketWarning {
344        exchange: ExchangeId,
345        symbol: String,
346        point: MarketWarningPoint,
347    },
348    RiskLimit {
349        exchange: ExchangeId,
350        symbol: String,
351        point: RiskLimitPoint,
352    },
353    PredictedFunding {
354        exchange: ExchangeId,
355        symbol: String,
356        point: PredictedFundingPoint,
357    },
358    FundingSettlement {
359        exchange: ExchangeId,
360        symbol: String,
361        point: FundingSettlementPoint,
362    },
363    MarkPriceKline {
364        exchange: ExchangeId,
365        symbol: String,
366        timeframe: KlineInterval,
367        point: MarkPriceKlinePoint,
368    },
369    IndexPriceKline {
370        exchange: ExchangeId,
371        symbol: String,
372        timeframe: KlineInterval,
373        point: IndexPriceKlinePoint,
374    },
375    PremiumIndexKline {
376        exchange: ExchangeId,
377        symbol: String,
378        timeframe: KlineInterval,
379        point: PremiumIndexKlinePoint,
380    },
381    // --- derived bar aggregator events ---
382    /// Footprint bar update. Emitted on every trade (open bar, same upsert
383    /// semantics as `Event::Bar`). `Kind::Footprint` carries the interval.
384    Footprint {
385        exchange: ExchangeId,
386        symbol: String,
387        point: FootprintPoint,
388    },
389    // --- connector lifecycle events (meta, not data stream) ---
390    /// Emitted once when `hub.connect_public(exchange)` succeeds, or
391    /// immediately if it was already connected when `warmup()` called.
392    ConnectorReady {
393        exchange: ExchangeId,
394    },
395    /// Emitted once per `(exchange, account_type)` after REST
396    /// `get_exchange_info` succeeds. Multiple emits per exchange if both
397    /// Spot and Futures resolve. `symbols` is the raw exchange response.
398    SymbolsLoaded {
399        exchange: ExchangeId,
400        account_type: AccountType,
401        symbols: Vec<SymbolInfo>,
402    },
403    // --- private (auth-required) events ---
404    /// Order lifecycle event (create/fill/cancel/expire).
405    /// `symbol` is the instrument symbol from the order, or `""` if the
406    /// exchange did not include it in this event.
407    OrderUpdate {
408        exchange: ExchangeId,
409        account_type: AccountType,
410        symbol: String,
411        point: OrderUpdatePoint,
412    },
413    /// Account balance change event.
414    /// `symbol` is the asset ticker (e.g. `"USDT"`), not a trading pair.
415    BalanceUpdate {
416        exchange: ExchangeId,
417        account_type: AccountType,
418        symbol: String,
419        point: BalanceUpdatePoint,
420    },
421    /// Futures position change event.
422    /// `symbol` is the instrument symbol for the position.
423    PositionUpdate {
424        exchange: ExchangeId,
425        account_type: AccountType,
426        symbol: String,
427        point: PositionUpdatePoint,
428    },
429}
430
431impl Event {
432    pub fn exchange(&self) -> ExchangeId {
433        match self {
434            Event::Trade { exchange, .. } | Event::AggTrade { exchange, .. } |
435            Event::Bar { exchange, .. } | Event::Ticker { exchange, .. } |
436            Event::OrderbookSnapshot { exchange, .. } | Event::OrderbookDelta { exchange, .. } |
437            Event::MarkPrice { exchange, .. } |
438            Event::FundingRate { exchange, .. } | Event::OpenInterest { exchange, .. } |
439            Event::Liquidation { exchange, .. } |
440            Event::BlockTrade { exchange, .. } | Event::IndexPrice { exchange, .. } |
441            Event::CompositeIndex { exchange, .. } | Event::OptionGreeks { exchange, .. } |
442            Event::VolatilityIndex { exchange, .. } | Event::HistoricalVolatility { exchange, .. } |
443            Event::LongShortRatio { exchange, .. } |
444            Event::Basis { exchange, .. } | Event::InsuranceFund { exchange, .. } |
445            Event::OrderbookL3 { exchange, .. } | Event::SettlementEvent { exchange, .. } |
446            Event::MarketWarning { exchange, .. } |
447            Event::RiskLimit { exchange, .. } | Event::PredictedFunding { exchange, .. } |
448            Event::FundingSettlement { exchange, .. } |
449            Event::MarkPriceKline { exchange, .. } | Event::IndexPriceKline { exchange, .. } |
450            Event::PremiumIndexKline { exchange, .. } |
451            Event::Footprint { exchange, .. } => *exchange,
452            Event::OrderUpdate { exchange, .. } | Event::BalanceUpdate { exchange, .. } |
453            Event::PositionUpdate { exchange, .. } => *exchange,
454            Event::ConnectorReady { exchange } => *exchange,
455            Event::SymbolsLoaded { exchange, .. } => *exchange,
456        }
457    }
458    pub fn symbol(&self) -> &str {
459        match self {
460            Event::Trade { symbol, .. } | Event::AggTrade { symbol, .. } |
461            Event::Bar { symbol, .. } | Event::Ticker { symbol, .. } |
462            Event::OrderbookSnapshot { symbol, .. } | Event::OrderbookDelta { symbol, .. } |
463            Event::MarkPrice { symbol, .. } |
464            Event::FundingRate { symbol, .. } | Event::OpenInterest { symbol, .. } |
465            Event::Liquidation { symbol, .. } |
466            Event::BlockTrade { symbol, .. } | Event::IndexPrice { symbol, .. } |
467            Event::CompositeIndex { symbol, .. } | Event::OptionGreeks { symbol, .. } |
468            Event::VolatilityIndex { symbol, .. } | Event::HistoricalVolatility { symbol, .. } |
469            Event::LongShortRatio { symbol, .. } |
470            Event::Basis { symbol, .. } | Event::InsuranceFund { symbol, .. } |
471            Event::OrderbookL3 { symbol, .. } | Event::SettlementEvent { symbol, .. } |
472            Event::MarketWarning { symbol, .. } |
473            Event::RiskLimit { symbol, .. } | Event::PredictedFunding { symbol, .. } |
474            Event::FundingSettlement { symbol, .. } |
475            Event::MarkPriceKline { symbol, .. } | Event::IndexPriceKline { symbol, .. } |
476            Event::PremiumIndexKline { symbol, .. } |
477            Event::Footprint { symbol, .. } => symbol,
478            Event::OrderUpdate { symbol, .. } | Event::BalanceUpdate { symbol, .. } |
479            Event::PositionUpdate { symbol, .. } => symbol,
480            // Lifecycle events carry no symbol.
481            Event::ConnectorReady { .. } | Event::SymbolsLoaded { .. } => "",
482        }
483    }
484
485    /// Replace the symbol label on this event in-place.
486    ///
487    /// Used by `Station::subscribe` so each `SubscriptionHandle` sees the
488    /// user-input symbol it passed in `SubscriptionSet::add(...)`, regardless
489    /// of which other consumer first established the underlying multiplex.
490    /// The routing key (raw exchange-native) is unaffected; this only changes
491    /// the cosmetic label that `Event.symbol()` returns to the consumer.
492    pub(crate) fn set_symbol(&mut self, new_symbol: String) {
493        match self {
494            Event::Trade { symbol, .. }
495            | Event::AggTrade { symbol, .. }
496            | Event::Bar { symbol, .. }
497            | Event::Ticker { symbol, .. }
498            | Event::OrderbookSnapshot { symbol, .. }
499            | Event::OrderbookDelta { symbol, .. }
500            | Event::MarkPrice { symbol, .. }
501            | Event::FundingRate { symbol, .. }
502            | Event::OpenInterest { symbol, .. }
503            | Event::Liquidation { symbol, .. }
504            | Event::BlockTrade { symbol, .. }
505            | Event::IndexPrice { symbol, .. }
506            | Event::CompositeIndex { symbol, .. }
507            | Event::OptionGreeks { symbol, .. }
508            | Event::VolatilityIndex { symbol, .. }
509            | Event::HistoricalVolatility { symbol, .. }
510            | Event::LongShortRatio { symbol, .. }
511            | Event::Basis { symbol, .. }
512            | Event::InsuranceFund { symbol, .. }
513            | Event::OrderbookL3 { symbol, .. }
514            | Event::SettlementEvent { symbol, .. }
515            | Event::MarketWarning { symbol, .. }
516            | Event::RiskLimit { symbol, .. }
517            | Event::PredictedFunding { symbol, .. }
518            | Event::FundingSettlement { symbol, .. }
519            | Event::MarkPriceKline { symbol, .. }
520            | Event::IndexPriceKline { symbol, .. }
521            | Event::PremiumIndexKline { symbol, .. }
522            | Event::Footprint { symbol, .. }
523            | Event::OrderUpdate { symbol, .. }
524            | Event::BalanceUpdate { symbol, .. }
525            | Event::PositionUpdate { symbol, .. } => *symbol = new_symbol,
526            // Lifecycle events have no symbol field — no-op.
527            Event::ConnectorReady { .. } | Event::SymbolsLoaded { .. } => {}
528        }
529    }
530    pub fn timestamp_ms(&self) -> i64 {
531        use crate::series::DataPoint;
532        match self {
533            Event::Trade { point, .. } => point.timestamp_ms(),
534            Event::AggTrade { point, .. } => point.timestamp_ms(),
535            Event::Bar { point, .. } => point.timestamp_ms(),
536            Event::Ticker { point, .. } => point.timestamp_ms(),
537            Event::OrderbookSnapshot { point, .. } => point.timestamp_ms(),
538            Event::OrderbookDelta { point, .. } => point.timestamp_ms(),
539            Event::MarkPrice { point, .. } => point.timestamp_ms(),
540            Event::FundingRate { point, .. } => point.timestamp_ms(),
541            Event::OpenInterest { point, .. } => point.timestamp_ms(),
542            Event::Liquidation { point, .. } => point.timestamp_ms(),
543            Event::BlockTrade { point, .. } => point.timestamp_ms(),
544            Event::IndexPrice { point, .. } => point.timestamp_ms(),
545            Event::CompositeIndex { point, .. } => point.timestamp_ms(),
546            Event::OptionGreeks { point, .. } => point.timestamp_ms(),
547            Event::VolatilityIndex { point, .. } => point.timestamp_ms(),
548            Event::HistoricalVolatility { point, .. } => point.timestamp_ms(),
549            Event::LongShortRatio { point, .. } => point.timestamp_ms(),
550            Event::Basis { point, .. } => point.timestamp_ms(),
551            Event::InsuranceFund { point, .. } => point.timestamp_ms(),
552            Event::OrderbookL3 { point, .. } => point.timestamp_ms(),
553            Event::SettlementEvent { point, .. } => point.timestamp_ms(),
554            Event::MarketWarning { point, .. } => point.timestamp_ms(),
555            Event::RiskLimit { point, .. } => point.timestamp_ms(),
556            Event::PredictedFunding { point, .. } => point.timestamp_ms(),
557            Event::FundingSettlement { point, .. } => point.timestamp_ms(),
558            Event::MarkPriceKline { point, .. } => point.timestamp_ms(),
559            Event::IndexPriceKline { point, .. } => point.timestamp_ms(),
560            Event::PremiumIndexKline { point, .. } => point.timestamp_ms(),
561            Event::Footprint { point, .. } => point.timestamp_ms(),
562            Event::OrderUpdate { point, .. } => point.timestamp_ms(),
563            Event::BalanceUpdate { point, .. } => point.timestamp_ms(),
564            Event::PositionUpdate { point, .. } => point.timestamp_ms(),
565            // Lifecycle events: use current epoch ms.
566            Event::ConnectorReady { .. } | Event::SymbolsLoaded { .. } => {
567                chrono::Utc::now().timestamp_millis()
568            }
569        }
570    }
571}
572
573/// Outcome of [`crate::Station::warmup`].
574///
575/// `ok` holds every exchange that connected and had its exchange-info fetched
576/// successfully. `failed` holds exchanges that failed at either the connect or
577/// the REST stage.
578#[derive(Debug, Clone)]
579pub struct WarmupReport {
580    /// Exchanges that connected (and had `get_exchange_info` succeed for at
581    /// least one `AccountType`, if the connector supports it).
582    pub ok: Vec<ExchangeId>,
583    /// Exchanges that failed at connect or REST stage.
584    pub failed: Vec<(ExchangeId, String)>,
585}
586
587/// RAII handle returned from `Station::subscribe`. Dropping releases the
588/// per-StreamKey consumer ref count; when count hits zero the multiplexer
589/// shuts down.
590pub struct SubscriptionHandle {
591    pub(crate) rx: tokio::sync::mpsc::UnboundedReceiver<Event>,
592    pub(crate) _refs: Vec<MultiplexRef>,
593}
594
595impl std::fmt::Debug for SubscriptionHandle {
596    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
597        f.debug_struct("SubscriptionHandle").finish()
598    }
599}
600
601impl SubscriptionHandle {
602    pub async fn recv(&mut self) -> Option<Event> {
603        self.rx.recv().await
604    }
605}
606
607/// Per-stream subscribe failure reported by [`crate::Station::subscribe`].
608///
609/// Carries everything a consumer needs to log + skip without forcing them
610/// to parse a `Display` string. `error.is_not_supported()` distinguishes
611/// venue-doesn't-expose-this-stream (architectural, quiet) from transient
612/// failures (worth surfacing).
613#[derive(Debug)]
614pub struct FailedStream {
615    pub exchange: ExchangeId,
616    pub account_type: AccountType,
617    /// User-input symbol form (NOT the normalized exchange-native form).
618    pub symbol: String,
619    pub stream: Stream,
620    pub error: crate::StationError,
621}
622
623/// Outcome of [`crate::Station::subscribe`] in continue-on-error mode.
624///
625/// `handle` always exists and carries events for every stream in `ok`.
626/// `failed` is a per-stream list of subscribes that did not produce a
627/// live forwarder. The most common entry there is
628/// `StationError::StreamNotSupported` — the venue genuinely does not
629/// expose the requested stream on the WS wire. Other errors (transport,
630/// REST, symbol normalize) also land here so the consumer can log them
631/// without aborting the whole subscribe batch.
632///
633/// `failed` is empty on success — callers that want fail-fast semantics
634/// can simply `if !report.failed.is_empty() { return Err(...) }`.
635pub struct SubscribeReport {
636    pub handle: SubscriptionHandle,
637    pub ok: Vec<SeriesKey>,
638    pub failed: Vec<FailedStream>,
639}
640
641impl SubscribeReport {
642    /// True if every requested stream produced a live forwarder.
643    pub fn is_fully_ok(&self) -> bool { self.failed.is_empty() }
644    /// Convenience: total streams requested (`ok.len() + failed.len()`).
645    pub fn total(&self) -> usize { self.ok.len() + self.failed.len() }
646
647    /// Move all [`MultiplexRef`]s out of the inner [`SubscriptionHandle`]
648    /// into `dest`, returning `Self` with an empty `_refs` list.
649    ///
650    /// Used by [`crate::quota::ConsumerHandle::subscribe`] to take ownership
651    /// of the refs so that dropping the `ConsumerHandle` releases all the
652    /// consumer's subscriptions, even when the caller retains the
653    /// `SubscriptionHandle` for event `recv()`. The upstream broadcast keeps
654    /// emitting as long as any consumer holds a ref — ref extraction does not
655    /// interrupt the event stream.
656    pub(crate) fn take_refs_into(mut self, dest: &mut Vec<MultiplexRef>) -> Self {
657        dest.extend(std::mem::take(&mut self.handle._refs));
658        self
659    }
660}
661
662impl std::fmt::Debug for SubscribeReport {
663    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
664        f.debug_struct("SubscribeReport")
665            .field("ok", &self.ok.len())
666            .field("failed", &self.failed.len())
667            .finish()
668    }
669}
670
671pub(crate) struct MultiplexRef {
672    pub(crate) station: std::sync::Weak<crate::station::StationInner>,
673    pub(crate) key: SeriesKey,
674}
675
676impl Drop for MultiplexRef {
677    fn drop(&mut self) {
678        if let Some(inner) = self.station.upgrade() {
679            inner.release_consumer(&self.key);
680        }
681    }
682}
digdigdig3_station/subscription.rs

digdigdig3_station/
subscription.rs
