move pong handler to a lower level (#135)

src/peer/con.rs

@@ -72,6 +72,7 @@ impl Connection {
         let pinger = connection.writer();
         tokio::spawn(async move {
             loop {
+                log::debug!("sending a ping");
                 if let Err(err) = pinger.write(Message::Ping(Vec::default())).await {
                     log::error!("ping error: {}", err);
                 }
@@ -133,6 +134,10 @@ async fn retainer<S: Signer>(
                     last = Instant::now();
                     log::trace!("received a message from relay");
                     let message = match message {
+                        Ok(Message::Pong(_)) => {
+                            log::debug!("received a pong");
+                            continue 'receive;
+                        }
                         Ok(message) => message,
                         Err(err) => {
                             // todo: those errors probably mean we need to re-connect

src/peer/mod.rs

@@ -373,15 +373,14 @@ where
         }
     }
 
-    fn parse(&self, msg: Message) -> Result<Option<Envelope>, PeerError> {
+    fn parse(&self, msg: Message) -> Result<Envelope, PeerError> {
         let bytes = match msg {
-            Message::Pong(_) => return Ok(None),
             Message::Binary(bytes) => bytes,
             _ => return Err(PeerError::InvalidMessage),
         };
 
         let envelope = Envelope::parse_from_bytes(&bytes)?;
-        Ok(Some(envelope))
+        Ok(envelope)
     }
 
     async fn handle_envelope(&self, mut envelope: Envelope) -> Result<(), PeerError> {
@@ -451,11 +450,7 @@ where
     pub async fn start(self, mut reader: Connection) {
         while let Some(input) = reader.read().await {
             let envelope = match self.parse(input) {
-                Ok(Some(env)) => env,
-                Ok(_) => {
-                    log::trace!("received a pong message");
-                    continue;
-                }
+                Ok(env) => env,
                 Err(err) => {
                     log::error!("error while loading received message: {:#}", err);
                     continue;