WIP

.rustfmt.toml

@@ -0,0 +1 @@
+edition = "2018"

Cargo.lock

@@ -1283,6 +1283,7 @@ dependencies = [
  "thiserror 1.0.19 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio 0.2.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "url 2.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "uuid 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1336,6 +1337,15 @@ name = "utf-8"
 version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "uuid"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "rand 0.7.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.111 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "vcpkg"
 version = "0.2.9"
@@ -1628,6 +1638,7 @@ dependencies = [
 "checksum unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c"
 "checksum url 2.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "829d4a8476c35c9bf0bbce5a3b23f4106f79728039b726d292bb93bc106787cb"
 "checksum utf-8 0.7.5 (registry+https://github.com/rust-lang/crates.io-index)" = "05e42f7c18b8f902290b009cde6d651262f956c98bc51bca4cd1d511c9cd85c7"
+"checksum uuid 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)" = "9fde2f6a4bea1d6e007c4ad38c6839fa71cbb63b6dbf5b595aa38dc9b1093c11"
 "checksum vcpkg 0.2.9 (registry+https://github.com/rust-lang/crates.io-index)" = "55d1e41d56121e07f1e223db0a4def204e45c85425f6a16d462fd07c8d10d74c"
 "checksum vec_map 0.8.2 (registry+https://github.com/rust-lang/crates.io-index)" = "f1bddf1187be692e79c5ffeab891132dfb0f236ed36a43c7ed39f1165ee20191"
 "checksum version_check 0.9.2 (registry+https://github.com/rust-lang/crates.io-index)" = "b5a972e5669d67ba988ce3dc826706fb0a8b01471c088cb0b6110b805cc36aed"

Cargo.toml

@@ -23,6 +23,7 @@ fslock = "0.1"
 rusqlite = "0.23"
 thiserror = "1.0"
 pin-project = "0.4"
+uuid = { version = "0.8", features = ["serde", "v4"] }
 
 [dev-dependencies]
 criterion = "0.3"

coverage.sh

@@ -0,0 +1,7 @@
+#/bin/bash
+set -e
+export CARGO_INCREMENTAL=0
+export RUSTFLAGS="-Zprofile -Ccodegen-units=1 -Cinline-threshold=0 -Clink-dead-code -Coverflow-checks=off -Zno-landing-pads"
+cargo +nightly build
+cargo +nightly test
+grcov ./target/debug/ -s . -t html --llvm --branch --ignore-not-existing -o ./target/debug/coverage/

src/bin/multi-user-wrapper.rs

@@ -0,0 +1,5 @@
+#[tokio::main]
+async fn main() {
+    env_logger::init();
+    // TODO
+}

src/bin/server.rs

@@ -9,6 +9,8 @@ use log::*;
 use tokio::net::{TcpListener, TcpStream};
 use tokio::stream::StreamExt;
 
+use twfss::network::server::{RPCResponse, RPCServer};
+use twfss::protocol::PacketPayload;
 use twfss::{paths, Database};
 
 #[tokio::main]
@@ -63,3 +65,15 @@ async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
         }
     }
 }
+
+struct ServerConnection {
+    // TODO
+}
+
+impl RPCServer for ServerConnection {
+    type PacketType = PacketPayload;
+
+    fn incoming_call(&mut self, call: Self::PacketType, response: RPCResponse<Self::PacketType>) {
+        // TODO
+    }
+}

src/database.rs

@@ -1,10 +1,10 @@
 use std::ffi::OsString;
 
 use rusqlite::Connection;
+//use uuid::Uuid;
 
 pub struct Database {
     db: Connection,
-    // TODO
 }
 
 impl Database {
@@ -24,3 +24,7 @@ impl Database {
         Vec::new()
     }
 }
+
+pub struct SynchronizedDirectoryInfo {
+    //root_id: Uuid,
+}

src/lib.rs

@@ -2,7 +2,6 @@
 extern crate rmp_serde as rmps;
 
 use std::ffi::OsString;
-use std::io;
 
 use thiserror::Error;
 

src/network/integration_test.rs

