nautilus_binance/spot/websocket/streams/
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 WebSocket message handler.
17//!
18//! The handler is a stateless I/O boundary: it decodes raw SBE binary frames
19//! into venue-specific event types and emits them on the output channel.
20//! Domain conversion happens in the data client layer.
21
22use std::{
23    collections::VecDeque,
24    sync::{
25        Arc,
26        atomic::{AtomicBool, AtomicU64, Ordering},
27    },
28};
29
30use ahash::AHashMap;
31use nautilus_network::{
32    RECONNECTED,
33    websocket::{SubscriptionState, WebSocketClient},
34};
35use tokio_tungstenite::tungstenite::Message;
36use ustr::Ustr;
37
38pub use super::parse::{MarketDataMessage, decode_market_data};
39use super::{
40    messages::{
41        BinanceSpotServerShutdownMsg, BinanceSpotWsMessage, BinanceSpotWsStreamsCommand,
42        BinanceWsErrorMsg, BinanceWsErrorResponse, BinanceWsResponse, BinanceWsSubscription,
43    },
44    parse::decode_market_data as decode_sbe,
45};
46use crate::common::consts::BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION;
47
48/// Binance Spot WebSocket feed handler.
49///
50/// Decodes raw SBE binary frames into venue-specific event types without
51/// performing domain conversion. The data client layer owns instrument
52/// lookups and Nautilus type construction.
53pub(super) struct BinanceSpotWsFeedHandler {
54    #[allow(dead_code)]
55    signal: Arc<AtomicBool>,
56    inner: Option<WebSocketClient>,
57    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsStreamsCommand>,
58    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
59    #[allow(dead_code)]
60    out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsMessage>,
61    subscriptions: SubscriptionState,
62    request_id_counter: Arc<AtomicU64>,
63    pending_messages: VecDeque<BinanceSpotWsMessage>,
64    pending_requests: AHashMap<u64, Vec<String>>,
65}
66
67impl BinanceSpotWsFeedHandler {
68    /// Creates a new handler instance.
69    pub(super) fn new(
70        signal: Arc<AtomicBool>,
71        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<BinanceSpotWsStreamsCommand>,
72        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
73        out_tx: tokio::sync::mpsc::UnboundedSender<BinanceSpotWsMessage>,
74        subscriptions: SubscriptionState,
75        request_id_counter: Arc<AtomicU64>,
76    ) -> Self {
77        Self {
78            signal,
79            inner: None,
80            cmd_rx,
81            raw_rx,
82            out_tx,
83            subscriptions,
84            request_id_counter,
85            pending_messages: VecDeque::new(),
86            pending_requests: AHashMap::new(),
87        }
88    }
89
90    /// Returns the next message from the handler.
91    ///
92    /// Processes both commands and raw WebSocket messages.
93    pub(super) async fn next(&mut self) -> Option<BinanceSpotWsMessage> {
94        if let Some(message) = self.pending_messages.pop_front() {
95            return Some(message);
96        }
97
98        loop {
99            tokio::select! {
100                Some(cmd) = self.cmd_rx.recv() => {
101                    match cmd {
102                        BinanceSpotWsStreamsCommand::SetClient(client) => {
103                            log::debug!("Handler received WebSocket client");
104                            self.inner = Some(client);
105                        }
106                        BinanceSpotWsStreamsCommand::Disconnect => {
107                            log::debug!("Handler disconnecting WebSocket client");
108                            self.inner = None;
109                            return None;
110                        }
111                        BinanceSpotWsStreamsCommand::Subscribe { streams } => {
112                            if let Err(e) = self.handle_subscribe(streams).await {
113                                log::error!("Failed to handle subscribe command: {e}");
114                            }
115                        }
116                        BinanceSpotWsStreamsCommand::Unsubscribe { streams } => {
117                            if let Err(e) = self.handle_unsubscribe(streams).await {
118                                log::error!("Failed to handle unsubscribe command: {e}");
119                            }
120                        }
121                    }
122                }
123                Some(msg) = self.raw_rx.recv() => {
124                    if let Message::Text(ref text) = msg
125                        && text.as_str() == RECONNECTED
126                    {
127                        log::debug!("Handler received reconnection signal");
128                        return Some(BinanceSpotWsMessage::Reconnected);
129                    }
130
131                    let messages = self.handle_message(msg);
132                    if !messages.is_empty() {
133                        let mut iter = messages.into_iter();
134                        let first = iter.next();
135                        self.pending_messages.extend(iter);
136
137                        if let Some(msg) = first {
138                            return Some(msg);
139                        }
140                    }
141                }
142                else => {
143                    return None;
144                }
145            }
146        }
147    }
148
149    fn handle_message(&mut self, msg: Message) -> Vec<BinanceSpotWsMessage> {
150        match msg {
151            Message::Binary(data) => self.handle_binary_frame(&data),
152            Message::Text(text) => self.handle_text_frame(&text),
153            Message::Close(_) => {
154                log::debug!("Received close frame");
155                vec![]
156            }
157            Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => vec![],
158        }
159    }
160
161    fn handle_binary_frame(&self, data: &[u8]) -> Vec<BinanceSpotWsMessage> {
162        match decode_sbe(data) {
163            Ok(MarketDataMessage::Trades(event)) => {
164                vec![BinanceSpotWsMessage::Trades(event)]
165            }
166            Ok(MarketDataMessage::BestBidAsk(event)) => {
167                vec![BinanceSpotWsMessage::BestBidAsk(event)]
168            }
169            Ok(MarketDataMessage::DepthSnapshot(event)) => {
170                vec![BinanceSpotWsMessage::DepthSnapshot(event)]
171            }
172            Ok(MarketDataMessage::DepthDiff(event)) => {
173                vec![BinanceSpotWsMessage::DepthDiff(event)]
174            }
175            Err(e) => {
176                log::error!("SBE decode error: {e}");
177                vec![BinanceSpotWsMessage::RawBinary(data.to_vec())]
178            }
179        }
180    }
181
182    fn handle_text_frame(&mut self, text: &str) -> Vec<BinanceSpotWsMessage> {
183        if let Ok(error) = serde_json::from_str::<BinanceWsErrorResponse>(text) {
184            if let Some(id) = error.id
185                && let Some(streams) = self.pending_requests.remove(&id)
186            {
187                for stream in &streams {
188                    self.subscriptions.mark_failure(stream);
189                }
190                log::warn!(
191                    "Subscription request failed: id={id}, streams={streams:?}, code={}, msg={}",
192                    error.code,
193                    error.msg
194                );
195            }
196            return vec![BinanceSpotWsMessage::Error(BinanceWsErrorMsg {
197                code: error.code,
198                msg: error.msg,
199            })];
200        }
201
202        if let Ok(response) = serde_json::from_str::<BinanceWsResponse>(text) {
203            self.handle_subscription_response(&response);
204            return vec![];
205        }
206
207        classify_unsolicited_json(text)
208    }
209
210    fn handle_subscription_response(&mut self, response: &BinanceWsResponse) {
211        if let Some(streams) = self.pending_requests.remove(&response.id) {
212            if response.result.is_none() {
213                for stream in &streams {
214                    self.subscriptions.confirm_subscribe(stream);
215                }
216                log::debug!("Subscription confirmed: streams={streams:?}");
217            } else {
218                for stream in &streams {
219                    self.subscriptions.mark_failure(stream);
220                }
221                log::warn!(
222                    "Subscription failed: streams={streams:?}, result={:?}",
223                    response.result
224                );
225            }
226        } else {
227            log::debug!("Received response for unknown request: id={}", response.id);
228        }
229    }
230
231    async fn handle_subscribe(&mut self, streams: Vec<String>) -> anyhow::Result<()> {
232        let request_id = self.request_id_counter.fetch_add(1, Ordering::SeqCst);
233        let request = BinanceWsSubscription::subscribe(streams.clone(), request_id);
234        let payload = serde_json::to_string(&request)?;
235
236        self.pending_requests.insert(request_id, streams.clone());
237
238        for stream in &streams {
239            self.subscriptions.mark_subscribe(stream);
240        }
241
242        self.send_text(
243            payload,
244            Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()),
245        )
246        .await?;
247        Ok(())
248    }
249
250    async fn handle_unsubscribe(&self, streams: Vec<String>) -> anyhow::Result<()> {
251        let request_id = self.request_id_counter.fetch_add(1, Ordering::SeqCst);
252        let request = BinanceWsSubscription::unsubscribe(streams.clone(), request_id);
253        let payload = serde_json::to_string(&request)?;
254
255        self.send_text(
256            payload,
257            Some(BINANCE_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()),
258        )
259        .await?;
260
261        for stream in &streams {
262            self.subscriptions.mark_unsubscribe(stream);
263            self.subscriptions.confirm_unsubscribe(stream);
264        }
265
266        Ok(())
267    }
268
269    async fn send_text(
270        &self,
271        payload: String,
272        rate_limit_keys: Option<&[Ustr]>,
273    ) -> anyhow::Result<()> {
274        let Some(client) = &self.inner else {
275            anyhow::bail!("No active WebSocket client");
276        };
277        client
278            .send_text(payload, rate_limit_keys)
279            .await
280            .map_err(|e| anyhow::anyhow!("Failed to send message: {e}"))?;
281        Ok(())
282    }
283}
284
285/// Classifies a JSON text frame that did not match a subscription response or
286/// known error envelope. Recognises the `serverShutdown` event; otherwise
287/// emits `RawJson` for parseable payloads or an empty vector for garbage.
288fn classify_unsolicited_json(text: &str) -> Vec<BinanceSpotWsMessage> {
289    let Ok(value) = serde_json::from_str::<serde_json::Value>(text) else {
290        log::warn!("Failed to parse JSON message: {text}");
291        return vec![];
292    };
293
294    if value.get("e").and_then(|v| v.as_str()) == Some("serverShutdown")
295        && let Ok(msg) = serde_json::from_value::<BinanceSpotServerShutdownMsg>(value.clone())
296    {
297        log::warn!(
298            "Binance server shutdown notice received (event_time={}); disconnect expected ~10 minutes from event",
299            msg.event_time,
300        );
301        return vec![BinanceSpotWsMessage::ServerShutdown(msg)];
302    }
303
304    vec![BinanceSpotWsMessage::RawJson(value)]
305}
306
307#[cfg(test)]
308mod tests {
309    use std::sync::{
310        Arc,
311        atomic::{AtomicBool, AtomicU64},
312    };
313
314    use nautilus_network::websocket::SubscriptionState;
315    use rstest::rstest;
316
317    use super::*;
318
319    #[rstest]
320    fn test_classify_unsolicited_json_server_shutdown_emits_variant() {
321        let text = r#"{"e":"serverShutdown","E":1700000000000}"#;
322        let out = classify_unsolicited_json(text);
323        assert_eq!(out.len(), 1);
324        match &out[0] {
325            BinanceSpotWsMessage::ServerShutdown(msg) => {
326                assert_eq!(msg.event_type, "serverShutdown");
327                assert_eq!(msg.event_time, 1_700_000_000_000);
328            }
329            other => panic!("expected ServerShutdown variant, was {other:?}"),
330        }
331    }
332
333    #[rstest]
334    fn test_classify_unsolicited_json_unrelated_emits_raw_json() {
335        let text = r#"{"e":"trade","p":"50000"}"#;
336        let out = classify_unsolicited_json(text);
337        assert_eq!(out.len(), 1);
338        assert!(matches!(out[0], BinanceSpotWsMessage::RawJson(_)));
339    }
340
341    #[rstest]
342    fn test_classify_unsolicited_json_invalid_returns_empty() {
343        let out = classify_unsolicited_json("not json");
344        assert!(out.is_empty());
345    }
346
347    #[rstest]
348    fn test_handle_text_frame_error_with_id_emits_error_and_clears_pending_request() {
349        let signal = Arc::new(AtomicBool::new(false));
350        let request_id_counter = Arc::new(AtomicU64::new(2));
351        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
352        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
353        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
354        let subscriptions = SubscriptionState::new('@');
355
356        let mut handler = BinanceSpotWsFeedHandler::new(
357            signal,
358            cmd_rx,
359            raw_rx,
360            out_tx,
361            subscriptions,
362            request_id_counter,
363        );
364        handler
365            .pending_requests
366            .insert(1, vec!["btcusdt@trade".to_string()]);
367
368        let out = handler.handle_text_frame(r#"{"code":2,"msg":"Invalid request","id":1}"#);
369        assert_eq!(out.len(), 1);
370        match &out[0] {
371            BinanceSpotWsMessage::Error(err) => {
372                assert_eq!(err.code, 2);
373                assert_eq!(err.msg, "Invalid request");
374            }
375            other => panic!("expected Error variant, was {other:?}"),
376        }
377        assert!(handler.pending_requests.is_empty());
378    }
379}
nautilus_binance/spot/websocket/streams/handler.rs

nautilus_binance/spot/websocket/streams/
handler.rs
