Make application take unparsed messages

Author: davidsteiner

crates/hotfix-message/src/message.rs

@@ -10,6 +10,7 @@ use crate::parts::{Body, Header, Part, RepeatingGroup, Trailer};
 use crate::session_fields::{BEGIN_STRING, BODY_LENGTH, CHECK_SUM, MSG_TYPE};
 use hotfix_dictionary::{FieldLocation, IsFieldDefinition};
 
+#[derive(Clone)]
 pub struct Message {
     pub(crate) header: Header,
     pub(crate) body: Body,

crates/hotfix-message/src/parts/body.rs

@@ -1,7 +1,7 @@
 use crate::field_map::FieldMap;
 use crate::parts::Part;
 
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct Body {
     pub(crate) fields: FieldMap,
 }

crates/hotfix-message/src/parts/header.rs

@@ -4,7 +4,7 @@ use crate::field_map::FieldMap;
 use crate::parts::Part;
 use crate::session_fields;
 
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct Header {
     pub fields: FieldMap,
 }

crates/hotfix-message/src/parts/trailer.rs

@@ -3,7 +3,7 @@ use crate::parts::Part;
 use crate::session_fields;
 use hotfix_dictionary::IsFieldDefinition;
 
-#[derive(Default)]
+#[derive(Clone, Default)]
 pub struct Trailer {
     pub(crate) fields: FieldMap,
 }

crates/hotfix/src/application.rs

@@ -1,14 +1,16 @@
+use hotfix_message::message::Message;
+
 #[async_trait::async_trait]
 /// The application users of HotFIX can implement to hook into the engine.
-pub trait Application<Inbound, Outbound>: Send + Sync + 'static {
+pub trait Application<Outbound>: Send + Sync + 'static {
     /// Called when a message is sent to the engine to be sent to the counterparty.
     ///
     /// This is invoked before the raw message is persisted in the message store.
     async fn on_outbound_message(&self, msg: &Outbound) -> OutboundDecision;
     /// Called when a message is received from the counterparty.
     ///
-    /// This is invoked after the message is verified and parsed into a typed message.
-    async fn on_inbound_message(&self, msg: Inbound) -> InboundDecision;
+    /// This is invoked after the message is verified by the session layer.
+    async fn on_inbound_message(&self, msg: &Message) -> InboundDecision;
     /// Called when the session is logged out.
     async fn on_logout(&mut self, reason: &str);
     /// Called when the session is logged on.

crates/hotfix/src/initiator.rs

@@ -13,7 +13,7 @@ use tracing::{debug, warn};
 
 use crate::application::Application;
 use crate::config::SessionConfig;
-use crate::message::{InboundMessage, OutboundMessage};
+use crate::message::OutboundMessage;
 use crate::session::error::{SendError, SendOutcome, SessionCreationError};
 use crate::session::{InternalSessionRef, SessionHandle};
 use crate::store::MessageStore;
@@ -27,9 +27,9 @@ pub struct Initiator<Outbound> {
 }
 
 impl<Outbound: OutboundMessage> Initiator<Outbound> {
-    pub async fn start<Inbound: InboundMessage>(
+    pub async fn start(
         config: SessionConfig,
-        application: impl Application<Inbound, Outbound>,
+        application: impl Application<Outbound>,
         store: impl MessageStore + 'static,
     ) -> Result<Self, SessionCreationError> {
         let session_ref = InternalSessionRef::new(config.clone(), application, store)?;
@@ -157,10 +157,10 @@ async fn establish_connection<Outbound: OutboundMessage>(
 mod tests {
     use super::*;
     use crate::application::{Application, InboundDecision, OutboundDecision};
+    use crate::message::generate_message;
     use crate::message::logon::{Logon, ResetSeqNumConfig};
     use crate::message::logout::Logout;
     use crate::message::parser::Parser;
-    use crate::message::{InboundMessage, generate_message};
     use crate::store::in_memory::InMemoryMessageStore;
     use hotfix_message::Part;
     use hotfix_message::message::Message;
@@ -180,21 +180,15 @@ mod tests {
         }
     }
 
-    impl InboundMessage for DummyMessage {
-        fn parse(_message: &Message) -> Self {
-            DummyMessage
-        }
-    }
-
     // No-op application
     struct NoOpApp;
 
     #[async_trait::async_trait]
-    impl Application<DummyMessage, DummyMessage> for NoOpApp {
+    impl Application<DummyMessage> for NoOpApp {
         async fn on_outbound_message(&self, _msg: &DummyMessage) -> OutboundDecision {
             OutboundDecision::Send
         }
-        async fn on_inbound_message(&self, _msg: DummyMessage) -> InboundDecision {
+        async fn on_inbound_message(&self, _msg: &Message) -> InboundDecision {
             InboundDecision::Accept
         }
         async fn on_logout(&mut self, _reason: &str) {}
@@ -366,13 +360,10 @@ mod tests {
         let mut config = create_test_config("127.0.0.1", port);
         config.reconnect_interval = 1; // Short interval for test
 
-        let _initiator = Initiator::<DummyMessage>::start::<DummyMessage>(
-            config,
-            NoOpApp,
-            InMemoryMessageStore::default(),
-        )
-        .await
-        .unwrap();
+        let _initiator =
+            Initiator::<DummyMessage>::start(config, NoOpApp, InMemoryMessageStore::default())
+                .await
+                .unwrap();
 
         // Accept first connection
         let (conn1, _) = tokio::time::timeout(Duration::from_secs(2), listener.accept())

crates/hotfix/src/message.rs

@@ -25,10 +25,6 @@ pub trait OutboundMessage: Clone + Send + 'static {
     fn message_type(&self) -> &str;
 }
 
-pub trait InboundMessage: Clone + Send + 'static {
-    fn parse(message: &Message) -> Self;
-}
-
 pub fn generate_message(
     begin_string: &str,
     sender_comp_id: &str,

crates/hotfix/src/message/heartbeat.rs

@@ -1,4 +1,4 @@
-use crate::message::{InboundMessage, OutboundMessage};
+use crate::message::OutboundMessage;
 use hotfix_message::Part;
 use hotfix_message::message::Message;
 use hotfix_message::session_fields::TEST_REQ_ID;
@@ -27,10 +27,3 @@ impl OutboundMessage for Heartbeat {
         "0"
     }
 }
-
-impl InboundMessage for Heartbeat {
-    fn parse(_message: &Message) -> Self {
-        // TODO: this needs to be implemented properly when we're implementing Test Requests
-        Heartbeat { test_req_id: None }
-    }
-}

crates/hotfix/src/message/reject.rs

@@ -1,4 +1,4 @@
-use crate::message::{InboundMessage, OutboundMessage};
+use crate::message::OutboundMessage;
 use hotfix_message::Part;
 use hotfix_message::message::Message;
 use hotfix_message::session_fields::{
@@ -50,6 +50,20 @@ impl Reject {
         self.text = Some(text.to_string());
         self
     }
+
+    #[cfg(test)]
+    fn parse(message: &Message) -> Self {
+        Self {
+            #[allow(clippy::expect_used)]
+            ref_seq_num: message
+                .get(REF_SEQ_NUM)
+                .expect("ref_seq_num should be present"),
+            ref_tag_id: message.get(REF_TAG_ID).ok(),
+            ref_msg_type: message.get(REF_MSG_TYPE).ok(),
+            session_reject_reason: message.get(SESSION_REJECT_REASON).ok(),
+            text: message.get::<&str>(TEXT).ok().map(|s| s.to_string()),
+        }
+    }
 }
 
 impl OutboundMessage for Reject {
@@ -75,22 +89,6 @@ impl OutboundMessage for Reject {
     }
 }
 
-impl InboundMessage for Reject {
-    fn parse(message: &Message) -> Self {
-        Self {
-            // TODO: how do we handle errors in parsing messages?
-            #[allow(clippy::expect_used)]
-            ref_seq_num: message
-                .get(REF_SEQ_NUM)
-                .expect("ref_seq_num should be present"),
-            ref_tag_id: message.get(REF_TAG_ID).ok(),
-            ref_msg_type: message.get(REF_MSG_TYPE).ok(),
-            session_reject_reason: message.get(SESSION_REJECT_REASON).ok(),
-            text: message.get::<&str>(TEXT).ok().map(|s| s.to_string()),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/hotfix/src/session.rs

@@ -20,6 +20,7 @@ use crate::Application;
 use crate::application::{InboundDecision, OutboundDecision};
 use crate::config::SessionConfig;
 use crate::message::OutboundMessage;
+use crate::message::generate_message;
 use crate::message::heartbeat::Heartbeat;
 use crate::message::logon::{Logon, ResetSeqNumConfig};
 use crate::message::logout::Logout;
@@ -30,7 +31,6 @@ use crate::message::sequence_reset::SequenceReset;
 use crate::message::test_request::TestRequest;
 use crate::message::verification::verify_message;
 use crate::message::verification_error::{CompIdType, MessageVerificationError};
-use crate::message::{InboundMessage, generate_message};
 use crate::message_utils::{is_admin, prepare_message_for_resend};
 use crate::session::admin_request::AdminRequest;
 use crate::session::error::SessionCreationError;
@@ -57,7 +57,7 @@ use hotfix_message::session_fields::{
 
 const SCHEDULE_CHECK_INTERVAL: u64 = 1;
 
-struct Session<A, I, O, S> {
+struct Session<A, O, S> {
     message_config: MessageConfig,
     config: SessionConfig,
     schedule: SessionSchedule,
@@ -67,21 +67,20 @@ struct Session<A, I, O, S> {
     store: S,
     schedule_check_timer: Pin<Box<Sleep>>,
     reset_on_next_logon: bool,
-    _phantom: std::marker::PhantomData<fn() -> (I, O)>,
+    _phantom: std::marker::PhantomData<fn() -> O>,
 }
 
-impl<App, Inbound, Outbound, Store> Session<App, Inbound, Outbound, Store>
+impl<App, Outbound, Store> Session<App, Outbound, Store>
 where
-    App: Application<Inbound, Outbound>,
-    Inbound: InboundMessage,
+    App: Application<Outbound>,
     Outbound: OutboundMessage,
     Store: MessageStore,
 {
     fn new(
         config: SessionConfig,
         application: App,
         store: Store,
-    ) -> Result<Session<App, Inbound, Outbound, Store>, SessionCreationError> {
+    ) -> Result<Session<App, Outbound, Store>, SessionCreationError> {
         let schedule_check_timer = sleep(Duration::from_secs(SCHEDULE_CHECK_INTERVAL));
 
         let dictionary = Self::get_data_dictionary(&config)?;
@@ -245,9 +244,8 @@ where
     ) -> Result<(), SessionOperationError> {
         match self.verify_message(message, true) {
             Ok(_) => {
-                let parsed_message = Inbound::parse(message);
                 if matches!(
-                    self.application.on_inbound_message(parsed_message).await,
+                    self.application.on_inbound_message(message).await,
                     InboundDecision::TerminateSession
                 ) {
                     error!("failed to send inbound message to application");
@@ -1131,14 +1129,13 @@ fn get_msg_seq_num(message: &Message) -> u64 {
         .expect("MsgSeqNum missing from validated message - parser bug")
 }
 
-async fn run_session<App, Inbound, Outbound, Store>(
-    mut session: Session<App, Inbound, Outbound, Store>,
+async fn run_session<App, Outbound, Store>(
+    mut session: Session<App, Outbound, Store>,
     mut event_receiver: mpsc::Receiver<SessionEvent>,
     mut outbound_message_receiver: mpsc::Receiver<OutboundRequest<Outbound>>,
     mut admin_request_receiver: mpsc::Receiver<AdminRequest>,
 ) where
-    App: Application<Inbound, Outbound>,
-    Inbound: InboundMessage,
+    App: Application<Outbound>,
     Outbound: OutboundMessage,
     Store: MessageStore + Send + 'static,
 {
@@ -1196,7 +1193,7 @@ async fn run_session<App, Inbound, Outbound, Store>(
 mod tests {
     use super::*;
     use crate::application::{InboundDecision, OutboundDecision};
-    use crate::message::{InboundMessage, OutboundMessage};
+    use crate::message::OutboundMessage;
     use crate::store::{Result as StoreResult, StoreError};
     use chrono::{DateTime, Datelike, NaiveDate, NaiveTime, TimeDelta, Timelike};
     use chrono_tz::Tz;
@@ -1293,21 +1290,15 @@ mod tests {
         }
     }
 
-    impl InboundMessage for DummyMessage {
-        fn parse(_message: &Message) -> Self {
-            DummyMessage
-        }
-    }
-
     /// Minimal no-op application for testing
     struct NoOpApp;
 
     #[async_trait::async_trait]
-    impl Application<DummyMessage, DummyMessage> for NoOpApp {
+    impl Application<DummyMessage> for NoOpApp {
         async fn on_outbound_message(&self, _: &DummyMessage) -> OutboundDecision {
             OutboundDecision::Send
         }
-        async fn on_inbound_message(&self, _: DummyMessage) -> InboundDecision {
+        async fn on_inbound_message(&self, _: &Message) -> InboundDecision {
             InboundDecision::Accept
         }
         async fn on_logout(&mut self, _: &str) {}
@@ -1341,7 +1332,7 @@ mod tests {
         schedule: SessionSchedule,
         state: SessionState,
         store: TestStore,
-    ) -> Session<NoOpApp, DummyMessage, DummyMessage, TestStore> {
+    ) -> Session<NoOpApp, DummyMessage, TestStore> {
         let config = create_test_config();
         let message_config = MessageConfig::default();
         let dictionary = Dictionary::fix44();