Implement client RPC logic.

src/network/client.rs

@@ -1,57 +1,106 @@
+use std::collections::BTreeMap;
+use std::fmt;
 use std::time::Duration;
 
-use async_tungstenite::{tokio::connect_async, tungstenite::Message, WebSocketStream};
 use core::pin::Pin;
 use futures::future::{self, Either, Future};
+use futures::sink::{Sink, SinkExt};
+use futures::stream::{Stream, StreamExt};
 use futures::task::{Context, Poll};
-use tokio::stream::Stream;
-use tokio::stream::StreamExt;
+use rmps::Serializer;
+use serde::de::DeserializeOwned;
+use serde::Serialize;
 use tokio::sync::{mpsc, oneshot};
 use tokio::time::{delay_for, Delay};
-use url::Url;
 
-use super::packet::IncomingPacket;
+use super::idalloc::IdAllocator;
+use super::packet::{Packet, SERVER_EVENT_ID};
 
 pub fn create<StreamType, ErrorType, PacketType>(
     stream: StreamType,
 ) -> (
-    RPCInterface<StreamType, PacketType>,
+    RPCInterface<PacketType, ErrorType>,
     ServerEventStream<PacketType>,
 )
 where
-    StreamType: Stream<Item = Result<Vec<u8>, ErrorType>> + Unpin + Send + 'static,
+    StreamType: Stream<Item = Result<Vec<u8>, ErrorType>>
+        + Sink<Vec<u8>, Error = ErrorType>
+        + Unpin
+        + Send
+        + 'static,
+    PacketType: DeserializeOwned + Serialize + Send + 'static,
+    ErrorType: fmt::Debug + Send + 'static,
 {
-    let (cmd_send, cmd_receive) = mpsc::channel(1);
+    // TODO: Does this channel have to be unbounded? We can just drop the sender, and the receiver
+    // will notice.
+    let (cmd_send, cmd_receive) = mpsc::unbounded_channel();
+    let (event_send, event_receive) = mpsc::channel(1);
     tokio::spawn(async move {
-        // TODO: Remove type annotations.
-        client_connection_task::<StreamType, ErrorType, PacketType>(stream, cmd_receive).await;
+        client_connection_task(stream, cmd_receive, event_send).await;
     });
-    // TODO
-    panic!("Not yet implemented.");
-}
 
-enum ClientTaskCommand {
-    Close,
-    SendPackets,
+    (
+        RPCInterface { cmd_send },
+        ServerEventStream {
+            packets: Box::pin(event_receive),
+        },
+    )
 }
 
