Create structure for awaiting session times

Author: davidsteiner

crates/hotfix/src/initiator.rs

@@ -58,10 +58,13 @@ async fn establish_connection<M: FixMessage>(config: SessionConfig, session_ref:
             warn!("session indicated we shouldn't reconnect");
             break;
         }
+        session_ref.await_active_session_time().await;
 
         match FixConnection::connect(&config, session_ref.clone()).await {
             Ok(conn) => {
                 session_ref.register_writer(conn.get_writer()).await;
+
+                // TODO: should this ask the session about disconnects?
                 conn.run_until_disconnect().await;
 
                 warn!("session connection dropped, attempting to reconnect");

crates/hotfix/src/session.rs

@@ -30,6 +30,7 @@ use crate::message::resend_request::ResendRequest;
 use crate::message::sequence_reset::SequenceReset;
 use crate::message::test_request::TestRequest;
 use crate::message_utils::is_admin;
+use crate::session::event::AwaitingActiveSessionResponse;
 use crate::session::state::AwaitingResendState;
 use crate::session_schedule::SessionSchedule;
 use event::SessionEvent;
@@ -94,6 +95,15 @@ impl<M: FixMessage> SessionRef<M> {
             .unwrap();
         receiver.await.expect("to receive a response")
     }
+
+    pub async fn await_active_session_time(&self) {
+        let (sender, receiver) = oneshot::channel::<AwaitingActiveSessionResponse>();
+        self.sender
+            .send(SessionEvent::AwaitingActiveSession(sender))
+            .await
+            .unwrap();
+        receiver.await.expect("to receive a response");
+    }
 }
 
 struct Session<M, S> {
@@ -137,6 +147,7 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
             state: SessionState::Disconnected {
                 reconnect: true,
                 _reason: "initialising".to_string(),
+                session_awaiter: None,
             },
             application,
             store,
@@ -323,12 +334,14 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
                 self.state = SessionState::Disconnected {
                     reconnect: true,
                     _reason: reason,
+                    session_awaiter: None,
                 }
             }
             SessionState::LoggedOut { reconnect } => {
                 self.state = SessionState::Disconnected {
                     reconnect,
                     _reason: "logged out".to_string(),
+                    session_awaiter: None,
                 }
             }
             SessionState::Disconnected { .. } => {
@@ -647,6 +660,9 @@ impl<M: FixMessage, S: MessageStore> Session<M, S> {
                     .send(self.state.should_reconnect())
                     .expect("be able to respond");
             }
+            SessionEvent::AwaitingActiveSession(responder) => {
+                self.state.register_session_awaiter(responder);
+            }
         }
     }
 

crates/hotfix/src/session/event.rs

@@ -15,4 +15,19 @@ pub enum SessionEvent<M> {
     Connected(WriterRef),
     /// Ask the session whether we should attempt to reconnect.
     ShouldReconnect(oneshot::Sender<bool>),
+    /// Ask the session to notify us when the session is active.
+    AwaitingActiveSession(oneshot::Sender<AwaitingActiveSessionResponse>),
+}
+
+/// The response sent by the session to AwaitingActiveSession messages.
+///
+/// This doesn't include an Inactive variant, as the session won't respond until
+/// it's active or in a state that indicates it should just be shut down due to an
+/// unrecoverable error.
+#[derive(Debug, Clone, Copy)]
+pub enum AwaitingActiveSessionResponse {
+    /// The session is now active and ready to connect.
+    Active,
+    /// The session should be shut down due to an unrecoverable error.
+    Shutdown,
 }

crates/hotfix/src/session/state.rs

@@ -1,8 +1,10 @@
 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 {
@@ -17,7 +19,14 @@ pub enum SessionState {
     /// The peer has logged us out.
     LoggedOut { reconnect: bool },
     /// The TCP connection has been dropped.
-    Disconnected { reconnect: bool, _reason: String },
+    ///
+    /// This is also the state we're in if we purposefully disconnected due to the current
+    /// time being out of session hours.
+    Disconnected {
+        reconnect: bool,
+        session_awaiter: Option<oneshot::Sender<AwaitingActiveSessionResponse>>,
+        _reason: String,
+    },
 }
 
 impl SessionState {
@@ -92,6 +101,34 @@ impl SessionState {
             _ => false,
         }
     }
+
+    pub fn register_session_awaiter(
+        &self,
+        responder: oneshot::Sender<AwaitingActiveSessionResponse>,
+    ) {
+        match self {
+            SessionState::Disconnected {
+                reconnect: true,
+                session_awaiter,
+                _reason,
+            } => {
+                if session_awaiter.is_some() {
+                    error!("session awaiter already registered");
+                    if let Err(err) = responder.send(AwaitingActiveSessionResponse::Shutdown) {
+                        error!("failed to send session awaiter response: {err:?}");
+                    }
+                } else {
+                    // set the session_awaiter on the current state
+                }
+            }
+            _ => {
+                error!("session awaiter can only be registered on disconnected sessions");
+                if let Err(err) = responder.send(AwaitingActiveSessionResponse::Shutdown) {
+                    error!("failed to send session awaiter response: {err:?}");
+                }
+            }
+        }
+    }
 }
 
 /// Session state we're in while processing messages we requested to be resent.