Skip to main content

nautilus_architect_ax/websocket/orders/
handler.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//! Orders WebSocket message handler for Ax.
17
18use std::{
19    collections::VecDeque,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26use ahash::AHashMap;
27use dashmap::DashMap;
28use nautilus_model::identifiers::{ClientOrderId, VenueOrderId};
29use nautilus_network::websocket::{AuthTracker, WebSocketClient};
30use tokio_tungstenite::tungstenite::Message;
31use ustr::Ustr;
32
33use crate::{
34    common::enums::AxOrderRequestType,
35    websocket::{
36        messages::{
37            AxOrdersWsFrame, AxOrdersWsMessage, AxWsCancelOrder, AxWsError, AxWsGetOpenOrders,
38            AxWsOrderEvent, AxWsOrderResponse, AxWsPlaceOrder, OrderMetadata,
39        },
40        parse::parse_order_message,
41    },
42};
43
44/// Simple tracking info for pending WebSocket orders.
45#[derive(Clone, Debug)]
46pub struct WsOrderInfo {
47    /// Client order ID for correlation.
48    pub client_order_id: ClientOrderId,
49    /// Instrument symbol.
50    pub symbol: Ustr,
51    /// Numeric AX client ID.
52    pub cid: u64,
53}
54
55/// Commands sent from the outer client to the inner orders handler.
56#[derive(Debug)]
57pub enum HandlerCommand {
58    /// Set the WebSocket client for this handler.
59    SetClient(WebSocketClient),
60    /// Disconnect the WebSocket connection.
61    Disconnect,
62    /// Mark the current handshake-authenticated session as ready.
63    SessionAuthenticated,
64    /// Place an order.
65    PlaceOrder {
66        /// Request ID for correlation.
67        request_id: i64,
68        /// Order placement message.
69        order: AxWsPlaceOrder,
70        /// Order info for tracking.
71        order_info: WsOrderInfo,
72    },
73    /// Cancel an order.
74    CancelOrder {
75        /// Request ID for correlation.
76        request_id: i64,
77        /// Order ID to cancel.
78        order_id: String,
79    },
80    /// Get open orders.
81    GetOpenOrders {
82        /// Request ID for correlation.
83        request_id: i64,
84    },
85}
86
87/// Orders feed handler that processes WebSocket messages.
88///
89/// Runs in a dedicated Tokio task and owns the WebSocket client exclusively.
90/// Emits raw venue types for downstream consumers to parse into domain events.
91pub(crate) struct AxOrdersWsFeedHandler {
92    signal: Arc<AtomicBool>,
93    inner: Option<WebSocketClient>,
94    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
95    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
96    auth_tracker: AuthTracker,
97    pending_orders: AHashMap<i64, WsOrderInfo>,
98    message_queue: VecDeque<AxOrdersWsMessage>,
99    orders_metadata: Arc<DashMap<ClientOrderId, OrderMetadata>>,
100    venue_to_client_order_id: Arc<DashMap<VenueOrderId, ClientOrderId>>,
101    has_authenticated_session: bool,
102    needs_session_restore: bool,
103}
104
105impl AxOrdersWsFeedHandler {
106    /// Creates a new [`AxOrdersWsFeedHandler`] instance.
107    #[must_use]
108    pub(crate) fn new(
109        signal: Arc<AtomicBool>,
110        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
111        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
112        auth_tracker: AuthTracker,
113        orders_metadata: Arc<DashMap<ClientOrderId, OrderMetadata>>,
114        venue_to_client_order_id: Arc<DashMap<VenueOrderId, ClientOrderId>>,
115    ) -> Self {
116        Self {
117            signal,
118            inner: None,
119            cmd_rx,
120            raw_rx,
121            auth_tracker,
122            pending_orders: AHashMap::new(),
123            message_queue: VecDeque::new(),
124            orders_metadata,
125            venue_to_client_order_id,
126            has_authenticated_session: false,
127            needs_session_restore: false,
128        }
129    }
130
131    fn restore_authenticated_session(&mut self) {
132        if self.has_authenticated_session {
133            log::debug!("Restoring authenticated session after reconnection");
134
135            // The reconnect handshake has already succeeded with the current Bearer header.
136            self.auth_tracker.succeed();
137            self.message_queue
138                .push_back(AxOrdersWsMessage::Authenticated);
139            log::debug!("Authenticated session restored");
140        } else {
141            log::warn!("Cannot restore authentication before the initial session succeeds");
142        }
143    }
144
145    /// Returns the next message from the handler.
146    ///
147    /// This method blocks until a message is available or the handler is stopped.
148    pub(crate) async fn next(&mut self) -> Option<AxOrdersWsMessage> {
149        loop {
150            if self.needs_session_restore && self.message_queue.is_empty() {
151                self.needs_session_restore = false;
152                self.restore_authenticated_session();
153            }
154
155            if let Some(msg) = self.message_queue.pop_front() {
156                return Some(msg);
157            }
158
159            tokio::select! {
160                Some(cmd) = self.cmd_rx.recv() => {
161                    self.handle_command(cmd).await;
162                }
163
164                () = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
165                    if self.signal.load(Ordering::Acquire) {
166                        log::debug!("Stop signal received during idle period");
167                        return None;
168                    }
169                }
170
171                msg = self.raw_rx.recv() => {
172                    let msg = match msg {
173                        Some(msg) => msg,
174                        None => {
175                            log::debug!("WebSocket stream closed");
176                            return None;
177                        }
178                    };
179
180                    if let Message::Ping(data) = &msg {
181                        log::trace!("Received ping frame with {} bytes", data.len());
182
183                        if let Some(client) = &self.inner
184                            && let Err(e) = client.send_pong(data.to_vec()).await
185                        {
186                            log::warn!("Failed to send pong frame: {e}");
187                        }
188                        continue;
189                    }
190
191                    if let Some(messages) = self.parse_raw_message(msg) {
192                        self.message_queue.extend(messages);
193                    }
194
195                    if self.signal.load(Ordering::Acquire) {
196                        log::debug!("Stop signal received");
197                        return None;
198                    }
199                }
200            }
201        }
202    }
203
204    async fn handle_command(&mut self, cmd: HandlerCommand) {
205        match cmd {
206            HandlerCommand::SetClient(client) => {
207                log::debug!("WebSocketClient received by handler");
208                self.inner = Some(client);
209            }
210            HandlerCommand::Disconnect => {
211                log::debug!("Disconnect command received");
212                self.auth_tracker.fail("Disconnected");
213
214                if let Some(inner) = self.inner.take() {
215                    inner.disconnect().await;
216                }
217            }
218            HandlerCommand::SessionAuthenticated => {
219                log::debug!("Session authenticated command received");
220                self.has_authenticated_session = true;
221                self.auth_tracker.succeed();
222                self.message_queue
223                    .push_back(AxOrdersWsMessage::Authenticated);
224            }
225            HandlerCommand::PlaceOrder {
226                request_id,
227                order,
228                order_info,
229            } => {
230                log::debug!(
231                    "PlaceOrder command received: request_id={request_id}, symbol={}",
232                    order.s
233                );
234                self.pending_orders.insert(request_id, order_info.clone());
235
236                if let Err(e) = self.send_json(&order).await {
237                    log::error!("Failed to send place order message: {e}");
238                    self.message_queue
239                        .push_back(AxOrdersWsMessage::Error(AxWsError::new(format!(
240                            "Failed to send place order for {}: {e}",
241                            order_info.client_order_id
242                        ))));
243                }
244            }
245            HandlerCommand::CancelOrder {
246                request_id,
247                order_id,
248            } => {
249                log::debug!(
250                    "CancelOrder command received: request_id={request_id}, order_id={order_id}"
251                );
252                self.send_cancel_order(request_id, &order_id).await;
253            }
254            HandlerCommand::GetOpenOrders { request_id } => {
255                log::debug!("GetOpenOrders command received: request_id={request_id}");
256                self.send_get_open_orders(request_id).await;
257            }
258        }
259    }
260
261    async fn send_cancel_order(&mut self, request_id: i64, order_id: &str) {
262        let msg = AxWsCancelOrder {
263            rid: request_id,
264            t: AxOrderRequestType::CancelOrder,
265            oid: order_id.to_string(),
266        };
267
268        if let Err(e) = self.send_json(&msg).await {
269            log::error!("Failed to send cancel order message: {e}");
270            self.message_queue
271                .push_back(AxOrdersWsMessage::Error(AxWsError::new(format!(
272                    "Failed to send cancel for order {order_id}: {e}"
273                ))));
274        }
275    }
276
277    async fn send_get_open_orders(&mut self, request_id: i64) {
278        let msg = AxWsGetOpenOrders {
279            rid: request_id,
280            t: AxOrderRequestType::GetOpenOrders,
281        };
282
283        if let Err(e) = self.send_json(&msg).await {
284            log::error!("Failed to send get open orders message: {e}");
285            self.message_queue
286                .push_back(AxOrdersWsMessage::Error(AxWsError::new(format!(
287                    "Failed to send get open orders request: {e}"
288                ))));
289        }
290    }
291
292    async fn send_json<T: serde::Serialize>(&self, msg: &T) -> Result<(), String> {
293        let Some(inner) = &self.inner else {
294            return Err("No WebSocket client available".to_string());
295        };
296
297        let payload = serde_json::to_string(msg).map_err(|e| e.to_string())?;
298        log::trace!("Sending WebSocket payload ({} bytes)", payload.len());
299
300        inner
301            .send_text(payload, None)
302            .await
303            .map_err(|e| e.to_string())
304    }
305
306    fn parse_raw_message(&mut self, msg: Message) -> Option<Vec<AxOrdersWsMessage>> {
307        match msg {
308            Message::Text(text) => {
309                if text == nautilus_network::RECONNECTED {
310                    log::info!("Received WebSocket reconnected signal");
311                    self.auth_tracker.fail("Reconnecting");
312                    self.needs_session_restore = true;
313                    return Some(vec![AxOrdersWsMessage::Reconnected]);
314                }
315
316                log::trace!("Raw websocket message: {text}");
317
318                let raw_msg: AxOrdersWsFrame = match parse_order_message(&text) {
319                    Ok(v) => v,
320                    Err(e) => {
321                        log::error!("Failed to parse WebSocket message: {e}: {text}");
322                        return None;
323                    }
324                };
325
326                self.handle_raw_message(raw_msg)
327            }
328            Message::Binary(data) => {
329                log::debug!("Received binary message with {} bytes", data.len());
330                None
331            }
332            Message::Close(_) => {
333                log::debug!("Received close message, waiting for reconnection");
334                None
335            }
336            _ => None,
337        }
338    }
339
340    fn handle_raw_message(&mut self, raw_msg: AxOrdersWsFrame) -> Option<Vec<AxOrdersWsMessage>> {
341        match raw_msg {
342            AxOrdersWsFrame::Error(err) => {
343                log::warn!(
344                    "Order error response: rid={} code={:?} msg={:?}",
345                    err.rid,
346                    err.err.code,
347                    err.err.msg
348                );
349
350                Some(vec![AxOrdersWsMessage::Error(err.into())])
351            }
352            AxOrdersWsFrame::Response(resp) => self.handle_response(resp),
353            AxOrdersWsFrame::Event(event) => self.handle_event(*event),
354        }
355    }
356
357    fn handle_response(&mut self, resp: AxWsOrderResponse) -> Option<Vec<AxOrdersWsMessage>> {
358        match resp {
359            AxWsOrderResponse::PlaceOrder(msg) => {
360                log::debug!("Place order response: rid={} oid={}", msg.rid, msg.res.oid);
361                let Some(order_info) = self.pending_orders.remove(&msg.rid) else {
362                    log::warn!("Ignoring unsolicited place order response: rid={}", msg.rid);
363                    return Some(vec![AxOrdersWsMessage::PlaceOrderResponse(msg)]);
364                };
365
366                let venue_order_id = match VenueOrderId::new_checked(&msg.res.oid) {
367                    Ok(venue_order_id) => venue_order_id,
368                    Err(e) => {
369                        log::warn!(
370                            "Invalid venue order ID in place response for {}: {e}",
371                            order_info.client_order_id,
372                        );
373                        return Some(vec![AxOrdersWsMessage::PlaceOrderResponse(msg)]);
374                    }
375                };
376
377                if let Some(mut metadata) =
378                    self.orders_metadata.get_mut(&order_info.client_order_id)
379                {
380                    if metadata
381                        .venue_order_id
382                        .is_none_or(|current| current == venue_order_id)
383                    {
384                        metadata.venue_order_id = Some(venue_order_id);
385                        self.venue_to_client_order_id
386                            .insert(venue_order_id, order_info.client_order_id);
387                    }
388                } else {
389                    log::debug!(
390                        "Order tracking already cleared before place response: {}",
391                        order_info.client_order_id,
392                    );
393                }
394
395                Some(vec![AxOrdersWsMessage::PlaceOrderResponse(msg)])
396            }
397            AxWsOrderResponse::CancelOrder(msg) => {
398                log::debug!(
399                    "Cancel order response: rid={} accepted={}",
400                    msg.rid,
401                    msg.res.cxl_rx
402                );
403                Some(vec![AxOrdersWsMessage::CancelOrderResponse(msg)])
404            }
405            AxWsOrderResponse::OpenOrders(msg) => {
406                log::debug!("Open orders response: {} orders", msg.res.orders.len());
407                Some(vec![AxOrdersWsMessage::OpenOrdersResponse(msg)])
408            }
409            AxWsOrderResponse::List(msg) => {
410                let order_count = msg.res.o.as_ref().map_or(0, |o| o.len());
411                log::debug!(
412                    "List subscription response: rid={} li={} orders={}",
413                    msg.rid,
414                    msg.res.li,
415                    order_count
416                );
417                None
418            }
419        }
420    }
421
422    fn handle_event(&self, event: AxWsOrderEvent) -> Option<Vec<AxOrdersWsMessage>> {
423        if matches!(event, AxWsOrderEvent::Heartbeat) {
424            log::trace!("Received heartbeat");
425            return None;
426        }
427        Some(vec![AxOrdersWsMessage::Event(Box::new(event))])
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use std::sync::{Arc, atomic::AtomicBool};
434
435    use dashmap::DashMap;
436    use nautilus_core::time::get_atomic_clock_realtime;
437    use nautilus_model::{
438        identifiers::{AccountId, InstrumentId, StrategyId, TraderId},
439        types::{Currency, Money},
440    };
441    use nautilus_network::websocket::AuthTracker;
442    use rstest::rstest;
443    use rust_decimal::Decimal;
444    use ustr::Ustr;
445
446    use super::*;
447    use crate::{
448        common::enums::{AxOrderSide, AxTimeInForce},
449        execution::create_order_filled,
450        websocket::{
451            messages::{
452                AxWsOrderError, AxWsOrderErrorResponse, AxWsOrderFilled, AxWsPlaceOrderResponse,
453                AxWsPlaceOrderResult,
454            },
455            orders::OrdersCaches,
456        },
457    };
458
459    fn test_handler() -> AxOrdersWsFeedHandler {
460        let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
461        let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
462        AxOrdersWsFeedHandler::new(
463            Arc::new(AtomicBool::new(false)),
464            cmd_rx,
465            raw_rx,
466            AuthTracker::default(),
467            Arc::new(DashMap::new()),
468            Arc::new(DashMap::new()),
469        )
470    }
471
472    #[rstest]
473    fn test_place_order_response_records_venue_identity() {
474        let mut handler = test_handler();
475        let request_id = 11;
476        let cid = 1011;
477        let client_order_id = ClientOrderId::from("CID-11");
478        let venue_order_id = VenueOrderId::from("OID-11");
479        handler
480            .orders_metadata
481            .insert(client_order_id, test_order_metadata(client_order_id));
482        handler.pending_orders.insert(
483            request_id,
484            WsOrderInfo {
485                client_order_id,
486                symbol: Ustr::from("EURUSD-PERP"),
487                cid,
488            },
489        );
490
491        let response = AxWsOrderResponse::PlaceOrder(AxWsPlaceOrderResponse {
492            rid: request_id,
493            res: AxWsPlaceOrderResult {
494                oid: "OID-11".to_string(),
495            },
496        });
497
498        let messages = handler.handle_response(response).unwrap();
499
500        assert_eq!(messages.len(), 1);
501        assert!(handler.pending_orders.get(&request_id).is_none());
502        assert_eq!(
503            handler
504                .orders_metadata
505                .get(&client_order_id)
506                .and_then(|metadata| metadata.venue_order_id),
507            Some(venue_order_id),
508        );
509        assert_eq!(
510            handler
511                .venue_to_client_order_id
512                .get(&venue_order_id)
513                .map(|client_order_id| *client_order_id),
514            Some(client_order_id),
515        );
516    }
517
518    #[rstest]
519    fn test_late_place_order_response_does_not_restore_cleared_tracking() {
520        let mut handler = test_handler();
521        let request_id = 12;
522        let client_order_id = ClientOrderId::from("CID-12");
523        handler.pending_orders.insert(
524            request_id,
525            WsOrderInfo {
526                client_order_id,
527                symbol: Ustr::from("EURUSD-PERP"),
528                cid: 1012,
529            },
530        );
531
532        let response = AxWsOrderResponse::PlaceOrder(AxWsPlaceOrderResponse {
533            rid: request_id,
534            res: AxWsPlaceOrderResult {
535                oid: "OID-12".to_string(),
536            },
537        });
538
539        let messages = handler.handle_response(response).unwrap();
540
541        assert_eq!(messages.len(), 1);
542        assert!(handler.pending_orders.get(&request_id).is_none());
543        assert!(!handler.orders_metadata.contains_key(&client_order_id));
544        assert!(handler.venue_to_client_order_id.is_empty());
545    }
546
547    #[rstest]
548    fn test_place_order_error_preserves_cid_for_reconciliation() {
549        let mut handler = test_handler();
550        let request_id = 13;
551        let cid = 1013;
552        let client_order_id = ClientOrderId::from("CID-13");
553        handler
554            .orders_metadata
555            .insert(client_order_id, test_order_metadata(client_order_id));
556        handler.pending_orders.insert(
557            request_id,
558            WsOrderInfo {
559                client_order_id,
560                symbol: Ustr::from("EURUSD-PERP"),
561                cid,
562            },
563        );
564
565        let messages = handler
566            .handle_raw_message(AxOrdersWsFrame::Error(AxWsOrderErrorResponse {
567                rid: request_id,
568                err: AxWsOrderError {
569                    code: Some(400),
570                    msg: Some("invalid order".to_string()),
571                },
572            }))
573            .unwrap();
574
575        assert_eq!(messages.len(), 1);
576        assert_eq!(
577            handler
578                .pending_orders
579                .get(&request_id)
580                .unwrap()
581                .client_order_id,
582            client_order_id,
583        );
584        assert!(handler.orders_metadata.contains_key(&client_order_id));
585        handler.handle_response(AxWsOrderResponse::PlaceOrder(AxWsPlaceOrderResponse {
586            rid: request_id,
587            res: AxWsPlaceOrderResult {
588                oid: "OID-13".to_string(),
589            },
590        }));
591
592        assert_eq!(
593            handler
594                .orders_metadata
595                .get(&client_order_id)
596                .unwrap()
597                .venue_order_id,
598            Some(VenueOrderId::from("OID-13"))
599        );
600        assert_eq!(
601            *handler
602                .venue_to_client_order_id
603                .get(&VenueOrderId::from("OID-13"))
604                .unwrap(),
605            client_order_id
606        );
607    }
608
609    #[rstest]
610    #[case::send_failure(true)]
611    #[case::request_error(false)]
612    #[tokio::test]
613    async fn test_uncertain_place_error_then_fill(#[case] send_failure: bool) {
614        let mut handler = test_handler();
615        let client_order_id = ClientOrderId::from("CID-UNCERTAIN");
616        let request_id = 17;
617        let cid = 1017;
618        handler
619            .orders_metadata
620            .insert(client_order_id, test_order_metadata(client_order_id));
621
622        let order_info = WsOrderInfo {
623            client_order_id,
624            symbol: Ustr::from("EURUSD-PERP"),
625            cid,
626        };
627
628        if send_failure {
629            handler
630                .handle_command(HandlerCommand::PlaceOrder {
631                    request_id,
632                    order: AxWsPlaceOrder {
633                        rid: request_id,
634                        t: AxOrderRequestType::PlaceOrder,
635                        s: order_info.symbol,
636                        d: AxOrderSide::Buy,
637                        q: 100,
638                        p: Decimal::from(50000),
639                        tif: AxTimeInForce::Gtc,
640                        po: false,
641                        tag: None,
642                        cid: Some(cid),
643                    },
644                    order_info,
645                })
646                .await;
647
648            assert!(matches!(
649                handler.message_queue.pop_front(),
650                Some(AxOrdersWsMessage::Error(_))
651            ));
652        } else {
653            handler.pending_orders.insert(request_id, order_info);
654            handler.handle_raw_message(AxOrdersWsFrame::Error(AxWsOrderErrorResponse {
655                rid: request_id,
656                err: AxWsOrderError {
657                    code: Some(500),
658                    msg: Some("unresolved request".to_string()),
659                },
660            }));
661        }
662
663        let caches = OrdersCaches {
664            orders_metadata: handler.orders_metadata.clone(),
665            venue_to_client_id: handler.venue_to_client_order_id.clone(),
666            cid_to_client_order_id: Arc::new(DashMap::new()),
667            ..OrdersCaches::default()
668        };
669
670        caches.cid_to_client_order_id.insert(cid, client_order_id);
671        let mut fill: AxWsOrderFilled =
672            serde_json::from_str(include_str!("../../../test_data/ws_order_filled.json")).unwrap();
673        fill.o.cid = Some(cid);
674        let event = create_order_filled(
675            &fill.o,
676            &fill.xs,
677            fill.ts,
678            fill.tn,
679            &caches,
680            AccountId::from("AX-001"),
681            Money::from("1.23 USD"),
682            get_atomic_clock_realtime(),
683        )
684        .expect("fill after uncertain error must remain tracked");
685
686        assert_eq!(event.client_order_id, client_order_id);
687        assert_eq!(event.trader_id, TraderId::from("TRADER-001"));
688        assert_eq!(event.strategy_id, StrategyId::from("S-001"));
689        assert_eq!(event.instrument_id, InstrumentId::from("EURUSD-PERP.AX"));
690        assert_eq!(event.last_qty.as_decimal(), Decimal::from(100));
691        assert_eq!(event.last_px.as_decimal(), Decimal::from(50000));
692        assert_eq!(event.commission, Some(Money::from("1.23 USD")));
693        assert_eq!(
694            handler
695                .pending_orders
696                .get(&request_id)
697                .unwrap()
698                .client_order_id,
699            client_order_id
700        );
701        assert_eq!(
702            *caches.cid_to_client_order_id.get(&cid).unwrap(),
703            client_order_id
704        );
705    }
706
707    #[rstest]
708    fn test_late_place_response_preserves_replacement_binding() {
709        let mut handler = test_handler();
710        let client_order_id = ClientOrderId::from("CID-REPLACED");
711        let replacement_id = VenueOrderId::from("OID-CHILD");
712        let mut metadata = test_order_metadata(client_order_id);
713        metadata.venue_order_id = Some(replacement_id);
714        handler.orders_metadata.insert(client_order_id, metadata);
715        handler
716            .venue_to_client_order_id
717            .insert(replacement_id, client_order_id);
718        handler.pending_orders.insert(
719            18,
720            WsOrderInfo {
721                client_order_id,
722                symbol: Ustr::from("EURUSD-PERP"),
723                cid: 1018,
724            },
725        );
726
727        handler.handle_response(AxWsOrderResponse::PlaceOrder(AxWsPlaceOrderResponse {
728            rid: 18,
729            res: AxWsPlaceOrderResult {
730                oid: "OID-PARENT".to_string(),
731            },
732        }));
733
734        assert_eq!(
735            handler
736                .orders_metadata
737                .get(&client_order_id)
738                .unwrap()
739                .venue_order_id,
740            Some(replacement_id)
741        );
742        assert_eq!(
743            *handler
744                .venue_to_client_order_id
745                .get(&replacement_id)
746                .unwrap(),
747            client_order_id
748        );
749        assert_eq!(handler.venue_to_client_order_id.len(), 1);
750    }
751
752    #[rstest]
753    fn test_handle_event_forwards_venue_event() {
754        let handler = test_handler();
755
756        let event = AxWsOrderEvent::Heartbeat;
757        let result = handler.handle_event(event);
758        assert!(result.is_none());
759    }
760
761    #[tokio::test]
762    async fn test_authenticated_session_is_restored_after_reconnect() {
763        let mut handler = test_handler();
764
765        handler
766            .handle_command(HandlerCommand::SessionAuthenticated)
767            .await;
768        let initial = handler.message_queue.pop_front();
769        handler.auth_tracker.fail("Reconnecting");
770        handler.restore_authenticated_session();
771        let restored = handler.message_queue.pop_front();
772
773        assert!(handler.has_authenticated_session);
774        assert!(matches!(initial, Some(AxOrdersWsMessage::Authenticated)));
775        assert!(matches!(restored, Some(AxOrdersWsMessage::Authenticated)));
776        assert!(handler.auth_tracker.is_authenticated());
777    }
778
779    fn test_order_metadata(client_order_id: ClientOrderId) -> OrderMetadata {
780        OrderMetadata {
781            trader_id: TraderId::from("TRADER-001"),
782            strategy_id: StrategyId::from("S-001"),
783            instrument_id: InstrumentId::from("EURUSD-PERP.AX"),
784            client_order_id,
785            order_init: None,
786            is_accepted: false,
787            venue_order_id: None,
788            ts_init: 0.into(),
789            size_precision: 0,
790            price_precision: 2,
791            quote_currency: Currency::USD(),
792        }
793    }
794}