wickra_data/live/
binance.rs

1//! Binance spot WebSocket kline feed.
2//!
3//! Subscribes to Binance's `<symbol>@kline_<interval>` stream and emits a
4//! [`KlineEvent`] every time the server pushes a new tick. The event tells you
5//! whether the current candle is still open or has just closed.
6//!
7//! Example (requires the `live-binance` feature):
8//!
9//! ```no_run
10//! use wickra_data::live::binance::{BinanceKlineStream, Interval};
11//! # async fn run() -> wickra_data::Result<()> {
12//! let mut stream = BinanceKlineStream::connect(&["BTCUSDT".to_string()], Interval::OneMinute).await?;
13//! while let Some(event) = stream.next_event().await? {
14//!     if event.is_closed {
15//!         println!("closed {} @ {}", event.symbol, event.candle.close);
16//!     }
17//! }
18//! # Ok(()) }
19//! ```
20
21use std::time::Duration;
22
23use futures_util::SinkExt;
24use futures_util::StreamExt;
25use serde::Deserialize;
26use tokio::net::TcpStream;
27use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
28use tokio_tungstenite::tungstenite::Message;
29use tokio_tungstenite::MaybeTlsStream;
30use tokio_tungstenite::WebSocketStream;
31
32use crate::error::{Error, Result};
33use wickra_core::Candle;
34
35/// Tunable knobs for a [`BinanceKlineStream`]. The defaults match Binance's
36/// public production endpoint and are right for almost every caller; the
37/// fields exist so an integration test or a Binance Testnet user can point
38/// the stream at a different base URL and shrink the reconnect timing.
39#[derive(Debug, Clone)]
40pub struct BinanceConfig {
41    /// WebSocket endpoint **without** path (e.g. `"wss://stream.binance.com:9443"`).
42    /// The combined-stream path `/stream?streams=…` is appended internally.
43    pub base_url: String,
44    /// Maximum time to wait for the next inbound frame before treating the
45    /// connection as stalled. Binance pings roughly every 3 minutes, so a
46    /// healthy but quiet stream stays comfortably inside the 300 s default.
47    pub read_timeout: Duration,
48    /// Delay before the first reconnect attempt; doubles on each failure up
49    /// to [`Self::max_reconnect_backoff`].
50    pub initial_reconnect_delay: Duration,
51    /// Upper bound on the exponential reconnect backoff.
52    pub max_reconnect_backoff: Duration,
53    /// How many times [`BinanceKlineStream::next_event`] retries a dropped
54    /// connection before surfacing the last error. Must be `>= 1`.
55    pub max_reconnect_attempts: u32,
56    /// Upper bound on an inbound WebSocket message. Kline frames are tiny;
57    /// this only caps a pathological or hostile server from forcing an
58    /// unbounded allocation.
59    pub max_message_size: usize,
60    /// Upper bound on a single inbound WebSocket frame.
61    pub max_frame_size: usize,
62}
63
64impl Default for BinanceConfig {
65    fn default() -> Self {
66        Self {
67            base_url: "wss://stream.binance.com:9443".to_string(),
68            read_timeout: Duration::from_secs(300),
69            initial_reconnect_delay: Duration::from_secs(1),
70            max_reconnect_backoff: Duration::from_secs(30),
71            max_reconnect_attempts: 6,
72            max_message_size: 8 << 20,
73            max_frame_size: 2 << 20,
74        }
75    }
76}
77
78/// Supported Binance kline intervals. The `as_str` value matches Binance's
79/// wire-format strings (`"1m"`, `"5m"`, `"1h"`, etc.).
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub enum Interval {
82    OneSecond,
83    OneMinute,
84    ThreeMinutes,
85    FiveMinutes,
86    FifteenMinutes,
87    ThirtyMinutes,
88    OneHour,
89    TwoHours,
90    FourHours,
91    SixHours,
92    EightHours,
93    TwelveHours,
94    OneDay,
95    ThreeDays,
96    OneWeek,
97    OneMonth,
98}
99
100impl Interval {
101    /// Wire-format string used in the stream name.
102    pub fn as_str(self) -> &'static str {
103        match self {
104            Self::OneSecond => "1s",
105            Self::OneMinute => "1m",
106            Self::ThreeMinutes => "3m",
107            Self::FiveMinutes => "5m",
108            Self::FifteenMinutes => "15m",
109            Self::ThirtyMinutes => "30m",
110            Self::OneHour => "1h",
111            Self::TwoHours => "2h",
112            Self::FourHours => "4h",
113            Self::SixHours => "6h",
114            Self::EightHours => "8h",
115            Self::TwelveHours => "12h",
116            Self::OneDay => "1d",
117            Self::ThreeDays => "3d",
118            Self::OneWeek => "1w",
119            Self::OneMonth => "1M",
120        }
121    }
122}
123
124/// One push from the Binance kline stream.
125#[derive(Debug, Clone)]
126pub struct KlineEvent {
127    /// Symbol in lowercase form as sent by Binance (e.g. `"btcusdt"`).
128    pub symbol: String,
129    /// Interval the candle belongs to.
130    pub interval: Interval,
131    /// Candle in its current state (may still be open).
132    pub candle: Candle,
133    /// Whether the candle has been closed by the server. Closed events are the
134    /// only ones safe to use for bar-completion logic.
135    pub is_closed: bool,
136}
137
138/// A live Binance kline stream.
139#[derive(Debug)]
140pub struct BinanceKlineStream {
141    socket: WebSocketStream<MaybeTlsStream<TcpStream>>,
142    /// Lowercased symbols the stream is subscribed to. Retained so the
143    /// connection can be rebuilt on a reconnect.
144    symbols: Vec<String>,
145    /// Interval requested at connect time. Used to tag every event.
146    interval: Interval,
147    /// `true` once the caller invoked [`close`](Self::close). A closed stream
148    /// is never polled or reconnected again.
149    closed: bool,
150    /// Timing / sizing knobs. Retained so reconnects honour the same config.
151    config: BinanceConfig,
152}
153
154/// Wire-format representation of an incoming Binance kline tick. Public so callers
155/// can deserialize it themselves if they prefer.
156#[derive(Debug, Clone, Deserialize)]
157pub struct RawWsEnvelope {
158    /// Stream name, e.g. `"btcusdt@kline_1m"`.
159    pub stream: String,
160    pub data: RawKlinePayload,
161}
162
163#[derive(Debug, Clone, Deserialize)]
164pub struct RawKlinePayload {
165    #[serde(rename = "e")]
166    pub event_type: String,
167    #[serde(rename = "E")]
168    pub event_time: i64,
169    #[serde(rename = "s")]
170    pub symbol: String,
171    #[serde(rename = "k")]
172    pub kline: RawKline,
173}
174
175#[derive(Debug, Clone, Deserialize)]
176pub struct RawKline {
177    #[serde(rename = "t")]
178    pub open_time: i64,
179    #[serde(rename = "T")]
180    pub close_time: i64,
181    #[serde(rename = "s")]
182    pub symbol: String,
183    #[serde(rename = "i")]
184    pub interval: String,
185    #[serde(rename = "o")]
186    pub open: String,
187    #[serde(rename = "c")]
188    pub close: String,
189    #[serde(rename = "h")]
190    pub high: String,
191    #[serde(rename = "l")]
192    pub low: String,
193    #[serde(rename = "v")]
194    pub volume: String,
195    #[serde(rename = "x")]
196    pub is_closed: bool,
197}
198
199impl BinanceKlineStream {
200    /// Open a raw combined-stream WebSocket for the given (already-lowercased)
201    /// symbols.
202    async fn open(
203        symbols: &[String],
204        interval: Interval,
205        config: &BinanceConfig,
206    ) -> Result<WebSocketStream<MaybeTlsStream<TcpStream>>> {
207        let streams: Vec<String> = symbols
208            .iter()
209            .map(|s| format!("{}@kline_{}", s, interval.as_str()))
210            .collect();
211        let url = format!("{}/stream?streams={}", config.base_url, streams.join("/"));
212        let url = url::Url::parse(&url).map_err(|e| Error::Malformed(e.to_string()))?;
213        // tokio-tungstenite 0.29's WebSocketConfig is #[non_exhaustive],
214        // so the only way to set fields is via the builder-style methods on
215        // a fresh `default()` value.
216        let ws_config = WebSocketConfig::default()
217            .max_message_size(Some(config.max_message_size))
218            .max_frame_size(Some(config.max_frame_size));
219        let (socket, _) =
220            tokio_tungstenite::connect_async_with_config(url.as_str(), Some(ws_config), false)
221                .await?;
222        Ok(socket)
223    }
224
225    /// Connect to Binance's combined-stream endpoint for one or more symbols.
226    ///
227    /// Symbols may be passed in either case; they are lowercased to match
228    /// Binance's stream-name conventions. A dropped or stalled connection is
229    /// re-established transparently by [`next_event`](Self::next_event).
230    pub async fn connect(symbols: &[String], interval: Interval) -> Result<Self> {
231        Self::connect_with_config(symbols, interval, BinanceConfig::default()).await
232    }
233
234    /// Connect with a custom [`BinanceConfig`]. Useful for Binance Testnet
235    /// (`"wss://testnet.binance.vision"`) or for shrinking the reconnect
236    /// timing in integration tests.
237    pub async fn connect_with_config(
238        symbols: &[String],
239        interval: Interval,
240        config: BinanceConfig,
241    ) -> Result<Self> {
242        if symbols.is_empty() {
243            return Err(Error::Malformed(
244                "BinanceKlineStream requires at least one symbol".into(),
245            ));
246        }
247        let symbols: Vec<String> = symbols.iter().map(|s| s.to_lowercase()).collect();
248        let socket = Self::open(&symbols, interval, &config).await?;
249        Ok(Self {
250            socket,
251            symbols,
252            interval,
253            closed: false,
254            config,
255        })
256    }
257
258    /// Whether the caller has closed the stream. Once closed, every further
259    /// [`next_event`](Self::next_event) call yields `Ok(None)` immediately.
260    pub fn is_closed(&self) -> bool {
261        self.closed
262    }
263
264    /// Re-establish a dropped connection with exponential backoff. Returns the
265    /// last error if every attempt fails.
266    async fn reconnect(&mut self) -> Result<()> {
267        let mut delay = self.config.initial_reconnect_delay;
268        let mut last_err = None;
269        for _ in 0..self.config.max_reconnect_attempts {
270            tokio::time::sleep(delay).await;
271            match Self::open(&self.symbols, self.interval, &self.config).await {
272                Ok(socket) => {
273                    self.socket = socket;
274                    return Ok(());
275                }
276                Err(e) => {
277                    last_err = Some(e);
278                    delay = delay
279                        .saturating_mul(2)
280                        .min(self.config.max_reconnect_backoff);
281                }
282            }
283        }
284        Err(last_err.expect("max_reconnect_attempts is non-zero"))
285    }
286
287    /// Receive the next kline event. A dropped, errored or stalled connection
288    /// is re-established transparently (exponential backoff, up to
289    /// [`BinanceConfig::max_reconnect_attempts`]); an exhausted reconnect
290    /// surfaces as `Err`. `Ok(None)` is returned only after the caller has
291    /// [`close`](Self::close)d the stream.
292    pub async fn next_event(&mut self) -> Result<Option<KlineEvent>> {
293        if self.closed {
294            return Ok(None);
295        }
296        loop {
297            // A protocol error, a clean server close, or a read stall are all
298            // transient: reconnect with backoff and resume reading.
299            let Ok(Some(Ok(msg))) =
300                tokio::time::timeout(self.config.read_timeout, self.socket.next()).await
301            else {
302                self.reconnect().await?;
303                continue;
304            };
305            match msg {
306                Message::Text(text) => {
307                    if let Some(event) = Self::parse_frame(&text, self.interval)? {
308                        return Ok(Some(event));
309                    }
310                    // Non-kline frame (subscription ack / heartbeat / error):
311                    // skip it and keep reading.
312                }
313                Message::Binary(bytes) => {
314                    let text = String::from_utf8_lossy(&bytes);
315                    if let Some(event) = Self::parse_frame(&text, self.interval)? {
316                        return Ok(Some(event));
317                    }
318                }
319                Message::Ping(payload) => {
320                    // Best-effort Pong reply. If the write fails the
321                    // connection is already dead — the next read will
322                    // surface the error and trigger reconnect through
323                    // the timeout/err arm above.
324                    let _ = self.socket.send(Message::Pong(payload)).await;
325                }
326                Message::Pong(_) | Message::Frame(_) => {}
327                Message::Close(_) => {
328                    self.reconnect().await?;
329                }
330            }
331        }
332    }
333
334    /// Parse one raw WebSocket text frame.
335    ///
336    /// Returns `Ok(Some(event))` for a kline frame, `Ok(None)` for any other
337    /// frame (subscription acknowledgements, error objects, heartbeats), and
338    /// `Err` only when a frame that *is* a kline fails to decode.
339    fn parse_frame(text: &str, interval: Interval) -> Result<Option<KlineEvent>> {
340        let value: serde_json::Value = serde_json::from_str(text)?;
341        // Combined-stream kline frames carry `data.e == "kline"`. Everything
342        // else on the socket is control traffic that must not abort the feed.
343        let is_kline = value
344            .get("data")
345            .and_then(|d| d.get("e"))
346            .and_then(serde_json::Value::as_str)
347            == Some("kline");
348        if !is_kline {
349            return Ok(None);
350        }
351        let envelope: RawWsEnvelope = serde_json::from_value(value)?;
352        Ok(Some(envelope.into_event(interval)?))
353    }
354
355    /// Close the underlying socket cleanly and mark the stream closed. After
356    /// this, [`next_event`](Self::next_event) yields `Ok(None)` and never
357    /// reconnects.
358    pub async fn close(&mut self) -> Result<()> {
359        self.closed = true;
360        self.socket.close(None).await?;
361        Ok(())
362    }
363}
364
365impl RawWsEnvelope {
366    fn into_event(self, interval: Interval) -> Result<KlineEvent> {
367        let k = self.data.kline;
368        let open: f64 = k
369            .open
370            .parse()
371            .map_err(|_| Error::Malformed(format!("bad open '{}'", k.open)))?;
372        let high: f64 = k
373            .high
374            .parse()
375            .map_err(|_| Error::Malformed(format!("bad high '{}'", k.high)))?;
376        let low: f64 = k
377            .low
378            .parse()
379            .map_err(|_| Error::Malformed(format!("bad low '{}'", k.low)))?;
380        let close: f64 = k
381            .close
382            .parse()
383            .map_err(|_| Error::Malformed(format!("bad close '{}'", k.close)))?;
384        let volume: f64 = k
385            .volume
386            .parse()
387            .map_err(|_| Error::Malformed(format!("bad volume '{}'", k.volume)))?;
388        let candle = Candle::new(open, high, low, close, volume, k.open_time)?;
389        Ok(KlineEvent {
390            symbol: self.data.symbol.to_lowercase(),
391            interval,
392            candle,
393            is_closed: k.is_closed,
394        })
395    }
396}
397
398#[cfg(test)]
399mod tests {
400    use super::*;
401
402    #[test]
403    fn interval_as_str_covers_every_variant() {
404        // Wire-format strings are part of Binance's public protocol — pin
405        // every mapping so a typo here is caught immediately.
406        let pairs: &[(Interval, &str)] = &[
407            (Interval::OneSecond, "1s"),
408            (Interval::OneMinute, "1m"),
409            (Interval::ThreeMinutes, "3m"),
410            (Interval::FiveMinutes, "5m"),
411            (Interval::FifteenMinutes, "15m"),
412            (Interval::ThirtyMinutes, "30m"),
413            (Interval::OneHour, "1h"),
414            (Interval::TwoHours, "2h"),
415            (Interval::FourHours, "4h"),
416            (Interval::SixHours, "6h"),
417            (Interval::EightHours, "8h"),
418            (Interval::TwelveHours, "12h"),
419            (Interval::OneDay, "1d"),
420            (Interval::ThreeDays, "3d"),
421            (Interval::OneWeek, "1w"),
422            (Interval::OneMonth, "1M"),
423        ];
424        for (iv, expected) in pairs {
425            assert_eq!(iv.as_str(), *expected);
426        }
427    }
428
429    #[test]
430    fn binance_config_default_matches_production_endpoint() {
431        let cfg = BinanceConfig::default();
432        assert_eq!(cfg.base_url, "wss://stream.binance.com:9443");
433        assert_eq!(cfg.read_timeout, Duration::from_secs(300));
434        assert_eq!(cfg.initial_reconnect_delay, Duration::from_secs(1));
435        assert_eq!(cfg.max_reconnect_backoff, Duration::from_secs(30));
436        assert_eq!(cfg.max_reconnect_attempts, 6);
437        assert_eq!(cfg.max_message_size, 8 << 20);
438        assert_eq!(cfg.max_frame_size, 2 << 20);
439    }
440
441    #[tokio::test]
442    async fn connect_rejects_an_empty_symbol_list() {
443        let err = BinanceKlineStream::connect(&[], Interval::OneMinute)
444            .await
445            .unwrap_err();
446        assert!(matches!(err, Error::Malformed(_)));
447    }
448
449    #[test]
450    fn parses_real_binance_payload() {
451        // Sample event format from Binance's public docs (truncated).
452        let json = r#"{
453            "stream": "btcusdt@kline_1m",
454            "data": {
455              "e": "kline",
456              "E": 1700000000000,
457              "s": "BTCUSDT",
458              "k": {
459                "t": 1700000000000,
460                "T": 1700000059999,
461                "s": "BTCUSDT",
462                "i": "1m",
463                "f": 1,
464                "L": 100,
465                "o": "30000.0",
466                "c": "30050.0",
467                "h": "30100.0",
468                "l": "29950.0",
469                "v": "12.5",
470                "n": 50,
471                "x": false,
472                "q": "375000.0",
473                "V": "6.25",
474                "Q": "187500.0",
475                "B": "0"
476              }
477            }
478        }"#;
479        let env: RawWsEnvelope = serde_json::from_str(json).unwrap();
480        let evt = env.into_event(Interval::OneMinute).unwrap();
481        assert_eq!(evt.symbol, "btcusdt");
482        assert_eq!(evt.candle.open, 30_000.0);
483        assert_eq!(evt.candle.close, 30_050.0);
484        assert!(!evt.is_closed);
485        assert_eq!(evt.interval, Interval::OneMinute);
486    }
487
488    #[test]
489    fn rejects_non_parsable_numbers() {
490        let json = r#"{
491            "stream": "btcusdt@kline_1m",
492            "data": {
493              "e": "kline", "E": 0, "s": "BTCUSDT",
494              "k": {
495                "t": 0, "T": 0, "s": "BTCUSDT", "i": "1m",
496                "f": 0, "L": 0,
497                "o": "not-a-number", "c": "0", "h": "0", "l": "0",
498                "v": "0", "n": 0, "x": false, "q": "0", "V": "0", "Q": "0", "B": "0"
499              }
500            }
501        }"#;
502        let env: RawWsEnvelope = serde_json::from_str(json).unwrap();
503        let err = env.into_event(Interval::OneMinute).unwrap_err();
504        assert!(matches!(err, Error::Malformed(_)));
505    }
506
507    #[test]
508    fn skips_non_kline_frames() {
509        // Subscription acknowledgement: skipped, never an error.
510        let ack = r#"{"result":null,"id":1}"#;
511        assert!(BinanceKlineStream::parse_frame(ack, Interval::OneMinute)
512            .unwrap()
513            .is_none());
514        // Error object: also skipped.
515        let err = r#"{"error":{"code":2,"msg":"Invalid request"}}"#;
516        assert!(BinanceKlineStream::parse_frame(err, Interval::OneMinute)
517            .unwrap()
518            .is_none());
519    }
520
521    #[test]
522    fn parse_frame_decodes_a_kline() {
523        let json = r#"{
524            "stream": "btcusdt@kline_1m",
525            "data": {
526              "e": "kline", "E": 1700000000000, "s": "BTCUSDT",
527              "k": {
528                "t": 1700000000000, "T": 1700000059999, "s": "BTCUSDT", "i": "1m",
529                "f": 1, "L": 100, "o": "30000.0", "c": "30050.0", "h": "30100.0",
530                "l": "29950.0", "v": "12.5", "n": 50, "x": true,
531                "q": "375000.0", "V": "6.25", "Q": "187500.0", "B": "0"
532              }
533            }
534        }"#;
535        let event = BinanceKlineStream::parse_frame(json, Interval::OneMinute)
536            .unwrap()
537            .expect("a kline frame yields an event");
538        assert_eq!(event.symbol, "btcusdt");
539        assert!(event.is_closed);
540    }
541
542    // ====================================================================
543    // Mock WebSocket server: drives the async / reconnect / control-frame
544    // paths against a `127.0.0.1` listener instead of the real Binance
545    // endpoint. Each test gets its own port (`bind("127.0.0.1:0")`).
546    // ====================================================================
547
548    use std::sync::atomic::{AtomicU32, Ordering};
549    use std::sync::Arc;
550    use tokio::net::TcpListener;
551
552    /// A kline JSON frame matching Binance's combined-stream envelope. Always
553    /// reports `is_closed = true` so the test can assert on the flag.
554    fn sample_kline_text() -> String {
555        r#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":1700000000000,"s":"BTCUSDT","k":{"t":1700000000000,"T":1700000059999,"s":"BTCUSDT","i":"1m","f":1,"L":100,"o":"30000.0","c":"30050.0","h":"30100.0","l":"29950.0","v":"12.5","n":50,"x":true,"q":"375000.0","V":"6.25","Q":"187500.0","B":"0"}}}"#.to_string()
556    }
557
558    /// Test-tuned [`BinanceConfig`]: aim at the given mock and shrink every
559    /// timer so a failing reconnect loop completes in milliseconds.
560    fn test_config(base_url: String) -> BinanceConfig {
561        BinanceConfig {
562            base_url,
563            read_timeout: Duration::from_millis(200),
564            initial_reconnect_delay: Duration::from_millis(5),
565            max_reconnect_backoff: Duration::from_millis(10),
566            max_reconnect_attempts: 3,
567            ..BinanceConfig::default()
568        }
569    }
570
571    /// Spawn a mock WS server that accepts one connection, drops the
572    /// listener (so any reconnect lands on a refused port), and then hands
573    /// the upgraded socket to `handler`. Every step `.unwrap()`s — a failure
574    /// here is a bug in the test scaffolding, not in the production code.
575    async fn one_shot_server<F, Fut>(handler: F) -> String
576    where
577        F: FnOnce(WebSocketStream<TcpStream>) -> Fut + Send + 'static,
578        Fut: std::future::Future<Output = ()> + Send + 'static,
579    {
580        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
581        let base_url = format!("ws://{}", listener.local_addr().unwrap());
582        tokio::spawn(async move {
583            let (stream, _) = listener.accept().await.unwrap();
584            drop(listener);
585            let ws = tokio_tungstenite::accept_async(stream).await.unwrap();
586            handler(ws).await;
587        });
588        base_url
589    }
590
591    /// Spawn a mock WS server that accepts exactly `n_accepts` connections
592    /// and invokes `handler` for each (with a zero-based index). Returns a
593    /// [`JoinHandle`](tokio::task::JoinHandle) the test can await so every
594    /// spawned handler is guaranteed to reach its closing brace before the
595    /// runtime is torn down.
596    async fn multi_shot_server<F, Fut>(
597        n_accepts: u32,
598        handler: F,
599    ) -> (String, tokio::task::JoinHandle<()>)
600    where
601        F: Fn(u32, WebSocketStream<TcpStream>) -> Fut + Send + Sync + 'static,
602        Fut: std::future::Future<Output = ()> + Send + 'static,
603    {
604        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
605        let base_url = format!("ws://{}", listener.local_addr().unwrap());
606        let handler = Arc::new(handler);
607        let h = tokio::spawn(async move {
608            let mut joins = Vec::with_capacity(n_accepts as usize);
609            for index in 0..n_accepts {
610                let (stream, _) = listener.accept().await.unwrap();
611                let handler = handler.clone();
612                joins.push(tokio::spawn(async move {
613                    let ws = tokio_tungstenite::accept_async(stream).await.unwrap();
614                    handler(index, ws).await;
615                }));
616            }
617            for j in joins {
618                j.await.unwrap();
619            }
620        });
621        (base_url, h)
622    }
623
624    #[tokio::test]
625    async fn next_event_decodes_a_text_kline_frame() {
626        let kline = sample_kline_text();
627        let base = one_shot_server(move |mut ws| async move {
628            let _ = ws.send(Message::Text(kline.into())).await;
629            while let Some(Ok(_)) = ws.next().await {}
630        })
631        .await;
632        let mut stream = BinanceKlineStream::connect_with_config(
633            &["BTCUSDT".to_string()],
634            Interval::OneMinute,
635            test_config(base),
636        )
637        .await
638        .unwrap();
639        assert!(!stream.is_closed());
640        let event = stream
641            .next_event()
642            .await
643            .unwrap()
644            .expect("server pushes a kline");
645        assert_eq!(event.symbol, "btcusdt");
646        assert!(event.is_closed);
647    }
648
649    #[tokio::test]
650    async fn next_event_decodes_a_binary_kline_frame() {
651        let kline = sample_kline_text();
652        let base = one_shot_server(move |mut ws| async move {
653            let bytes: Vec<u8> = kline.into_bytes();
654            let _ = ws.send(Message::Binary(bytes.into())).await;
655            while let Some(Ok(_)) = ws.next().await {}
656        })
657        .await;
658        let mut stream = BinanceKlineStream::connect_with_config(
659            &["BTCUSDT".to_string()],
660            Interval::OneMinute,
661            test_config(base),
662        )
663        .await
664        .unwrap();
665        let event = stream
666            .next_event()
667            .await
668            .unwrap()
669            .expect("server pushes a kline as Binary");
670        assert_eq!(event.symbol, "btcusdt");
671    }
672
673    #[tokio::test]
674    async fn next_event_replies_to_a_ping_with_a_pong() {
675        let kline = sample_kline_text();
676        let base = one_shot_server(move |mut ws| async move {
677            let _ = ws
678                .send(Message::Ping(b"binance-ping".as_slice().into()))
679                .await;
680            let _ = ws.send(Message::Text(kline.into())).await;
681            while let Some(Ok(_)) = ws.next().await {}
682        })
683        .await;
684        let mut stream = BinanceKlineStream::connect_with_config(
685            &["BTCUSDT".to_string()],
686            Interval::OneMinute,
687            test_config(base),
688        )
689        .await
690        .unwrap();
691        // If the client never replied to the Ping the server's drain would
692        // observe nothing — but for line coverage it's enough that the
693        // client received the Ping, sent a Pong, then read the next frame.
694        let event = stream
695            .next_event()
696            .await
697            .unwrap()
698            .expect("kline arrives right after the ping");
699        assert_eq!(event.symbol, "btcusdt");
700    }
701
702    #[tokio::test]
703    async fn next_event_skips_inbound_pong_frames() {
704        let kline = sample_kline_text();
705        let base = one_shot_server(move |mut ws| async move {
706            let _ = ws
707                .send(Message::Pong(b"unsolicited".as_slice().into()))
708                .await;
709            let _ = ws.send(Message::Text(kline.into())).await;
710            while let Some(Ok(_)) = ws.next().await {}
711        })
712        .await;
713        let mut stream = BinanceKlineStream::connect_with_config(
714            &["BTCUSDT".to_string()],
715            Interval::OneMinute,
716            test_config(base),
717        )
718        .await
719        .unwrap();
720        let event = stream
721            .next_event()
722            .await
723            .unwrap()
724            .expect("kline follows the ignored Pong");
725        assert_eq!(event.symbol, "btcusdt");
726    }
727
728    #[tokio::test]
729    async fn next_event_reconnects_after_a_server_close_frame() {
730        let kline = sample_kline_text();
731        let (base, server_done) = multi_shot_server(2, move |index, mut ws| {
732            let kline = kline.clone();
733            async move {
734                let msg = if index == 0 {
735                    // First connection: send a clean Close so the client
736                    // exercises the Message::Close reconnect path.
737                    Message::Close(None)
738                } else {
739                    Message::Text(kline.into())
740                };
741                let _ = ws.send(msg).await;
742            }
743        })
744        .await;
745        let mut stream = BinanceKlineStream::connect_with_config(
746            &["BTCUSDT".to_string()],
747            Interval::OneMinute,
748            test_config(base),
749        )
750        .await
751        .unwrap();
752        let event = stream
753            .next_event()
754            .await
755            .unwrap()
756            .expect("reconnect succeeds and the second connection serves a kline");
757        assert_eq!(event.symbol, "btcusdt");
758        // Wait for every spawned handler to reach its final state.
759        tokio::time::timeout(Duration::from_secs(1), server_done)
760            .await
761            .unwrap()
762            .unwrap();
763    }
764
765    #[tokio::test]
766    async fn next_event_reconnects_after_a_read_timeout() {
767        let kline = sample_kline_text();
768        let stall_token = Arc::new(AtomicU32::new(0));
769        let stall_token_h = stall_token.clone();
770        let (base, server_done) = multi_shot_server(2, move |index, mut ws| {
771            let kline = kline.clone();
772            let stall_token = stall_token_h.clone();
773            async move {
774                if index == 0 {
775                    // First connection: never write anything. Outlast the
776                    // client's 80 ms read_timeout but bounded so the
777                    // handler still completes for the coverage check.
778                    stall_token.fetch_add(1, Ordering::SeqCst);
779                    tokio::time::sleep(Duration::from_millis(250)).await;
780                } else {
781                    let _ = ws.send(Message::Text(kline.into())).await;
782                }
783            }
784        })
785        .await;
786        let cfg = BinanceConfig {
787            read_timeout: Duration::from_millis(80),
788            ..test_config(base)
789        };
790        let mut stream = BinanceKlineStream::connect_with_config(
791            &["BTCUSDT".to_string()],
792            Interval::OneMinute,
793            cfg,
794        )
795        .await
796        .unwrap();
797        let event = stream
798            .next_event()
799            .await
800            .unwrap()
801            .expect("client times out, reconnects, and reads the kline");
802        assert_eq!(event.symbol, "btcusdt");
803        assert!(stall_token.load(Ordering::SeqCst) >= 1);
804        tokio::time::timeout(Duration::from_secs(1), server_done)
805            .await
806            .unwrap()
807            .unwrap();
808    }
809
810    #[tokio::test]
811    async fn next_event_yields_none_after_close() {
812        let base = one_shot_server(|mut ws| async move {
813            // Stay open until the client closes; this lets close() complete
814            // its handshake cleanly.
815            while let Some(Ok(_)) = ws.next().await {}
816        })
817        .await;
818        let mut stream = BinanceKlineStream::connect_with_config(
819            &["BTCUSDT".to_string()],
820            Interval::OneMinute,
821            test_config(base),
822        )
823        .await
824        .unwrap();
825        stream.close().await.unwrap();
826        assert!(stream.is_closed());
827        assert!(stream.next_event().await.unwrap().is_none());
828    }
829
830    #[tokio::test]
831    async fn next_event_surfaces_an_error_when_reconnect_attempts_are_exhausted() {
832        // After the first accept the listener is dropped (one_shot_server
833        // does this), so every reconnect attempt lands on a closed port.
834        let base = one_shot_server(|mut ws| async move {
835            let _ = ws.send(Message::Close(None)).await;
836            // Returning here also drops the socket, but the listener has
837            // already been released — the client's subsequent connects
838            // will be refused.
839        })
840        .await;
841        let cfg = BinanceConfig {
842            max_reconnect_attempts: 2,
843            initial_reconnect_delay: Duration::from_millis(1),
844            max_reconnect_backoff: Duration::from_millis(2),
845            ..test_config(base)
846        };
847        let mut stream = BinanceKlineStream::connect_with_config(
848            &["BTCUSDT".to_string()],
849            Interval::OneMinute,
850            cfg,
851        )
852        .await
853        .unwrap();
854        let err = stream
855            .next_event()
856            .await
857            .expect_err("reconnect attempts are exhausted");
858        // Either a WS-layer error or a Malformed error from URL parsing —
859        // we only care that the call surfaced as Err rather than panicked.
860        let _ = err;
861    }
862
863    #[tokio::test]
864    async fn next_event_skips_non_kline_frames_and_returns_the_next_kline() {
865        // Drives the loop through the Text- *and* Binary-arm "frame was
866        // not a kline, keep reading" fall-throughs before serving the
867        // actual kline.
868        let kline = sample_kline_text();
869        let base = one_shot_server(move |mut ws| async move {
870            let _ = ws
871                .send(Message::Text(r#"{"result":null,"id":1}"#.into()))
872                .await;
873            let _ = ws
874                .send(Message::Binary(b"{\"id\":2}".to_vec().into()))
875                .await;
876            let _ = ws.send(Message::Text(kline.into())).await;
877        })
878        .await;
879        let mut stream = BinanceKlineStream::connect_with_config(
880            &["BTCUSDT".to_string()],
881            Interval::OneMinute,
882            test_config(base),
883        )
884        .await
885        .unwrap();
886        let event = stream
887            .next_event()
888            .await
889            .unwrap()
890            .expect("kline arrives after the two skipped control frames");
891        assert_eq!(event.symbol, "btcusdt");
892    }
893
894    #[tokio::test]
895    async fn next_event_propagates_a_parse_error_from_a_malformed_kline() {
896        // A "kline" envelope whose open field is not a number — parse_frame
897        // identifies it as a kline, into_event then fails, and next_event
898        // bubbles the error rather than skipping the frame.
899        let bad = r#"{"stream":"btcusdt@kline_1m","data":{"e":"kline","E":0,"s":"BTCUSDT","k":{"t":0,"T":0,"s":"BTCUSDT","i":"1m","o":"not-a-number","c":"0","h":"0","l":"0","v":"0","x":false}}}"#.to_string();
900        let base = one_shot_server(move |mut ws| async move {
901            let _ = ws.send(Message::Text(bad.into())).await;
902            while let Some(Ok(_)) = ws.next().await {}
903        })
904        .await;
905        let mut stream = BinanceKlineStream::connect_with_config(
906            &["BTCUSDT".to_string()],
907            Interval::OneMinute,
908            test_config(base),
909        )
910        .await
911        .unwrap();
912        let err = stream.next_event().await.unwrap_err();
913        assert!(matches!(err, Error::Malformed(_)));
914    }
915}
wickra_data/live/binance.rs

wickra_data/live/
binance.rs
