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::{
27        AxMdErrorResponse, AxMdMessage, AxOrdersWsFrame, AxWsOrderEvent, AxWsOrderResponse,
28    },
29};
30
31#[inline]
32fn peek_type_tag(bytes: &[u8]) -> Option<u8> {
33    if bytes.len() > 7
34        && bytes[0] == b'{'
35        && bytes[1] == b'"'
36        && bytes[2] == b't'
37        && bytes[3] == b'"'
38        && bytes[4] == b':'
39        && bytes[5] == b'"'
40        && bytes[7] == b'"'
41    {
42        Some(bytes[6])
43    } else {
44        None
45    }
46}
47
48#[inline]
49fn has_type_tag_prefix(bytes: &[u8]) -> bool {
50    bytes.len() > 5 && bytes[0] == b'{' && bytes[1] == b'"' && bytes[2] == b't' && bytes[3] == b'"'
51}
52
53/// Parses a raw JSON string into an [`AxMdMessage`].
54///
55/// Uses a fast byte-scan to extract the type discriminator without
56/// allocating an intermediate `serde_json::Value` tree, then dispatches
57/// directly to the target struct deserializer.
58///
59/// # Errors
60///
61/// Returns an error if the JSON is malformed or has an unknown type tag.
62pub fn parse_md_message(raw: &str) -> Result<AxMdMessage, serde_json::Error> {
63    if let Some(tag) = peek_type_tag(raw.as_bytes()) {
64        return match tag {
65            b'1' => serde_json::from_str(raw).map(AxMdMessage::BookL1),
66            b'2' => serde_json::from_str(raw).map(AxMdMessage::BookL2),
67            b'3' => serde_json::from_str(raw).map(AxMdMessage::BookL3),
68            b's' => serde_json::from_str(raw).map(AxMdMessage::Ticker),
69            b't' => serde_json::from_str(raw).map(AxMdMessage::Trade),
70            b'c' => serde_json::from_str(raw).map(AxMdMessage::Candle),
71            b'h' => serde_json::from_str(raw).map(AxMdMessage::Heartbeat),
72            b'e' => serde_json::from_str::<AxWsErrorResponse>(raw)
73                .map(|resp| AxMdMessage::Error(resp.into())),
74            tag => Err(serde_json::Error::custom(format!(
75                "unknown MD message type tag: '{}'",
76                tag as char
77            ))),
78        };
79    }
80
81    // Slow path: subscription responses and errors (no "t" field, rare)
82    let value: serde_json::Value = serde_json::from_str(raw)?;
83
84    if value.get("result").is_some() {
85        return serde_json::from_value(value).map(AxMdMessage::SubscriptionResponse);
86    }
87
88    if value.get("error").is_some() {
89        return serde_json::from_value::<AxMdErrorResponse>(value)
90            .map(|resp| AxMdMessage::Error(resp.into()));
91    }
92
93    // Fallback: "t" exists but wasn't at position 0
94    if let Some(t) = value.get("t").and_then(|v| v.as_str()) {
95        match t {
96            "1" => serde_json::from_value(value).map(AxMdMessage::BookL1),
97            "2" => serde_json::from_value(value).map(AxMdMessage::BookL2),
98            "3" => serde_json::from_value(value).map(AxMdMessage::BookL3),
99            "s" => serde_json::from_value(value).map(AxMdMessage::Ticker),
100            "t" => serde_json::from_value(value).map(AxMdMessage::Trade),
101            "c" => serde_json::from_value(value).map(AxMdMessage::Candle),
102            "h" => serde_json::from_value(value).map(AxMdMessage::Heartbeat),
103            "e" => serde_json::from_value::<AxWsErrorResponse>(value)
104                .map(|resp| AxMdMessage::Error(resp.into())),
105            other => Err(serde_json::Error::custom(format!(
106                "unknown MD message type: {other}"
107            ))),
108        }
109    } else {
110        Err(serde_json::Error::custom(
111            "MD message has no 't', 'result', or 'error' field",
112        ))
113    }
114}
115
116/// Parses a raw JSON string into an [`AxOrdersWsFrame`].
117///
118/// Events (most frequent) get a fast byte-scan to detect the `"t"` field
119/// and dispatch directly to `AxWsOrderEvent` (internally tagged).
120/// Responses and errors (infrequent) use a single `Value` parse with
121/// field inspection, avoiding the sequential-try overhead of `untagged`.
122pub(crate) fn parse_order_message(raw: &str) -> Result<AxOrdersWsFrame, serde_json::Error> {
123    // Fast path: event messages start with {"t":"
124    if has_type_tag_prefix(raw.as_bytes()) {
125        return parse_order_event(raw).map(|e| AxOrdersWsFrame::Event(Box::new(e)));
126    }
127
128    // Slow path: responses and errors (infrequent, use Value dispatch)
129    let value: serde_json::Value = serde_json::from_str(raw)?;
130
131    if value.get("err").is_some() {
132        return serde_json::from_value(value).map(AxOrdersWsFrame::Error);
133    }
134
135    if let Some(res) = value.get("res") {
136        if res.get("orders").is_some() {
137            return serde_json::from_value(value)
138                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::OpenOrders(r)));
139        }
140
141        if res.get("oid").is_some() {
142            return serde_json::from_value(value)
143                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::PlaceOrder(r)));
144        }
145
146        if res.get("cxl_rx").is_some() {
147            return serde_json::from_value(value)
148                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::CancelOrder(r)));
149        }
150
151        if res.get("li").is_some() {
152            return serde_json::from_value(value)
153                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::List(r)));
154        }
155
156        return Err(serde_json::Error::custom(
157            "unrecognized order response shape",
158        ));
159    }
160
161    // Fallback: may be an event with "t" not at position 0
162    if value.get("t").is_some() {
163        return parse_order_event(raw).map(|e| AxOrdersWsFrame::Event(Box::new(e)));
164    }
165
166    Err(serde_json::Error::custom(
167        "order WS message has no 't', 'err', or 'res' field",
168    ))
169}
170
171fn parse_order_event(raw: &str) -> Result<AxWsOrderEvent, serde_json::Error> {
172    match serde_json::from_str(raw) {
173        Ok(event) => Ok(event),
174        Err(e) => {
175            // Live orders WS sends undocumented `{"t":"pu"}` about every 2s.
176            let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
177                return Err(e);
178            };
179
180            if value.get("t").and_then(|v| v.as_str()) == Some("pu") {
181                return Ok(AxWsOrderEvent::Heartbeat);
182            }
183
184            Err(e)
185        }
186    }
187}
188
189#[cfg(test)]
190mod tests {
191    use rstest::rstest;
192
193    use super::*;
194    use crate::websocket::messages::{
195        AxMdMessage, AxOrdersWsFrame, AxWsOrderEvent, AxWsOrderResponse,
196    };
197
198    #[rstest]
199    fn test_parse_md_message_unknown_tag_errors() {
200        let raw = r#"{"t":"X","s":"EURUSD-PERP"}"#;
201        let err = parse_md_message(raw).expect_err("unknown tag should error");
202        assert!(err.to_string().contains("unknown MD message type tag"));
203    }
204
205    #[rstest]
206    fn test_parse_md_message_slow_path_subscription_response() {
207        let raw = r#"{"rid":1,"result":{"subscribed":"EURUSD-PERP"}}"#;
208        let msg = parse_md_message(raw).expect("should parse subscription response");
209        assert!(matches!(msg, AxMdMessage::SubscriptionResponse(_)));
210    }
211
212    #[rstest]
213    fn test_parse_md_message_slow_path_error_response() {
214        let raw = r#"{"rid":2,"error":{"code":400,"message":"bad"}}"#;
215        let msg = parse_md_message(raw).expect("should parse error response");
216        match msg {
217            AxMdMessage::Error(err) => {
218                assert_eq!(err.message, "bad");
219                assert_eq!(err.request_id, Some(2));
220            }
221            other => panic!("expected Error variant, was {other:?}"),
222        }
223    }
224
225    #[rstest]
226    fn test_parse_md_message_no_recognized_fields_errors() {
227        let raw = r#"{"foo":"bar"}"#;
228        let err = parse_md_message(raw).expect_err("should reject unknown shape");
229        assert!(
230            err.to_string()
231                .contains("no 't', 'result', or 'error' field")
232        );
233    }
234
235    #[rstest]
236    fn test_parse_md_message_malformed_json_errors() {
237        let raw = "not json";
238        assert!(parse_md_message(raw).is_err());
239    }
240
241    #[rstest]
242    fn test_parse_order_message_unrecognized_res_errors() {
243        let raw = r#"{"rid":1,"res":{"foo":"bar"}}"#;
244        let err = parse_order_message(raw).expect_err("unrecognized res shape should error");
245        assert!(
246            err.to_string()
247                .contains("unrecognized order response shape")
248        );
249    }
250
251    #[rstest]
252    fn test_parse_order_message_no_recognized_fields_errors() {
253        let raw = r#"{"foo":"bar"}"#;
254        let err = parse_order_message(raw).expect_err("unknown shape should error");
255        assert!(err.to_string().contains("no 't', 'err', or 'res' field"));
256    }
257
258    #[rstest]
259    fn test_parse_order_message_pu_keep_alive() {
260        let msg = parse_order_message(r#"{"t":"pu"}"#).expect("should parse keep-alive");
261        assert!(matches!(
262            msg,
263            AxOrdersWsFrame::Event(event) if matches!(*event, AxWsOrderEvent::Heartbeat)
264        ));
265    }
266
267    #[rstest]
268    fn test_parse_order_message_unknown_tag_errors() {
269        let err = parse_order_message(r#"{"t":"zz"}"#).expect_err("unknown tag should error");
270        assert!(err.to_string().contains("unknown variant `zz`"));
271    }
272
273    #[rstest]
274    fn test_parse_order_message_malformed_json_errors() {
275        let raw = "not json";
276        assert!(parse_order_message(raw).is_err());
277    }
278
279    #[rstest]
280    fn test_parse_order_message_list_response_with_orders() {
281        let raw = r#"{"rid":0,"res":{"li":"01KCQM-4WP1-0000","o":[]}}"#;
282        let msg = parse_order_message(raw).expect("should parse list response");
283        assert!(matches!(
284            msg,
285            AxOrdersWsFrame::Response(AxWsOrderResponse::List(_))
286        ));
287    }
288
289    #[rstest]
290    fn test_parse_order_message_replaced_live_shape() {
291        let raw = include_str!("../../test_data/ws_order_replaced_live.json");
292        let msg = parse_order_message(raw).expect("should parse live replaced event");
293
294        let AxOrdersWsFrame::Event(event) = msg else {
295            panic!("expected Event frame");
296        };
297        let AxWsOrderEvent::Replaced(replaced) = *event else {
298            panic!("expected Replaced event");
299        };
300
301        assert_eq!(replaced.no.oid, "O-01KWY01WX8JT4DABKC6FRS5NT4");
302        assert_eq!(replaced.no.rq, 100);
303    }
304
305    #[rstest]
306    #[case::direct_open_orders_array(
307        include_str!("../../test_data/ws_order_open_orders_response_invalid_direct_array.json"),
308        "unrecognized order response shape",
309    )]
310    #[case::single_replacement_order(
311        include_str!("../../test_data/ws_order_replaced_invalid_single_order.json"),
312        "missing field `ro`",
313    )]
314    fn test_parse_order_message_rejects_obsolete_shapes(
315        #[case] raw: &str,
316        #[case] expected_error: &str,
317    ) {
318        let error = parse_order_message(raw).expect_err("obsolete shape should be rejected");
319        assert!(
320            error.to_string().contains(expected_error),
321            "expected {expected_error:?} in {error}",
322        );
323    }
324}
nautilus_architect_ax/websocket/parse.rs

nautilus_architect_ax/websocket/
parse.rs
