Remove all unsafe code via pin_project.

Cargo.lock

@@ -1273,6 +1273,7 @@ dependencies = [
  "fswatcher 0.1.0 (git+https://github.com/mgottschlag/fswatcher-rs.git)",
  "futures 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pin-project 0.4.20 (registry+https://github.com/rust-lang/crates.io-index)",
  "range-set 0.0.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "rmp-serde 0.14.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "rusqlite 0.23.1 (registry+https://github.com/rust-lang/crates.io-index)",

Cargo.toml

@@ -22,6 +22,7 @@ range-set = "*"
 fslock = "0.1"
 rusqlite = "0.23"
 thiserror = "1.0"
+pin-project = "0.4"
 
 [dev-dependencies]
 criterion = "0.3"

src/network/client.rs

@@ -8,6 +8,7 @@ use futures::sink::{Sink, SinkExt};
 use futures::stream::{Stream, StreamExt};
 use futures::task::{Context, Poll};
 use log::error;
+use pin_project::pin_project;
 use rmps::Serializer;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
@@ -240,7 +241,7 @@ impl<PacketType, ErrorType> RPCInterface<PacketType, ErrorType> {
         }
 
         RPC {
-            result: Box::pin(receiver),
+            result: receiver,
             timeout: None,
         }
     }
@@ -260,14 +261,17 @@ struct CallData<PacketType, ErrorType> {
     result: oneshot::Sender<Result<PacketType, RPCError<ErrorType>>>,
 }
 
+#[pin_project]
 pub struct RPC<PacketType, ErrorType> {
-    result: Pin<Box<oneshot::Receiver<Result<PacketType, RPCError<ErrorType>>>>>,
-    timeout: Option<Pin<Box<Delay>>>,
+    #[pin]
+    result: oneshot::Receiver<Result<PacketType, RPCError<ErrorType>>>,
+    #[pin]
+    timeout: Option<Delay>,
 }
 
 impl<PacketType, ErrorType> RPC<PacketType, ErrorType> {
     pub fn with_timeout(mut self, timeout: Duration) -> Self {
-        self.timeout = Some(Box::pin(delay_for(timeout)));
+        self.timeout = Some(delay_for(timeout));
         self
     }
 }
@@ -276,11 +280,10 @@ impl<PacketType, ErrorType> Future for RPC<PacketType, 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_.
-        let self_ = unsafe { self.get_unchecked_mut() };
+        let this = self.project();
 
         // Try receiving from the receiver.
