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

digdigdig3_station/
subscription.rs
