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.get("orders").is_some() {
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::{
175        AxMdMessage, AxOrdersWsFrame, AxWsOrderEvent, AxWsOrderResponse,
176    };
177
178    #[rstest]
179    fn test_parse_md_message_unknown_tag_errors() {
180        let raw = r#"{"t":"X","s":"EURUSD-PERP"}"#;
181        let err = parse_md_message(raw).expect_err("unknown tag should error");
182        assert!(err.to_string().contains("unknown MD message type tag"));
183    }
184
185    #[rstest]
186    fn test_parse_md_message_slow_path_subscription_response() {
187        let raw = r#"{"rid":1,"result":{"subscribed":"EURUSD-PERP"}}"#;
188        let msg = parse_md_message(raw).expect("should parse subscription response");
189        assert!(matches!(msg, AxMdMessage::SubscriptionResponse(_)));
190    }
191
192    #[rstest]
193    fn test_parse_md_message_slow_path_error_response() {
194        let raw = r#"{"rid":2,"error":{"code":400,"message":"bad"}}"#;
195        let msg = parse_md_message(raw).expect("should parse error response");
196        match msg {
197            AxMdMessage::Error(err) => {
198                assert_eq!(err.message, "bad");
199                assert_eq!(err.request_id, Some(2));
200            }
201            other => panic!("expected Error variant, was {other:?}"),
202        }
203    }
204
205    #[rstest]
206    fn test_parse_md_message_no_recognized_fields_errors() {
207        let raw = r#"{"foo":"bar"}"#;
208        let err = parse_md_message(raw).expect_err("should reject unknown shape");
209        assert!(
210            err.to_string()
211                .contains("no 't', 'result', or 'error' field")
212        );
213    }
214
215    #[rstest]
216    fn test_parse_md_message_malformed_json_errors() {
217        let raw = "not json";
218        assert!(parse_md_message(raw).is_err());
219    }
220
221    #[rstest]
222    fn test_parse_order_message_unrecognized_res_errors() {
223        let raw = r#"{"rid":1,"res":{"foo":"bar"}}"#;
224        let err = parse_order_message(raw).expect_err("unrecognized res shape should error");
225        assert!(
226            err.to_string()
227                .contains("unrecognized order response shape")
228        );
229    }
230
231    #[rstest]
232    fn test_parse_order_message_no_recognized_fields_errors() {
233        let raw = r#"{"foo":"bar"}"#;
234        let err = parse_order_message(raw).expect_err("unknown shape should error");
235        assert!(err.to_string().contains("no 't', 'err', or 'res' field"));
236    }
237
238    #[rstest]
239    fn test_parse_order_message_malformed_json_errors() {
240        let raw = "not json";
241        assert!(parse_order_message(raw).is_err());
242    }
243
244    #[rstest]
245    fn test_parse_order_message_list_response_with_orders() {
246        let raw = r#"{"rid":0,"res":{"li":"01KCQM-4WP1-0000","o":[]}}"#;
247        let msg = parse_order_message(raw).expect("should parse list response");
248        assert!(matches!(
249            msg,
250            AxOrdersWsFrame::Response(AxWsOrderResponse::List(_))
251        ));
252    }
253
254    #[rstest]
255    fn test_parse_order_message_replaced_live_shape() {
256        let raw = include_str!("../../test_data/ws_order_replaced_live.json");
257        let msg = parse_order_message(raw).expect("should parse live replaced event");
258
259        let AxOrdersWsFrame::Event(event) = msg else {
260            panic!("expected Event frame");
261        };
262        let AxWsOrderEvent::Replaced(replaced) = *event else {
263            panic!("expected Replaced event");
264        };
265
266        assert_eq!(replaced.no.oid, "O-01KWY01WX8JT4DABKC6FRS5NT4");
267        assert_eq!(replaced.no.rq, 100);
268    }
269
270    #[rstest]
271    #[case::direct_open_orders_array(
272        include_str!("../../test_data/ws_order_open_orders_response_invalid_direct_array.json"),
273        "unrecognized order response shape",
274    )]
275    #[case::single_replacement_order(
276        include_str!("../../test_data/ws_order_replaced_invalid_single_order.json"),
277        "missing field `ro`",
278    )]
279    fn test_parse_order_message_rejects_obsolete_shapes(
280        #[case] raw: &str,
281        #[case] expected_error: &str,
282    ) {
283        let error = parse_order_message(raw).expect_err("obsolete shape should be rejected");
284        assert!(
285            error.to_string().contains(expected_error),
286            "expected {expected_error:?} in {error}",
287        );
288    }
289}
nautilus_architect_ax/websocket/parse.rs

nautilus_architect_ax/websocket/
parse.rs
