C2: skip non-kline Binance frames instead of aborting the stream
next_event deserialized every text frame straight into RawWsEnvelope, so a subscription acknowledgement, heartbeat or error object propagated a decode Err and killed the feed. Frames are now routed through parse_frame, which inspects data.e: kline frames yield an event, everything else is skipped, and an Err is raised only when a genuine kline frame fails to decode. Adds tests for skipped acks/errors.
This commit is contained in:
@@ -173,12 +173,17 @@ impl BinanceKlineStream {
|
|||||||
};
|
};
|
||||||
match msg {
|
match msg {
|
||||||
Message::Text(text) => {
|
Message::Text(text) => {
|
||||||
let envelope: RawWsEnvelope = serde_json::from_str(&text)?;
|
if let Some(event) = Self::parse_frame(&text, self.interval)? {
|
||||||
return Ok(Some(envelope.into_event(self.interval)?));
|
return Ok(Some(event));
|
||||||
|
}
|
||||||
|
// Non-kline frame (subscription ack / heartbeat / error):
|
||||||
|
// skip it and keep reading.
|
||||||
}
|
}
|
||||||
Message::Binary(bytes) => {
|
Message::Binary(bytes) => {
|
||||||
let envelope: RawWsEnvelope = serde_json::from_slice(&bytes)?;
|
let text = String::from_utf8_lossy(&bytes);
|
||||||
return Ok(Some(envelope.into_event(self.interval)?));
|
if let Some(event) = Self::parse_frame(&text, self.interval)? {
|
||||||
|
return Ok(Some(event));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
Message::Ping(payload) => {
|
Message::Ping(payload) => {
|
||||||
self.socket.send(Message::Pong(payload)).await?;
|
self.socket.send(Message::Pong(payload)).await?;
|
||||||
@@ -189,6 +194,27 @@ impl BinanceKlineStream {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Parse one raw WebSocket text frame.
|
||||||
|
///
|
||||||
|
/// Returns `Ok(Some(event))` for a kline frame, `Ok(None)` for any other
|
||||||
|
/// frame (subscription acknowledgements, error objects, heartbeats), and
|
||||||
|
/// `Err` only when a frame that *is* a kline fails to decode.
|
||||||
|
fn parse_frame(text: &str, interval: Interval) -> Result<Option<KlineEvent>> {
|
||||||
|
let value: serde_json::Value = serde_json::from_str(text)?;
|
||||||
|
// Combined-stream kline frames carry `data.e == "kline"`. Everything
|
||||||
|
// else on the socket is control traffic that must not abort the feed.
|
||||||
|
let is_kline = value
|
||||||
|
.get("data")
|
||||||
|
.and_then(|d| d.get("e"))
|
||||||
|
.and_then(serde_json::Value::as_str)
|
||||||
|
== Some("kline");
|
||||||
|
if !is_kline {
|
||||||
|
return Ok(None);
|
||||||
|
}
|
||||||
|
let envelope: RawWsEnvelope = serde_json::from_value(value)?;
|
||||||
|
Ok(Some(envelope.into_event(interval)?))
|
||||||
|
}
|
||||||
|
|
||||||
/// Close the underlying socket cleanly.
|
/// Close the underlying socket cleanly.
|
||||||
pub async fn close(mut self) -> Result<()> {
|
pub async fn close(mut self) -> Result<()> {
|
||||||
self.socket.close(None).await?;
|
self.socket.close(None).await?;
|
||||||
@@ -290,4 +316,39 @@ mod tests {
|
|||||||
let err = env.into_event(Interval::OneMinute).unwrap_err();
|
let err = env.into_event(Interval::OneMinute).unwrap_err();
|
||||||
assert!(matches!(err, Error::Malformed(_)));
|
assert!(matches!(err, Error::Malformed(_)));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn skips_non_kline_frames() {
|
||||||
|
// Subscription acknowledgement: skipped, never an error.
|
||||||
|
let ack = r#"{"result":null,"id":1}"#;
|
||||||
|
assert!(BinanceKlineStream::parse_frame(ack, Interval::OneMinute)
|
||||||
|
.unwrap()
|
||||||
|
.is_none());
|
||||||
|
// Error object: also skipped.
|
||||||
|
let err = r#"{"error":{"code":2,"msg":"Invalid request"}}"#;
|
||||||
|
assert!(BinanceKlineStream::parse_frame(err, Interval::OneMinute)
|
||||||
|
.unwrap()
|
||||||
|
.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn parse_frame_decodes_a_kline() {
|
||||||
|
let json = 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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}"#;
|
||||||
|
let event = BinanceKlineStream::parse_frame(json, Interval::OneMinute)
|
||||||
|
.unwrap()
|
||||||
|
.expect("a kline frame yields an event");
|
||||||
|
assert_eq!(event.symbol, "btcusdt");
|
||||||
|
assert!(event.is_closed);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user