Correctly reconnect when we enter a new active session time

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -97,12 +97,14 @@ impl<M: FixMessage> SessionRef<M> {
     }
 
     pub async fn await_active_session_time(&self) {
+        debug!("awaiting active session time");
         let (sender, receiver) = oneshot::channel::<AwaitingActiveSessionResponse>();
         self.sender
             .send(SessionEvent::AwaitingActiveSession(sender))
             .await
             .unwrap();
         receiver.await.expect("to receive a response");
+        debug!("resuming connection as session is active");
     }
 }
 
@@ -399,14 +401,19 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
     async fn on_logout(&mut self) {
         if let SessionState::AwaitingLogout { .. } = &self.state {
             self.state.disconnect().await;
+            self.state = SessionState::Disconnected {
+                reconnect: true,
+                _reason: "we logged out gracefully".to_string(),
+                session_awaiter: None,
+            };
+        } else {
+            // TODO: reconnect = false isn't always valid, this should be more sophisticated
+            self.state.disconnect().await;
+            self.state = SessionState::LoggedOut { reconnect: false };
+            self.application
+                .send_logout("peer has logged us out".to_string())
+                .await;
         }
-
-        // TODO: reconnect = false isn't always valid, this should be more sophisticated
-        self.state.disconnect().await;
-        self.state = SessionState::LoggedOut { reconnect: false };
-        self.application
-            .send_logout("peer has logged us out".to_string())
-            .await;
     }
 
     async fn on_heartbeat(&mut self, message: &Message) {

crates/hotfix/src/session/state.rs

@@ -1,12 +1,11 @@
+use crate::message::parser::RawFixMessage;
+use crate::session::event::AwaitingActiveSessionResponse;
+use crate::transport::socket_writer::WriterRef;
 use hotfix_message::message::Message;
 use std::collections::VecDeque;
 use tokio::sync::oneshot;
 use tracing::{debug, error};
 
-use crate::message::parser::RawFixMessage;
-use crate::session::event::AwaitingActiveSessionResponse;
-use crate::transport::socket_writer::WriterRef;
-
 pub enum SessionState {
     /// We have established a connection, sent a logon message and await a response.
     AwaitingLogon { writer: WriterRef, logon_sent: bool },
@@ -119,6 +118,7 @@ impl SessionState {
                     }
                 } else {
                     *session_awaiter = Some(responder);
+                    debug!("registered session awaiter");
                 }
             }
             _ => {
@@ -138,6 +138,8 @@ impl SessionState {
             if let Some(awaiter) = session_awaiter.take() {
                 if let Err(err) = awaiter.send(AwaitingActiveSessionResponse::Active) {
                     error!("failed to send session awaiter response: {err:?}");
+                } else {
+                    debug!("notified session awaiter");
                 }
             }
         }