Revert "Try without buffer"

This reverts commit 301ba13.

Author: tiagocoutinho

src/lib.rs

@@ -3,7 +3,7 @@ extern crate log;
 
 use futures::future::join_all;
 use serde::Deserialize;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{mpsc, oneshot};
 
@@ -28,11 +28,25 @@ type ChannelTx = mpsc::Sender<Message>;
 type Error = Box<dyn std::error::Error + Send + Sync + 'static>;
 type Result<T> = std::result::Result<T, Error>;
 
+type TcpReader = BufReader<tokio::net::tcp::OwnedReadHalf>;
+type TcpWriter = BufWriter<tokio::net::tcp::OwnedWriteHalf>;
+
 fn frame_size(frame: &[u8]) -> Result<usize> {
     Ok(u16::from_be_bytes(frame[4..6].try_into()?) as usize)
 }
 
-async fn read_frame(stream: &mut (impl AsyncRead + Unpin)) -> Result<Frame> {
+fn split_connection(stream: TcpStream) -> (TcpReader, TcpWriter) {
+    let (reader, writer) = stream.into_split();
+    (BufReader::new(reader), BufWriter::new(writer))
+}
+
+async fn create_connection(url: &str) -> Result<(TcpReader, TcpWriter)> {
+    let stream = TcpStream::connect(url).await?;
+    stream.set_nodelay(true)?;
+    Ok(split_connection(stream))
+}
+
+async fn read_frame(stream: &mut TcpReader) -> Result<Frame> {
     let mut buf = vec![0u8; 6];
     // Read header
     stream.read_exact(&mut buf).await?;
@@ -55,7 +69,7 @@ struct Modbus {
 
 struct Device {
     url: String,
-    stream: Option<TcpStream>,
+    stream: Option<(TcpReader, TcpWriter)>,
 }
 
 impl Device {
@@ -67,17 +81,16 @@ impl Device {
     }
 
     async fn connect(&mut self) -> Result<()> {
-        match TcpStream::connect(&self.url).await {
-            Ok(stream) => {
-                stream.set_nodelay(true)?;
+        match create_connection(&self.url).await {
+            Ok(connection) => {
                 info!("modbus connection to {} sucessfull", self.url);
-                self.stream = Some(stream);
+                self.stream = Some(connection);
                 Ok(())
             }
             Err(error) => {
                 self.stream = None;
                 info!("modbus connection to {} error: {} ", self.url, error);
-                Err(Box::new(error))
+                Err(error)
             }
         }
     }
@@ -91,10 +104,10 @@ impl Device {
     }
 
     async fn raw_write_read(&mut self, frame: &Frame) -> Result<Frame> {
-        let stream = self.stream.as_mut().ok_or("no modbus connection")?;
-        stream.write_all(&frame).await?;
-        stream.flush().await?;
-        read_frame(stream).await
+        let (reader, writer) = self.stream.as_mut().ok_or("no modbus connection")?;
+        writer.write_all(&frame).await?;
+        writer.flush().await?;
+        read_frame(reader).await
     }
 
     async fn write_read(&mut self, frame: &Frame) -> Result<Frame> {
@@ -176,27 +189,25 @@ impl Bridge {
             &self.listen.bind, &self.modbus.url
         );
         loop {
-            let (mut client, _) = listener.accept().await.unwrap();
-            client.set_nodelay(true).unwrap();
+            let (client, _) = listener.accept().await.unwrap();
             let tx = tx.clone();
             tokio::spawn(async move {
-                if let Err(err) = Self::handle_client(&mut client, tx).await {
+                if let Err(err) = Self::handle_client(client, tx).await {
                     error!("Client error: {:?}", err);
                 }
             });
         }
     }
 
-    async fn handle_client(
-        client: &mut (impl AsyncRead + AsyncWrite + Unpin),
-        channel: ChannelTx,
-    ) -> Result<()> {
+    async fn handle_client(client: TcpStream, channel: ChannelTx) -> Result<()> {
+        client.set_nodelay(true)?;
         channel.send(Message::Connection).await?;
-        while let Ok(buf) = read_frame(client).await {
+        let (mut reader, mut writer) = split_connection(client);
+        while let Ok(buf) = read_frame(&mut reader).await {
             let (tx, rx) = oneshot::channel();
             channel.send(Message::Packet(buf, tx)).await?;
-            client.write_all(&rx.await?).await?;
-            client.flush().await?;
+            writer.write_all(&rx.await?).await?;
+            writer.flush().await?;
         }
         channel.send(Message::Disconnection).await?;
         Ok(())
@@ -227,24 +238,3 @@ impl Server {
         Ok(Self::new(config_file)?.run().await)
     }
 }
-
-#[test]
-fn test_device_not_connected() {
-    let device = Device::new("some url");
-    assert_eq!(device.url, "some url");
-    assert!(device.stream.is_none());
-    assert_eq!(device.is_connected(), false);
-}
-
-#[test]
-fn test_device_not_connected_disconnects() {
-    let mut device = Device::new("some url");
-    assert_eq!(device.is_connected(), false);
-    device.disconnect();
-    assert_eq!(device.is_connected(), false);
-}
-
-#[tokio::test]
-async fn test_device_connection() {
-    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
-}