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
85        .get("result")
86        .or_else(|| value.get("res"))
87        .is_some_and(|v| !v.is_null())
88    {
89        return serde_json::from_value(value).map(AxMdMessage::SubscriptionResponse);
90    }
91
92    if value
93        .get("error")
94        .or_else(|| value.get("err"))
95        .is_some_and(|v| !v.is_null())
96    {
97        return serde_json::from_value::<AxMdErrorResponse>(value)
98            .map(|resp| AxMdMessage::Error(resp.into()));
99    }
100
101    // Fallback: "t" exists but wasn't at position 0
102    if let Some(t) = value.get("t").and_then(|v| v.as_str()) {
103        match t {
104            "1" => serde_json::from_value(value).map(AxMdMessage::BookL1),
105            "2" => serde_json::from_value(value).map(AxMdMessage::BookL2),
106            "3" => serde_json::from_value(value).map(AxMdMessage::BookL3),
107            "s" => serde_json::from_value(value).map(AxMdMessage::Ticker),
108            "t" => serde_json::from_value(value).map(AxMdMessage::Trade),
109            "c" => serde_json::from_value(value).map(AxMdMessage::Candle),
110            "h" => serde_json::from_value(value).map(AxMdMessage::Heartbeat),
111            "e" => serde_json::from_value::<AxWsErrorResponse>(value)
112                .map(|resp| AxMdMessage::Error(resp.into())),
113            other => Err(serde_json::Error::custom(format!(
114                "unknown MD message type: {other}"
115            ))),
116        }
117    } else {
118        Err(serde_json::Error::custom(
119            "MD message has no 't', 'result', or 'error' field",
120        ))
121    }
122}
123
124/// Parses a raw JSON string into an [`AxOrdersWsFrame`].
125///
126/// Events (most frequent) get a fast byte-scan to detect the `"t"` field
127/// and dispatch directly to `AxWsOrderEvent` (internally tagged).
128/// Responses and errors (infrequent) use a single `Value` parse with
129/// field inspection, avoiding the sequential-try overhead of `untagged`.
130pub(crate) fn parse_order_message(raw: &str) -> Result<AxOrdersWsFrame, serde_json::Error> {
131    // Fast path: event messages start with {"t":"
132    if has_type_tag_prefix(raw.as_bytes()) {
133        return parse_order_event(raw).map(|e| AxOrdersWsFrame::Event(Box::new(e)));
134    }
135
136    // Slow path: responses and errors (infrequent, use Value dispatch)
137    let value: serde_json::Value = serde_json::from_str(raw)?;
138
139    if value
140        .get("err")
141        .or_else(|| value.get("error"))
142        .is_some_and(|v| !v.is_null())
143    {
144        return serde_json::from_value(value).map(AxOrdersWsFrame::Error);
145    }
146
147    if let Some(res) = value.get("res").or_else(|| value.get("result")) {
148        if res.get("orders").is_some() {
149            return serde_json::from_value(value)
150                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::OpenOrders(r)));
151        }
152
153        if res.get("oid").is_some() {
154            return serde_json::from_value(value)
155                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::PlaceOrder(r)));
156        }
157
158        if res.get("cxl_rx").is_some() {
159            return serde_json::from_value(value)
160                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::CancelOrder(r)));
161        }
162
163        if res.get("li").is_some() {
164            return serde_json::from_value(value)
165                .map(|r| AxOrdersWsFrame::Response(AxWsOrderResponse::List(r)));
166        }
167
168        return Err(serde_json::Error::custom(
169            "unrecognized order response shape",
170        ));
171    }
172
173    // Fallback: may be an event with "t" not at position 0
174    if value.get("t").is_some() {
175        return parse_order_event(raw).map(|e| AxOrdersWsFrame::Event(Box::new(e)));
176    }
177
178    Err(serde_json::Error::custom(
179        "order WS message has no 't', 'err', or 'res' field",
180    ))
181}
182
183fn parse_order_event(raw: &str) -> Result<AxWsOrderEvent, serde_json::Error> {
184    match serde_json::from_str(raw) {
185        Ok(event) => Ok(event),
186        Err(e) => {
187            // Live orders WS sends undocumented `{"t":"pu"}` about every 2s.
188            let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
189                return Err(e);
190            };
191
192            if value.get("t").and_then(|v| v.as_str()) == Some("pu") {
193                return Ok(AxWsOrderEvent::Heartbeat);
194            }
195
196            Err(e)
197        }
198    }
199}
200
201#[cfg(test)]
202mod tests {
203    use rstest::rstest;
204
205    use super::*;
206    use crate::websocket::messages::{
207        AxMdMessage, AxOrdersWsFrame, AxWsOrderEvent, AxWsOrderResponse,
208    };
209
210    #[rstest]
211    fn test_parse_md_message_unknown_tag_errors() {
212        let raw = r#"{"t":"X","s":"EURUSD-PERP"}"#;
213        let err = parse_md_message(raw).expect_err("unknown tag should error");
214        assert!(err.to_string().contains("unknown MD message type tag"));
215    }
216
217    #[rstest]
218    fn test_parse_md_message_slow_path_subscription_response() {
219        let raw = r#"{"rid":1,"result":{"subscribed":"EURUSD-PERP"}}"#;
220        let msg = parse_md_message(raw).expect("should parse subscription response");
221        assert!(matches!(msg, AxMdMessage::SubscriptionResponse(_)));
222    }
223
224    #[rstest]
225    fn test_parse_md_message_slow_path_error_response() {
226        let raw = r#"{"rid":2,"error":{"code":400,"message":"bad"}}"#;
227        let msg = parse_md_message(raw).expect("should parse error response");
228        match msg {
229            AxMdMessage::Error(err) => {
230                assert_eq!(err.message, "bad");
231                assert_eq!(err.request_id, Some(2));
232            }
233            other => panic!("expected Error variant, was {other:?}"),
234        }
235    }
236
237    #[rstest]
238    fn test_parse_md_message_no_recognized_fields_errors() {
239        let raw = r#"{"foo":"bar"}"#;
240        let err = parse_md_message(raw).expect_err("should reject unknown shape");
241        assert!(
242            err.to_string()
243                .contains("no 't', 'result', or 'error' field")
244        );
245    }
246
247    #[rstest]
248    fn test_parse_md_message_malformed_json_errors() {
249        let raw = "not json";
250        assert!(parse_md_message(raw).is_err());
251    }
252
253    #[rstest]
254    fn test_parse_order_message_unrecognized_res_errors() {
255        let raw = r#"{"rid":1,"res":{"foo":"bar"}}"#;
256        let err = parse_order_message(raw).expect_err("unrecognized res shape should error");
257        assert!(
258            err.to_string()
259                .contains("unrecognized order response shape")
260        );
261    }
262
263    #[rstest]
264    fn test_parse_order_message_no_recognized_fields_errors() {
265        let raw = r#"{"foo":"bar"}"#;
266        let err = parse_order_message(raw).expect_err("unknown shape should error");
267        assert!(err.to_string().contains("no 't', 'err', or 'res' field"));
268    }
269
270    #[rstest]
271    fn test_parse_order_message_pu_keep_alive() {
272        let msg = parse_order_message(r#"{"t":"pu"}"#).expect("should parse keep-alive");
273        assert!(matches!(
274            msg,
275            AxOrdersWsFrame::Event(event) if matches!(*event, AxWsOrderEvent::Heartbeat)
276        ));
277    }
278
279    #[rstest]
280    fn test_parse_order_message_unknown_tag_errors() {
281        let err = parse_order_message(r#"{"t":"zz"}"#).expect_err("unknown tag should error");
282        assert!(err.to_string().contains("unknown variant `zz`"));
283    }
284
285    #[rstest]
286    fn test_parse_order_message_malformed_json_errors() {
287        let raw = "not json";
288        assert!(parse_order_message(raw).is_err());
289    }
290
291    #[rstest]
292    fn test_parse_order_message_list_response_with_orders() {
293        let raw = r#"{"rid":0,"res":{"li":"01KCQM-4WP1-0000","o":[]}}"#;
294        let msg = parse_order_message(raw).expect("should parse list response");
295        assert!(matches!(
296            msg,
297            AxOrdersWsFrame::Response(AxWsOrderResponse::List(_))
298        ));
299    }
300
301    #[rstest]
302    fn test_parse_order_message_replaced_live_shape() {
303        let raw = include_str!("../../test_data/ws_order_replaced_live.json");
304        let msg = parse_order_message(raw).expect("should parse live replaced event");
305
306        let AxOrdersWsFrame::Event(event) = msg else {
307            panic!("expected Event frame");
308        };
309        let AxWsOrderEvent::Replaced(replaced) = *event else {
310            panic!("expected Replaced event");
311        };
312
313        assert_eq!(
314            replaced.no.as_ref().unwrap().oid,
315            "O-01KWY01WX8JT4DABKC6FRS5NT4"
316        );
317        assert_eq!(replaced.no.as_ref().unwrap().rq, 100);
318    }
319
320    #[rstest]
321    #[case::direct_open_orders_array(
322        include_str!("../../test_data/ws_order_open_orders_response_invalid_direct_array.json"),
323        "unrecognized order response shape",
324    )]
325    #[case::single_replacement_order(
326        include_str!("../../test_data/ws_order_replaced_invalid_single_order.json"),
327        "missing field `ro`",
328    )]
329    fn test_parse_order_message_rejects_obsolete_shapes(
330        #[case] raw: &str,
331        #[case] expected_error: &str,
332    ) {
333        let error = parse_order_message(raw).expect_err("obsolete shape should be rejected");
334        assert!(
335            error.to_string().contains(expected_error),
336            "expected {expected_error:?} in {error}",
337        );
338    }
339
340    #[rstest]
341    #[case("rid", "result", "error")]
342    #[case("request_id", "res", "err")]
343    fn test_md_response_aliases(#[case] id: &str, #[case] result: &str, #[case] error: &str) {
344        let mut success: serde_json::Value = serde_json::from_str(include_str!(
345            "../../test_data/captured/ws-md-response-1.json"
346        ))
347        .unwrap();
348        let mut failure: serde_json::Value = serde_json::from_str(include_str!(
349            "../../test_data/captured/ws-md-response-6.json"
350        ))
351        .unwrap();
352
353        for value in [&mut success, &mut failure] {
354            let obj = value.as_object_mut().unwrap();
355            let rid = obj.remove("rid").unwrap();
356            obj.insert(id.into(), rid);
357        }
358
359        let res = success.as_object_mut().unwrap().remove("result").unwrap();
360        success[result] = res;
361        let err = failure.as_object_mut().unwrap().remove("error").unwrap();
362        failure[error] = err;
363
364        let AxMdMessage::SubscriptionResponse(response) =
365            parse_md_message(&success.to_string()).unwrap()
366        else {
367            panic!("expected subscription response")
368        };
369
370        let AxMdMessage::Error(response_error) = parse_md_message(&failure.to_string()).unwrap()
371        else {
372            panic!("expected error response")
373        };
374
375        assert_eq!(response.rid, 1);
376        assert_eq!(response_error.request_id, Some(6));
377        assert_eq!(
378            response_error.message,
379            failure[error]["message"].as_str().unwrap()
380        );
381    }
382
383    #[rstest]
384    #[case("rid", "res", "err")]
385    #[case("request_id", "result", "error")]
386    fn test_orders_response_aliases_and_null_error(
387        #[case] id: &str,
388        #[case] result: &str,
389        #[case] error: &str,
390    ) {
391        let mut value: serde_json::Value = serde_json::from_str(include_str!(
392            "../../test_data/captured/ws-orders-response-0.json"
393        ))
394        .unwrap();
395        let obj = value.as_object_mut().unwrap();
396        let rid = obj.remove("rid").unwrap();
397        let res = obj.remove("res").unwrap();
398        obj.remove("err");
399        obj.insert(id.into(), rid);
400        obj.insert(result.into(), res);
401        obj.insert(error.into(), serde_json::Value::Null);
402
403        let AxOrdersWsFrame::Response(AxWsOrderResponse::List(response)) =
404            parse_order_message(&value.to_string()).unwrap()
405        else {
406            panic!("expected login response")
407        };
408
409        assert_eq!(response.rid, 0);
410        assert_eq!(response.res.li, value[result]["li"].as_str().unwrap());
411        assert_eq!(response.res.cod, value[result]["cod"].as_bool());
412        assert_eq!(response.res.chb, value[result]["chb"].as_u64());
413    }
414
415    #[rstest]
416    fn test_in_place_amendment_has_no_replacement_identity() {
417        let raw = include_str!("../../test_data/ws_order_amended.json");
418
419        let AxOrdersWsFrame::Event(event) = parse_order_message(raw).unwrap() else {
420            panic!("expected event")
421        };
422
423        let AxWsOrderEvent::Replaced(message) = *event else {
424            panic!("expected amendment")
425        };
426
427        assert_eq!(message.noid, None);
428        assert!(message.no.is_none());
429        assert_eq!(message.ro.q, 150);
430        assert_eq!(message.ro.rq, 150);
431        assert_eq!(message.ro.o, crate::common::enums::AxOrderStatus::Accepted);
432    }
433
434    #[rstest]
435    #[case(include_str!("../../test_data/captured/ws-md-1.json"))]
436    #[case(include_str!("../../test_data/captured/ws-md-2.json"))]
437    #[case(include_str!("../../test_data/captured/ws-md-3.json"))]
438    #[case(include_str!("../../test_data/captured/ws-md-c.json"))]
439    #[case(include_str!("../../test_data/captured/ws-md-h.json"))]
440    #[case(include_str!("../../test_data/captured/ws-md-response-1.json"))]
441    #[case(include_str!("../../test_data/captured/ws-md-response-2.json"))]
442    #[case(include_str!("../../test_data/captured/ws-md-response-3.json"))]
443    #[case(include_str!("../../test_data/captured/ws-md-response-4.json"))]
444    #[case(include_str!("../../test_data/captured/ws-md-response-6.json"))]
445    #[case(include_str!("../../test_data/captured/ws-md-s.json"))]
446    #[case(include_str!("../../test_data/captured/ws-md-t.json"))]
447    fn test_current_market_data_capture(#[case] raw: &str) {
448        let value: serde_json::Value = serde_json::from_str(raw).unwrap();
449        let parsed = parse_md_message(raw).unwrap();
450
451        let (tag, ts) = match parsed {
452            AxMdMessage::BookL1(book) => ("1", book.ts),
453            AxMdMessage::BookL2(book) => {
454                assert!(book.st);
455                ("2", book.ts)
456            }
457            AxMdMessage::BookL3(book) => {
458                assert!(book.st);
459                ("3", book.ts)
460            }
461            AxMdMessage::Trade(trade) => ("t", trade.ts),
462            AxMdMessage::Candle(candle) => ("c", candle.ts),
463            AxMdMessage::Ticker(ticker) => ("s", ticker.ts),
464            AxMdMessage::Heartbeat(heartbeat) => ("h", heartbeat.ts),
465            AxMdMessage::SubscriptionResponse(response) => {
466                assert_eq!(response.rid, value["rid"].as_i64().unwrap());
467                assert_eq!(
468                    response
469                        .result
470                        .subscribed
471                        .as_deref()
472                        .or(response.result.subscribed_candle.as_deref()),
473                    value["result"]["subscribed"]
474                        .as_str()
475                        .or(value["result"]["subscribed_candle"].as_str())
476                );
477                return;
478            }
479            AxMdMessage::Error(error) => {
480                assert_eq!(error.request_id, value["rid"].as_i64());
481                assert_eq!(error.message, value["error"]["message"].as_str().unwrap());
482                return;
483            }
484        };
485
486        assert_eq!(tag, value["t"].as_str().unwrap());
487        assert_eq!(ts, value["ts"].as_i64().unwrap());
488    }
489
490    #[rstest]
491    #[case(include_str!("../../test_data/captured/ws-orders-response-2.json"))]
492    #[case(include_str!("../../test_data/captured/ws-orders-response-3.json"))]
493    fn test_orders_error_alias(#[case] raw: &str) {
494        let mut value: serde_json::Value = serde_json::from_str(raw).unwrap();
495        let obj = value.as_object_mut().unwrap();
496        let rid = obj.remove("rid").unwrap();
497        let error = obj.remove("err").unwrap();
498        obj.insert("request_id".into(), rid);
499        obj.insert("error".into(), error);
500
501        let AxOrdersWsFrame::Error(response) = parse_order_message(&value.to_string()).unwrap()
502        else {
503            panic!("expected error")
504        };
505
506        assert_eq!(response.rid, value["request_id"].as_i64().unwrap());
507        assert_eq!(response.err.code, value["error"]["code"].as_i64());
508        assert_eq!(response.err.msg.as_deref(), value["error"]["msg"].as_str());
509    }
510    #[rstest]
511    #[case(false, false)]
512    #[case(true, false)]
513    #[case(false, true)]
514    #[case(true, true)]
515    fn test_sparse_order_error_preserves_request_identity(
516        #[case] code: bool,
517        #[case] message: bool,
518    ) {
519        let mut value: serde_json::Value =
520            serde_json::from_str(include_str!("../../test_data/ws_order_error_response.json"))
521                .unwrap();
522        let expected_id = value["rid"].as_i64();
523        let expected_code = code.then(|| value["err"]["code"].as_i64().unwrap().to_string());
524
525        let expected_message = if message {
526            value["err"]["msg"].as_str().unwrap().to_owned()
527        } else {
528            "AX request failed without an error message".to_owned()
529        };
530
531        if !code {
532            value["err"].as_object_mut().unwrap().remove("code");
533        }
534
535        if !message {
536            value["err"].as_object_mut().unwrap().remove("msg");
537        }
538
539        let AxOrdersWsFrame::Error(response) = parse_order_message(&value.to_string()).unwrap()
540        else {
541            panic!("expected error")
542        };
543
544        let error: crate::websocket::messages::AxWsError = response.into();
545
546        assert_eq!(error.request_id, expected_id);
547        assert_eq!(error.code, expected_code);
548        assert_eq!(error.message, expected_message);
549    }
550}
nautilus_architect_ax/websocket/parse.rs

nautilus_architect_ax/websocket/
parse.rs
