116: Decoupling TCP Reader

Author: Diego Pardo

crates/hotfix/src/transport.rs

@@ -1,4 +1,5 @@
 mod connection;
+pub mod reader;
 pub mod socket;
 pub mod writer;
 

crates/hotfix/src/transport/connection.rs

@@ -1,4 +1,4 @@
-use crate::transport::socket_reader::ReaderRef;
+use crate::transport::reader::ReaderRef;
 use crate::transport::writer::WriterRef;
 
 pub struct FixConnection {

crates/hotfix/src/transport/reader.rs

@@ -0,0 +1,21 @@
+use tokio::sync::oneshot;
+
+#[derive(Clone, Debug)]
+#[allow(dead_code)]
+pub struct ReaderMessage;
+
+pub struct ReaderRef {
+    disconnect_signal: oneshot::Receiver<()>,
+}
+
+impl ReaderRef {
+    pub fn new(disconnect_signal: oneshot::Receiver<()>) -> Self {
+        Self { disconnect_signal }
+    }
+
+    pub async fn wait_for_disconnect(self) {
+        self.disconnect_signal
+            .await
+            .expect("not to drop signal prematurely");
+    }
+}

crates/hotfix/src/transport/socket.rs

@@ -11,7 +11,7 @@ use crate::{
     message::FixMessage,
     session::SessionRef,
     transport::{
-        FixConnection, socket_reader::ReaderRef, socket_writer::spawn_socket_writer,
+        FixConnection, socket_reader::spawn_socket_reader, socket_writer::spawn_socket_writer,
         tcp::create_tcp_connection, tls::create_tcp_over_tls_connection,
     },
 };
@@ -42,7 +42,7 @@ where
     let (reader, writer) = tokio::io::split(stream);
 
     let writer_ref = spawn_socket_writer(writer);
-    let reader_ref = ReaderRef::new(reader, session_ref);
+    let reader_ref = spawn_socket_reader(reader, session_ref);
 
     FixConnection::new(writer_ref, reader_ref)
 }

crates/hotfix/src/transport/socket/socket_reader.rs

@@ -5,34 +5,17 @@ use tracing::debug;
 use crate::message::FixMessage;
 use crate::message::parser::Parser;
 use crate::session::SessionRef;
+use crate::transport::reader::ReaderRef;
 
-#[derive(Clone, Debug)]
-#[allow(dead_code)]
-pub struct ReaderMessage;
+pub fn spawn_socket_reader(
+    reader: ReadHalf<impl AsyncRead + Send + 'static>,
+    session_ref: SessionRef<impl FixMessage>,
+) -> ReaderRef {
+    let (dc_sender, dc_receiver) = oneshot::channel();
+    let actor = ReaderActor::new(reader, session_ref, dc_sender);
+    tokio::spawn(run_reader(actor));
 
-pub struct ReaderRef {
-    disconnect_signal: oneshot::Receiver<()>,
-}
-
-impl ReaderRef {
-    pub fn new<M: FixMessage>(
-        reader: ReadHalf<impl AsyncRead + Send + 'static>,
-        session_ref: SessionRef<M>,
-    ) -> Self {
-        let (dc_sender, dc_receiver) = oneshot::channel();
-        let actor = ReaderActor::new(reader, session_ref, dc_sender);
-        tokio::spawn(run_reader(actor));
-
-        Self {
-            disconnect_signal: dc_receiver,
-        }
-    }
-
-    pub async fn wait_for_disconnect(self) {
-        self.disconnect_signal
-            .await
-            .expect("not to drop signal prematurely");
-    }
+    ReaderRef::new(dc_receiver)
 }
 
 struct ReaderActor<M, R> {