Avoid cancelling a read call (#144)

Stream read is not cancellation safe, it can then put the
socket in a bad state and reads will get blocked. forever

Instead we run the read in own routine, and then pipe read messages
to the connection loop. which is cancellation safe.

Fixes #143

src/peer/con.rs

@@ -6,7 +6,7 @@ use futures_util::stream::StreamExt;
 use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::time::Instant;
-use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::{Error, Message};
 use url::Url;
 
 const PING_INTERVAL: Duration = Duration::from_secs(20);
@@ -104,9 +104,10 @@ async fn retainer<S: Signer>(
             }
         };
 
-        let (mut write, mut read) = ws.split();
+        let (mut write, read) = ws.split();
         let mut last = Instant::now();
 
+        let mut read = read_stream(read);
         'receive: loop {
             // we check here when was the last time a message was received
             // from the relay. we expect to receive PONG answers (because we
@@ -129,7 +130,15 @@ async fn retainer<S: Signer>(
                         break 'receive;
                     }
                 },
-                Some(message) =  read.next() => {
+                message = read.recv() => {
+                    let message = match message {
+                        None=> {
+                            log::debug!("read stream ended")  ;
+                            break 'receive;
+                        },
+                        Some(message) => message,
+                    };
+
                     // we take a note with when a message was received
                     last = Instant::now();
                     log::trace!("received a message from relay");
@@ -158,3 +167,27 @@ async fn retainer<S: Signer>(
         log::info!("retrying connection");
     }
 }
+
+fn read_stream(
+    mut stream: futures_util::stream::SplitStream<
+        tokio_tungstenite::WebSocketStream<
+            tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
+        >,
+    >,
+) -> mpsc::Receiver<Result<Message, Error>> {
+    let (sender, receiver) = mpsc::channel(1);
+    tokio::spawn(async move {
+        loop {
+            match stream.next().await {
+                None => return,
+                Some(result) => {
+                    if sender.send(result).await.is_err() {
+                        return;
+                    }
+                }
+            }
+        }
+    });
+
+    receiver
+}

src/peer/mod.rs

@@ -259,7 +259,7 @@ where
                     // otherwise, we clear up the payload
                     // and set the error instead
                     envelope.payload = None;
-                    let mut e = envelope.mut_error();
+                    let e = envelope.mut_error();
                     e.code = err.code();
                     e.message = err.to_string();
                 }

src/relay/api.rs

@@ -411,7 +411,7 @@ impl<M: Metrics> Stream<M> {
         let mut resp = Envelope::new();
         resp.uid = envelope.uid;
         resp.destination = Some((&self.id).into()).into();
-        let mut e = resp.mut_error();
+        let e = resp.mut_error();
         e.message = err.to_string();
 
         let bytes = match resp.write_to_bytes() {

src/types/mod.rs

@@ -7,7 +7,7 @@ use std::time::{Duration, SystemTime};
 
 include!(concat!(env!("OUT_DIR"), "/protos/mod.rs"));
 
-pub use peer::*;
+pub use peer::Backlog;
 pub use types::*;
 
 #[derive(thiserror::Error, Debug)]