Flesh out structure for session flow tests

Author: davidsteiner

crates/hotfix/src/lib.rs

@@ -29,10 +29,10 @@ pub(crate) mod error;
 pub mod initiator;
 pub mod message;
 mod message_utils;
-pub(crate) mod session;
+pub mod session;
 mod session_schedule;
 pub mod store;
-pub(crate) mod transport;
+pub mod transport;
 
 pub use application::Application;
 pub use hotfix_message::field_types;

crates/hotfix/src/message.rs

@@ -13,6 +13,8 @@ pub(crate) mod resend_request;
 pub(crate) mod sequence_reset;
 pub(crate) mod test_request;
 
+pub use parser::RawFixMessage;
+
 pub trait FixMessage: Clone + Send + 'static {
     fn write(&self, msg: &mut Message);
 

crates/hotfix/src/transport/connection.rs

@@ -2,22 +2,19 @@ use crate::transport::reader::ReaderRef;
 use crate::transport::writer::WriterRef;
 
 pub struct FixConnection {
-    _writer: WriterRef,
-    _reader: ReaderRef,
+    writer: WriterRef,
+    reader: ReaderRef,
 }
 
 impl FixConnection {
     pub fn new(writer: WriterRef, reader: ReaderRef) -> Self {
-        Self {
-            _writer: writer,
-            _reader: reader,
-        }
+        Self { writer, reader }
     }
     pub fn get_writer(&self) -> WriterRef {
-        self._writer.clone()
+        self.writer.clone()
     }
 
     pub async fn run_until_disconnect(self) {
-        self._reader.wait_for_disconnect().await
+        self.reader.wait_for_disconnect().await
     }
 }

crates/hotfix/tests/common/mock_application.rs

@@ -0,0 +1,13 @@
+use crate::common::test_messages::TestMessage;
+use hotfix::Application;
+
+pub struct MockApplication {}
+
+#[async_trait::async_trait]
+impl Application<TestMessage> for MockApplication {
+    async fn on_message_from_app(&self, _msg: TestMessage) {}
+
+    async fn on_message_to_app(&self, _msg: TestMessage) {}
+
+    async fn on_logout(&mut self, _reason: &str) {}
+}

crates/hotfix/tests/common/mock_counterparty.rs

@@ -0,0 +1,89 @@
+use hotfix::message::{FixMessage, RawFixMessage};
+use hotfix::session::SessionRef;
+use hotfix::transport::FixConnection;
+use hotfix::transport::reader::ReaderRef;
+use hotfix::transport::writer::{WriterMessage, WriterRef};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::sync::mpsc::Receiver;
+use tokio::sync::{Mutex, mpsc, oneshot};
+
+pub struct MockCounterparty {
+    received_messages: Arc<Mutex<Vec<RawFixMessage>>>,
+    _connection: FixConnection,
+    _dc_sender: oneshot::Sender<()>,
+}
+
+type MessageStore = Arc<Mutex<Vec<RawFixMessage>>>;
+
+impl MockCounterparty {
+    pub async fn start(session_ref: SessionRef<impl FixMessage>) -> Self {
+        let (writer_ref, received_messages) = Self::spawn_writer();
+        let (reader_ref, dc_sender) = Self::create_reader();
+        let connection = FixConnection::new(writer_ref, reader_ref);
+
+        session_ref.register_writer(connection.get_writer()).await;
+
+        Self {
+            received_messages,
+            _connection: connection,
+            _dc_sender: dc_sender,
+        }
+    }
+
+    pub async fn assert_message_count(&self, expected_count: usize, timeout_secs: f32) {
+        let timeout_duration = Duration::from_secs_f32(timeout_secs);
+        let start_time = Instant::now();
+
+        loop {
+            {
+                let messages = self.received_messages.lock().await;
+                if messages.len() >= expected_count {
+                    return;
+                }
+            }
+
+            if start_time.elapsed() >= timeout_duration {
+                let current_count = self.received_messages.lock().await.len();
+                panic!(
+                    "Expected {expected_count} messages, but only received {current_count} within {timeout_secs} seconds"
+                );
+            }
+
+            tokio::time::sleep(Duration::from_millis(1)).await;
+        }
+    }
+
+    fn spawn_writer() -> (WriterRef, MessageStore) {
+        // mock implementation to receive messages from the session
+        // and hold on to them for test assertions
+        let received_messages: MessageStore = Arc::new(Mutex::new(vec![]));
+        let (sender, mailbox) = mpsc::channel(10);
+        tokio::spawn(Self::run_writer(mailbox, received_messages.clone()));
+
+        (WriterRef::new(sender), received_messages)
+    }
+
+    fn create_reader() -> (ReaderRef, oneshot::Sender<()>) {
+        let (dc_sender, dc_receiver) = oneshot::channel();
+        (ReaderRef::new(dc_receiver), dc_sender)
+    }
+
+    async fn run_writer(
+        mut mailbox: Receiver<WriterMessage>,
+        received_messages: Arc<Mutex<Vec<RawFixMessage>>>,
+    ) {
+        while let Some(msg) = mailbox.recv().await {
+            match msg {
+                WriterMessage::SendMessage(fix_message) => {
+                    println!("Received message from session: {fix_message:?}");
+                    received_messages.lock().await.push(fix_message);
+                }
+                WriterMessage::Disconnect => {
+                    println!("Disconnecting");
+                    break;
+                }
+            }
+        }
+    }
+}

crates/hotfix/tests/common/mod.rs

@@ -0,0 +1,3 @@
+pub mod mock_application;
+pub mod mock_counterparty;
+pub mod test_messages;

crates/hotfix/tests/common/test_messages.rs

@@ -0,0 +1,17 @@
+use hotfix::Message as HotfixMessage;
+use hotfix::message::FixMessage;
+
+#[derive(Debug, Clone)]
+pub struct TestMessage;
+
+impl FixMessage for TestMessage {
+    fn write(&self, _msg: &mut HotfixMessage) {}
+
+    fn message_type(&self) -> &str {
+        unimplemented!()
+    }
+
+    fn parse(_msg: &HotfixMessage) -> Self {
+        TestMessage
+    }
+}

crates/hotfix/tests/session_tests.rs

@@ -0,0 +1,40 @@
+use crate::common::mock_application::MockApplication;
+use crate::common::mock_counterparty::MockCounterparty;
+use crate::common::test_messages::TestMessage;
+use hotfix::application::ApplicationRef;
+use hotfix::config::SessionConfig;
+use hotfix::session::SessionRef;
+use hotfix::store::in_memory::InMemoryMessageStore;
+
+mod common;
+
+#[tokio::test]
+async fn test_happy_login_flow() {
+    let session = create_session();
+    let mock_counterparty = MockCounterparty::start(session).await;
+    mock_counterparty.assert_message_count(1, 0.5).await;
+}
+
+fn create_session() -> SessionRef<TestMessage> {
+    let config = create_session_config();
+    let application_ref = ApplicationRef::new(MockApplication {});
+    let message_store = InMemoryMessageStore::default();
+
+    SessionRef::new(config, application_ref, message_store)
+}
+
+fn create_session_config() -> SessionConfig {
+    SessionConfig {
+        begin_string: "FIX.4.4".to_string(),
+        sender_comp_id: "dummy-initiator".to_string(),
+        target_comp_id: "dummy-acceptor".to_string(),
+        data_dictionary_path: None,
+        connection_host: "".to_string(),
+        connection_port: 0,
+        tls_config: None,
+        heartbeat_interval: 30,
+        reconnect_interval: 30,
+        reset_on_logon: false,
+        schedule: None,
+    }
+}