Remove anyhow from session handle and initiator APIs

Author: davidsteiner

crates/hotfix-web/src/session_controller.rs

@@ -22,11 +22,13 @@ impl<Outbound: OutboundMessage> SessionController for HttpSessionController<Outb
     }
 
     async fn request_reset_on_next_logon(&self) -> anyhow::Result<()> {
-        self.session_handle.request_reset_on_next_logon().await
+        self.session_handle.request_reset_on_next_logon().await?;
+        Ok(())
     }
 
     async fn shutdown(&self, reconnect: bool) -> anyhow::Result<()> {
-        self.session_handle.shutdown(reconnect).await
+        self.session_handle.shutdown(reconnect).await?;
+        Ok(())
     }
 }
 

crates/hotfix/src/initiator.rs

@@ -6,7 +6,6 @@
 //! The initiator establishes the transport layer connection with
 //! the peer, and sends the initial Logon (35=A) message. For transport,
 //! `HotFIX` supports plain TCP and encrypted TLS over TCP connections.
-use anyhow::Result;
 use std::time::Duration;
 use tokio::sync::watch;
 use tokio::time::sleep;
@@ -15,7 +14,7 @@ use tracing::{debug, warn};
 use crate::application::Application;
 use crate::config::SessionConfig;
 use crate::message::{InboundMessage, OutboundMessage};
-use crate::session::error::{SendError, SendOutcome};
+use crate::session::error::{SendError, SendOutcome, SessionCreationError};
 use crate::session::{InternalSessionRef, SessionHandle};
 use crate::store::MessageStore;
 use crate::transport::connect;
@@ -32,7 +31,7 @@ impl<Outbound: OutboundMessage> Initiator<Outbound> {
         config: SessionConfig,
         application: impl Application<Inbound, Outbound>,
         store: impl MessageStore + 'static,
-    ) -> Result<Self> {
+    ) -> Result<Self, SessionCreationError> {
         let session_ref = InternalSessionRef::new(config.clone(), application, store)?;
         let (completion_tx, completion_rx) = watch::channel(false);
 
@@ -76,9 +75,11 @@ impl<Outbound: OutboundMessage> Initiator<Outbound> {
         self.session_handle.clone()
     }
 
-    pub async fn shutdown(self, reconnect: bool) -> Result<()> {
+    pub async fn shutdown(self, reconnect: bool) -> Result<(), SendError> {
         self.session_handle.shutdown(reconnect).await?;
-        tokio::time::timeout(Duration::from_secs(5), self.wait_for_shutdown()).await?;
+        tokio::time::timeout(Duration::from_secs(5), self.wait_for_shutdown())
+            .await
+            .map_err(|_| SendError::SessionGone)?;
 
         Ok(())
     }

crates/hotfix/src/session/error.rs

@@ -52,6 +52,18 @@ pub enum SendError {
     SessionGone,
 }
 
+impl<T> From<tokio::sync::mpsc::error::SendError<T>> for SendError {
+    fn from(_: tokio::sync::mpsc::error::SendError<T>) -> Self {
+        SendError::SessionGone
+    }
+}
+
+impl From<tokio::sync::oneshot::error::RecvError> for SendError {
+    fn from(_: tokio::sync::oneshot::error::RecvError) -> Self {
+        SendError::SessionGone
+    }
+}
+
 /// Error that can occur when sending a message internally within the session.
 ///
 /// This is a subset of `SendError` without `SessionTerminated` and `SessionGone`,

crates/hotfix/src/session/session_handle.rs

@@ -36,12 +36,8 @@ impl<Outbound> SessionHandle<Outbound> {
             message: msg,
             confirm: Some(tx),
         };
-        self.outbound_message_sender
-            .send(request)
-            .await
-            .map_err(|_| SendError::Disconnected)?;
-
-        rx.await.map_err(|_| SendError::SessionGone)?
+        self.outbound_message_sender.send(request).await?;
+        rx.await?
     }
 
     /// Sends a message without waiting for confirmation.
@@ -53,27 +49,21 @@ impl<Outbound> SessionHandle<Outbound> {
             message: msg,
             confirm: None,
         };
-        self.outbound_message_sender
-            .send(request)
-            .await
-            .map_err(|_| SendError::Disconnected)?;
-
+        self.outbound_message_sender.send(request).await?;
         Ok(())
     }
 
