Add first parts of the server side of the RPC protocol.

src/bin/low_level_integration_test.rs → src/bin/rpc_integration_test.rs

@@ -1,8 +1,7 @@
 use structopt::StructOpt;
 use tokio::signal;
 
-use twfss::network::client::low_level_integration_test_client;
-use twfss::network::server::LowLevelIntegrationTestServer;
+use twfss::network::integration_test::{rpc_integration_test_client, RpcIntegrationTestServer};
 
 #[derive(Debug, StructOpt)]
 #[structopt(about = "Integration tester for the low-level packet protocol.")]
@@ -15,7 +14,7 @@ enum Options {
 async fn main() {
     match Options::from_args() {
         Options::Server { bind_address } => {
-            let server = LowLevelIntegrationTestServer::start(&bind_address).await;
+            let server = RpcIntegrationTestServer::start(&bind_address).await;
             println!("Listening at {}...", bind_address);
             signal::ctrl_c().await.unwrap();
             println!("Ctrl+C received, stopping...");
@@ -23,7 +22,7 @@ async fn main() {
         }
         Options::Client { connect_address } => {
             println!("Connecting to {}...", connect_address);
-            low_level_integration_test_client(connect_address).await;
+            rpc_integration_test_client(&connect_address).await;
         }
     }
 }

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/network/client.rs

@@ -435,14 +435,3 @@ mod tests {
     // - Test whether the client correctly closes the connection and whether futures return an error
     // in this case.
 }
-
-pub async fn low_level_integration_test_client(_address: String) {
-    /*use super::websocket::WebSocketConnection;
-
-    let url = Url::parse(&address).unwrap();
-    let (mut ws_stream, _) = connect_async(url).await.unwrap();
-    let mut conn = WebSocketConnection::new(ws_stream);
-    let (rpc_interface, server_events) = create::<_, _, u32>(conn);*/
-    // TODO
-    panic!("Not yet implemented.");
-}

src/network/integration_test.rs

@@ -0,0 +1,64 @@
+use std::net::SocketAddr;
+
+use async_tungstenite::accept_async;
+use async_tungstenite::tokio::TokioAdapter;
+use futures::SinkExt;
+use log::*;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::stream::StreamExt;
+
+pub async fn rpc_integration_test_client(_address: &str) {
+    /*use super::websocket::WebSocketConnection;
+
+    let url = Url::parse(&address).unwrap();
+    let (mut ws_stream, _) = connect_async(url).await.unwrap();
+    let mut conn = WebSocketConnection::new(ws_stream);
+    let (rpc_interface, server_events) = create::<_, _, u32>(conn);*/
+    // TODO
+    panic!("Not yet implemented.");
+}
+
+pub struct RpcIntegrationTestServer {
+    // TODO
+}
+
+impl RpcIntegrationTestServer {
+    pub async fn start(_bind_address: &str) -> RpcIntegrationTestServer {
+        // 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));
+        }
+        RpcIntegrationTestServer {}
+    }
+
+    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]

tests/rpc.rs

@@ -0,0 +1,21 @@
+use std::net::SocketAddr;
+
+use async_tungstenite::accept_async;
+use async_tungstenite::tokio::TokioAdapter;
+use futures::SinkExt;
+use log::*;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::stream::StreamExt;
+
+use twfss::network::integration_test::{rpc_integration_test_client, RpcIntegrationTestServer};
+
+#[tokio::test]
+async fn test_rpc() {
+    /*const ADDRESS: &str = "localhost:12345";
+    let server = RpcIntegrationTestServer::start(ADDRESS).await;
+
+    // TODO: Error handling, usable asserts.
+    rpc_integration_test_client(&ADDRESS).await;
+
+    server.stop().await;*/
+}