homework: power socket tokio client

smart-house/src/bin/power_socket_mock.rs

@@ -1,8 +1,7 @@
-use std::io;
-use std::io::{Read, Write};
-use std::net::{SocketAddr, TcpListener, TcpStream};
-use std::sync::{Arc, RwLock};
+use std::net::SocketAddr;
+use std::sync::Arc;
 use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 fn parse_args() -> SocketAddr {
     let mut args = std::env::args();
@@ -13,6 +12,7 @@ fn parse_args() -> SocketAddr {
         .expect("server address should be valid")
 }
 
+#[derive(Debug)]
 struct RealPowerSocket {
     power_rate: f32,
     on: bool,
@@ -23,46 +23,40 @@ const CMD_TURN_ON: u8 = 2;
 const CMD_TURN_OFF: u8 = 3;
 const CMD_GET_POWER: u8 = 4;
 
-fn handle_connection(mut stream: TcpStream, real_power_socket: Arc<RwLock<RealPowerSocket>>) {
+const TIMEOUT: Duration = Duration::from_secs(5);
+
+async fn handle_connection(mut socket: tokio::net::TcpStream, real_power_socket: Arc<tokio::sync::RwLock<RealPowerSocket>>) -> Result<(), std::io::Error> {
+    let mut buf = [0u8; 1];
     loop {
-        let mut buf = [0u8; 1];
-        let result = stream.read_exact(&mut buf);
-        match result {
-            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
-                std::thread::sleep(Duration::from_millis(250));
-                continue;
-            }
-            Err(e) if [io::ErrorKind::UnexpectedEof, io::ErrorKind::ConnectionReset].contains(&e.kind()) => {
-                println!("connection is over");
-                break;
-            }
-            Err(e) => panic!("error on reading socket: {e}"),
-            Ok(_) => {}
+        let read = tokio::time::timeout(TIMEOUT, socket.read(&mut buf)).await??;
+        if read == 0 {
+            println!("connection closed");
+            return Ok(());
         }
         match buf {
             [CMD_GET_ON] => {
                 println!("handling CMD_GET_ON");
-                let power_socket = real_power_socket.read().expect("power socket lock is poisoned");
+                let power_socket = real_power_socket.try_read().map_err(|e| std::io::Error::other(e))?;
                 buf = if power_socket.on { [1u8; 1] } else { [0u8; 1] };
-                stream.write_all(&buf).expect("response write error");
+                let _ = tokio::time::timeout(TIMEOUT, socket.write(&buf)).await?;
             }
             [CMD_TURN_ON] => {
                 println!("handling CMD_TURN_ON");
-                let mut power_socket = real_power_socket.write().expect("power socket lock is poisoned");
+                let mut power_socket = real_power_socket.try_write().map_err(|e| std::io::Error::other(e))?;
                 power_socket.on = true;
-                stream.write_all(&buf).expect("response write error");
+                let _ = tokio::time::timeout(TIMEOUT, socket.write(&buf)).await?;
             }
             [CMD_TURN_OFF] => {
                 println!("handling CMD_TURN_OFF");
-                let mut power_socket = real_power_socket.write().expect("power socket lock is poisoned");
+                let mut power_socket = real_power_socket.try_write().map_err(|e| std::io::Error::other(e))?;
                 power_socket.on = false;
-                stream.write_all(&buf).expect("response write error");
+                let _ = tokio::time::timeout(TIMEOUT, socket.write(&buf)).await?;
             }
             [CMD_GET_POWER] => {
                 println!("handling CMD_GET_POWER");
-                let power_socket = real_power_socket.write().expect("power socket lock is poisoned");
+                let power_socket = real_power_socket.try_read().map_err(|e| std::io::Error::other(e))?;
                 let data_buf = power_socket.power_rate.to_le_bytes();
-                stream.write_all(&data_buf).expect("response write error");
+                let _ = tokio::time::timeout(TIMEOUT, socket.write(&data_buf)).await?;
             }
             _ => {
                 println!("unknown command {} - ignore it", buf[0]);
@@ -71,19 +65,17 @@ fn handle_connection(mut stream: TcpStream, real_power_socket: Arc<RwLock<RealPo
     }
 }
 
-const TIMEOUT: Duration = Duration::from_secs(5);
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let real_power_socket = Arc::new(tokio::sync::RwLock::new(RealPowerSocket { power_rate: 12.0, on: false }));
+    let rt = tokio::runtime::Builder::new_current_thread().enable_all().build()?;
+    rt.block_on(async {
+        let listener = tokio::net::TcpListener::bind(parse_args()).await?;
 
-fn main() {
-    let real_power_socket = Arc::new(RwLock::new(RealPowerSocket { power_rate: 12.0, on: false }));
-    let listener = TcpListener::bind(parse_args()).expect("address 127.0.0.1:10001 should be available for listening");
-    for connection in listener.incoming() {
-        println!("new connection");
-        let real_power_socket = real_power_socket.clone();
-        let stream = connection.expect("connection should not fail");
-        stream.set_write_timeout(Some(TIMEOUT)).expect("set_write_timeout failed");
-        stream.set_nonblocking(true).expect("set_nonblocking call failed");
-        std::thread::spawn(move || {
-            handle_connection(stream, real_power_socket);
-        });
-    }
+        loop {
+            let (socket, _) = listener.accept().await?;
+            println!("new connection");
+            let real_power_socket = real_power_socket.clone();
+            tokio::spawn(async move { handle_connection(socket, real_power_socket).await });
+        }
+    })
 }

smart-house/src/main.rs

@@ -1,6 +1,10 @@
-use smart_house::Thermometer;
+use smart_house::PowerSocket;
+use std::time::Duration;
 
 fn main() {
-    let thermometer = Thermometer::connect("127.0.0.1:10001").unwrap();
-    println!("{}", thermometer.display());
+    let mut power_socket = PowerSocket::connect("127.0.0.1:10001");
+    println!("{}", power_socket.display());
+    std::thread::sleep(Duration::from_secs(2));
+    power_socket.set_on(!power_socket.is_on());
+    println!("{}", power_socket.display());
 }