nautilus_okx/websocket/
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//! WebSocket message handler for OKX.
17//!
18//! The handler is a thin I/O boundary between the network layer and the client. It owns the
19//! `WebSocketClient`, deserializes raw venue messages into `OKXWsMessage` events, and handles
20//! subscription management, authentication, and retry logic.
21//!
22//! All domain parsing (venue types to Nautilus types) occurs outside the handler:
23//! - Data parsing in `PyOKXWebSocketClient` (uses an instruments cache)
24//! - Execution parsing in `execution.rs` (uses the system Cache)
25
26use std::{
27    collections::VecDeque,
28    sync::{
29        Arc,
30        atomic::{AtomicBool, Ordering},
31    },
32};
33
34use nautilus_model::identifiers::ClientOrderId;
35use nautilus_network::{
36    RECONNECTED,
37    retry::{RetryManager, create_websocket_retry_manager},
38    websocket::{AuthTracker, SubscriptionState, TEXT_PING, TEXT_PONG, WebSocketClient},
39};
40use serde_json::Value;
41use tokio_tungstenite::tungstenite::Message;
42use ustr::Ustr;
43
44use super::{
45    enums::{OKXSubscriptionEvent, OKXWsChannel, OKXWsOperation},
46    error::OKXWsError,
47    messages::{
48        OKXOrderMsg, OKXSubscription, OKXSubscriptionArg, OKXWebSocketArg, OKXWebSocketError,
49        OKXWsFrame, OKXWsMessage,
50    },
51    subscription::topic_from_websocket_arg,
52};
53use crate::{
54    common::consts::{OKX_FIELD_SMSG, OKX_SUCCESS_CODE, should_retry_error_code},
55    websocket::client::OKX_RATE_LIMIT_KEY_SUBSCRIPTION,
56};
57
58/// Commands sent from the outer client to the inner message handler.
59#[derive(Debug)]
60pub enum HandlerCommand {
61    /// Set the WebSocketClient for the handler to use.
62    SetClient(WebSocketClient),
63    /// Disconnect the WebSocket connection.
64    Disconnect,
65    /// Send authentication payload to the WebSocket.
66    Authenticate { payload: String },
67    /// Subscribe to the given channels.
68    Subscribe { args: Vec<OKXSubscriptionArg> },
69    /// Unsubscribe from the given channels.
70    Unsubscribe { args: Vec<OKXSubscriptionArg> },
71    /// Send a pre-serialized payload (used for order operations).
72    Send {
73        payload: String,
74        rate_limit_keys: Option<Vec<Ustr>>,
75        request_id: Option<String>,
76        client_order_id: Option<ClientOrderId>,
77        op: Option<OKXWsOperation>,
78    },
79}
80
81pub(super) struct OKXWsFeedHandler {
82    signal: Arc<AtomicBool>,
83    inner: Option<WebSocketClient>,
84    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
85    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
86    out_tx: tokio::sync::mpsc::UnboundedSender<OKXWsMessage>,
87    auth_tracker: AuthTracker,
88    subscriptions_state: SubscriptionState,
89    retry_manager: RetryManager<OKXWsError>,
90    pending_messages: VecDeque<OKXWsMessage>,
91}
92
93impl OKXWsFeedHandler {
94    /// Creates a new [`OKXWsFeedHandler`] instance.
95    pub(super) fn new(
96        signal: Arc<AtomicBool>,
97        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
98        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
99        out_tx: tokio::sync::mpsc::UnboundedSender<OKXWsMessage>,
100        auth_tracker: AuthTracker,
101        subscriptions_state: SubscriptionState,
102    ) -> Self {
103        Self {
104            signal,
105            inner: None,
106            cmd_rx,
107            raw_rx,
108            out_tx,
109            auth_tracker,
110            subscriptions_state,
111            retry_manager: create_websocket_retry_manager(),
112            pending_messages: VecDeque::new(),
113        }
114    }
115
116    pub(super) fn is_stopped(&self) -> bool {
117        self.signal.load(Ordering::Acquire)
118    }
119
120    pub(super) fn send(&self, msg: OKXWsMessage) -> Result<(), ()> {
121        self.out_tx.send(msg).map_err(|_| ())
122    }
123
124    async fn send_with_retry(
125        &self,
126        payload: String,
127        rate_limit_keys: Option<&[Ustr]>,
128    ) -> Result<(), OKXWsError> {
129        if let Some(client) = &self.inner {
130            let keys_owned: Option<Vec<Ustr>> = rate_limit_keys.map(|k| k.to_vec());
131            self.retry_manager
132                .execute_with_retry(
133                    "websocket_send",
134                    || {
135                        let payload = payload.clone();
136                        let keys = keys_owned.clone();
137                        async move {
138                            client
139                                .send_text(payload, keys.as_deref())
140                                .await
141                                .map_err(|e| OKXWsError::ClientError(format!("Send failed: {e}")))
142                        }
143                    },
144                    should_retry_okx_error,
145                    create_okx_timeout_error,
146                )
147                .await
148        } else {
149            Err(OKXWsError::ClientError(
150                "No active WebSocket client".to_string(),
151            ))
152        }
153    }
154
155    pub(super) async fn send_pong(&self) -> anyhow::Result<()> {
156        match self.send_with_retry(TEXT_PONG.to_string(), None).await {
157            Ok(()) => {
158                log::trace!("Sent pong response to OKX text ping");
159                Ok(())
160            }
161            Err(e) => {
162                log::warn!("Failed to send pong after retries: error={e}");
163                Err(anyhow::anyhow!("Failed to send pong: {e}"))
164            }
165        }
166    }
167
168    pub(super) async fn next(&mut self) -> Option<OKXWsMessage> {
169        if let Some(message) = self.pending_messages.pop_front() {
170            return Some(message);
171        }
172
173        loop {
174            tokio::select! {
175                Some(cmd) = self.cmd_rx.recv() => {
176                    match cmd {
177                        HandlerCommand::SetClient(client) => {
178                            log::debug!("Handler received WebSocket client");
179                            self.inner = Some(client);
180                        }
181                        HandlerCommand::Disconnect => {
182                            log::debug!("Handler disconnecting WebSocket client");
183                            self.inner = None;
184                            return None;
185                        }
186                        HandlerCommand::Authenticate { payload } => {
187                            if let Err(e) = self.send_with_retry(
188                                payload,
189                                Some(OKX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()),
190                            ).await {
191                                log::error!(
192                                    "Failed to send authentication message after retries: error={e}"
193                                );
194                            }
195                        }
196                        HandlerCommand::Subscribe { args } => {
197                            if let Err(e) = self.handle_subscribe(args).await {
198                                log::error!("Failed to handle subscribe command: error={e}");
199                            }
200                        }
201                        HandlerCommand::Unsubscribe { args } => {
202                            if let Err(e) = self.handle_unsubscribe(args).await {
203                                log::error!("Failed to handle unsubscribe command: error={e}");
204                            }
205                        }
206                        HandlerCommand::Send {
207                            payload,
208                            rate_limit_keys,
209                            request_id,
210                            client_order_id,
211                            op,
212                        } => {
213                            if let Err(e) = self.send_with_retry(
214                                payload,
215                                rate_limit_keys.as_deref(),
216                            ).await {
217                                log::error!("Failed to send message after retries: error={e}");
218
219                                if let Some(request_id) = request_id {
220                                    self.pending_messages.push_back(OKXWsMessage::SendFailed {
221                                        request_id,
222                                        client_order_id,
223                                        op,
224                                        error: format!("{e}"),
225                                    });
226                                }
227                            }
228                        }
229                    }
230                }
231
232                () = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
233                    if self.signal.load(Ordering::Acquire) {
234                        log::debug!("Stop signal received during idle period");
235                        return None;
236                    }
237                }
238
239                msg = self.raw_rx.recv() => {
240                    let event = match msg {
241                        Some(msg) => match Self::parse_raw_message(msg) {
242                            Some(event) => event,
243                            None => continue,
244                        },
245                        None => {
246                            log::debug!("WebSocket stream closed");
247                            return None;
248                        }
249                    };
250
251                    match event {
252                        OKXWsFrame::Ping => {
253                            if let Err(e) = self.send_pong().await {
254                                log::warn!("Failed to send pong response: error={e}");
255                            }
256                        }
257                        OKXWsFrame::Login {
258                            code, msg, conn_id, ..
259                        } => {
260                            if code == OKX_SUCCESS_CODE {
261                                self.auth_tracker.succeed();
262                                return Some(OKXWsMessage::Authenticated);
263                            }
264
265                            log::error!("WebSocket authentication failed: error={msg}");
266                            self.auth_tracker.fail(msg.clone());
267
268                            let error = OKXWebSocketError {
269                                code,
270                                message: msg,
271                                conn_id: Some(conn_id),
272                                timestamp: nautilus_core::time::get_atomic_clock_realtime()
273                                    .get_time_ns()
274                                    .as_u64(),
275                            };
276                            self.pending_messages.push_back(OKXWsMessage::Error(error));
277                        }
278                        OKXWsFrame::BookData { arg, action, data } => {
279                            return Some(OKXWsMessage::BookData { arg, action, data });
280                        }
281                        OKXWsFrame::OrderResponse {
282                            id, op, code, msg, data,
283                        } => {
284                            return Some(OKXWsMessage::OrderResponse {
285                                id, op, code, msg, data,
286                            });
287                        }
288                        OKXWsFrame::Data { arg, data } => {
289                            if let Some(output) = self.route_data_message(arg, data) {
290                                return Some(output);
291                            }
292                        }
293                        OKXWsFrame::Error { code, msg } => {
294                            let error = OKXWebSocketError {
295                                code,
296                                message: msg,
297                                conn_id: None,
298                                timestamp: nautilus_core::time::get_atomic_clock_realtime()
299                                    .get_time_ns()
300                                    .as_u64(),
301                            };
302                            return Some(OKXWsMessage::Error(error));
303                        }
304                        OKXWsFrame::Reconnected => {
305                            self.auth_tracker.invalidate();
306                            return Some(OKXWsMessage::Reconnected);
307                        }
308                        OKXWsFrame::Subscription {
309                            event, arg, code, msg, ..
310                        } => {
311                            self.handle_subscription_ack(&event, &arg, code.as_deref(), msg.as_deref());
312                        }
313                        OKXWsFrame::ChannelConnCount { .. } => {}
314                    }
315                }
316
317                else => {
318                    log::debug!("Handler shutting down: stream ended or command channel closed");
319                    return None;
320                }
321            }
322        }
323    }
324
325    fn route_data_message(&self, arg: OKXWebSocketArg, data: Value) -> Option<OKXWsMessage> {
326        let OKXWebSocketArg {
327            channel, inst_id, ..
328        } = arg;
329
330        match channel {
331            OKXWsChannel::Account => Some(OKXWsMessage::Account(data)),
332            OKXWsChannel::Positions => Some(OKXWsMessage::Positions(data)),
333            OKXWsChannel::Orders => {
334                parse_array_items(data, "orders", false).map(OKXWsMessage::Orders)
335            }
336            OKXWsChannel::SprdOrders => {
337                parse_array_items(data, "spread orders", false).map(OKXWsMessage::SpreadOrders)
338            }
339            OKXWsChannel::OrdersAlgo | OKXWsChannel::AlgoAdvance => {
340                parse_array_items(data, "algo orders", false).map(OKXWsMessage::AlgoOrders)
341            }
342            OKXWsChannel::Instruments => {
343                parse_array_items(data, "instruments", true).map(OKXWsMessage::Instruments)
344            }
345            _ => Some(OKXWsMessage::ChannelData {
346                channel,
347                inst_id,
348                data,
349            }),
350        }
351    }
352
353    fn handle_subscription_ack(
354        &self,
355        event: &OKXSubscriptionEvent,
356        arg: &OKXWebSocketArg,
357        code: Option<&str>,
358        msg: Option<&str>,
359    ) {
360        let topic = topic_from_websocket_arg(arg);
361        let success = code.is_none_or(|c| c == OKX_SUCCESS_CODE);
362
363        match event {
364            OKXSubscriptionEvent::Subscribe => {
365                if success {
366                    self.subscriptions_state.confirm_subscribe(&topic);
367                } else {
368                    log::warn!(
369                        "Subscription failed: topic={topic:?}, error={msg:?}, code={code:?}"
370                    );
371                    self.subscriptions_state.mark_failure(&topic);
372                }
373            }
374            OKXSubscriptionEvent::Unsubscribe => {
375                if success {
376                    self.subscriptions_state.confirm_unsubscribe(&topic);
377                } else {
378                    log::warn!(
379                        "Unsubscription failed - restoring subscription: \
380                         topic={topic:?}, error={msg:?}, code={code:?}"
381                    );
382                    self.subscriptions_state.confirm_unsubscribe(&topic);
383                    self.subscriptions_state.mark_subscribe(&topic);
384                    self.subscriptions_state.confirm_subscribe(&topic);
385                }
386            }
387        }
388    }
389
390    async fn handle_subscribe(&self, args: Vec<OKXSubscriptionArg>) -> anyhow::Result<()> {
391        for arg in &args {
392            log::debug!(
393                "Subscribing to channel: channel={:?}, inst_id={:?}",
394                arg.channel,
395                arg.inst_id
396            );
397        }
398
399        let message = OKXSubscription {
400            op: OKXWsOperation::Subscribe,
401            args,
402        };
403
404        let json_txt = serde_json::to_string(&message)
405            .map_err(|e| anyhow::anyhow!("Failed to serialize subscription: {e}"))?;
406
407        self.send_with_retry(json_txt, Some(OKX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
408            .await
409            .map_err(|e| anyhow::anyhow!("Failed to send subscription after retries: {e}"))?;
410        Ok(())
411    }
412
413    async fn handle_unsubscribe(&self, args: Vec<OKXSubscriptionArg>) -> anyhow::Result<()> {
414        for arg in &args {
415            log::debug!(
416                "Unsubscribing from channel: channel={:?}, inst_id={:?}",
417                arg.channel,
418                arg.inst_id
419            );
420        }
421
422        let message = OKXSubscription {
423            op: OKXWsOperation::Unsubscribe,
424            args,
425        };
426
427        let json_txt = serde_json::to_string(&message)
428            .map_err(|e| anyhow::anyhow!("Failed to serialize unsubscription: {e}"))?;
429
430        self.send_with_retry(json_txt, Some(OKX_RATE_LIMIT_KEY_SUBSCRIPTION.as_slice()))
431            .await
432            .map_err(|e| anyhow::anyhow!("Failed to send unsubscription after retries: {e}"))?;
433        Ok(())
434    }
435
436    pub(crate) fn parse_raw_message(
437        msg: tokio_tungstenite::tungstenite::Message,
438    ) -> Option<OKXWsFrame> {
439        match msg {
440            tokio_tungstenite::tungstenite::Message::Text(text) => {
441                if text == TEXT_PONG {
442                    log::trace!("Received pong from OKX");
443                    return None;
444                }
445
446                if text == TEXT_PING {
447                    log::trace!("Received ping from OKX (text)");
448                    return Some(OKXWsFrame::Ping);
449                }
450
451                if text == RECONNECTED {
452                    log::debug!("Received WebSocket reconnection signal");
453                    return Some(OKXWsFrame::Reconnected);
454                }
455                log::trace!("Received WebSocket message: {text}");
456
457                match serde_json::from_str(&text) {
458                    Ok(ws_event) => match &ws_event {
459                        OKXWsFrame::Error { code, msg } => {
460                            if should_retry_error_code(code) {
461                                log::warn!("WebSocket error: {code} - {msg}");
462                            } else {
463                                log::error!("WebSocket error: {code} - {msg}");
464                            }
465                            Some(ws_event)
466                        }
467                        OKXWsFrame::Login {
468                            event,
469                            code,
470                            msg,
471                            conn_id,
472                        } => {
473                            if code == OKX_SUCCESS_CODE {
474                                log::debug!("WebSocket authenticated: conn_id={conn_id}");
475                            } else {
476                                log::error!(
477                                    "WebSocket authentication failed: \
478                                     event={event}, code={code}, error={msg}"
479                                );
480                            }
481                            Some(ws_event)
482                        }
483                        OKXWsFrame::Subscription {
484                            event,
485                            arg,
486                            conn_id,
487                            ..
488                        } => {
489                            let channel_str = serde_json::to_string(&arg.channel)
490                                .expect("Invalid OKX websocket channel")
491                                .trim_matches('"')
492                                .to_string();
493                            log::debug!("{event}d: channel={channel_str}, conn_id={conn_id}");
494                            Some(ws_event)
495                        }
496                        OKXWsFrame::ChannelConnCount {
497                            channel,
498                            conn_count,
499                            conn_id,
500                            ..
501                        } => {
502                            let channel_str = serde_json::to_string(channel)
503                                .expect("Invalid OKX websocket channel")
504                                .trim_matches('"')
505                                .to_string();
506                            log::debug!(
507                                "Channel connection status: \
508                                 channel={channel_str}, connections={conn_count}, conn_id={conn_id}",
509                            );
510                            None
511                        }
512                        OKXWsFrame::Ping => {
513                            log::trace!("Ignoring ping event parsed from text payload");
514                            None
515                        }
516                        OKXWsFrame::Data { .. } | OKXWsFrame::BookData { .. } => Some(ws_event),
517                        OKXWsFrame::OrderResponse {
518                            id, op, code, data, ..
519                        } => {
520                            if code == OKX_SUCCESS_CODE {
521                                log::debug!(
522                                    "Order operation successful: id={id:?}, op={op}, code={code}"
523                                );
524
525                                if let Some(order_data) = data.first() {
526                                    let success_msg = order_data
527                                        .get(OKX_FIELD_SMSG)
528                                        .and_then(|s| s.as_str())
529                                        .unwrap_or("Order operation successful");
530                                    log::debug!("Order success details: {success_msg}");
531                                }
532                            }
533                            Some(ws_event)
534                        }
535                        OKXWsFrame::Reconnected => {
536                            log::warn!("Unexpected Reconnected event from deserialization");
537                            None
538                        }
539                    },
540                    Err(e) => {
541                        log::error!("Failed to parse message: {e}: {text}");
542                        None
543                    }
544                }
545            }
546            Message::Ping(_payload) => {
547                log::trace!("Received binary ping frame from OKX");
548                Some(OKXWsFrame::Ping)
549            }
550            Message::Pong(payload) => {
551                log::trace!("Received pong frame from OKX ({} bytes)", payload.len());
552                None
553            }
554            Message::Binary(msg) => {
555                log::debug!("Raw binary frame ({} bytes)", msg.len());
556                log::trace!("Raw binary: {msg:?}");
557                None
558            }
559            Message::Close(_) => {
560                log::debug!("Received close message");
561                None
562            }
563            msg => {
564                log::warn!("Unexpected message: {msg}");
565                None
566            }
567        }
568    }
569}
570
571/// Returns `true` when an OKX WebSocket order message represents a post-only auto-cancel.
572pub fn is_post_only_auto_cancel(msg: &OKXOrderMsg) -> bool {
573    use crate::common::{consts::OKX_POST_ONLY_CANCEL_SOURCE, enums::OKXOrderStatus};
574
575    if msg.state != OKXOrderStatus::Canceled {
576        return false;
577    }
578
579    let cancel_source_matches = matches!(
580        msg.cancel_source.as_deref(),
581        Some(source) if source == OKX_POST_ONLY_CANCEL_SOURCE
582    );
583
584    let reason_matches = matches!(
585        msg.cancel_source_reason.as_deref(),
586        Some(reason) if reason.contains("POST_ONLY")
587    );
588
589    if !(cancel_source_matches || reason_matches) {
590        return false;
591    }
592
593    msg.acc_fill_sz
594        .as_ref()
595        .is_none_or(|filled| filled == "0" || filled.is_empty())
596}
597
598// Per-item deserialization so one malformed entry does not drop the batch.
599fn parse_array_items<T: serde::de::DeserializeOwned>(
600    data: Value,
601    label: &str,
602    warn_on_parse_error: bool,
603) -> Option<Vec<T>> {
604    let Value::Array(items) = data else {
605        if warn_on_parse_error {
606            log::warn!("Expected {label} payload to be a JSON array");
607        } else {
608            log::error!("Expected {label} payload to be a JSON array");
609        }
610        return None;
611    };
612
613    let mut parsed = Vec::with_capacity(items.len());
614    for (idx, item) in items.into_iter().enumerate() {
615        match serde_json::from_value::<T>(item) {
616            Ok(value) => parsed.push(value),
617            Err(e) => {
618                if warn_on_parse_error {
619                    log::warn!("Failed to parse {label} item at index {idx}: {e}");
620                } else {
621                    log::error!("Failed to parse {label} item at index {idx}: {e}");
622                }
623            }
624        }
625    }
626
627    if parsed.is_empty() {
628        None
629    } else {
630        Some(parsed)
631    }
632}
633
634#[inline]
635fn contains_ignore_ascii_case(haystack: &str, needle: &str) -> bool {
636    haystack
637        .as_bytes()
638        .windows(needle.len())
639        .any(|window| window.eq_ignore_ascii_case(needle.as_bytes()))
640}
641
642// Specific phrases rather than bare "connection"/"network", which appear
643// in permanent errors too (e.g. "no active WebSocket client connection").
644const RETRYABLE_CLIENT_ERROR_PHRASES: &[&str] = &[
645    "timeout",
646    "timed out",
647    "connection reset",
648    "connection refused",
649    "connection closed",
650    "connection aborted",
651    "broken pipe",
652    "network unreachable",
653    "network is unreachable",
654    "no route to host",
655];
656
657fn should_retry_okx_error(error: &OKXWsError) -> bool {
658    match error {
659        OKXWsError::OkxError { error_code, .. } => should_retry_error_code(error_code),
660        OKXWsError::TungsteniteError(_) => true,
661        OKXWsError::ClientError(msg) => RETRYABLE_CLIENT_ERROR_PHRASES
662            .iter()
663            .any(|phrase| contains_ignore_ascii_case(msg, phrase)),
664        OKXWsError::AuthenticationError(_)
665        | OKXWsError::JsonError(_)
666        | OKXWsError::ParsingError(_) => false,
667    }
668}
669
670fn create_okx_timeout_error(msg: String) -> OKXWsError {
671    OKXWsError::ClientError(msg)
672}
673
674#[cfg(test)]
675mod tests {
676    use std::sync::{Arc, atomic::AtomicBool};
677
678    use nautilus_network::websocket::{AuthTracker, SubscriptionState};
679    use rstest::rstest;
680    use serde_json::json;
681
682    use super::*;
683    use crate::common::{consts::OKX_WS_TOPIC_DELIMITER, testing::load_test_json};
684
685    fn create_handler() -> OKXWsFeedHandler {
686        let signal = Arc::new(AtomicBool::new(false));
687        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
688        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
689        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
690
691        OKXWsFeedHandler::new(
692            signal,
693            cmd_rx,
694            raw_rx,
695            out_tx,
696            AuthTracker::new(),
697            SubscriptionState::new(OKX_WS_TOPIC_DELIMITER),
698        )
699    }
700
701    #[rstest]
702    #[case("Connection reset by peer", true)]
703    #[case("send timeout after 30s", true)]
704    #[case("Connection closed unexpectedly", true)]
705    #[case("Broken pipe", true)]
706    #[case("Network unreachable", true)]
707    #[case("No active WebSocket client connection", false)]
708    #[case("network protocol upgrade required", false)]
709    #[case("invalid frame format", false)]
710    fn test_should_retry_client_error(#[case] msg: &str, #[case] expected: bool) {
711        let err = OKXWsError::ClientError(msg.to_string());
712        assert_eq!(should_retry_okx_error(&err), expected);
713    }
714
715    #[derive(serde::Deserialize, Debug, PartialEq)]
716    struct ParseArrayItem {
717        value: i64,
718    }
719
720    #[rstest]
721    fn test_parse_array_items_keeps_good_items_when_one_fails() {
722        let data = json!([
723            {"value": 1},
724            {"value": "not a number"},
725            {"value": 3},
726        ]);
727
728        let parsed: Vec<ParseArrayItem> =
729            parse_array_items(data, "test", false).expect("non-empty");
730        assert_eq!(
731            parsed,
732            vec![ParseArrayItem { value: 1 }, ParseArrayItem { value: 3 }],
733        );
734    }
735
736    #[rstest]
737    fn test_parse_array_items_returns_none_when_payload_not_array() {
738        let data = json!({"not": "an array"});
739        let parsed: Option<Vec<ParseArrayItem>> = parse_array_items(data, "test", false);
740        assert!(parsed.is_none());
741    }
742
743    #[rstest]
744    fn test_parse_array_items_returns_none_when_all_items_fail() {
745        let data = json!([{"value": "bad"}]);
746        let parsed: Option<Vec<ParseArrayItem>> = parse_array_items(data, "test", false);
747        assert!(parsed.is_none());
748    }
749
750    #[rstest]
751    fn test_route_instruments_keeps_valid_items_when_one_item_fails() {
752        let handler = create_handler();
753        let mut frame: Value =
754            serde_json::from_str(&load_test_json("ws_instruments.json")).expect("valid fixture");
755        let data = frame
756            .get_mut("data")
757            .and_then(Value::as_array_mut)
758            .expect("data array");
759        let mut invalid_item = data[0].clone();
760        invalid_item["tickSz"] = json!(7);
761        data.insert(0, invalid_item);
762
763        let arg: OKXWebSocketArg = serde_json::from_value(frame["arg"].clone()).expect("valid arg");
764        let msg = handler
765            .route_data_message(arg, frame["data"].clone())
766            .expect("instruments message");
767
768        match msg {
769            OKXWsMessage::Instruments(instruments) => {
770                assert_eq!(instruments.len(), 1);
771                assert_eq!(instruments[0].inst_id.as_str(), "BTC-USDT-SWAP");
772            }
773            other => panic!("Expected Instruments, was {other:?}"),
774        }
775    }
776}
nautilus_okx/websocket/handler.rs

nautilus_okx/websocket/
handler.rs
