From b4fc1fc57fb973ee856d97079918b0051c93858b Mon Sep 17 00:00:00 2001 From: Alexandre Bury Date: Sun, 23 Jun 2019 20:46:46 -0400 Subject: Add doc and comments --- src/pipe.rs | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) (limited to 'src/pipe.rs') diff --git a/src/pipe.rs b/src/pipe.rs index 710be1f..2fd9fac 100644 --- a/src/pipe.rs +++ b/src/pipe.rs @@ -3,12 +3,18 @@ use futures::sink::{Sink, Wait}; use futures::sync::mpsc::Sender; use std::io::{Error, ErrorKind, Result, Write}; +/// Adapter to implement the `std::io::Write` trait on a `Sender` from a futures channel. +/// +/// It uses an intermediate buffer to transfer packets. pub struct Pipe { + // Wrapping the sender in `Wait` makes it blocking, so we can implement blocking-style + // io::Write over the async-style Sender. dest: Wait>, bytes: BytesMut, } impl Pipe { + /// Wrap the given sender in a `Pipe`. pub fn new(destination: Sender) -> Self { Pipe { dest: destination.wait(), @@ -19,17 +25,24 @@ impl Pipe { impl Drop for Pipe { fn drop(&mut self) { + // This is the correct thing to do, but is not super important since the `Sink` + // implementation of `Sender` just returns `Ok` without doing anything else. let _ = self.dest.close(); } } impl Write for Pipe { fn write(&mut self, buf: &[u8]) -> Result { + // We are given a slice of bytes we do not own, so we must start by copying it. self.bytes.extend_from_slice(buf); - match self.dest.send(self.bytes.take().into()) { - Ok(_) => Ok(buf.len()), - Err(e) => Err(Error::new(ErrorKind::UnexpectedEof, e)), - } + + // Then, take the buffer and send it in the channel. + self.dest + .send(self.bytes.take().into()) + .map_err(|e| Error::new(ErrorKind::UnexpectedEof, e))?; + + // Return how much we sent - all of it. + Ok(buf.len()) } fn flush(&mut self) -> Result<()> { -- cgit v1.2.3