-struct OutgoingCallMap {
-    // TODO
+enum ClientTaskCommand<PacketType, ErrorType> {
+    Close,
+    SendPacket(CallData<PacketType, ErrorType>),
 }
 
 async fn client_connection_task<StreamType, ErrorType, PacketType>(
     mut stream: StreamType,
-    mut cmd_receive: mpsc::Receiver<ClientTaskCommand>,
+    mut cmd_receive: mpsc::UnboundedReceiver<ClientTaskCommand<PacketType, ErrorType>>,
+    mut event_send: mpsc::Sender<PacketType>,
 ) where
-    StreamType: Stream<Item = Result<Vec<u8>, ErrorType>> + Unpin,
+    PacketType: DeserializeOwned + Serialize,
+    StreamType:
+        Stream<Item = Result<Vec<u8>, ErrorType>> + Sink<Vec<u8>, Error = ErrorType> + Unpin,
+    ErrorType: fmt::Debug,
 {
+    let mut free_ids = IdAllocator::new();
+    let mut calls_in_progress = BTreeMap::<u32, oneshot::Sender<Result<PacketType, RPCError<ErrorType>>>>::new();
+
+    // TODO: Refactor this function.
+
     let mut connection_closed = false;
     loop {
         match future::select(Box::pin(stream.next()), Box::pin(cmd_receive.recv())).await {
             Either::Left((packet, _cmd_future)) => {
                 match packet {
-                    Some(data) => {
-                        // TODO
+                    Some(Ok(data)) => {
+                        match rmps::decode::from_slice(&data) as Result<Packet<PacketType>, _> {
+                            Ok(deserialized) => {
+                                if deserialized.id == SERVER_EVENT_ID {
+                                    // We received an asynchronous event from the server.
+                                    event_send.send(deserialized.payload).await.ok();
+                                } else {
+                                    // We received a response to a call - find the call and forward
+                                    // the reply.
+                                    if let Some(call) = calls_in_progress.remove(&deserialized.id) {
+                                        call.send(Ok(deserialized.payload)).ok();
+                                    } else {
+                                        // TODO: Use proper logging functionality.
+                                        eprintln!("Received a reply for an unknown call ({:?})!", deserialized.id);
+                                    }
+                                }
+                            }
+                            Err(e) => {
+                                // We received an invalid packet. We cannot determine for which
+                                // call the packet was, so we just ignore it. The corresponding
+                                // call will freeze until the connection is dropped.
+                                // TODO: Use proper logging functionality.
+                                eprintln!("Invalid packet received: {:?}", e);
+                            }
+                        };
+                    }
+                    Some(Err(e)) => {
+                        // TODO: Use proper logging functionality.
+                        eprintln!("Network error: {:?}", e);
                     }
                     None => {
                         // Stream closed. We still need to return errors for any
@@ -68,14 +117,38 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
                     Some(ClientTaskCommand::Close) => {
                         // The RPC interface for this connection was dropped, so
                         // we just exit the loop and close the connection.
-
-                        // We need to return errors for all pending packets
-                        // first, though.
-                        // TODO
                         break;
                     }
-                    Some(ClientTaskCommand::SendPackets) => {
-                        // TODO
+                    Some(ClientTaskCommand::SendPacket(call)) => {
+                        // Add an ID for the outgoing packet.
+                        let id = free_ids.alloc();
+                        let packet = Packet {
+                            id: id,
+                            response: false,
+                            payload: call.data,
+                        };
+                        let mut serialized = Vec::new();
+                        packet
+                            .serialize(
+                                &mut Serializer::new(&mut serialized)
+                                    .with_struct_map()
+                                    .with_string_variants(),
+                            )
+                            .unwrap();
+
+                        // Send the serialized packet.
+                        // TODO: Send without flushing?
+                        match stream.send(serialized).await {
+                            Ok(()) => {
+                                calls_in_progress.insert(id, call.result);
+                            }
+                            Err(e) => {
+                                // We potentially leak the ID here, but that is
+                                // okay as the state of the call is not well
+                                // defined anymore.
+                                call.result.send(Err(RPCError::Stream(e))).ok();
+                            }
+                        }
                     }
                     None => {
                         // This should never happen.
@@ -86,6 +159,11 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
         };
     }
 
+    // Return errors for all pending packets as there will be no responses.
+    for (_, result) in calls_in_progress {
+        result.send(Err(RPCError::Closed)).ok();
+    }
+
     if connection_closed {
         // The stream was closed, but the command channel was not. We still need
         // to return errors for any packet sent by the client, so do not exit
@@ -99,17 +177,19 @@ async fn client_connection_task<StreamType, ErrorType, PacketType>(
     }
 }
 
-async fn client_connection_closed(mut cmd_receive: mpsc::Receiver<ClientTaskCommand>) {
+async fn client_connection_closed<PacketType, ErrorType>(
+    mut cmd_receive: mpsc::UnboundedReceiver<ClientTaskCommand<PacketType, ErrorType>>,
+) {
     loop {
         match cmd_receive.recv().await {
             Some(ClientTaskCommand::Close) => {
                 // The connection was already closed.
                 break;
             }
-            Some(ClientTaskCommand::SendPackets) => {
+            Some(ClientTaskCommand::SendPacket(call)) => {
                 // The connection was closed, so return an error for any
                 // packet sent.
-                // TODO
+                call.result.send(Err(RPCError::Closed)).ok();
             }
             None => {
                 // This should never happen.
@@ -119,22 +199,47 @@ async fn client_connection_closed(mut cmd_receive: mpsc::Receiver<ClientTaskComm
     }
 }
 
-pub struct RPCInterface<StreamType, PacketType> {
-    stream: StreamType,
-    _packet: Option<PacketType>, // TODO: Remove.
+pub struct RPCInterface<PacketType, ErrorType> {
+    cmd_send: mpsc::UnboundedSender<ClientTaskCommand<PacketType, ErrorType>>,
 }
 
-impl<StreamType, PacketType> RPCInterface<StreamType, PacketType> {
-    fn call(_call: &PacketType) -> RPC<StreamType, PacketType> {
-        // TODO
-        panic!("Not yet implemented.");
+impl<PacketType, ErrorType> RPCInterface<PacketType, ErrorType> {
+    pub async fn call(&mut self, call: PacketType) -> RPC<PacketType, ErrorType> {
+        let (sender, receiver) = oneshot::channel();
+        let call_data = CallData {
+            data: call,
+            result: sender,
+        };
+        if let Err(_) = self.cmd_send.send(ClientTaskCommand::SendPacket(call_data)) {
+            panic!("could not send packet to connection task");
+        }
+
+        RPC {
+            result: Box::pin(receiver),
+            timeout: None,
+        }
     }
 }
 
+impl<PacketType, ErrorType> Drop for RPCInterface<PacketType, ErrorType> {
+    fn drop(&mut self) {
+        // Stop the connection loop and close the connection.
+        if let Err(_) = self.cmd_send.send(ClientTaskCommand::Close) {
+            panic!("could not send close signal to connection task");
+        }
+    }
+}
+
+struct CallData<PacketType, ErrorType> {
+    data: PacketType,
+    result: oneshot::Sender<Result<PacketType, RPCError<ErrorType>>>,
+}
+
 pub struct RPC<PacketType, ErrorType> {
-    result: Pin<Box<oneshot::Receiver<Result<IncomingPacket<PacketType>, ErrorType>>>>,
+    result: Pin<Box<oneshot::Receiver<Result<PacketType, RPCError<ErrorType>>>>>,
     timeout: Option<Pin<Box<Delay>>>,
 }
+
 impl<PacketType, ErrorType> RPC<PacketType, ErrorType> {
     pub fn with_timeout(mut self, timeout: Duration) -> Self {
         self.timeout = Some(Box::pin(delay_for(timeout)));
@@ -143,7 +248,7 @@ impl<PacketType, ErrorType> RPC<PacketType, ErrorType> {
 }
 
 impl<PacketType, ErrorType> Future for RPC<PacketType, ErrorType> {
-    type Output = Result<IncomingPacket<PacketType>, RPCError<ErrorType>>;
+    type Output = Result<PacketType, RPCError<ErrorType>>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
         // Safe, as we will not move self_.
@@ -151,7 +256,7 @@ impl<PacketType, ErrorType> Future for RPC<PacketType, ErrorType> {
 
         // Try receiving from the receiver.
         match Pin::as_mut(&mut self_.result).poll(cx) {
-            Poll::Ready(Ok(result)) => return Poll::Ready(result.map_err(|x| RPCError::Stream(x))),
+            Poll::Ready(Ok(result)) => return Poll::Ready(result),
             Poll::Ready(Err(_)) => {
                 // The channel was closed.
                 return Poll::Ready(Err(RPCError::Closed));
@@ -173,11 +278,11 @@ impl<PacketType, ErrorType> Future for RPC<PacketType, ErrorType> {
 
 // TODO: Do we even need this type?
 pub struct ServerEventStream<PacketType> {
-    packets: Pin<Box<mpsc::Receiver<IncomingPacket<PacketType>>>>,
+    packets: Pin<Box<mpsc::Receiver<PacketType>>>,
 }
 
 impl<PacketType> Stream for ServerEventStream<PacketType> {
-    type Item = IncomingPacket<PacketType>;
+    type Item = PacketType;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         // Safe, as we will not move self_.
@@ -187,19 +292,20 @@ impl<PacketType> Stream for ServerEventStream<PacketType> {
     }
 }
 
+#[derive(Debug)]
 pub enum RPCError<StreamError> {
     Closed,
     Timeout,
-    Stream(StreamError)
+    Stream(StreamError),
 }
 
-pub async fn low_level_integration_test_client(address: String) {
-    use super::websocket::WebSocketConnection;
+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);
+    let (rpc_interface, server_events) = create::<_, _, u32>(conn);*/
     // TODO
     panic!("Not yet implemented.");
 }

src/network/packet.rs

@@ -27,26 +27,10 @@
 //! a way that no such temporal information is necessary to process the events.
 //! TODO: Example to demonstrate the problem.
 
-//use futures::sink::Sink;
-//use futures::stream::Stream;
-//use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 
 pub const SERVER_EVENT_ID: u32 = u32::max_value();
 
-//pub trait PacketConnection: Sink<&Self::Payload> + Stream<IncomingPacket<Payload>> {
-//    type Payload: Serialize + DeserializeOwned;
-//}
-
-/// Type which holds a byte vector received via the network and which allows access to the
-/// deserialized packet.
-///
-/// The type can be used for efficient zero-copy deserialization of large packets.
-pub struct IncomingPacket<Payload> {
-    // TODO
-    _packet: Payload,
-}
-
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
 pub struct Packet<Payload> {
     pub id: u32,

src/network/server.rs

@@ -1,15 +1,10 @@
-use super::packet::IncomingPacket;
-
 // Trait that is implemented by any RPC server to process incoming calls.
 trait RPCServer {
     type PacketType;
     type NetworkError;
 
-    fn incoming_call<Response>(
-        &mut self,
-        call: IncomingPacket<Self::PacketType>,
-        response: Response,
-    ) where
+    fn incoming_call<Response>(&mut self, call: Self::PacketType, response: Response)
+    where
         Response: RPCResponse<PacketType = Self::PacketType, NetworkError = Self::NetworkError>;
 }
 
@@ -29,11 +24,10 @@ pub struct LowLevelIntegrationTestServer {
 impl LowLevelIntegrationTestServer {
     pub async fn start(_bind_address: &str) -> LowLevelIntegrationTestServer {
         // TODO
-        LowLevelIntegrationTestServer{}
+        LowLevelIntegrationTestServer {}
     }
 
     pub async fn stop(self) {
         // TODO
     }
 }
-