Disconnect counterparty if they respond with non-logon message to logon

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -198,6 +198,8 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
     }
 
     async fn process_message(&mut self, message: Message) -> Result<()> {
+        let message_type = message.header().get(fix44::MSG_TYPE)?;
+
         if let SessionState::AwaitingResend(state) = &mut self.state {
             // TODO: consider what messages won't have a sequence number?
             // e.g. SequenceReset?
@@ -210,9 +212,17 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
                 return Ok(());
             }
         }
+
+        if let SessionState::AwaitingLogon { .. } = &mut self.state {
+            // TODO: should this (and all inbound message processing) logic be pushed into the state?
+            if message_type != "A" {
+                self.state.disconnect().await;
+                return Ok(());
+            }
+        }
+
         // TODO: should we verify messages here?
 
-        let message_type = message.header().get(fix44::MSG_TYPE)?;
         match message_type {
             "0" => {
                 self.on_heartbeat(&message).await;

crates/hotfix/tests/common/test_messages.rs

@@ -1,17 +1,94 @@
 use hotfix::Message as HotfixMessage;
 use hotfix::message::FixMessage;
+use hotfix_message::{Part, fix44};
 
+/// Business messages used for testing.
 #[derive(Debug, Clone)]
-pub struct TestMessage;
+pub enum TestMessage {
+    /// A minimal implementation of a valid execution report.
+    MinimalExecutionReport {
+        order_id: String,
+        exec_id: String,
+        exec_type: fix44::ExecType,
+        ord_status: fix44::OrdStatus,
+        side: fix44::Side,
+        symbol: String,
+        order_qty: f64,
+        price: f64,
+    },
+}
+
+impl TestMessage {
+    pub fn dummy_execution_report() -> Self {
+        Self::MinimalExecutionReport {
+            order_id: "123456789".to_string(),
+            exec_id: "123456789".to_string(),
+            exec_type: fix44::ExecType::New,
+            ord_status: fix44::OrdStatus::New,
+            side: fix44::Side::Buy,
+            symbol: "ABC".to_string(),
+            order_qty: 100.0,
+            price: 100.0,
+        }
+    }
+}
 
 impl FixMessage for TestMessage {
-    fn write(&self, _msg: &mut HotfixMessage) {}
+    fn write(&self, msg: &mut HotfixMessage) {
+        match self {
+            TestMessage::MinimalExecutionReport {
+                order_id,
+                exec_id,
+                exec_type,
+                ord_status,
+                side,
+                symbol,
+                order_qty,
+                price,
+            } => {
+                msg.set(fix44::ORDER_ID, order_id.as_str());
+                msg.set(fix44::EXEC_ID, exec_id.as_str());
+                msg.set(fix44::EXEC_TYPE, *exec_type);
+                msg.set(fix44::ORD_STATUS, *ord_status);
+                msg.set(fix44::SIDE, *side);
+                msg.set(fix44::SYMBOL, symbol.as_str());
+                msg.set(fix44::ORDER_QTY, *order_qty);
+                msg.set(fix44::PRICE, *price);
+            }
+        }
+    }
 
     fn message_type(&self) -> &str {
-        unimplemented!()
+        match self {
+            TestMessage::MinimalExecutionReport { .. } => "8",
+        }
     }
 
-    fn parse(_msg: &HotfixMessage) -> Self {
-        TestMessage
+    fn parse(msg: &HotfixMessage) -> Self {
+        let msg_type: &str = msg.get(fix44::MSG_TYPE).unwrap();
+        if msg_type != "8" {
+            // not an execution report
+            panic!("Invalid message type: {}", msg_type);
+        }
+
+        let order_id: &str = msg.get(fix44::ORDER_ID).unwrap();
+        let exec_id: &str = msg.get(fix44::EXEC_ID).unwrap();
+        let exec_type = msg.get(fix44::EXEC_TYPE).unwrap();
+        let ord_status = msg.get(fix44::ORD_STATUS).unwrap();
+        let side = msg.get(fix44::SIDE).unwrap();
+        let symbol: &str = msg.get(fix44::SYMBOL).unwrap();
+        let order_qty = msg.get(fix44::ORDER_QTY).unwrap();
+        let price = msg.get(fix44::PRICE).unwrap();
+
+        Self::MinimalExecutionReport {
+            order_id: order_id.to_string(),
+            exec_id: exec_id.to_string(),
+            exec_type,
+            ord_status,
+            side,
+            symbol: symbol.to_string(),
+            order_qty,
+            price,
+        }
     }
 }

crates/hotfix/tests/session_test_cases/logon_tests.rs

@@ -1,6 +1,7 @@
 use crate::common::session_actions::SessionActions;
 use crate::common::session_assertions::SessionAssertions;
 use crate::common::setup::given_a_connected_session;
+use crate::common::test_messages::TestMessage;
 use hotfix::session::Status;
 use hotfix_message::Part;
 use hotfix_message::fix44::MSG_TYPE;
@@ -25,3 +26,21 @@ async fn test_happy_logon() {
     session.when_disconnect_is_requested().await;
     mock_counterparty.then_gets_disconnected().await;
 }
+
+#[tokio::test]
+async fn test_non_logon_response_to_logon() {
+    let (session, mut mock_counterparty) = given_a_connected_session().await;
+
+    // assert that a logon message is received (type 'A')
+    mock_counterparty
+        .then_receives(|msg| assert_eq!(msg.header().get::<&str>(MSG_TYPE).unwrap(), "A"))
+        .await;
+    session.then_status_changes_to(Status::AwaitingLogon).await;
+
+    // counterparty sends an execution report without ever responding to our logon
+    let dummy_report = TestMessage::dummy_execution_report();
+    mock_counterparty.when_message_is_sent(dummy_report).await;
+
+    // we disconnect them as a result
+    mock_counterparty.then_gets_disconnected().await;
+}