feat: allow one off restarts with sequence number reset (#241)

* Split up event queue for sessions so that messages can be prioritised

* Split out app facing session handle which doesn't support internal actions

* Support action to reset sequence numbers as a one-off action

* Allow shutdown action to set the reconnect flag

* Expose reset on next logon action via public interface

* Add test case for requesting a one-off reset and then reconnecting

* Add unit test suite for socket reader

* Add unit test suite for socket writer

* Add doc comment for SessionHandle

Author: davidsteiner

Cargo.lock

@@ -20,6 +20,7 @@ ui = ["askama", "mime_guess", "rust-embed"]
 [dependencies]
 hotfix = { version = "0.2.9", path = "../hotfix" }
 
+anyhow = { workspace = true }
 askama = { workspace = true, features = ["serde_json"], optional = true }
 async-trait = { workspace = true }
 axum = { workspace = true }

crates/hotfix-status/Cargo.toml

@@ -1,5 +1,6 @@
 use crate::AppState;
 use crate::data_provider::DataProvider;
+use crate::error::AppResult;
 use axum::extract::State;
 use axum::routing::get;
 use axum::{Json, Router};
@@ -30,8 +31,8 @@ struct SessionInfoResponse {
 
 async fn get_session_info<P: DataProvider>(
     State(state): State<AppState<P>>,
-) -> Json<SessionInfoResponse> {
-    let session_info = state.data_provider.get_session_info().await;
+) -> AppResult<Json<SessionInfoResponse>> {
+    let session_info = state.data_provider.get_session_info().await?;
 
-    Json(SessionInfoResponse { session_info })
+    Ok(Json(SessionInfoResponse { session_info }))
 }

crates/hotfix-status/src/api.rs

@@ -1,19 +1,19 @@
 use hotfix::message::FixMessage;
-use hotfix::session::{SessionInfo, SessionRef};
+use hotfix::session::{SessionHandle, SessionInfo};
 
 #[async_trait::async_trait]
 pub trait DataProvider: Clone + Send + Sync {
-    async fn get_session_info(&self) -> SessionInfo;
+    async fn get_session_info(&self) -> anyhow::Result<SessionInfo>;
 }
 
 #[derive(Clone)]
 pub struct SessionDataProvider<M> {
-    pub(crate) session_ref: SessionRef<M>,
+    pub(crate) session_handle: SessionHandle<M>,
 }
 
 #[async_trait::async_trait]
 impl<M: FixMessage> DataProvider for SessionDataProvider<M> {
-    async fn get_session_info(&self) -> SessionInfo {
-        self.session_ref.get_session_info().await
+    async fn get_session_info(&self) -> anyhow::Result<SessionInfo> {
+        self.session_handle.get_session_info().await
     }
 }

crates/hotfix-status/src/data_provider.rs

@@ -3,6 +3,8 @@ use axum::response::{IntoResponse, Response};
 
 #[derive(Debug, displaydoc::Display, thiserror::Error)]
 pub enum AppError {
+    /// General anyhow errors
+    Anyhow(#[from] anyhow::Error),
     #[cfg(feature = "ui")]
     /// could not render the template
     Render(#[from] askama::Error),

crates/hotfix-status/src/error.rs

@@ -9,15 +9,15 @@ use crate::api::build_api_router;
 use crate::data_provider::{DataProvider, SessionDataProvider};
 use axum::Router;
 use hotfix::message::FixMessage;
-use hotfix::session::SessionRef;
+use hotfix::session::SessionHandle;
 
 #[derive(Clone)]
 struct AppState<P> {
     data_provider: P,
 }
 
-pub fn build_router<M: FixMessage>(session_ref: SessionRef<M>) -> Router {
-    let data_provider = SessionDataProvider { session_ref };
+pub fn build_router<M: FixMessage>(session_handle: SessionHandle<M>) -> Router {
+    let data_provider = SessionDataProvider { session_handle };
     build_router_with_provider(data_provider)
 }
 
@@ -55,8 +55,8 @@ mod tests {
 
     #[async_trait::async_trait]
     impl DataProvider for FakeDataProvider {
-        async fn get_session_info(&self) -> SessionInfo {
-            self.session_info.clone()
+        async fn get_session_info(&self) -> anyhow::Result<SessionInfo> {
+            Ok(self.session_info.clone())
         }
     }
 

crates/hotfix-status/src/lib.rs

@@ -19,7 +19,7 @@ struct DashboardTemplate<'a> {
 pub(crate) async fn dashboard_handler<P: DataProvider>(
     State(state): State<AppState<P>>,
 ) -> AppResult<impl IntoResponse> {
-    let session_info = state.data_provider.get_session_info().await;
+    let session_info = state.data_provider.get_session_info().await?;
     let timestamp_string = Utc::now().to_rfc3339();
 
     let template = DashboardTemplate {

crates/hotfix-status/src/ui/dashboard.rs

@@ -12,8 +12,10 @@ keywords.workspace = true
 categories.workspace = true
 
 [features]
+default = ["test-utils"]
 redb = ["dep:redb"]
 mongodb = ["dep:mongodb"]
+test-utils = []
 
 [lints]
 workspace = true

crates/hotfix/Cargo.toml

@@ -15,14 +15,14 @@ use tracing::{debug, warn};
 use crate::application::Application;
 use crate::config::SessionConfig;
 use crate::message::FixMessage;
-use crate::session::SessionRef;
+use crate::session::{InternalSessionRef, SessionHandle};
 use crate::store::MessageStore;
 use crate::transport::connect;
 
 #[derive(Clone)]
 pub struct Initiator<M> {
     pub config: SessionConfig,
-    session: SessionRef<M>,
+    session_handle: SessionHandle<M>,
     completion_rx: watch::Receiver<bool>,
 }
 
@@ -32,7 +32,7 @@ impl<M: FixMessage> Initiator<M> {
         application: impl Application<M>,
         store: impl MessageStore + Send + Sync + 'static,
     ) -> Self {
-        let session_ref = SessionRef::new(config.clone(), application, store);
+        let session_ref = InternalSessionRef::new(config.clone(), application, store);
         let (completion_tx, completion_rx) = watch::channel(false);
 
         tokio::spawn({
@@ -43,25 +43,27 @@ impl<M: FixMessage> Initiator<M> {
 
         Self {
             config,
-            session: session_ref,
+            session_handle: session_ref.into(),
             completion_rx,
         }
     }
 
-    pub async fn send_message(&self, msg: M) {
-        self.session.send_message(msg).await;
+    pub async fn send_message(&self, msg: M) -> anyhow::Result<()> {
+        self.session_handle.send_message(msg).await?;
+
+        Ok(())
     }
 
     pub fn is_interested(&self, sender_comp_id: &str, target_comp_id: &str) -> bool {
         self.config.sender_comp_id == sender_comp_id && self.config.target_comp_id == target_comp_id
     }
 
-    pub fn session_ref(&self) -> SessionRef<M> {
-        self.session.clone()
+    pub fn session_handle(&self) -> SessionHandle<M> {
+        self.session_handle.clone()
     }
 
-    pub async fn shutdown(self) -> Result<(), Elapsed> {
-        self.session.shutdown().await;
+    pub async fn shutdown(self, reconnect: bool) -> Result<(), Elapsed> {
+        self.session_handle.shutdown(reconnect).await;
         tokio::time::timeout(Duration::from_secs(5), self.wait_for_shutdown()).await
     }
 
@@ -84,7 +86,7 @@ impl<M: FixMessage> Initiator<M> {
 
 async fn establish_connection<M: FixMessage>(
     config: SessionConfig,
-    session_ref: SessionRef<M>,
+    session_ref: InternalSessionRef<M>,
     completion_tx: watch::Sender<bool>,
 ) {
     loop {