-        match Pin::as_mut(&mut self_.result).poll(cx) {
+        match this.result.poll(cx) {
             Poll::Ready(Ok(result)) => return Poll::Ready(result),
             Poll::Ready(Err(_)) => {
                 // The channel was closed.
@@ -290,8 +293,8 @@ impl<PacketType, ErrorType> Future for RPC<PacketType, ErrorType> {
         }
 
         // Nothing received, check the timeout instead.
-        if self_.timeout.is_some() {
-            match Pin::as_mut(&mut self_.timeout.as_mut().unwrap()).poll(cx) {
+        if this.timeout.is_some() {
+            match this.timeout.as_pin_mut().unwrap().poll(cx) {
                 Poll::Ready(()) => return Poll::Ready(Err(RPCError::Timeout)),
                 Poll::Pending => (),
             }

src/network/mod.rs

@@ -49,12 +49,15 @@ pub mod test_utils {
     use futures::stream::Stream;
     use futures::task::Context;
     use futures::task::Poll;
+    use pin_project::pin_project;
     use std::pin::Pin;
     use std::time::Duration;
     use tokio::sync::mpsc;
 
+    #[pin_project]
     pub struct Pipe {
         sender: Pin<Box<mpsc::UnboundedSender<Result<Vec<u8>, PipeError>>>>,
+        #[pin]
         receiver: Pin<Box<mpsc::UnboundedReceiver<Result<Vec<u8>, PipeError>>>>,
     }
 
@@ -103,10 +106,7 @@ pub mod test_utils {
         type Item = Result<Vec<u8>, PipeError>;
 
         fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-            // Safe, as we will not move self_.
-            let self_ = unsafe { self.get_unchecked_mut() };
-
-            Pin::as_mut(&mut self_.receiver).poll_next(cx)
+            self.project().receiver.poll_next(cx)
         }
     }
 

src/network/websocket.rs

@@ -1,3 +1,5 @@
+use std::pin::Pin;
+
 use async_tungstenite::tungstenite::{Error as WsError, Message};
 use async_tungstenite::WebSocketStream;
 use futures::prelude::*;
@@ -5,17 +7,17 @@ use futures::sink::Sink;
 use futures::stream::Stream;
 use futures::task::Context;
 use futures::task::Poll;
-use std::pin::Pin;
+use pin_project::pin_project;
 
+#[pin_project]
 pub struct WebSocketConnection<S> {
-    stream: Pin<Box<WebSocketStream<S>>>,
+    #[pin]
+    stream: WebSocketStream<S>,
 }
 
 impl<S> WebSocketConnection<S> {
-    pub fn new(s: WebSocketStream<S>) -> Self {
-        Self {
-            stream: Box::pin(s),
-        }
+    pub fn new(stream: WebSocketStream<S>) -> Self {
+        Self { stream }
     }
 }
 
@@ -26,31 +28,19 @@ where
     type Error = WsError;
 
     fn poll_ready(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
-        // Safe, as we will not move self_.
-        let self_ = unsafe { self.get_unchecked_mut() };
-
-        Pin::as_mut(&mut self_.stream).poll_ready(cx)
+        self.project().stream.poll_ready(cx)
     }
 
     fn start_send(self: Pin<&mut Self>, item: Vec<u8>) -> Result<(), Self::Error> {
-        // Safe, as we will not move self_.
-        let self_ = unsafe { self.get_unchecked_mut() };
-
-        Pin::as_mut(&mut self_.stream).start_send(Message::Binary(item))
+        self.project().stream.start_send(Message::Binary(item))
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
-        // Safe, as we will not move self_.
-        let self_ = unsafe { self.get_unchecked_mut() };
-
-        Pin::as_mut(&mut self_.stream).poll_flush(cx)
+        self.project().stream.poll_flush(cx)
     }
 
     fn poll_close(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
-        // Safe, as we will not move self_.
-        let self_ = unsafe { self.get_unchecked_mut() };
-
-        Pin::as_mut(&mut self_.stream).poll_close(cx)
+        self.project().stream.poll_close(cx)
     }
 }
 
@@ -64,18 +54,17 @@ where
         self: Pin<&mut Self>,
         cx: &mut Context,
     ) -> Poll<Option<<WebSocketConnection<S> as Stream>::Item>> {
-        // Safe, as we will not move self_.
-        let self_ = unsafe { self.get_unchecked_mut() };
-
-        match Pin::as_mut(&mut self_.stream).poll_next(cx) {
-            Poll::Ready(Some(Ok(Message::Binary(data)))) => Poll::Ready(Some(Ok(data))),
-            Poll::Ready(Some(Ok(_))) => unsafe {
-                // Ignore other message types, try again.
-                Pin::new_unchecked(self_).poll_next(cx)
-            },
-            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(e))),
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Pending => Poll::Pending,
+        let mut this = self.project();
+        loop {
+            match this.stream.as_mut().poll_next(cx) {
+                Poll::Ready(Some(Ok(Message::Binary(data)))) => return Poll::Ready(Some(Ok(data))),
+                Poll::Ready(Some(Ok(_))) => {
+                    // Ignore other message types, try again.
+                }
+                Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
+                Poll::Ready(None) => return Poll::Ready(None),
+                Poll::Pending => return Poll::Pending,
+            }
         }
     }
 }