Switch to async-std and add a websocket server.

Cargo.toml

@@ -11,3 +11,8 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 dirs = "2.0"
 byteorder = "1.3"
+async-std = "1.0"
+async-tungstenite = { version = "0.3", features = ["async-std-runtime"] }
+futures = "0.3"
+env_logger = "0.7"
+log = "0.4"

src/bin/client.rs

@@ -1,8 +1,10 @@
 use std::fs::create_dir_all;
-use std::io::{self, Read, Write};
-use std::os::unix::net::{UnixListener, UnixStream};
+use std::io;
 use std::sync::{Arc, Mutex};
-use std::thread;
+
+use async_std::os::unix::net::{UnixListener, UnixStream};
+use async_std::prelude::*;
+use async_std::task;
 
 use twfss::{Database, SynchronizedDirectory};
 
@@ -27,6 +29,10 @@ fn db_path() -> String {
 }
 
 fn main() {
+    task::block_on(run());
+}
+
+async fn run() {
     // Create the data directories if necessary.
     create_dir_all(config_dir()).unwrap();
     create_dir_all(data_dir()).unwrap();
@@ -47,21 +53,20 @@ fn main() {
 
     // Listen for CLI commands.
     // TODO: Graceful shutdown on errors.
-    let listener = UnixListener::bind(cli::socket_path()).unwrap();
-    for stream in listener.incoming() {
+    let listener = UnixListener::bind(cli::socket_path()).await.unwrap();
+    let mut incoming = listener.incoming();
+    while let Some(stream) = incoming.next().await {
         match stream {
             Ok(stream) => {
-                thread::spawn(move || {
-                    let mut stream = stream;
-                    match handle_cli_client(&mut stream) {
-                        Ok(()) => {}
-                        Err(e) => {
-                            // Log error and try to send it to the stream.
-                            // TODO
-                            write!(stream, "Error: {:?}", e).ok();
-                        }
-                    };
-                });
+                let mut stream = stream;
+                match handle_cli_client(&mut stream).await {
+                    Ok(()) => {}
+                    Err(e) => {
+                        // Log error and try to send it to the stream.
+                        // TODO
+                        write!(stream, "Error: {:?}", e).await.ok();
+                    }
+                };
             }
             Err(err) => {
                 eprintln!("Error while listening on the local unix socket: {:?}", err);
@@ -71,9 +76,9 @@ fn main() {
     }
 }
 
-fn handle_cli_client(stream: &mut UnixStream) -> Result<(), Error> {
+async fn handle_cli_client(stream: &mut UnixStream) -> Result<(), Error> {
     let mut request = String::new();
-    stream.read_to_string(&mut request)?;
+    stream.read_to_string(&mut request).await?;
 
     let options: cli::Options = serde_json::from_str(&request)?;
     let _verbose = options.verbose;

src/bin/server.rs

@@ -1,3 +1,66 @@
+use async_std::net::{SocketAddr, TcpListener, TcpStream};
+use async_std::task;
+use async_tungstenite::accept_async;
+use futures::{SinkExt, StreamExt};
+use log::*;
+
+fn config_dir() -> String {
+    format!(
+        "{}/.config/twfss",
+        dirs::home_dir().unwrap().to_str().unwrap().to_owned()
+    )
+}
+
+fn data_dir() -> String {
+    format!(
+        "{}/.local/share/twfss-server",
+        dirs::home_dir().unwrap().to_str().unwrap().to_owned()
+    )
+}
+
+fn db_path() -> String {
+    format!("{}/server.db", data_dir())
+}
+
 fn main() {
+    task::block_on(run());
+}
+
+async fn run() {
+    env_logger::init();
+
+    // Read configuration.
+    // TODO
+
+    // Start a thread that periodically checks whether the maximum database size
+    // has been exceeded, and if has, whether a directory has not been used for
+    // a long time. In that case, the file database for that directory is deleted.
     // TODO
+
+    // Start the websocket server and listen for connections.
+    let addr = "127.0.0.1:1199";
+    let listener = TcpListener::bind(&addr).await.unwrap();
+    info!("Listening on {}.", addr);
+
+    while let Ok((stream, _)) = listener.accept().await {
+        let peer = stream
+            .peer_addr()
+            .expect("connected streams should have a peer address");
+
+        async_std::task::spawn(accept_connection(peer, stream));
+    }
+}
+
+async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
+    let mut ws_stream = accept_async(stream).await.expect("Failed to accept");
+
+    info!("New WebSocket connection from {}.", peer);
+
+    while let Some(msg) = ws_stream.next().await {
+        let msg = msg.expect("Failed to get request");
+        // TODO
+        if msg.is_text() || msg.is_binary() {
+            ws_stream.send(msg).await.expect("Failed to send response");
+        }
+    }
 }