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, AxOrdersWsFrame, AxWsOrderResponse},
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 [`AxOrdersWsFrame`].
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<AxOrdersWsFrame, 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| AxOrdersWsFrame::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(AxOrdersWsFrame::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| AxOrdersWsFrame::Response(AxWsOrderResponse::OpenOrders(r)));
137        }
138
139        if res.get("oid").is_some() {
140            return serde_json::from_value(value)
141                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::PlaceOrder(r)));
142        }
143
144        if res.get("cxl_rx").is_some() {
145            return serde_json::from_value(value)
146                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::CancelOrder(r)));
147        }
148
149        if res.get("li").is_some() {
150            return serde_json::from_value(value)
151                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::List(r)));
152        }
153
154        return Err(serde_json::Error::custom(
155            "unrecognized order response shape",
156        ));
157    }
158
159    // Fallback: may be an event with "t" not at position 0
160    if value.get("t").is_some() {
161        return serde_json::from_value(value).map(|e| AxOrdersWsFrame::Event(Box::new(e)));
162    }
163
164    Err(serde_json::Error::custom(
165        "order WS message has no 't', 'err', or 'res' field",
166    ))
167}
168
169#[cfg(test)]
170mod tests {
171    use rstest::rstest;
172
173    use super::*;
174    use crate::websocket::messages::{AxMdMessage, AxOrdersWsFrame, AxWsOrderResponse};
175
176    #[rstest]
177    fn test_parse_md_message_unknown_tag_errors() {
178        let raw = r#"{"t":"X","s":"EURUSD-PERP"}"#;
179        let err = parse_md_message(raw).expect_err("unknown tag should error");
180        assert!(err.to_string().contains("unknown MD message type tag"));
181    }
182
183    #[rstest]
184    fn test_parse_md_message_slow_path_subscription_response() {
185        let raw = r#"{"rid":1,"result":{"subscribed":"EURUSD-PERP"}}"#;
186        let msg = parse_md_message(raw).expect("should parse subscription response");
187        assert!(matches!(msg, AxMdMessage::SubscriptionResponse(_)));
188    }
189
190    #[rstest]
191    fn test_parse_md_message_slow_path_error_response() {
192        let raw = r#"{"rid":2,"error":{"code":400,"message":"bad"}}"#;
193        let msg = parse_md_message(raw).expect("should parse error response");
194        match msg {
195            AxMdMessage::Error(err) => {
196                assert_eq!(err.message, "bad");
197                assert_eq!(err.request_id, Some(2));
198            }
199            other => panic!("expected Error variant, was {other:?}"),
200        }
201    }
202
203    #[rstest]
204    fn test_parse_md_message_no_recognized_fields_errors() {
205        let raw = r#"{"foo":"bar"}"#;
206        let err = parse_md_message(raw).expect_err("should reject unknown shape");
207        assert!(
208            err.to_string()
209                .contains("no 't', 'result', or 'error' field")
210        );
211    }
212
213    #[rstest]
214    fn test_parse_md_message_malformed_json_errors() {
215        let raw = "not json";
216        assert!(parse_md_message(raw).is_err());
217    }
218
219    #[rstest]
220    fn test_parse_order_message_unrecognized_res_errors() {
221        let raw = r#"{"rid":1,"res":{"foo":"bar"}}"#;
222        let err = parse_order_message(raw).expect_err("unrecognized res shape should error");
223        assert!(
224            err.to_string()
225                .contains("unrecognized order response shape")
226        );
227    }
228
229    #[rstest]
230    fn test_parse_order_message_no_recognized_fields_errors() {
231        let raw = r#"{"foo":"bar"}"#;
232        let err = parse_order_message(raw).expect_err("unknown shape should error");
233        assert!(err.to_string().contains("no 't', 'err', or 'res' field"));
234    }
235
236    #[rstest]
237    fn test_parse_order_message_malformed_json_errors() {
238        let raw = "not json";
239        assert!(parse_order_message(raw).is_err());
240    }
241
242    #[rstest]
243    fn test_parse_order_message_list_response_with_orders() {
244        let raw = r#"{"rid":0,"res":{"li":"01KCQM-4WP1-0000","o":[]}}"#;
245        let msg = parse_order_message(raw).expect("should parse list response");
246        assert!(matches!(
247            msg,
248            AxOrdersWsFrame::Response(AxWsOrderResponse::List(_))
249        ));
250    }
251}
nautilus_architect_ax/websocket/parse.rs

nautilus_architect_ax/websocket/
parse.rs
