From ca8a1b688431034b4d7951d298f0299d7e155601 Mon Sep 17 00:00:00 2001 From: Milo Turner Date: Mon, 9 Mar 2020 11:59:09 -0400 Subject: [PATCH] implement other log functions and use them --- hptp-recv/src/main.rs | 12 +++++------- hptp-send/src/main.rs | 10 ++++++---- hptp/src/lib.rs | 2 +- hptp/src/{log.rs => logger.rs} | 20 ++++++++++++++++++++ 4 files changed, 32 insertions(+), 12 deletions(-) rename hptp/src/{log.rs => logger.rs} (83%) diff --git a/hptp-recv/src/main.rs b/hptp-recv/src/main.rs index 6679126..b1b0ee6 100644 --- a/hptp-recv/src/main.rs +++ b/hptp-recv/src/main.rs @@ -4,7 +4,7 @@ extern crate tokio; #[macro_use] extern crate thiserror; -use hptp::log::Logger; +use hptp::logger::Logger; use hptp::msg::Msg; use hptp::peer::{self, Peer}; use tokio::io::AsyncWrite; @@ -63,21 +63,19 @@ where let mut pos = 0; loop { match peer.recv().await { - Ok(Msg::Ack) => log.debug_msg("not expected an ack...").await, + Ok(Msg::Ack) => log.debug_msg("not expecting an ack...").await, Ok(Msg::Data(data)) => { let len = data.len(); out.write_all(&data).await?; out.flush().await?; - log.recv_data_accepted(pos, len, hptp::log::InOrder).await; + log.recv_data_accepted(pos, len, hptp::logger::InOrder) + .await; send(peer, Msg::Ack).await?; pos += len; } - Err(peer::RecvError::InvalidMessage { .. }) => { - log.debug_msg(format!("got an invalid message; discarding")) - .await - } + Err(peer::RecvError::InvalidMessage { .. }) => log.recv_corrupt().await, Err(peer::RecvError::Io { source }) => return Err(source.into()), } diff --git a/hptp-send/src/main.rs b/hptp-send/src/main.rs index ce5e959..4bd39ea 100644 --- a/hptp-send/src/main.rs +++ b/hptp-send/src/main.rs @@ -4,7 +4,7 @@ extern crate tokio; #[macro_use] extern crate thiserror; -use hptp::log::Logger; +use hptp::logger::Logger; use hptp::msg::{self, Msg}; use hptp::peer::{self, Peer}; use std::net::SocketAddr; @@ -96,7 +96,7 @@ async fn recv(log: &mut Logger, peer: &mut Peer) -> Result, Error> { Ok(m) => Ok(Some(m)), Err(peer::RecvError::Io { source }) => Err(source.into()), Err(peer::RecvError::InvalidMessage { .. }) => { - log.debug_msg("invalid message; discarding").await; + log.recv_corrupt().await; Ok(None) } } @@ -114,13 +114,15 @@ where send(peer, Msg::Data(data)).await?; log.send_data(pos, len).await; if let Some(Msg::Ack) = recv(log, peer).await? { - log.debug_msg("got ack").await; + log.recv_ack(pos).await; } else { log.debug_msg("didn't get ack??").await; } pos += len; } - None => return Ok(()), + None => break, } } + log.completed().await; + Ok(()) } diff --git a/hptp/src/lib.rs b/hptp/src/lib.rs index d3d937d..8bc9bf7 100644 --- a/hptp/src/lib.rs +++ b/hptp/src/lib.rs @@ -2,7 +2,7 @@ extern crate rand; #[macro_use] extern crate thiserror; -pub mod log; +pub mod logger; pub mod msg; pub mod peer; diff --git a/hptp/src/log.rs b/hptp/src/logger.rs similarity index 83% rename from hptp/src/log.rs rename to hptp/src/logger.rs index de933a1..b56645f 100644 --- a/hptp/src/log.rs +++ b/hptp/src/logger.rs @@ -53,9 +53,21 @@ impl Logger { self.log_payload(LogPayload::RecvDataIgnored).await } + pub async fn recv_ack(&mut self, last_offset: usize) { + self.log_payload(LogPayload::RecvAck { last_offset }).await + } + + pub async fn recv_corrupt(&mut self) { + self.log_payload(LogPayload::RecvCorrupt).await + } + pub async fn send_data(&mut self, start: usize, len: usize) { self.log_payload(LogPayload::SendData { start, len }).await } + + pub async fn completed(&mut self) { + self.log_payload(LogPayload::Completed).await + } } struct LogMessage<'a> { @@ -76,10 +88,15 @@ enum LogPayload<'a> { order: AcceptedOrder, }, RecvDataIgnored, + RecvAck { + last_offset: usize, + }, + RecvCorrupt, SendData { start: usize, len: usize, }, + Completed, } #[derive(Copy, Clone, Eq, PartialEq)] @@ -106,7 +123,10 @@ impl Display for LogPayload<'_> { write!(f, "[recv data] {} ({}) ACCEPTED ({})", start, len, order) } LogPayload::RecvDataIgnored => write!(f, "[recv data] IGNORED"), + LogPayload::RecvCorrupt => write!(f, "[recv corrupt packet]"), + LogPayload::RecvAck { last_offset } => write!(f, "[recv ack] {}", last_offset), LogPayload::SendData { start, len } => write!(f, "[send data] {} ({})", start, len), + LogPayload::Completed => write!(f, "[completed]"), } } }