nautilus_binance/spot/websocket/public_json/
handler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Binance Spot public JSON WebSocket handler.
17
18use std::{
19    collections::VecDeque,
20    fmt::Debug,
21    sync::{
22        Arc,
23        atomic::{AtomicBool, AtomicU64, Ordering},
24    },
25};
26
27use ahash::AHashMap;
28use nautilus_network::{
29    RECONNECTED,
30    websocket::{SubscriptionState, WebSocketClient},
31};
32use ustr::Ustr;
33
34use super::messages::{
35    BinanceCombinedStreamEvent, BinanceSpotBookTickerMsg, BinanceSpotDepthDiffMsg,
36    BinanceSpotKlineMsg, BinanceSpotPartialDepthMsg, BinanceSpotPartialDepthPayload,
37    BinanceSpotPublicWsCommand, BinanceSpotPublicWsMessage, BinanceSpotServerShutdownMsg,
38    BinanceSpotTradeMsg, BinanceSpotWsErrorResponse, BinanceSpotWsResponse, BinanceWsSubscription,
39};
40use crate::common::{consts::BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION, enums::BinanceWsEventType};
41
42/// Handler for Binance Spot public JSON WebSocket streams.
43pub(super) struct BinanceSpotPublicWsHandler {
44    signal: Arc<AtomicBool>,
45    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotPublicWsCommand>,
46    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
47    inner: Option<WebSocketClient>,
48    pending_messages: VecDeque<BinanceSpotPublicWsMessage>,
49    subscriptions: SubscriptionState,
50    request_id_counter: Arc<AtomicU64>,
51    pending_requests: AHashMap<u64, Vec<String>>,
52}
53
54impl Debug for BinanceSpotPublicWsHandler {
55    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
56        f.debug_struct(stringify!(BinanceSpotPublicWsHandler))
57            .field("pending_requests", &self.pending_requests.len())
58            .finish_non_exhaustive()
59    }
60}
61
62impl BinanceSpotPublicWsHandler {
63    pub(super) fn new(
64        signal: Arc<AtomicBool>,
65        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotPublicWsCommand>,
66        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
67        subscriptions: SubscriptionState,
68        request_id_counter: Arc<AtomicU64>,
69    ) -> Self {
70        Self {
71            signal,
72            cmd_rx,
73            raw_rx,
74            inner: None,
75            pending_messages: VecDeque::new(),
76            subscriptions,
77            request_id_counter,
78            pending_requests: AHashMap::new(),
79        }
80    }
81
82    pub(super) async fn next(&mut self) -> Option<BinanceSpotPublicWsMessage> {
83        loop {
84            if let Some(msg) = self.pending_messages.pop_front() {
85                return Some(msg);
86            }
87
88            if self.signal.load(Ordering::Relaxed) {
89                return None;
90            }
91
92            tokio::select! {
93                Some(cmd) = self.cmd_rx.recv() => {
94                    self.handle_command(cmd).await;
95                }
96                Some(raw) = self.raw_rx.recv() => {
97                    let out = self.handle_raw_message(raw).await;
98                    if !out.is_empty() {
99                        let mut iter = out.into_iter();
100                        let first = iter.next();
101                        self.pending_messages.extend(iter);
102
103                        if let Some(msg) = first {
104                            return Some(msg);
105                        }
106                    }
107                }
108                else => {
109                    return None;
110                }
111            }
112        }
113    }
114
115    async fn handle_command(&mut self, cmd: BinanceSpotPublicWsCommand) {
116        match cmd {
117            BinanceSpotPublicWsCommand::SetClient(client) => {
118                self.inner = Some(client);
119            }
120            BinanceSpotPublicWsCommand::Disconnect => {
121                if let Some(client) = &self.inner {
122                    let () = client.disconnect().await;
123                }
124                self.inner = None;
125            }
126            BinanceSpotPublicWsCommand::Subscribe { streams } => {
127                self.send_subscribe(streams).await;
128            }
129            BinanceSpotPublicWsCommand::Unsubscribe { streams } => {
130                self.send_unsubscribe(streams).await;
131            }
132        }
133    }
134
135    async fn send_subscribe(&mut self, streams: Vec<String>) {
136        let Some(client) = &self.inner else {
137            log::warn!("Cannot subscribe: no client connected");
138            return;
139        };
140
141        let request_id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
142        self.pending_requests.insert(request_id, streams.clone());
143
144        for stream in &streams {
145            self.subscriptions.mark_subscribe(stream);
146        }
147
148        let request = BinanceWsSubscription::subscribe(streams, request_id);
149        let json = match serde_json::to_string(&request) {
150            Ok(j) => j,
151            Err(e) => {
152                log::error!("Failed to serialize subscribe request: {e}");
153                return;
154            }
155        };
156
157        if let Err(e) = client
158            .send_text(json, Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
159            .await
160        {
161            log::error!("Failed to send subscribe request: {e}");
162        }
163    }
164
165    async fn send_unsubscribe(&self, streams: Vec<String>) {
166        let Some(client) = &self.inner else {
167            log::warn!("Cannot unsubscribe: no client connected");
168            return;
169        };
170
171        let request_id = self.request_id_counter.fetch_add(1, Ordering::Relaxed);
172
173        let request = BinanceWsSubscription::unsubscribe(streams.clone(), request_id);
174        let json = match serde_json::to_string(&request) {
175            Ok(j) => j,
176            Err(e) => {
177                log::error!("Failed to serialize unsubscribe request: {e}");
178                return;
179            }
180        };
181
182        if let Err(e) = client
183            .send_text(json, Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
184            .await
185        {
186            log::error!("Failed to send unsubscribe request: {e}");
187        }
188
189        for stream in &streams {
190            self.subscriptions.mark_unsubscribe(stream);
191            self.subscriptions.confirm_unsubscribe(stream);
192        }
193    }
194
195    async fn handle_raw_message(&mut self, raw: Vec<u8>) -> Vec<BinanceSpotPublicWsMessage> {
196        if let Ok(text) = std::str::from_utf8(&raw)
197            && text == RECONNECTED
198        {
199            log::debug!("WebSocket reconnected signal received");
200            return vec![BinanceSpotPublicWsMessage::Reconnected];
201        }
202
203        let json: serde_json::Value = match serde_json::from_slice(&raw) {
204            Ok(j) => j,
205            Err(e) => {
206                log::warn!("Failed to parse Spot public JSON message: {e}");
207                return vec![];
208            }
209        };
210
211        if let Some(code) = json.get("code")
212            && let Some(code) = code.as_i64()
213        {
214            self.handle_subscription_response(&json);
215            let msg = json
216                .get("msg")
217                .and_then(|m| m.as_str())
218                .unwrap_or("Unknown error")
219                .to_string();
220            return vec![BinanceSpotPublicWsMessage::Error(
221                crate::spot::websocket::streams::messages::BinanceWsErrorMsg {
222                    code: code as i32,
223                    msg,
224                },
225            )];
226        }
227
228        if json.get("result").is_some() || json.get("id").is_some() {
229            self.handle_subscription_response(&json);
230            return vec![];
231        }
232
233        self.handle_stream_data(&json)
234    }
235
236    fn handle_subscription_response(&mut self, json: &serde_json::Value) {
237        if json.get("result").is_some()
238            && let Ok(response) = serde_json::from_value::<BinanceSpotWsResponse>(json.clone())
239        {
240            if let Some(streams) = self.pending_requests.remove(&response.id) {
241                if response.result.is_none() {
242                    for stream in &streams {
243                        self.subscriptions.confirm_subscribe(stream);
244                    }
245                    log::debug!("Subscription confirmed: streams={streams:?}");
246                } else {
247                    for stream in &streams {
248                        self.subscriptions.mark_failure(stream);
249                    }
250                    log::warn!(
251                        "Subscription failed: streams={streams:?}, result={:?}",
252                        response.result
253                    );
254                }
255            }
256        } else if let Ok(error) = serde_json::from_value::<BinanceSpotWsErrorResponse>(json.clone())
257        {
258            if let Some(id) = error.id
259                && let Some(streams) = self.pending_requests.remove(&id)
260            {
261                for stream in &streams {
262                    self.subscriptions.mark_failure(stream);
263                }
264            }
265            log::warn!(
266                "WebSocket error response: code={}, msg={}",
267                error.code,
268                error.msg
269            );
270        }
271    }
272
273    fn handle_stream_data(&self, json: &serde_json::Value) -> Vec<BinanceSpotPublicWsMessage> {
274        let (stream_name, payload) = split_combined_payload(json);
275
276        if let Some(depth) = parse_partial_depth_with_symbol(&payload, stream_name.as_deref()) {
277            return vec![BinanceSpotPublicWsMessage::DepthSnapshot(depth)];
278        }
279
280        if let Some(stream_name) = stream_name.as_deref()
281            && stream_name.ends_with("@bookTicker")
282        {
283            return serde_json::from_value::<BinanceSpotBookTickerMsg>(payload)
284                .map(BinanceSpotPublicWsMessage::BookTicker)
285                .map_err(|e| log::warn!("Failed to parse Spot bookTicker: {e}"))
286                .ok()
287                .into_iter()
288                .collect();
289        }
290
291        // `serverShutdown` is not a `BinanceWsEventType` variant (it deserializes to
292        // `Unknown` via `#[serde(other)]`), so detect it from the raw `e` field before
293        // enum dispatch, mirroring the SBE streams handler.
294        if payload.get("e").and_then(|v| v.as_str()) == Some("serverShutdown") {
295            return serde_json::from_value::<BinanceSpotServerShutdownMsg>(payload)
296                .map(BinanceSpotPublicWsMessage::ServerShutdown)
297                .map_err(|e| log::warn!("Failed to parse Spot server shutdown event: {e}"))
298                .ok()
299                .into_iter()
300                .collect();
301        }
302
303        let Some(event_type) = extract_event_type(&payload) else {
304            return vec![BinanceSpotPublicWsMessage::RawJson(payload)];
305        };
306
307        match event_type {
308            BinanceWsEventType::Trade => serde_json::from_value::<BinanceSpotTradeMsg>(payload)
309                .map(BinanceSpotPublicWsMessage::Trade)
310                .map_err(|e| log::warn!("Failed to parse Spot trade: {e}"))
311                .ok()
312                .into_iter()
313                .collect(),
314            BinanceWsEventType::BookTicker => {
315                serde_json::from_value::<BinanceSpotBookTickerMsg>(payload)
316                    .map(BinanceSpotPublicWsMessage::BookTicker)
317                    .map_err(|e| log::warn!("Failed to parse Spot bookTicker: {e}"))
318                    .ok()
319                    .into_iter()
320                    .collect()
321            }
322            BinanceWsEventType::DepthUpdate => {
323                serde_json::from_value::<BinanceSpotDepthDiffMsg>(payload)
324                    .map(BinanceSpotPublicWsMessage::DepthDiff)
325                    .map_err(|e| log::warn!("Failed to parse Spot depthUpdate: {e}"))
326                    .ok()
327                    .into_iter()
328                    .collect()
329            }
330            BinanceWsEventType::Kline => serde_json::from_value::<BinanceSpotKlineMsg>(payload)
331                .map(BinanceSpotPublicWsMessage::Kline)
332                .map_err(|e| log::warn!("Failed to parse Spot kline: {e}"))
333                .ok()
334                .into_iter()
335                .collect(),
336            _ => vec![BinanceSpotPublicWsMessage::RawJson(payload)],
337        }
338    }
339}
340
341fn split_combined_payload(json: &serde_json::Value) -> (Option<String>, serde_json::Value) {
342    if let Ok(wrapper) = serde_json::from_value::<BinanceCombinedStreamEvent>(json.clone()) {
343        (Some(wrapper.stream), wrapper.data)
344    } else {
345        (None, json.clone())
346    }
347}
348
349fn extract_event_type(json: &serde_json::Value) -> Option<BinanceWsEventType> {
350    json.get("e")
351        .and_then(|v| serde_json::from_value(v.clone()).ok())
352}
353
354fn parse_partial_depth_with_symbol(
355    payload: &serde_json::Value,
356    stream_name: Option<&str>,
357) -> Option<BinanceSpotPartialDepthMsg> {
358    let parsed = serde_json::from_value::<BinanceSpotPartialDepthPayload>(payload.clone()).ok()?;
359
360    let symbol = stream_name
361        .and_then(|stream| stream.split('@').next())
362        .map(|s| Ustr::from(s.to_uppercase().as_str()))?;
363
364    Some(BinanceSpotPartialDepthMsg {
365        symbol,
366        last_update_id: parsed.last_update_id,
367        bids: parsed.bids,
368        asks: parsed.asks,
369    })
370}
371
372#[cfg(test)]
373mod tests {
374    use std::sync::{
375        Arc,
376        atomic::{AtomicBool, AtomicU64},
377    };
378
379    use nautilus_network::{RECONNECTED, websocket::SubscriptionState};
380    use rstest::rstest;
381    use serde_json::json;
382    use ustr::Ustr;
383
384    use super::*;
385
386    #[rstest]
387    fn test_parse_partial_depth_with_symbol_uppercases_symbol_from_stream_name() {
388        let payload = json!({
389            "lastUpdateId": 12345,
390            "bids": [["42000.1", "0.5"]],
391            "asks": [["42000.2", "0.8"]]
392        });
393
394        let parsed = parse_partial_depth_with_symbol(&payload, Some("btcusdt@depth20"))
395            .expect("depth payload should parse");
396
397        assert_eq!(parsed.symbol, Ustr::from("BTCUSDT"));
398        assert_eq!(parsed.last_update_id, 12345);
399        assert_eq!(parsed.bids.len(), 1);
400        assert_eq!(parsed.asks.len(), 1);
401    }
402
403    #[tokio::test]
404    async fn test_handle_raw_message_emits_reconnected_signal() {
405        let signal = Arc::new(AtomicBool::new(false));
406        let request_id_counter = Arc::new(AtomicU64::new(1));
407        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
408        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
409        let subscriptions = SubscriptionState::new('@');
410
411        let mut handler = BinanceSpotPublicWsHandler::new(
412            signal,
413            cmd_rx,
414            raw_rx,
415            subscriptions,
416            request_id_counter,
417        );
418
419        let out = handler
420            .handle_raw_message(RECONNECTED.as_bytes().to_vec())
421            .await;
422        assert_eq!(out.len(), 1);
423        assert!(matches!(out[0], BinanceSpotPublicWsMessage::Reconnected));
424    }
425
426    #[tokio::test]
427    async fn test_handle_raw_message_error_with_id_emits_error() {
428        let signal = Arc::new(AtomicBool::new(false));
429        let request_id_counter = Arc::new(AtomicU64::new(2));
430        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
431        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
432        let subscriptions = SubscriptionState::new('@');
433
434        let mut handler = BinanceSpotPublicWsHandler::new(
435            signal,
436            cmd_rx,
437            raw_rx,
438            subscriptions,
439            request_id_counter,
440        );
441        handler
442            .pending_requests
443            .insert(1, vec!["btcusdt@trade".to_string()]);
444
445        let payload = json!({
446            "code": 2,
447            "msg": "Invalid request",
448            "id": 1
449        });
450
451        let out = handler
452            .handle_raw_message(payload.to_string().into_bytes())
453            .await;
454        assert_eq!(out.len(), 1);
455        match &out[0] {
456            BinanceSpotPublicWsMessage::Error(err) => {
457                assert_eq!(err.code, 2);
458                assert_eq!(err.msg, "Invalid request");
459            }
460            other => panic!("expected Error variant, was {other:?}"),
461        }
462        assert!(handler.pending_requests.is_empty());
463    }
464
465    #[rstest]
466    fn test_handle_stream_data_parses_book_ticker_without_event_type() {
467        let signal = Arc::new(AtomicBool::new(false));
468        let request_id_counter = Arc::new(AtomicU64::new(1));
469        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
470        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
471        let subscriptions = SubscriptionState::new('@');
472
473        let handler = BinanceSpotPublicWsHandler::new(
474            signal,
475            cmd_rx,
476            raw_rx,
477            subscriptions,
478            request_id_counter,
479        );
480
481        let payload = json!({
482            "stream": "btcusdt@bookTicker",
483            "data": {
484                "u": 94528182161_u64,
485                "s": "BTCUSDT",
486                "b": "73650.51000000",
487                "B": "2.95126000",
488                "a": "73650.52000000",
489                "A": "1.38108000"
490            }
491        });
492
493        let out = handler.handle_stream_data(&payload);
494        assert_eq!(out.len(), 1);
495        assert!(matches!(out[0], BinanceSpotPublicWsMessage::BookTicker(_)));
496    }
497
498    #[rstest]
499    fn test_handle_stream_data_emits_server_shutdown() {
500        let signal = Arc::new(AtomicBool::new(false));
501        let request_id_counter = Arc::new(AtomicU64::new(1));
502        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
503        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
504        let subscriptions = SubscriptionState::new('@');
505
506        let handler = BinanceSpotPublicWsHandler::new(
507            signal,
508            cmd_rx,
509            raw_rx,
510            subscriptions,
511            request_id_counter,
512        );
513
514        // `serverShutdown` is not a BinanceWsEventType variant, so it must be
515        // recognized from the raw `e` field rather than dropped as RawJson.
516        let payload = json!({"e": "serverShutdown", "E": 1_700_000_000_000_i64});
517
518        let out = handler.handle_stream_data(&payload);
519        assert_eq!(out.len(), 1);
520        assert!(matches!(
521            out[0],
522            BinanceSpotPublicWsMessage::ServerShutdown(_)
523        ));
524    }
525}
nautilus_binance/spot/websocket/public_json/handler.rs

nautilus_binance/spot/websocket/public_json/
handler.rs
