Decoupling TCP Writer

Author: Diego Pardo

crates/hotfix/src/initiator.rs

@@ -15,7 +15,7 @@ use crate::config::SessionConfig;
 use crate::message::FixMessage;
 use crate::session::SessionRef;
 use crate::store::MessageStore;
-use crate::transport::FixConnection;
+use crate::transport::connect;
 
 pub struct Initiator<M> {
     pub config: SessionConfig,
@@ -60,7 +60,7 @@ async fn establish_connection<M: FixMessage>(config: SessionConfig, session_ref:
         }
         session_ref.await_active_session_time().await;
 
-        match FixConnection::connect(&config, session_ref.clone()).await {
+        match connect(&config, session_ref.clone()).await {
             Ok(conn) => {
                 session_ref.register_writer(conn.get_writer()).await;
 

crates/hotfix/src/session.rs

@@ -22,7 +22,7 @@ use crate::message::heartbeat::Heartbeat;
 use crate::message::logon::{Logon, ResetSeqNumConfig};
 use crate::message::parser::RawFixMessage;
 use crate::store::MessageStore;
-use crate::transport::socket_writer::WriterRef;
+use crate::transport::writer::WriterRef;
 
 use crate::error::MessageVerificationError;
 use crate::message::logout::Logout;

crates/hotfix/src/session/event.rs

@@ -1,7 +1,7 @@
 use tokio::sync::oneshot;
 
 use crate::message::parser::RawFixMessage;
-use crate::transport::socket_writer::WriterRef;
+use crate::transport::writer::WriterRef;
 
 #[derive(Debug)]
 pub enum SessionEvent<M> {

crates/hotfix/src/session/state.rs

@@ -1,6 +1,6 @@
 use crate::message::parser::RawFixMessage;
 use crate::session::event::AwaitingActiveSessionResponse;
-use crate::transport::socket_writer::WriterRef;
+use crate::transport::writer::WriterRef;
 use hotfix_message::message::Message;
 use std::collections::VecDeque;
 use tokio::sync::oneshot;

crates/hotfix/src/transport.rs

@@ -1,7 +1,6 @@
 mod connection;
-pub(crate) mod socket_reader;
-pub(crate) mod socket_writer;
-mod tcp;
-mod tls;
+pub mod socket;
+pub mod writer;
 
 pub use connection::FixConnection;
+pub use socket::*;

crates/hotfix/src/transport/connection.rs

@@ -1,37 +1,18 @@
-use std::io;
-use tokio::io::{AsyncRead, AsyncWrite};
-
-use crate::config::SessionConfig;
-use crate::message::FixMessage;
-use crate::session::SessionRef;
 use crate::transport::socket_reader::ReaderRef;
-use crate::transport::socket_writer::WriterRef;
-use crate::transport::tcp::create_tcp_connection;
-use crate::transport::tls::create_tcp_over_tls_connection;
+use crate::transport::writer::WriterRef;
 
 pub struct FixConnection {
     _writer: WriterRef,
     _reader: ReaderRef,
 }
 
 impl FixConnection {
-    pub async fn connect(
-        config: &SessionConfig,
-        session_ref: SessionRef<impl FixMessage>,
-    ) -> io::Result<Self> {
-        let use_tls = config.tls_config.is_some();
-
-        let conn = if use_tls {
-            let stream = create_tcp_over_tls_connection(config).await?;
-            _create_io_refs(session_ref.clone(), stream).await
-        } else {
-            let stream = create_tcp_connection(config).await?;
-            _create_io_refs(session_ref.clone(), stream).await
-        };
-
-        Ok(conn)
+    pub fn new(writer: WriterRef, reader: ReaderRef) -> Self {
+        Self {
+            _writer: writer,
+            _reader: reader,
+        }
     }
-
     pub fn get_writer(&self) -> WriterRef {
         self._writer.clone()
     }
@@ -40,19 +21,3 @@ impl FixConnection {
         self._reader.wait_for_disconnect().await
     }
 }
-
-async fn _create_io_refs<M, Stream>(session_ref: SessionRef<M>, stream: Stream) -> FixConnection
-where
-    M: FixMessage,
-    Stream: AsyncRead + AsyncWrite + Send + 'static,
-{
-    let (reader, writer) = tokio::io::split(stream);
-
-    let writer_ref = WriterRef::new(writer);
-    let reader_ref = ReaderRef::new(reader, session_ref);
-
-    FixConnection {
-        _writer: writer_ref,
-        _reader: reader_ref,
-    }
-}

crates/hotfix/src/transport/socket.rs

@@ -0,0 +1,48 @@
+pub mod socket_reader;
+pub mod socket_writer;
+pub mod tcp;
+pub mod tls;
+
+use std::io;
+use tokio::io::{AsyncRead, AsyncWrite};
+
+use crate::{
+    config::SessionConfig,
+    message::FixMessage,
+    session::SessionRef,
+    transport::{
+        FixConnection, socket_reader::ReaderRef, socket_writer::spawn_socket_writer,
+        tcp::create_tcp_connection, tls::create_tcp_over_tls_connection,
+    },
+};
+
+/// Connect over TCP/TLS and return a FixConnection
+pub async fn connect(
+    config: &SessionConfig,
+    session_ref: SessionRef<impl FixMessage>,
+) -> io::Result<FixConnection> {
+    let use_tls = config.tls_config.is_some();
+
+    let conn = if use_tls {
+        let stream = create_tcp_over_tls_connection(config).await?;
+        _create_io_refs(session_ref.clone(), stream).await
+    } else {
+        let stream = create_tcp_connection(config).await?;
+        _create_io_refs(session_ref.clone(), stream).await
+    };
+
+    Ok(conn)
+}
+
+async fn _create_io_refs<M, Stream>(session_ref: SessionRef<M>, stream: Stream) -> FixConnection
+where
+    M: FixMessage,
+    Stream: AsyncRead + AsyncWrite + Send + 'static,
+{
+    let (reader, writer) = tokio::io::split(stream);
+
+    let writer_ref = spawn_socket_writer(writer);
+    let reader_ref = ReaderRef::new(reader, session_ref);
+
+    FixConnection::new(writer_ref, reader_ref)
+}

…es/hotfix/src/transport/socket_reader.rs …ix/src/transport/socket/socket_reader.rscrates/hotfix/src/transport/socket_reader.rs renamed to crates/hotfix/src/transport/socket/socket_reader.rs crates/hotfix/src/transport/socket_reader.rs renamed to crates/hotfix/src/transport/socket/socket_reader.rs

@@ -1,42 +1,14 @@
+use crate::transport::writer::{WriterMessage, WriterRef};
 use tokio::io::{AsyncWrite, AsyncWriteExt, WriteHalf};
 use tokio::sync::mpsc;
 use tracing::{debug, warn};
 
-use crate::message::parser::RawFixMessage;
+pub fn spawn_socket_writer(writer: WriteHalf<impl AsyncWrite + Send + 'static>) -> WriterRef {
+    let (sender, mailbox) = mpsc::channel(10);
+    let actor = WriterActor::new(writer, mailbox);
+    tokio::spawn(run_writer(actor));
 
-#[derive(Clone, Debug)]
-pub enum WriterMessage {
-    SendMessage(RawFixMessage),
-    Disconnect,
-}
-
-#[derive(Clone, Debug)]
-pub struct WriterRef {
-    sender: mpsc::Sender<WriterMessage>,
-}
-
-impl WriterRef {
-    pub fn new(writer: WriteHalf<impl AsyncWrite + Send + 'static>) -> Self {
-        let (sender, mailbox) = mpsc::channel(10);
-        let actor = WriterActor::new(writer, mailbox);
-        tokio::spawn(run_writer(actor));
-
-        Self { sender }
-    }
-
-    pub async fn send_raw_message(&self, msg: RawFixMessage) {
-        self.sender
-            .send(WriterMessage::SendMessage(msg))
-            .await
-            .expect("be able to send message");
-    }
-
-    pub async fn disconnect(&self) {
-        self.sender
-            .send(WriterMessage::Disconnect)
-            .await
-            .expect("be able to disconnect")
-    }
+    WriterRef::new(sender)
 }
 
 struct WriterActor<W> {