Move handle_incorrect_begin_string to inbound module

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -34,7 +34,7 @@ use crate::message::sequence_reset::SequenceReset;
 use crate::message::test_request::TestRequest;
 use crate::message::verification_error::{CompIdType, MessageVerificationError};
 use crate::session::admin_request::AdminRequest;
-use crate::session::ctx::SessionCtx;
+use crate::session::ctx::{SessionCtx, TransitionResult};
 use crate::session::error::SessionCreationError;
 use crate::session::error::{InternalSendError, InternalSendResultExt, SessionOperationError};
 pub use crate::session::error::{SendError, SendOutcome};
@@ -604,7 +604,12 @@ where
                     .await?;
             }
             MessageVerificationError::IncorrectBeginString(begin_string) => {
-                self.handle_incorrect_begin_string(begin_string).await;
+                if let Some(writer) = self.state.get_writer() {
+                    let result = inbound::handle_incorrect_begin_string(
+                        &mut self.ctx, writer, begin_string,
+                    ).await;
+                    self.apply_transition(result);
+                }
             }
             MessageVerificationError::IncorrectCompId {
                 comp_id,
@@ -664,14 +669,6 @@ where
         Ok(())
     }
 
-    async fn handle_incorrect_begin_string(&mut self, received_begin_string: String) {
-        self.state
-            .logout_and_terminate(
-                &mut self.ctx,
-                &format!("beginString={received_begin_string} is not supported"),
-            )
-            .await;
-    }
 
     async fn handle_incorrect_comp_id(
         &mut self,
@@ -753,6 +750,12 @@ where
         Ok(())
     }
 
+    fn apply_transition(&mut self, result: TransitionResult) {
+        if let TransitionResult::TransitionTo(new_state) = result {
+            self.state = new_state;
+        }
+    }
+
     fn reset_heartbeat_timer(&mut self) {
         self.state
             .reset_heartbeat_timer(self.ctx.config.heartbeat_interval);

crates/hotfix/src/session/ctx.rs

@@ -7,8 +7,14 @@ use crate::message::OutboundMessage;
 use crate::message::generate_message;
 use crate::message::parser::RawFixMessage;
 use crate::session::error::InternalSendError;
+use crate::session::state::SessionState;
 use crate::store::StoreError;
 
+pub(crate) enum TransitionResult {
+    Stay,
+    TransitionTo(SessionState),
+}
+
 pub(crate) struct SessionCtx<A, S> {
     pub config: SessionConfig,
     pub store: S,

crates/hotfix/src/session/inbound.rs

@@ -1,9 +1,12 @@
+use crate::message::logout::Logout;
 use crate::message::reject::Reject;
 use crate::message::verification::verify_message;
 use crate::message::verification_error::MessageVerificationError;
-use crate::session::ctx::SessionCtx;
+use crate::session::ctx::{SessionCtx, TransitionResult};
 use crate::session::outbound;
+use crate::session::state::SessionState;
 use crate::transport::writer::WriterRef;
+use tracing::warn;
 use hotfix_message::Part;
 use hotfix_message::message::Message;
 use hotfix_message::session_fields::{MSG_SEQ_NUM, SessionRejectReason};
@@ -47,6 +50,22 @@ pub(crate) async fn handle_sending_time_accuracy_problem<A, S: MessageStore>(
     }
 }
 
+pub(crate) async fn handle_incorrect_begin_string<A, S: MessageStore>(
+    ctx: &mut SessionCtx<A, S>,
+    writer: &WriterRef,
+    received_begin_string: String,
+) -> TransitionResult {
+    let logout = Logout::with_reason(format!(
+        "beginString={received_begin_string} is not supported"
+    ));
+    match ctx.prepare_message(logout).await {
+        Ok(prepared) => writer.send_raw_message(prepared.raw).await,
+        Err(err) => warn!("failed to send logout for incorrect begin string: {err}"),
+    }
+    writer.disconnect().await;
+    TransitionResult::TransitionTo(SessionState::new_disconnected(true, "incorrect begin string"))
+}
+
 pub(crate) async fn handle_invalid_msg_type<A, S: MessageStore>(
     ctx: &mut SessionCtx<A, S>,
     writer: &WriterRef,