@@ -0,0 +1,48 @@
+use log::*;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::stream::StreamExt;
+
+pub struct LowLevelIntegrationTestServer {
+    // TODO
+}
+
+impl LowLevelIntegrationTestServer {
+    pub async fn start(_bind_address: &str) -> LowLevelIntegrationTestServer {
+        // TODO
+        let addr = "127.0.0.1:12345";
+        let mut listener = TcpListener::bind(&addr)
+            .await
+            .expect("cannot bind to server port");
+        let mut incoming = listener.incoming();
+        info!("Listening on {}.", addr);
+
+        while let Some(Ok(stream)) = incoming.next().await {
+            let peer = stream
+                .peer_addr()
+                .expect("connected streams should have a peer address");
+
+            tokio::spawn(Self::accept_connection(peer, stream));
+        }
+        LowLevelIntegrationTestServer {}
+    }
+
+    pub async fn stop(self) {
+        // TODO
+    }
+
+    async fn accept_connection(peer: SocketAddr, stream: TcpStream) {
+        let mut ws_stream = accept_async(TokioAdapter(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");
+            }
+        }
+    }
+}

src/network/mod.rs

@@ -29,6 +29,7 @@
 
 pub mod client;
 pub mod idalloc;
+pub mod integration_test;
 pub mod server;
 pub mod websocket;
 

src/network/server.rs