-    pub async fn shutdown(&self, reconnect: bool) -> anyhow::Result<()> {
+    pub async fn shutdown(&self, reconnect: bool) -> Result<(), SendError> {
         self.admin_request_sender
             .send(AdminRequest::InitiateGracefulShutdown { reconnect })
             .await?;
-
         Ok(())
     }
 
-    pub async fn request_reset_on_next_logon(&self) -> anyhow::Result<()> {
+    pub async fn request_reset_on_next_logon(&self) -> Result<(), SendError> {
         self.admin_request_sender
             .send(AdminRequest::ResetSequenceNumbersOnNextLogon)
             .await?;
-
         Ok(())
     }
 }

crates/hotfix/src/session/session_ref.rs

@@ -1,4 +1,3 @@
-use anyhow::Result;
 use thiserror::Error;
 use tokio::sync::{mpsc, oneshot};
 use tracing::debug;
@@ -7,7 +6,7 @@ use crate::config::SessionConfig;
 use crate::message::{InboundMessage, OutboundMessage, RawFixMessage};
 use crate::session::Session;
 use crate::session::admin_request::AdminRequest;
-use crate::session::error::{SendError, SendOutcome};
+use crate::session::error::{SendError, SendOutcome, SessionCreationError};
 use crate::session::event::{AwaitingActiveSessionResponse, SessionEvent};
 use crate::store::MessageStore;
 use crate::transport::writer::WriterRef;
@@ -31,7 +30,7 @@ impl<Outbound: OutboundMessage> InternalSessionRef<Outbound> {
         config: SessionConfig,
         application: impl Application<Inbound, Outbound>,
         store: impl MessageStore + 'static,
-    ) -> Result<Self> {
+    ) -> Result<Self, SessionCreationError> {
         let (event_sender, event_receiver) = mpsc::channel::<SessionEvent>(100);
         let (outbound_message_sender, outbound_message_receiver) =
             mpsc::channel::<OutboundRequest<Outbound>>(10);

examples/load-testing/src/main.rs

@@ -94,17 +94,19 @@ async fn start_session(
     db_config: Database,
     app: LoadTestingApplication,
 ) -> Result<Initiator<OutboundMsg>> {
-    match db_config {
+    let initiator = match db_config {
         Database::Memory => {
             let store = hotfix::store::in_memory::InMemoryMessageStore::default();
-            Initiator::start(session_config, app, store).await
+            Initiator::start(session_config, app, store).await?
         }
         Database::File => {
             let store = hotfix::store::file::FileStore::new("data", "load-testing-store")
                 .expect("be able to create store");
-            Initiator::start(session_config, app, store).await
+            Initiator::start(session_config, app, store).await?
         }
-    }
+    };
+
+    Ok(initiator)
 }
 
 async fn submit_messages(session_handle: SessionHandle<OutboundMsg>, message_count: u32) {

examples/simple-new-order/src/main.rs

@@ -152,17 +152,19 @@ async fn start_session(
         .pop()
         .context("config must include a session")?;
 
-    match db_config {
+    let initiator = match db_config {
         Database::Memory => {
             let store = hotfix::store::in_memory::InMemoryMessageStore::default();
-            Initiator::start(session_config, app, store).await
+            Initiator::start(session_config, app, store).await?
         }
         Database::File => {
             let store = hotfix::store::file::FileStore::new("data", "simple-new-order-store")
                 .context("failed to create file store")?;
-            Initiator::start(session_config, app, store).await
+            Initiator::start(session_config, app, store).await?
         }
-    }
+    };
+
+    Ok(initiator)
 }
 
 async fn start_web_service(