nautilus_architect_ax/websocket/
parse.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//! Fast JSON message parsers for Ax WebSocket streams.
17//!
18//! Both parsers use byte-level prefix scanning to identify the message type
19//! tag before dispatching to the correct serde target struct, avoiding the
20//! intermediate `serde_json::Value` allocation on the hot path.
21
22use serde::de::Error;
23
24use super::{
25    error::AxWsErrorResponse,
26    messages::{AxMdErrorResponse, AxMdMessage, AxWsOrderResponse, AxWsRawMessage},
27};
28
29#[inline]
30fn peek_type_tag(bytes: &[u8]) -> Option<u8> {
31    if bytes.len() > 7
32        && bytes[0] == b'{'
33        && bytes[1] == b'"'
34        && bytes[2] == b't'
35        && bytes[3] == b'"'
36        && bytes[4] == b':'
37        && bytes[5] == b'"'
38        && bytes[7] == b'"'
39    {
40        Some(bytes[6])
41    } else {
42        None
43    }
44}
45
46#[inline]
47fn has_type_tag_prefix(bytes: &[u8]) -> bool {
48    bytes.len() > 5 && bytes[0] == b'{' && bytes[1] == b'"' && bytes[2] == b't' && bytes[3] == b'"'
49}
50
51/// Parses a raw JSON string into an [`AxMdMessage`].
52///
53/// Uses a fast byte-scan to extract the type discriminator without
54/// allocating an intermediate `serde_json::Value` tree, then dispatches
55/// directly to the target struct deserializer.
56///
57/// # Errors
58///
59/// Returns an error if the JSON is malformed or has an unknown type tag.
60pub fn parse_md_message(raw: &str) -> Result<AxMdMessage, serde_json::Error> {
61    if let Some(tag) = peek_type_tag(raw.as_bytes()) {
62        return match tag {
63            b'1' => serde_json::from_str(raw).map(AxMdMessage::BookL1),
64            b'2' => serde_json::from_str(raw).map(AxMdMessage::BookL2),
65            b'3' => serde_json::from_str(raw).map(AxMdMessage::BookL3),
66            b's' => serde_json::from_str(raw).map(AxMdMessage::Ticker),
67            b't' => serde_json::from_str(raw).map(AxMdMessage::Trade),
68            b'c' => serde_json::from_str(raw).map(AxMdMessage::Candle),
69            b'h' => serde_json::from_str(raw).map(AxMdMessage::Heartbeat),
70            b'e' => serde_json::from_str::<AxWsErrorResponse>(raw)
71                .map(|resp| AxMdMessage::Error(resp.into())),
72            tag => Err(serde_json::Error::custom(format!(
73                "unknown MD message type tag: '{}'",
74                tag as char
75            ))),
76        };
77    }
78
79    // Slow path: subscription responses and errors (no "t" field, rare)
80    let value: serde_json::Value = serde_json::from_str(raw)?;
81
82    if value.get("result").is_some() {
83        return serde_json::from_value(value).map(AxMdMessage::SubscriptionResponse);
84    }
85
86    if value.get("error").is_some() {
87        return serde_json::from_value::<AxMdErrorResponse>(value)
88            .map(|resp| AxMdMessage::Error(resp.into()));
89    }
90
91    // Fallback: "t" exists but wasn't at position 0
92    if let Some(t) = value.get("t").and_then(|v| v.as_str()) {
93        match t {
94            "1" => serde_json::from_value(value).map(AxMdMessage::BookL1),
95            "2" => serde_json::from_value(value).map(AxMdMessage::BookL2),
96            "3" => serde_json::from_value(value).map(AxMdMessage::BookL3),
97            "s" => serde_json::from_value(value).map(AxMdMessage::Ticker),
98            "t" => serde_json::from_value(value).map(AxMdMessage::Trade),
99            "c" => serde_json::from_value(value).map(AxMdMessage::Candle),
100            "h" => serde_json::from_value(value).map(AxMdMessage::Heartbeat),
101            "e" => serde_json::from_value::<AxWsErrorResponse>(value)
102                .map(|resp| AxMdMessage::Error(resp.into())),
103            other => Err(serde_json::Error::custom(format!(
104                "unknown MD message type: {other}"
105            ))),
106        }
107    } else {
108        Err(serde_json::Error::custom(
109            "MD message has no 't', 'result', or 'error' field",
110        ))
111    }
112}
113
114/// Parses a raw JSON string into an [`AxWsRawMessage`].
115///
116/// Events (most frequent) get a fast byte-scan to detect the `"t"` field
117/// and dispatch directly to `AxWsOrderEvent` (internally tagged).
118/// Responses and errors (infrequent) use a single `Value` parse with
119/// field inspection, avoiding the sequential-try overhead of `untagged`.
120pub(crate) fn parse_order_message(raw: &str) -> Result<AxWsRawMessage, serde_json::Error> {
121    // Fast path: event messages start with {"t":"
122    if has_type_tag_prefix(raw.as_bytes()) {
123        return serde_json::from_str(raw).map(|e| AxWsRawMessage::Event(Box::new(e)));
124    }
125
126    // Slow path: responses and errors (infrequent, use Value dispatch)
127    let value: serde_json::Value = serde_json::from_str(raw)?;
128
129    if value.get("err").is_some() {
130        return serde_json::from_value(value).map(AxWsRawMessage::Error);
131    }
132
133    if let Some(res) = value.get("res") {
134        if res.is_array() {
135            return serde_json::from_value(value)
136                .map(|r| AxWsRawMessage::Response(AxWsOrderResponse::OpenOrders(r)));
137        }
138        if res.get("oid").is_some() {
139            return serde_json::from_value(value)
140                .map(|r| AxWsRawMessage::Response(AxWsOrderResponse::PlaceOrder(r)));
141        }
142        if res.get("cxl_rx").is_some() {
143            return serde_json::from_value(value)
144                .map(|r| AxWsRawMessage::Response(AxWsOrderResponse::CancelOrder(r)));
145        }
146        if res.get("li").is_some() {
147            return serde_json::from_value(value)
148                .map(|r| AxWsRawMessage::Response(AxWsOrderResponse::List(r)));
149        }
150
151        return Err(serde_json::Error::custom(
152            "unrecognized order response shape",
153        ));
154    }
155
156    // Fallback: may be an event with "t" not at position 0
157    if value.get("t").is_some() {
158        return serde_json::from_value(value).map(|e| AxWsRawMessage::Event(Box::new(e)));
159    }
160
161    Err(serde_json::Error::custom(
162        "order WS message has no 't', 'err', or 'res' field",
163    ))
164}
nautilus_architect_ax/websocket/parse.rs

nautilus_architect_ax/websocket/
parse.rs