@@ -1,33 +1,194 @@
-// Trait that is implemented by any RPC server to process incoming calls.
-trait RPCServer {
+use std::fmt;
+
+use futures::sink::{Sink, SinkExt};
+use futures::stream::{Stream, StreamExt};
+use log::{error, info};
+use rmps::Serializer;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use tokio::sync::mpsc;
+
+use super::{Packet, SERVER_EVENT_ID};
+
+/// Trait that is implemented by any RPC server to process incoming calls.
+pub trait RPCServer {
     type PacketType;
-    type NetworkError;
 
-    fn incoming_call<Response>(&mut self, call: Self::PacketType, response: Response)
-    where
-        Response: RPCResponse<PacketType = Self::PacketType, NetworkError = Self::NetworkError>;
+    fn incoming_call(&mut self, call: Self::PacketType, response: RPCResponse<Self::PacketType>);
 }
 
 // TODO: Document that types implementing this trait should also implement Drop and should return
 // an error if the type is dropped without send() being called.
-trait RPCResponse {
-    type PacketType;
-    type NetworkError;
+pub struct RPCResponse<PacketType> {
+    send_packet: mpsc::UnboundedSender<Packet<PacketType>>,
+    id: u32,
+    error_value: Option<PacketType>,
+}
 
-    fn send(self, packet: &Self::PacketType) -> Result<(), Self::NetworkError>;
+impl<PacketType> RPCResponse<PacketType>
+where
+    PacketType: Serialize,
+{
+    pub fn send(self, packet: PacketType) {
+        // We ignore errors here, as they mean that the connection was just closed.
+        self.send_packet
+            .send(Packet {
+                payload: packet,
+                id: self.id,
+            })
+            .ok();
+    }
 }
 
-pub struct LowLevelIntegrationTestServer {
-    // TODO
+impl<PacketType> Drop for RPCResponse<PacketType> {
+    fn drop(&mut self) {
+        // Send an error response?
+        self.send_packet
+            .send(Packet {
+                payload: self.error_value.take().unwrap(),
+                id: self.id,
+            })
+            .ok();
+    }
 }
 
-impl LowLevelIntegrationTestServer {
-    pub async fn start(_bind_address: &str) -> LowLevelIntegrationTestServer {
-        // TODO
-        LowLevelIntegrationTestServer {}
+/// Wraps an `RPCServer` and adds the network protocol implementation around it.
+pub struct RPCServerWrapper<Server: RPCServer, Connection> {
+    server: Server,
+    connection: Connection,
+    server_events: ServerEventStream<Server::PacketType>,
+    send_packet: mpsc::UnboundedSender<Packet<Server::PacketType>>,
+    sent_packets: mpsc::UnboundedReceiver<Packet<Server::PacketType>>,
+    error_value: Server::PacketType,
+}
+
+impl<Server: RPCServer, Connection, ConnectionError> RPCServerWrapper<Server, Connection>
+where
+    Server::PacketType: DeserializeOwned + Serialize + Clone + Send + 'static,
+    Connection: Stream<Item = Vec<u8>> + Sink<Vec<u8>, Error = ConnectionError> + Unpin,
+    ConnectionError: fmt::Debug,
+{
+    pub fn new(
+        server: Server,
+        connection: Connection,
+        server_events: ServerEventStream<Server::PacketType>,
+        error_value: Server::PacketType,
+    ) -> Self {
+        let (send_packet, sent_packets) = mpsc::unbounded_channel();
+        Self {
+            server,
+            connection,
+            server_events,
+            send_packet,
+            sent_packets,
+            error_value,
+        }
     }
 
-    pub async fn stop(self) {
-        // TODO
+    pub async fn run(&mut self) {
+        loop {
+            tokio::select! {
+                packet = &mut self.connection.next() => {
+                    match packet {
+                        Some(data) => {
+                            if !self.packet_received(data) {
+                                // Disconnect on errors.
+                                break;
+                            }
+                        }
+                        None => {
+                            // Stream closed.
+                            println!("Disconnected.");
+                            break;
+                        }
+                    }
+                }
+                cmd = &mut self.sent_packets.next() => {
+                    match cmd {
+                        Some(packet) => {
+                            if !self.send_packet(packet).await {
+                                // Disconnect on errors.
+                                break;
+                            }
+                        }
+                        None => {
+                            // This should never happen.
+                            break;
+                        }
+                    }
+                }
+                cmd = &mut self.server_events.receive.next() => {
+                    match cmd {
+                        Some(packet) => {
+                            if !self.send_packet(Packet{payload: packet, id: SERVER_EVENT_ID}).await {
+                                // Disconnect on errors.
+                                break;
+                            }
+                        }
+                        None => {
+                            // This should never happen.
+                            break;
+                        }
+                    }
+                }
+            }
+        }
     }
+
+    fn packet_received(&mut self, packet: Vec<u8>) -> bool {
+        match rmps::decode::from_slice(&packet) as Result<Packet<Server::PacketType>, _> {
+            Ok(deserialized) => {
+                if deserialized.id == SERVER_EVENT_ID {
+                    // Clients must never send server events.
+                    info!("Client sent SERVER_EVENT_ID!");
+                    return false;
+                } else {
+                    let response = RPCResponse {
+                        send_packet: self.send_packet.clone(),
+                        id: deserialized.id,
+                        error_value: Some(self.error_value.clone()),
+                    };
+                    self.server.incoming_call(deserialized.payload, response);
+                }
+            }
+            Err(e) => {
+                error!("Invalid packet received: {:?}", e);
+                return false;
+            }
+        };
+        return true;
+    }
+
+    async fn send_packet(&mut self, packet: Packet<Server::PacketType>) -> bool {
+        let mut serialized = Vec::new();
+        packet
+            .serialize(
+                &mut Serializer::new(&mut serialized)
+                    .with_struct_map()
+                    .with_string_variants(),
+            )
+            .unwrap();
+        match self.connection.send(serialized).await {
+            Ok(()) => true,
+            Err(e) => {
+                info!("Could not send packet: {:?}", e);
+                false
+            }
+        }
+    }
+}
+
+pub fn server_events<PacketType>() -> (ServerEventSink<PacketType>, ServerEventStream<PacketType>) {
+    let (send, receive) = mpsc::unbounded_channel();
+    (ServerEventSink { send }, ServerEventStream { receive })
+}
+
+/// Stream of asynchronous packets sent to the clients.
+#[derive(Clone)]
+pub struct ServerEventSink<PacketType> {
+    send: mpsc::UnboundedSender<PacketType>,
+}
+
+pub struct ServerEventStream<PacketType> {
+    receive: mpsc::UnboundedReceiver<PacketType>,
 }

src/network/websocket.rs

@@ -5,8 +5,7 @@ use async_tungstenite::WebSocketStream;
 use futures::prelude::*;
 use futures::sink::Sink;
 use futures::stream::Stream;
-use futures::task::Context;
-use futures::task::Poll;
+use futures::task::{Context, Poll};
 use pin_project::pin_project;
 
 #[pin_project]

src/online_test.rs

@@ -0,0 +1,12 @@
+trait OnlineTest {
+    // TODO
+}
+
+/// Tests whether the system is online by connecting the specified address.
+struct GenericOnlineTest {
+    // TODO
+}
+
+impl OnlineTest for GenericOnlineTest {
+    // TODO
+}