rustrade_data/exchange/massive/
message.rs

1//! Massive WebSocket market data messages, and the transformer turning them into market events.
2//!
3//! A Massive frame is a JSON array of messages, each naming its channel in `ev`. The shared
4//! connection hands each stream frames holding only the messages of its own subscriptions, so a
5//! stream decodes every message in a frame as its kind's message, and each one separately: one
6//! that fails to decode is reported without losing the rest of the frame.
7
8use super::transformer::{AggregateProvenance, deserialize_first_condition};
9use crate::{
10    books::Level,
11    error::DataError,
12    event::MarketEvent,
13    exchange::Connector,
14    subscriber::shared::elements::excerpt,
15    subscription::{
16        Map, SubscriptionKind,
17        book::{OrderBookL1, OrderBooksL1},
18        candle::{Candle, Candles},
19        quote::{Quote, Quotes},
20        trade::{PublicTrade, PublicTrades},
21    },
22    transformer::ExchangeTransformer,
23};
24use chrono::{DateTime, TimeZone, Utc};
25use rust_decimal::Decimal;
26use rustrade_instrument::Side;
27use rustrade_integration::{
28    Transformer, protocol::websocket::WsMessage, subscription::SubscriptionId,
29};
30use serde::{Deserialize, Deserializer, de::DeserializeOwned};
31use serde_json::value::RawValue;
32use smol_str::{SmolStr, format_smolstr};
33use std::marker::PhantomData;
34use tokio::sync::mpsc;
35
36/// The crypto trade channel, the one whose conditions carry the aggressor's side.
37const CRYPTO_TRADES: &str = "XT";
38
39/// A frame of one stream's messages, each decoded on its own.
40#[derive(Debug)]
41pub struct MassiveMessages<Message>(pub Vec<Result<Message, DataError>>);
42
43impl<'de, Message> Deserialize<'de> for MassiveMessages<Message>
44where
45    Message: DeserializeOwned,
46{
47    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
48    where
49        D: Deserializer<'de>,
50    {
51        let elements = Vec::<&'de RawValue>::deserialize(deserializer)?;
52
53        Ok(Self(
54            elements
55                .into_iter()
56                .map(|raw| {
57                    serde_json::from_str(raw.get()).map_err(|error| {
58                        DataError::Socket(format!(
59                            "failed to decode a Massive message: {error}: {}",
60                            excerpt(raw.get())
61                        ))
62                    })
63                })
64                .collect(),
65        ))
66    }
67}
68
69/// A Massive market data message: what every kind's message has in common.
70pub trait MassiveMessage: DeserializeOwned {
71    /// The channel the message was published on: its `ev`.
72    fn event(&self) -> &str;
73
74    /// The market it concerns, spelled as the subscription spelled it.
75    fn symbol(&self) -> &str;
76
77    /// When the provider stamped it.
78    fn time_exchange(&self) -> DateTime<Utc>;
79
80    /// The subscription it belongs to, as the instrument map keys subscriptions.
81    fn subscription_id(&self) -> SubscriptionId {
82        // The spelling `ExchangeSub` gives the same channel and market.
83        SubscriptionId(format_smolstr!("{}|{}", self.event(), self.symbol()))
84    }
85}
86
87/// A subscription kind Massive's WebSocket serves, and the message it is decoded from.
88pub trait MassiveKind: SubscriptionKind {
89    /// The message the kind's channel publishes.
90    type Message: MassiveMessage;
91
92    /// The market event a message amounts to.
93    fn event(message: Self::Message) -> Self::Event;
94}
95
96/// A trade: `{"ev":"XT","pair":"BTC-USD","p":82857.53,"s":0.01,"t":1759056000000,"c":[2],"i":"1"}`.
97///
98/// Stocks and options spell the market `sym`, crypto `pair`.
99#[derive(Debug, Clone, Deserialize)]
100pub struct MassiveTrade {
101    #[serde(rename = "ev")]
102    pub event: SmolStr,
103    #[serde(alias = "sym", alias = "pair")]
104    pub symbol: SmolStr,
105    #[serde(rename = "p", with = "rust_decimal::serde::float")]
106    pub price: Decimal,
107    #[serde(rename = "s", with = "rust_decimal::serde::float")]
108    pub size: Decimal,
109    /// Milliseconds since the epoch.
110    #[serde(rename = "t")]
111    pub timestamp: i64,
112    /// The first trade condition. On crypto, `1` is a sell and `2` a buy; on stocks and options
113    /// the conditions describe the trade, not its side.
114    #[serde(
115        rename = "c",
116        default,
117        deserialize_with = "deserialize_first_condition"
118    )]
119    pub condition: Option<i32>,
120    /// The trade's identifier, a string on every cluster measured, accepted as a number too.
121    #[serde(rename = "i", default, deserialize_with = "deserialize_id")]
122    pub id: Option<SmolStr>,
123}
124
125/// A quote.
126///
127/// `{"ev":"XQ","pair":"BTC-USD","bp":82857.5,"bs":0.1,"ap":82858.0,"as":0.2,"t":1759056000000}`,
128/// or on forex, which spells the pair `p` and quotes no sizes,
129/// `{"ev":"C","p":"EUR/USD","b":1.1702,"a":1.1703,"t":1759056000000}`. Stocks and options spell the
130/// market `sym`.
131#[derive(Debug, Clone, Deserialize)]
132pub struct MassiveQuote {
133    #[serde(rename = "ev")]
134    pub event: SmolStr,
135    /// `p` is forex's pair. It never meets a price named `p`: no quote carries one.
136    #[serde(alias = "sym", alias = "pair", alias = "p")]
137    pub symbol: SmolStr,
138    #[serde(
139        alias = "bp",
140        alias = "b",
141        with = "rust_decimal::serde::float",
142        default
143    )]
144    pub bid_price: Decimal,
145    /// Absent on forex.
146    #[serde(alias = "bs", with = "rust_decimal::serde::float_option", default)]
147    pub bid_size: Option<Decimal>,
148    #[serde(
149        alias = "ap",
150        alias = "a",
151        with = "rust_decimal::serde::float",
152        default
153    )]
154    pub ask_price: Decimal,
155    /// Absent on forex.
156    #[serde(alias = "as", with = "rust_decimal::serde::float_option", default)]
157    pub ask_size: Option<Decimal>,
158    /// Milliseconds since the epoch.
159    #[serde(rename = "t")]
160    pub timestamp: i64,
161}
162
163/// A per-second or per-minute aggregate:
164/// `{"ev":"XAS","pair":"BTC-USD","o":1.0,"h":1.2,"l":0.9,"c":1.1,"v":10.5,"s":1759056000000,"e":1759056001000}`.
165///
166/// Stocks and options spell the market `sym`, crypto and forex `pair`.
167#[derive(Debug, Clone, Deserialize)]
168pub struct MassiveAggregate {
169    #[serde(rename = "ev")]
170    pub event: SmolStr,
171    #[serde(alias = "sym", alias = "pair")]
172    pub symbol: SmolStr,
173    #[serde(rename = "o", with = "rust_decimal::serde::float")]
174    pub open: Decimal,
175    #[serde(rename = "h", with = "rust_decimal::serde::float")]
176    pub high: Decimal,
177    #[serde(rename = "l", with = "rust_decimal::serde::float")]
178    pub low: Decimal,
179    #[serde(rename = "c", with = "rust_decimal::serde::float")]
180    pub close: Decimal,
181    /// On forex, a count of quote updates rather than a traded volume — see
182    /// [`provenance`](Self::provenance).
183    #[serde(rename = "v", with = "rust_decimal::serde::float")]
184    pub volume: Decimal,
185    /// The window's end in milliseconds since the epoch, exclusive: its start plus the interval.
186    #[serde(rename = "e")]
187    pub end_timestamp: i64,
188    // `z` is deliberately not read. It is the window's average trade *size*, not a trade count, and
189    // this message carries no count at all.
190}
191
192impl MassiveAggregate {
193    /// What the aggregate was built from. Spot FX has no consolidated tape, so Massive builds its
194    /// forex aggregates — `CA` and `CAS` — from quoted bid/ask updates.
195    pub fn provenance(&self) -> AggregateProvenance {
196        if self.event.starts_with('C') {
197            AggregateProvenance::QuoteTape
198        } else {
199            AggregateProvenance::TradeTape
200        }
201    }
202}
203
204impl MassiveMessage for MassiveTrade {
205    fn event(&self) -> &str {
206        &self.event
207    }
208
209    fn symbol(&self) -> &str {
210        &self.symbol
211    }
212
213    fn time_exchange(&self) -> DateTime<Utc> {
214        millis_to_datetime(self.timestamp)
215    }
216}
217
218impl MassiveMessage for MassiveQuote {
219    fn event(&self) -> &str {
220        &self.event
221    }
222
223    fn symbol(&self) -> &str {
224        &self.symbol
225    }
226
227    fn time_exchange(&self) -> DateTime<Utc> {
228        millis_to_datetime(self.timestamp)
229    }
230}
231
232impl MassiveMessage for MassiveAggregate {
233    fn event(&self) -> &str {
234        &self.event
235    }
236
237    fn symbol(&self) -> &str {
238        &self.symbol
239    }
240
241    /// The window's close, which [`Candle::close_time`] is too: the event is stamped when the
242    /// window it summarises ends.
243    fn time_exchange(&self) -> DateTime<Utc> {
244        millis_to_datetime(self.end_timestamp)
245    }
246}
247
248impl MassiveKind for PublicTrades {
249    type Message = MassiveTrade;
250
251    fn event(trade: MassiveTrade) -> PublicTrade {
252        // Only crypto's conditions say who took liquidity; on stocks and options `1` and `2` are
253        // unrelated trade conditions.
254        let side = (trade.event == CRYPTO_TRADES)
255            .then_some(trade.condition)
256            .flatten()
257            .and_then(|condition| match condition {
258                1 => Some(Side::Sell),
259                2 => Some(Side::Buy),
260                _ => None,
261            });
262
263        PublicTrade {
264            id: trade.id.unwrap_or_default(),
265            price: trade.price,
266            amount: trade.size,
267            side,
268        }
269    }
270}
271
272/// The quote as published. Forex quotes carry no sizes, so both amounts are zero there.
273///
274/// A [`Quote`] has no way to say a side is empty, so a side with nothing quoted on it — a zero
275/// price, or none — arrives as a zero price. Check both prices before using
276/// [`Quote::mid_price`] or [`Quote::spread`]; [`OrderBooksL1`] maps such a side to `None` instead.
277impl MassiveKind for Quotes {
278    type Message = MassiveQuote;
279
280    fn event(quote: MassiveQuote) -> Quote {
281        Quote {
282            bid_price: quote.bid_price,
283            bid_amount: quote.bid_size.unwrap_or(Decimal::ZERO),
284            ask_price: quote.ask_price,
285            ask_amount: quote.ask_size.unwrap_or(Decimal::ZERO),
286        }
287    }
288}
289
290/// The quote as a top of book, stamped with the quote's own time.
291///
292/// # A zero price is absent
293/// A side with nothing quoted on it can only arrive as a zero price, or no price at all. A
294/// zero-priced level would assert someone will trade at zero, and a mid-price averaged against it
295/// is wrong in a way nothing downstream can detect. So a zero price maps to `None`, the rule the
296/// other L1 decoders in this crate follow. A missing size at a real price — every forex quote — is
297/// a zero amount: the price is still a quote.
298impl MassiveKind for OrderBooksL1 {
299    type Message = MassiveQuote;
300
301    fn event(quote: MassiveQuote) -> OrderBookL1 {
302        let side = |price: Decimal, size: Option<Decimal>| {
303            (!price.is_zero()).then(|| Level::new(price, size.unwrap_or(Decimal::ZERO)))
304        };
305
306        OrderBookL1 {
307            // Must equal the event's `time_exchange`: downstream state orders L1 updates on it.
308            last_update_time: quote.time_exchange(),
309            best_bid: side(quote.bid_price, quote.bid_size),
310            best_ask: side(quote.ask_price, quote.ask_size),
311        }
312    }
313}
314
315/// The aggregate as a candle.
316///
317/// `close_time` is the wire's window end, which Massive supplies as the exclusive end of the
318/// period (`e == s + interval`), so it satisfies the [`Candle::close_time`] contract as it stands.
319/// A forex candle reports no volume, since its `v` counts quote updates, and no candle reports a
320/// trade count, since the message carries none.
321impl MassiveKind for Candles {
322    type Message = MassiveAggregate;
323
324    fn event(aggregate: MassiveAggregate) -> Candle {
325        let provenance = aggregate.provenance();
326
327        Candle {
328            close_time: aggregate.time_exchange(),
329            open: aggregate.open,
330            high: aggregate.high,
331            low: aggregate.low,
332            close: aggregate.close,
333            volume: provenance.volume(aggregate.volume),
334            trade_count: None,
335        }
336    }
337}
338
339/// Transforms one stream's Massive messages into market events of its kind.
340#[derive(Debug)]
341pub struct MassiveTransformer<Exchange, InstrumentKey, Kind> {
342    instrument_map: Map<InstrumentKey>,
343    phantom: PhantomData<(Exchange, Kind)>,
344}
345
346impl<Exchange, InstrumentKey, Kind> ExchangeTransformer<Exchange, InstrumentKey, Kind>
347    for MassiveTransformer<Exchange, InstrumentKey, Kind>
348where
349    Exchange: Connector + Send,
350    InstrumentKey: Clone + Send + Sync,
351    Kind: MassiveKind + Send,
352    Kind::Event: Send,
353{
354    async fn init(
355        instrument_map: Map<InstrumentKey>,
356        _: &[MarketEvent<InstrumentKey, Kind::Event>],
357        _: mpsc::UnboundedSender<WsMessage>,
358    ) -> Result<Self, DataError> {
359        Ok(Self {
360            instrument_map,
361            phantom: PhantomData,
362        })
363    }
364}
365
366impl<Exchange, InstrumentKey, Kind> Transformer
367    for MassiveTransformer<Exchange, InstrumentKey, Kind>
368where
369    Exchange: Connector,
370    InstrumentKey: Clone,
371    Kind: MassiveKind,
372{
373    type Error = DataError;
374    type Input = MassiveMessages<Kind::Message>;
375    type Output = MarketEvent<InstrumentKey, Kind::Event>;
376    type OutputIter = Vec<Result<Self::Output, Self::Error>>;
377
378    fn transform(&mut self, MassiveMessages(messages): Self::Input) -> Self::OutputIter {
379        let time_received = Utc::now();
380
381        messages
382            .into_iter()
383            .map(|message| {
384                let message = message?;
385                let instrument = self
386                    .instrument_map
387                    .find(&message.subscription_id())
388                    .map_err(|unidentified| DataError::Socket(unidentified.to_string()))?
389                    .clone();
390
391                Ok(MarketEvent {
392                    time_exchange: message.time_exchange(),
393                    time_received,
394                    exchange: Exchange::ID,
395                    instrument,
396                    kind: Kind::event(message),
397                })
398            })
399            .collect()
400    }
401}
402
403/// A trade identifier, as a string or a number.
404fn deserialize_id<'de, D>(deserializer: D) -> Result<Option<SmolStr>, D::Error>
405where
406    D: Deserializer<'de>,
407{
408    #[derive(Deserialize)]
409    #[serde(untagged)]
410    enum Id {
411        Text(SmolStr),
412        Number(u64),
413    }
414
415    Ok(Option::<Id>::deserialize(deserializer)?.map(|id| match id {
416        Id::Text(text) => text,
417        Id::Number(number) => format_smolstr!("{number}"),
418    }))
419}
420
421/// A millisecond timestamp as an instant, or the epoch, with a warning, if it is out of range.
422fn millis_to_datetime(millis: i64) -> DateTime<Utc> {
423    Utc.timestamp_millis_opt(millis)
424        .single()
425        .unwrap_or_else(|| {
426            tracing::warn!(
427                millis,
428                "out-of-range millisecond timestamp; using UNIX_EPOCH"
429            );
430            DateTime::<Utc>::UNIX_EPOCH
431        })
432}
433
434#[cfg(test)]
435#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
436mod tests {
437    use super::*;
438    use crate::{
439        Identifier,
440        exchange::massive::{
441            Massive, MassiveServerCrypto, MassiveServerForex, MassiveServerStocks,
442            channel::MassiveChannel,
443            market::{MassiveMarket, MassiveServer},
444        },
445        subscriber::mapper::{SubscriptionMapper, WebSocketSubMapper},
446        subscription::{Subscription, candle::CandleInterval},
447    };
448    use rust_decimal_macros::dec;
449    use rustrade_instrument::{
450        exchange::ExchangeId,
451        instrument::market_data::{MarketDataInstrument, kind::MarketDataInstrumentKind},
452    };
453
454    type Events<Kind> = Vec<
455        Result<MarketEvent<MarketDataInstrument, <Kind as SubscriptionKind>::Event>, DataError>,
456    >;
457
458    /// Transform `frame` for a stream of `kind` holding `pairs` on `Server`'s cluster, keyed by the
459    /// map the subscriber builds, so a message must name its subscription as the stream does.
460    fn transform<Server, Kind>(pairs: &[(&str, &str)], kind: Kind, frame: &str) -> Events<Kind>
461    where
462        Server: MassiveServer,
463        Kind: MassiveKind,
464        Subscription<Massive<Server>, MarketDataInstrument, Kind>:
465            Identifier<MassiveChannel> + Identifier<MassiveMarket>,
466    {
467        let subscriptions = pairs
468            .iter()
469            .map(|(base, quote)| {
470                Subscription::new(
471                    Massive::<Server>::default(),
472                    MarketDataInstrument::new(*base, *quote, MarketDataInstrumentKind::Spot),
473                    kind.clone(),
474                )
475            })
476            .collect::<Vec<_>>();
477
478        let mut transformer = MassiveTransformer::<Massive<Server>, _, Kind> {
479            instrument_map: WebSocketSubMapper::map(&subscriptions).instrument_map,
480            phantom: PhantomData,
481        };
482
483        transformer.transform(serde_json::from_str(frame).unwrap())
484    }
485
486    fn only<Event>(mut events: Vec<Result<Event, DataError>>) -> Event {
487        assert_eq!(events.len(), 1);
488        events.remove(0).unwrap()
489    }
490
491    #[test]
492    fn a_crypto_trade_carries_its_aggressor_side() {
493        let event = only(transform::<MassiveServerCrypto, _>(
494            &[("btc", "usd")],
495            PublicTrades,
496            r#"[{"ev":"XT","pair":"BTC-USD","p":45230.5,"s":0.5,"t":1704067200000,"c":[2],"i":"trade123"}]"#,
497        ));
498
499        assert_eq!(event.instrument.base.name().as_str(), "btc");
500        assert_eq!(event.exchange, ExchangeId::MassiveCrypto);
501        assert_eq!(event.time_exchange.timestamp_millis(), 1704067200000);
502        assert_eq!(event.kind.price, dec!(45230.5));
503        assert_eq!(event.kind.amount, dec!(0.5));
504        assert_eq!(event.kind.side, Some(Side::Buy));
505        assert_eq!(event.kind.id, "trade123");
506    }
507
508    /// On stocks and options, conditions 1 and 2 describe the trade, not who took liquidity.
509    #[test]
510    fn a_stock_trade_has_no_side_whatever_its_conditions() {
511        let event = only(transform::<MassiveServerStocks, _>(
512            &[("aapl", "usd")],
513            PublicTrades,
514            r#"[{"ev":"T","sym":"AAPL","p":150.25,"s":100,"t":1704067200000,"c":[2],"i":12345}]"#,
515        ));
516
517        assert_eq!(event.exchange, ExchangeId::MassiveStocks);
518        assert_eq!(event.kind.price, dec!(150.25));
519        assert_eq!(event.kind.amount, dec!(100));
520        assert_eq!(event.kind.side, None);
521        // A numeric identifier is accepted too.
522        assert_eq!(event.kind.id, "12345");
523    }
524
525    #[test]
526    fn a_crypto_quote_is_a_quote_and_a_top_of_book() {
527        let frame = r#"[{"ev":"XQ","pair":"BTC-USD","bp":45220.0,"bs":2.5,"ap":45240.0,"as":3.0,"lp":45230.0,"ls":0.1,"t":1704067200000}]"#;
528
529        let quote = only(transform::<MassiveServerCrypto, _>(
530            &[("btc", "usd")],
531            Quotes,
532            frame,
533        ));
534        assert_eq!(
535            quote.kind,
536            Quote {
537                bid_price: dec!(45220.0),
538                bid_amount: dec!(2.5),
539                ask_price: dec!(45240.0),
540                ask_amount: dec!(3.0),
541            }
542        );
543
544        let book = only(transform::<MassiveServerCrypto, _>(
545            &[("btc", "usd")],
546            OrderBooksL1,
547            frame,
548        ));
549        assert_eq!(
550            book.kind.best_bid,
551            Some(Level::new(dec!(45220.0), dec!(2.5)))
552        );
553        assert_eq!(
554            book.kind.best_ask,
555            Some(Level::new(dec!(45240.0), dec!(3.0)))
556        );
557        assert_eq!(book.kind.last_update_time, book.time_exchange);
558    }
559
560    /// Forex names the pair `p`, slashed, and quotes no sizes.
561    #[test]
562    fn a_forex_quote_has_zero_amounts() {
563        let frame =
564            r#"[{"ev":"C","p":"EUR/USD","b":1.085,"a":1.0852,"i":0,"x":48,"t":1704067200000}]"#;
565
566        let quote = only(transform::<MassiveServerForex, _>(
567            &[("eur", "usd")],
568            Quotes,
569            frame,
570        ));
571        assert_eq!(quote.exchange, ExchangeId::MassiveForex);
572        assert_eq!(quote.kind.bid_price, dec!(1.085));
573        assert_eq!(quote.kind.ask_price, dec!(1.0852));
574        assert_eq!(quote.kind.bid_amount, Decimal::ZERO);
575        assert_eq!(quote.kind.ask_amount, Decimal::ZERO);
576
577        let book = only(transform::<MassiveServerForex, _>(
578            &[("eur", "usd")],
579            OrderBooksL1,
580            frame,
581        ));
582        assert_eq!(
583            book.kind.best_bid,
584            Some(Level::new(dec!(1.085), Decimal::ZERO))
585        );
586    }
587
588    #[test]
589    fn a_zero_priced_side_is_absent_from_the_top_of_book() {
590        let book = only(transform::<MassiveServerCrypto, _>(
591            &[("btc", "usd")],
592            OrderBooksL1,
593            r#"[{"ev":"XQ","pair":"BTC-USD","bp":0,"bs":0,"ap":45240.0,"as":3.0,"t":1704067200000}]"#,
594        ));
595
596        assert_eq!(book.kind.best_bid, None);
597        assert!(book.kind.best_ask.is_some());
598    }
599
600    /// `e` is the window's exclusive end, `s` plus the interval, and is the candle's close time.
601    /// `z` is an average trade size, routinely fractional, and is not read as a count.
602    #[test]
603    fn an_aggregate_is_a_candle_closing_at_its_window_end() {
604        let start = 1704067200000_i64;
605        let candle = only(transform::<MassiveServerStocks, _>(
606            &[("aapl", "usd")],
607            Candles {
608                interval: CandleInterval::Min1,
609            },
610            &format!(
611                r#"[{{"ev":"AM","sym":"AAPL","o":150.1,"h":150.5,"l":150.05,"c":150.25,"v":1000,"vw":150.2,"s":{start},"e":{},"z":78.5}}]"#,
612                start + 60_000
613            ),
614        ));
615
616        assert_eq!(candle.time_exchange.timestamp_millis(), start + 60_000);
617        assert_eq!(candle.kind.close_time, candle.time_exchange);
618        assert_eq!(candle.kind.open, dec!(150.1));
619        assert_eq!(candle.kind.high, dec!(150.5));
620        assert_eq!(candle.kind.low, dec!(150.05));
621        assert_eq!(candle.kind.close, dec!(150.25));
622        assert_eq!(candle.kind.volume, Some(dec!(1000)));
623        assert_eq!(candle.kind.trade_count, None);
624    }
625
626    /// Spot FX has no trade tape: Massive builds forex aggregates from quote updates, so `v` counts
627    /// those, and reporting it as traded volume would be a silent lie.
628    #[test]
629    fn a_forex_aggregate_reports_no_volume() {
630        let candle = only(transform::<MassiveServerForex, _>(
631            &[("eur", "usd")],
632            Candles {
633                interval: CandleInterval::Sec1,
634            },
635            r#"[{"ev":"CAS","pair":"EUR/USD","o":1.1,"h":1.2,"l":1.0,"c":1.15,"v":4321,"s":1704067200000,"e":1704067201000}]"#,
636        ));
637
638        assert_eq!(candle.kind.volume, None);
639        assert_eq!(candle.kind.trade_count, None);
640    }
641
642    /// Each message decodes on its own: one that fails is reported, and the rest of its frame is
643    /// kept.
644    #[test]
645    fn a_message_that_fails_to_decode_loses_only_itself() {
646        let events = transform::<MassiveServerCrypto, _>(
647            &[("btc", "usd"), ("eth", "usd")],
648            PublicTrades,
649            r#"[{"ev":"XT","pair":"BTC-USD","p":1.0,"s":1.0,"t":1,"i":"1"},
650                {"ev":"XT","pair":"ETH-USD","p":"not a price","s":1.0,"t":1},
651                {"ev":"XT","pair":"ETH-USD","p":2.0,"s":1.0,"t":1,"i":"2"}]"#,
652        );
653
654        assert_eq!(events.len(), 3);
655        assert_eq!(events[0].as_ref().unwrap().kind.price, dec!(1.0));
656        let error = events[1].as_ref().unwrap_err().to_string();
657        assert!(
658            error.contains("failed to decode a Massive message"),
659            "{error}"
660        );
661        assert_eq!(events[2].as_ref().unwrap().kind.price, dec!(2.0));
662    }
663
664    #[test]
665    fn a_message_for_no_subscription_of_the_stream_is_an_error() {
666        let events = transform::<MassiveServerCrypto, _>(
667            &[("btc", "usd")],
668            PublicTrades,
669            r#"[{"ev":"XT","pair":"ETH-USD","p":1.0,"s":1.0,"t":1}]"#,
670        );
671
672        assert!(events[0].is_err());
673    }
674}
rustrade_data/exchange/massive/message.rs

rustrade_data/exchange/massive/
message.rs
