From 592a99bca4e760d00057fc3927c3c6f164e353e2 Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Sat, 19 Nov 2016 09:05:00 -0800 Subject: [PATCH 1/7] completed basic implementation of FramedUdp for streams and sink --- src/io/udp_frame.rs | 258 ++++++++++++++++++++++++++++++++++++++++++++ src/net/udp.rs | 14 +++ 2 files changed, 272 insertions(+) create mode 100644 src/io/udp_frame.rs diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs new file mode 100644 index 000000000..9c1eeb958 --- /dev/null +++ b/src/io/udp_frame.rs @@ -0,0 +1,258 @@ +use std::io; +use std::ops::{Deref, DerefMut}; +use std::sync::Arc; +use net::udp::UdpSocket +use futures::{Async, Poll, Stream, Sink, StartSend, AsyncSink}; +use futures::sync::BiLock; + +use io::Io; + +/// Encoding of frames via buffers. +/// +/// This trait is used when constructing an instance of `FramedUdp`. It provides +/// one type: `Out` for encoding outgoing frames according to a protocol. +/// +/// Because UDP is a connectionless protocol, the encode method will also be +/// responsible for determining the remote host to which the datagram should be +/// sent +/// +/// The trait itself is implemented on a type that can track state for decoding +/// or encoding, which is particularly useful for streaming parsers. In many +/// cases, though, this type will simply be a unit struct (e.g. `struct +/// HttpCodec`). +pub trait EncodeUdp { + + /// The type of frames to be encoded. + type Out; + + + /// Encodes a frame into the buffer provided. + /// + /// This method will encode `msg` into the byte buffer provided by `buf`. + /// The `buf` provided is an internal buffer of the `Framed` instance and + /// will be written out when possible. + /// + /// The codec also determines the destination to which the buffer should + /// be directed, which will be returned as a SocketAddr; + fn encode(&mut self, msg: Self::Out, buf: &mut Vec) -> SocketAddr; +} + +/// Decoding of frames via buffers. +/// +/// This trait is used when constructing an instance of `FramedUdp`. It provides +/// one type: `In` for decoding incoming frames from a Datagram +/// +/// Because UDP is a connectionless protocol, the decode method will also be +/// supplied with a SocketAddr of the remote host which sent the datagram +/// +/// The trait itself is implemented on a type that can track state for decoding +/// or encoding, which is particularly useful for streaming parsers. In many +/// cases, though, this type will simply be a unit struct (e.g. `struct +/// HttpCodec`). +pub trait DecodeUdp { + /// The type of decoded frames. + type In; + + /// Attempts to decode a frame from the provided buffer of bytes. + /// + /// This method is called by `FramedUdp` on a single datagram which has been + /// read from a socket. + /// + /// It is required that the Decoder empty the read buffer in every call to + /// decode, as the next poll_read that occurs will write the next datagram + /// into the buffer, without regard for what is already there. + /// + /// If the bytes look valid, but a frame isn't fully available yet, then + /// `Ok(None)` is returned. This indicates to the `Framed` instance that + /// it needs to read some more bytes before calling this method again. + /// In such a case, it is the decoder's responsibility to copy the data + /// into their own internal buffer for future use. + /// + /// Finally, if the bytes in the buffer are malformed then an error is + /// returned indicating why. This informs `Framed` that the stream is now + /// corrupt and should be terminated. + /// + /// When dealing with connectionless streams, there will likely be some sort + /// of state machine. + fn decode(&mut self, src: &SocketAddr, buf: &mut Vec) -> Result, io::Error>; +} + +/// A unified `Stream` and `Sink` interface to an underlying `Io` object, using +/// the `Encode` and `Decode` traits to encode and decode frames. +/// +/// You can acquire a `Framed` instance by using the `Io::framed` adapter. +pub struct FramedUdp { + socket: UdpSocket, + encoder: E, + decoder: D, + out_addr : Option, + rd: Vec, + wr: Vec, +} + +impl Stream for Framed { + type Item = D::In; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + loop { + + let before = self.rd.len(); + let ret = self.socket.recv_from(self.rd.mut_bytes(), &mut inaddr); + match ret { + Ok((n, addr)) => { + trace!("read {} bytes", n); + trace!("attempting to decode a frame"); + if let Some(frame) = try!(self.decoder.decode(&addr, &mut self.rd)) { + trace!("frame decoded from buffer"); + self.rd.clear(); + return Ok(Async::Ready(Some(frame))); + } + } + Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { + if self.rd.len() == before { + return Ok(Async::NotReady) + } + } + Err(e) => return Err(e), + } + } + } +} + +impl Sink for Framed { + type SinkItem = E::Out; + type SinkError = io::Error; + + fn start_send(&mut self, item: C::Out) -> StartSend { + if self.wr.len() > 0 { + try!(self.poll_complete()); + if self.wr.len() > 0 { + return Ok(AsyncSink::NotReady(item)); + } + } + + self.out_addr = Some(self.codec.encode(item, &mut self.wr)); + Ok(AsyncSink::Ready) + } + + fn poll_complete(&mut self) -> Poll<(), io::Error> { + trace!("flushing framed transport"); + + while !self.wr.is_empty() { + if let Some(outaddr) = self.out_addr.ref() { + trace!("writing; remaining={}", self.wr.len()); + let n = try_nb!(self.socket.send_to(&self.wr, outaddr)); + self.wr.clear(); + self.out_addr = None; + if n != self.wr.len() { + return Err(io::Error::new(io::ErrorKind::WriteZero, + "failed to write frame datagram to socket")); + } + } + else { + return Err(io::Error::new(io::ErrorKind::Other, + "outbound stream in invalid state: out_addr is not known")); + } + } + + return Ok(Async::Ready(())); + } +} + +pub fn framed_udp(socket : UdpSocket, decoder : D, encoder : E) -> Framed { + Framed { + socket: socket, + encoder: encoder, + decoder: decoder, + rd: Vec::with_capacity(64 * 1024), + wr: Vec::with_capacity(64 * 1024) + } +} + +impl FramedUdp { + /// Splits this `Stream + Sink` object into separate `Stream` and `Sink` + /// objects, which can be useful when you want to split ownership between + /// tasks, or allow direct interaction between the two objects (e.g. via + /// `Sink::send_all`). + pub fn split(self) -> (FramedRead, FramedWrite) { + let (a, b) = BiLock::new(self); + let read = FramedUdpRead { framed: a }; + let write = FramedUdpWrite { framed: b }; + (read, write) + } + + /// Returns a reference to the underlying I/O stream wrapped by `Framed`. + /// + /// Note that care should be taken to not tamper with the underlying stream + /// of data coming in as it may corrupt the stream of frames otherwise being + /// worked with. + pub fn get_ref(&self) -> &UdpSocket { + &self.socket + } + + /// Returns a mutable reference to the underlying I/O stream wrapped by + /// `Framed`. + /// + /// Note that care should be taken to not tamper with the underlying stream + /// of data coming in as it may corrupt the stream of frames otherwise being + /// worked with. + pub fn get_mut(&mut self) -> &mut UdpSocket { + &mut self.socket + } + + /// Consumes the `Framed`, returning its underlying I/O stream. + /// + /// Note that care should be taken to not tamper with the underlying stream + /// of data coming in as it may corrupt the stream of frames otherwise being + /// worked with. + pub fn into_inner(self) -> UdpSocket { + self.socket + } +} +/// A `Stream` interface to an underlying `Io` object, using the `Decode` trait +/// to decode frames. +pub struct FramedRead { + framed: BiLock>, +} + +impl Stream for FramedRead { + type Item = D::In; + type Error = io::Error; + + fn poll(&mut self) -> Poll, io::Error> { + if let Async::Ready(mut guard) = self.framed.poll_lock() { + guard.poll() + } else { + Ok(Async::NotReady) + } + } +} + +/// A `Sink` interface to an underlying `Io` object, using the `Encode` trait +/// to encode frames. +pub struct FramedWrite { + framed: BiLock>, +} + +impl Sink for FramedWrite { + type SinkItem = E::Out; + type SinkError = io::Error; + + fn start_send(&mut self, item: E::Out) -> StartSend { + if let Async::Ready(mut guard) = self.framed.poll_lock() { + guard.start_send(item) + } else { + Ok(AsyncSink::NotReady(item)) + } + } + + fn poll_complete(&mut self) -> Poll<(), io::Error> { + if let Async::Ready(mut guard) = self.framed.poll_lock() { + guard.poll_complete() + } else { + Ok(Async::NotReady) + } + } +} + diff --git a/src/net/udp.rs b/src/net/udp.rs index a88a8ca16..2a2e0a03a 100644 --- a/src/net/udp.rs +++ b/src/net/udp.rs @@ -42,6 +42,20 @@ impl UdpSocket { UdpSocket::new(udp, handle) } + /// Creates a FramedUdp object, which leverages a supplied `EncodeUdp` + /// and `DecodeUdp` to implement `Stream` and `Sink` + /// This moves the socket into the newly created FramedUdp object + pub fn framed(self, decoder : D, encoder : E) -> Framed { + FramedUdp { + socket: self, + encoder: encoder, + decoder: decoder, + is_readable: false, + rd: Vec::with_capacity(64 * 1024); + wr: Vec::with_capacity(64 * 1024), + } + } + /// Returns the local address that this stream is bound to. pub fn local_addr(&self) -> io::Result { self.io.get_ref().local_addr() From 71d8672aab2b6c4712942783920e01db578eac9c Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Sun, 20 Nov 2016 09:08:03 -0800 Subject: [PATCH 2/7] implemented moste of udp frames test --- src/io/mod.rs | 2 + src/io/udp_frame.rs | 128 ++++++++++++++++++--------------------- src/net/udp.rs | 18 +++--- tests/udp-line-frames.rs | 74 ++++++++++++++++++++++ 4 files changed, 144 insertions(+), 78 deletions(-) create mode 100644 tests/udp-line-frames.rs diff --git a/src/io/mod.rs b/src/io/mod.rs index 15aba3b74..694174892 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -33,6 +33,7 @@ macro_rules! try_nb { mod copy; mod frame; +mod udp_frame; mod flush; mod read_exact; mod read_to_end; @@ -43,6 +44,7 @@ mod window; mod write_all; pub use self::copy::{copy, Copy}; pub use self::frame::{EasyBuf, EasyBufMut, FramedRead, FramedWrite, Framed, Codec}; +pub use self::udp_frame::{FramedUdp, framed_udp, FramedUdpRead, FramedUdpWrite, CodecUdp}; pub use self::flush::{flush, Flush}; pub use self::read_exact::{read_exact, ReadExact}; pub use self::read_to_end::{read_to_end, ReadToEnd}; diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs index 9c1eeb958..a1b64db09 100644 --- a/src/io/udp_frame.rs +++ b/src/io/udp_frame.rs @@ -1,12 +1,9 @@ use std::io; -use std::ops::{Deref, DerefMut}; -use std::sync::Arc; -use net::udp::UdpSocket +use std::net::SocketAddr; +use net::UdpSocket; use futures::{Async, Poll, Stream, Sink, StartSend, AsyncSink}; use futures::sync::BiLock; -use io::Io; - /// Encoding of frames via buffers. /// /// This trait is used when constructing an instance of `FramedUdp`. It provides @@ -20,10 +17,13 @@ use io::Io; /// or encoding, which is particularly useful for streaming parsers. In many /// cases, though, this type will simply be a unit struct (e.g. `struct /// HttpCodec`). -pub trait EncodeUdp { +pub trait CodecUdp { /// The type of frames to be encoded. type Out; + + /// The type of decoded frames. + type In; /// Encodes a frame into the buffer provided. @@ -32,87 +32,64 @@ pub trait EncodeUdp { /// The `buf` provided is an internal buffer of the `Framed` instance and /// will be written out when possible. /// - /// The codec also determines the destination to which the buffer should + /// The encode method also determines the destination to which the buffer should /// be directed, which will be returned as a SocketAddr; fn encode(&mut self, msg: Self::Out, buf: &mut Vec) -> SocketAddr; -} - -/// Decoding of frames via buffers. -/// -/// This trait is used when constructing an instance of `FramedUdp`. It provides -/// one type: `In` for decoding incoming frames from a Datagram -/// -/// Because UDP is a connectionless protocol, the decode method will also be -/// supplied with a SocketAddr of the remote host which sent the datagram -/// -/// The trait itself is implemented on a type that can track state for decoding -/// or encoding, which is particularly useful for streaming parsers. In many -/// cases, though, this type will simply be a unit struct (e.g. `struct -/// HttpCodec`). -pub trait DecodeUdp { - /// The type of decoded frames. - type In; - + /// Attempts to decode a frame from the provided buffer of bytes. /// /// This method is called by `FramedUdp` on a single datagram which has been /// read from a socket. /// - /// It is required that the Decoder empty the read buffer in every call to + /// It is required that the decode method empty the read buffer in every call to /// decode, as the next poll_read that occurs will write the next datagram /// into the buffer, without regard for what is already there. /// /// If the bytes look valid, but a frame isn't fully available yet, then /// `Ok(None)` is returned. This indicates to the `Framed` instance that /// it needs to read some more bytes before calling this method again. - /// In such a case, it is the decoder's responsibility to copy the data + /// In such a case, it is decode's responsibility to copy the data /// into their own internal buffer for future use. /// /// Finally, if the bytes in the buffer are malformed then an error is /// returned indicating why. This informs `Framed` that the stream is now /// corrupt and should be terminated. /// - /// When dealing with connectionless streams, there will likely be some sort - /// of state machine. fn decode(&mut self, src: &SocketAddr, buf: &mut Vec) -> Result, io::Error>; } /// A unified `Stream` and `Sink` interface to an underlying `Io` object, using -/// the `Encode` and `Decode` traits to encode and decode frames. +/// the `CodecUdp` trait to encode and decode frames. /// /// You can acquire a `Framed` instance by using the `Io::framed` adapter. -pub struct FramedUdp { +pub struct FramedUdp { socket: UdpSocket, - encoder: E, - decoder: D, + codec: C, out_addr : Option, rd: Vec, wr: Vec, } -impl Stream for Framed { - type Item = D::In; +impl Stream for FramedUdp { + type Item = C::In; type Error = io::Error; fn poll(&mut self) -> Poll, io::Error> { loop { - let before = self.rd.len(); - let ret = self.socket.recv_from(self.rd.mut_bytes(), &mut inaddr); + let ret = self.socket.recv_from(self.rd.as_mut_slice()); match ret { Ok((n, addr)) => { trace!("read {} bytes", n); trace!("attempting to decode a frame"); - if let Some(frame) = try!(self.decoder.decode(&addr, &mut self.rd)) { + if let Some(frame) = try!(self.codec.decode(&addr, &mut self.rd)) { trace!("frame decoded from buffer"); self.rd.clear(); return Ok(Async::Ready(Some(frame))); } } Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - if self.rd.len() == before { - return Ok(Async::NotReady) - } + return Ok(Async::NotReady) } Err(e) => return Err(e), } @@ -120,11 +97,11 @@ impl Stream for Framed { } } -impl Sink for Framed { - type SinkItem = E::Out; +impl Sink for FramedUdp { + type SinkItem = C::Out; type SinkError = io::Error; - fn start_send(&mut self, item: C::Out) -> StartSend { + fn start_send(&mut self, item: C::Out) -> StartSend { if self.wr.len() > 0 { try!(self.poll_complete()); if self.wr.len() > 0 { @@ -140,9 +117,9 @@ impl Sink for Framed { trace!("flushing framed transport"); while !self.wr.is_empty() { - if let Some(outaddr) = self.out_addr.ref() { + if let Some(outaddr) = self.out_addr { trace!("writing; remaining={}", self.wr.len()); - let n = try_nb!(self.socket.send_to(&self.wr, outaddr)); + let n = try_nb!(self.socket.send_to(&self.wr, &outaddr)); self.wr.clear(); self.out_addr = None; if n != self.wr.len() { @@ -160,22 +137,37 @@ impl Sink for Framed { } } -pub fn framed_udp(socket : UdpSocket, decoder : D, encoder : E) -> Framed { - Framed { - socket: socket, - encoder: encoder, - decoder: decoder, - rd: Vec::with_capacity(64 * 1024), - wr: Vec::with_capacity(64 * 1024) - } +/// Helper function that Creates a new FramedUdp object. It moves the supplied socket, codec +/// into the resulting FramedUdp +pub fn framed_udp(socket : UdpSocket, codec : C) -> FramedUdp { + FramedUdp::new( + socket, + codec, + Vec::with_capacity(64 * 1024), + Vec::with_capacity(64 * 1024) + ) } -impl FramedUdp { +impl FramedUdp { + + /// Creates a new FramedUdp object. It moves the supplied socket, codec + /// supplied vecs. + pub fn new(sock : UdpSocket, codec : C, rd_buf : Vec, wr_buf : Vec) -> FramedUdp { + FramedUdp { + socket: sock, + codec : codec, + out_addr: None, + rd: rd_buf, + wr: wr_buf + } + } + + /// Splits this `Stream + Sink` object into separate `Stream` and `Sink` /// objects, which can be useful when you want to split ownership between /// tasks, or allow direct interaction between the two objects (e.g. via /// `Sink::send_all`). - pub fn split(self) -> (FramedRead, FramedWrite) { + pub fn split(self) -> (FramedUdpRead, FramedUdpWrite) { let (a, b) = BiLock::new(self); let read = FramedUdpRead { framed: a }; let write = FramedUdpWrite { framed: b }; @@ -210,17 +202,17 @@ impl FramedUdp { self.socket } } -/// A `Stream` interface to an underlying `Io` object, using the `Decode` trait +/// A `Stream` interface to an underlying `Io` object, using the `CodecUdp` trait /// to decode frames. -pub struct FramedRead { - framed: BiLock>, +pub struct FramedUdpRead { + framed: BiLock>, } -impl Stream for FramedRead { - type Item = D::In; +impl Stream for FramedUdpRead { + type Item = C::In; type Error = io::Error; - fn poll(&mut self) -> Poll, io::Error> { + fn poll(&mut self) -> Poll, io::Error> { if let Async::Ready(mut guard) = self.framed.poll_lock() { guard.poll() } else { @@ -229,17 +221,17 @@ impl Stream for FramedRead { } } -/// A `Sink` interface to an underlying `Io` object, using the `Encode` trait +/// A `Sink` interface to an underlying `Io` object, using the `CodecUdp` trait /// to encode frames. -pub struct FramedWrite { - framed: BiLock>, +pub struct FramedUdpWrite { + framed: BiLock>, } -impl Sink for FramedWrite { - type SinkItem = E::Out; +impl Sink for FramedUdpWrite { + type SinkItem = C::Out; type SinkError = io::Error; - fn start_send(&mut self, item: E::Out) -> StartSend { + fn start_send(&mut self, item: C::Out) -> StartSend { if let Async::Ready(mut guard) = self.framed.poll_lock() { guard.start_send(item) } else { diff --git a/src/net/udp.rs b/src/net/udp.rs index 2a2e0a03a..34de052c3 100644 --- a/src/net/udp.rs +++ b/src/net/udp.rs @@ -1,7 +1,7 @@ use std::io; use std::net::{self, SocketAddr, Ipv4Addr, Ipv6Addr}; use std::fmt; - +use io::FramedUdp; use futures::Async; use mio; @@ -45,15 +45,13 @@ impl UdpSocket { /// Creates a FramedUdp object, which leverages a supplied `EncodeUdp` /// and `DecodeUdp` to implement `Stream` and `Sink` /// This moves the socket into the newly created FramedUdp object - pub fn framed(self, decoder : D, encoder : E) -> Framed { - FramedUdp { - socket: self, - encoder: encoder, - decoder: decoder, - is_readable: false, - rd: Vec::with_capacity(64 * 1024); - wr: Vec::with_capacity(64 * 1024), - } + pub fn framed(self, codec : C) -> FramedUdp { + FramedUdp::new( + self, + codec, + Vec::with_capacity(64 * 1024), + Vec::with_capacity(64 * 1024) + ) } /// Returns the local address that this stream is bound to. diff --git a/tests/udp-line-frames.rs b/tests/udp-line-frames.rs new file mode 100644 index 000000000..90e8968d6 --- /dev/null +++ b/tests/udp-line-frames.rs @@ -0,0 +1,74 @@ +extern crate tokio_core; +extern crate env_logger; +extern crate futures; + +use std::io; +use std::net::{SocketAddr}; +use futures::{future, Future, Stream, Sink, IntoFuture}; +use tokio_core::io::{write_all, read, FramedUdp, CodecUdp, Io}; +use tokio_core::net::{UdpSocket}; +use tokio_core::reactor::{Core, Timeout}; +use std::time::Duration; +use std::str; + +pub struct LineCodec { + addr : Option +} + +impl CodecUdp for LineCodec { + type In = Vec; + type Out = Vec; + + fn decode(&mut self, addr : &SocketAddr, buf: &mut Vec) -> Result, io::Error> { + self.addr = Some(*addr); + match buf.as_slice().iter().position(|&b| b == b'\n') { + Some(i) => Ok(Some(buf[.. i + 1].into())), + None => Ok(None), + } + } + + fn encode(&mut self, item: Vec, into: &mut Vec) -> SocketAddr { + into.extend_from_slice(item.as_slice()); + into.push('\n' as u8); + + self.addr.unwrap() + } +} + +#[test] +fn echo() { + drop(env_logger::init()); + + let mut core = Core::new().unwrap(); + let handle = core.handle(); + + let srvcodec = LineCodec { addr : None }; + let clicodec = LineCodec { addr : None }; + + let srvaddr : SocketAddr = "127.0.0.1:31999".parse().unwrap(); + let clientaddr : SocketAddr = "127.0.0.1:32000".parse().unwrap(); + + let server = UdpSocket::bind(&srvaddr, &handle).unwrap(); + let client = UdpSocket::bind(&clientaddr, &handle).unwrap(); + + let job = client.send_to(b"PING", &srvaddr); + let _ = core.run(job.into_future()).unwrap(); + + let (srvstream, srvsink) = server.framed(srvcodec).split(); + let srvloop = srvstream.for_each(move |buf| { + println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + srvsink.send(b"PONG".to_vec()).map(|_| ()).wait() + }); + + let (clistream, clisink) = client.framed(clicodec).split(); + let cliloop = clistream.for_each(move |buf| { + println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + clisink.send(b"PING".to_vec()).map(|_| ()).wait() + }); + + let timeout = Timeout::new(Duration::from_millis(500), &handle).unwrap(); + + let wait = future::select_all(vec![timeout.boxed(), srvloop.boxed(), cliloop.boxed()]); + core.run(wait); + +} From 161811de8b59b66f156599807c779c89ba261817 Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Sun, 20 Nov 2016 11:40:43 -0800 Subject: [PATCH 3/7] moved udp test to examples, optimized buffer handling --- examples/udp-codec.rs | 114 +++++++++++++++++++++++++++++++++++++++ src/io/udp_frame.rs | 19 +++---- src/net/udp.rs | 87 +++++++++++++++++++++++++++++- tests/udp-line-frames.rs | 74 ------------------------- 4 files changed, 209 insertions(+), 85 deletions(-) create mode 100644 examples/udp-codec.rs delete mode 100644 tests/udp-line-frames.rs diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs new file mode 100644 index 000000000..fd6d98746 --- /dev/null +++ b/examples/udp-codec.rs @@ -0,0 +1,114 @@ +extern crate tokio_core; +extern crate env_logger; +extern crate futures; + +#[macro_use] +extern crate log; + +use std::io; +use std::net::{SocketAddr}; +use futures::{future, Future, Stream, Sink}; +use tokio_core::io::{CodecUdp}; +use tokio_core::net::{UdpSocket}; +use tokio_core::reactor::{Core, Timeout}; +use std::time::Duration; +use std::str; + +/// This is a basic example of leveraging `FramedUdp` to create +/// a simple UDP client and server which speak a custom Protocol. +/// `FramedUdp` applies a `Codec` to the input and output of an +/// `Evented` + +/// Simple Newline based parser, +/// This is for a connectionless server, it must keep track +/// of the Socket address of the last peer to contact it +/// so that it can respond back. +/// In the real world, one would probably +/// want an associative of remote peers to their state +pub struct LineCodec { + addr : Option +} + +impl CodecUdp for LineCodec { + type In = Vec; + type Out = Vec; + + fn decode(&mut self, addr : &SocketAddr, buf: &[u8]) -> Result, io::Error> { + trace!("decoding {} - {}", str::from_utf8(buf).unwrap(), addr); + self.addr = Some(*addr); + match buf.iter().position(|&b| b == b'\n') { + Some(i) => Ok(Some(buf[.. i].into())), + None => Ok(None), + } + } + + fn encode(&mut self, item: &Vec, into: &mut Vec) -> SocketAddr { + trace!("encoding {}", str::from_utf8(item.as_slice()).unwrap()); + into.extend_from_slice(item.as_slice()); + into.push('\n' as u8); + + self.addr.unwrap() + } +} + +fn main() { + drop(env_logger::init()); + + let mut core = Core::new().unwrap(); + let handle = core.handle(); + + //create the line codec parser for each + let srvcodec = LineCodec { addr : None }; + let clicodec = LineCodec { addr : None }; + + let srvaddr : SocketAddr = "127.0.0.1:31999".parse().unwrap(); + let clientaddr : SocketAddr = "127.0.0.1:32000".parse().unwrap(); + + //We bind each socket to a specific port + let server = UdpSocket::bind(&srvaddr, &handle).unwrap(); + let client = UdpSocket::bind(&clientaddr, &handle).unwrap(); + + //start things off by sending a ping from the client to the server + //This doesn't go through the codec to encode the message, but rather + //it sends raw data with the send_dgram future + { + let job = client.send_dgram(b"PING\n", &srvaddr); + core.run(job).unwrap(); + } + + //We create a FramedUdp instance, which associates a socket + //with a codec. We then immediate split that into the + //receiving side `Stream` and the writing side `Sink` + let (srvstream, srvsink) = server.framed(srvcodec).split(); + + //`Stream::fold` runs once per every received datagram. + //Note that we pass srvsink into fold, so that it can be + //supplied to every iteration. The reason for this is + //sink.send moves itself into `send` and then returns itself + let srvloop = srvstream.fold(srvsink, move |sink, buf| { + println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + sink.send(b"PONG".to_vec()) + }).map(|_| ()); + + //We create another FramedUdp instance, this time for the client socket + let (clistream, clisink) = client.framed(clicodec).split(); + + //And another infinite iteration + let cliloop = clistream.fold(clisink, move |sink, buf| { + println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + sink.send(b"PING".to_vec()) + }).map(|_| ()); + + let timeout = Timeout::new(Duration::from_millis(500), &handle).unwrap(); + + //`select_all` takes an `Iterable` of `Future` and returns a future itself + //This future waits until the first `Future` completes, it then returns + //that result. + let wait = future::select_all(vec![timeout.boxed(), srvloop.boxed(), cliloop.boxed()]); + + //Now we instruct `reactor::Core` to iterate, processing events until its future, `SelectAll` + //has completed + if let Err(e) core.run(wait) { + error!("{}", e.0); + } +} diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs index a1b64db09..1996ac593 100644 --- a/src/io/udp_frame.rs +++ b/src/io/udp_frame.rs @@ -34,7 +34,7 @@ pub trait CodecUdp { /// /// The encode method also determines the destination to which the buffer should /// be directed, which will be returned as a SocketAddr; - fn encode(&mut self, msg: Self::Out, buf: &mut Vec) -> SocketAddr; + fn encode(&mut self, msg: &Self::Out, buf: &mut Vec) -> SocketAddr; /// Attempts to decode a frame from the provided buffer of bytes. /// @@ -55,7 +55,7 @@ pub trait CodecUdp { /// returned indicating why. This informs `Framed` that the stream is now /// corrupt and should be terminated. /// - fn decode(&mut self, src: &SocketAddr, buf: &mut Vec) -> Result, io::Error>; + fn decode(&mut self, src: &SocketAddr, buf: &[u8]) -> Result, io::Error>; } /// A unified `Stream` and `Sink` interface to an underlying `Io` object, using @@ -82,9 +82,8 @@ impl Stream for FramedUdp { Ok((n, addr)) => { trace!("read {} bytes", n); trace!("attempting to decode a frame"); - if let Some(frame) = try!(self.codec.decode(&addr, &mut self.rd)) { + if let Some(frame) = try!(self.codec.decode(&addr, & self.rd[.. n])) { trace!("frame decoded from buffer"); - self.rd.clear(); return Ok(Async::Ready(Some(frame))); } } @@ -109,7 +108,7 @@ impl Sink for FramedUdp { } } - self.out_addr = Some(self.codec.encode(item, &mut self.wr)); + self.out_addr = Some(self.codec.encode(&item, &mut self.wr)); Ok(AsyncSink::Ready) } @@ -118,11 +117,13 @@ impl Sink for FramedUdp { while !self.wr.is_empty() { if let Some(outaddr) = self.out_addr { - trace!("writing; remaining={}", self.wr.len()); - let n = try_nb!(self.socket.send_to(&self.wr, &outaddr)); + let remaining = self.wr.len(); + trace!("writing; remaining={}", remaining); + let n = try_nb!(self.socket.send_to(self.wr.as_slice(), &outaddr)); + trace!("written {}", n); self.wr.clear(); self.out_addr = None; - if n != self.wr.len() { + if n != remaining { return Err(io::Error::new(io::ErrorKind::WriteZero, "failed to write frame datagram to socket")); } @@ -143,7 +144,7 @@ pub fn framed_udp(socket : UdpSocket, codec : C) -> FramedUdp { FramedUdp::new( socket, codec, - Vec::with_capacity(64 * 1024), + vec![0; 64 * 1024], Vec::with_capacity(64 * 1024) ) } diff --git a/src/net/udp.rs b/src/net/udp.rs index 34de052c3..109b759a9 100644 --- a/src/net/udp.rs +++ b/src/net/udp.rs @@ -2,8 +2,9 @@ use std::io; use std::net::{self, SocketAddr, Ipv4Addr, Ipv6Addr}; use std::fmt; use io::FramedUdp; -use futures::Async; +use futures::{Async, Future, Poll}; use mio; +use std::mem; use reactor::{Handle, PollEvented}; @@ -49,7 +50,7 @@ impl UdpSocket { FramedUdp::new( self, codec, - Vec::with_capacity(64 * 1024), + vec![0; 64 * 1024], Vec::with_capacity(64 * 1024) ) } @@ -97,6 +98,33 @@ impl UdpSocket { Err(e) => Err(e), } } + + /// Creates a future that will write the entire contents of the buffer `buf` to + /// the stream `a` provided. + /// + /// The returned future will return after data has been written to the outbound + /// socket. + /// The future will resolve to the stream as well as the buffer (for reuse if + /// needed). + /// + /// Any error which happens during writing will cause both the stream and the + /// buffer to get destroyed. + /// + /// The `buf` parameter here only requires the `AsRef<[u8]>` trait, which should + /// be broadly applicable to accepting data which can be converted to a slice. + /// The `Window` struct is also available in this crate to provide a different + /// window into a slice if necessary. + pub fn send_dgram<'a, T>(&'a self, buf: T, addr : &'a SocketAddr) -> SendDGram + where T: AsRef<[u8]>, + { + SendDGram { + state: UdpState::Writing { + sock: self, + addr: addr, + buf: buf, + }, + } + } /// Receives data from the socket. On success, returns the number of bytes /// read and the address from whence the data came. @@ -261,6 +289,61 @@ impl fmt::Debug for UdpSocket { } } +/// A future used to write the entire contents of some data to a stream. +/// +/// This is created by the [`write_all`] top-level method. +/// +/// [`write_all`]: fn.write_all.html +pub struct SendDGram<'a, T> { + state: UdpState<'a, T>, +} + +enum UdpState<'a, T> { + Writing { + sock: &'a UdpSocket, + buf: T, + addr: &'a SocketAddr, + }, + Empty, +} + + +fn zero_write() -> io::Error { + io::Error::new(io::ErrorKind::WriteZero, "zero-length write") +} + +fn incomplete_write(reason : &str) -> io::Error { + io::Error::new(io::ErrorKind::Other, reason) +} + +impl<'a, T> Future for SendDGram<'a, T> + where T: AsRef<[u8]>, +{ + type Item = T; + type Error = io::Error; + + fn poll(&mut self) -> Poll { + match self.state { + UdpState::Writing { ref sock, ref buf, ref addr} => { + let buf = buf.as_ref(); + let n = try_nb!(sock.send_to(&buf, addr)); + if n == 0 { + return Err(zero_write()) + } + if n != buf.len() { + return Err(incomplete_write("Failed to send entire message in datagram")) + } + } + UdpState::Empty => panic!("poll a SendAllTo after it's done"), + } + + match mem::replace(&mut self.state, UdpState::Empty) { + UdpState::Writing { buf, .. } => Ok((buf).into()), + UdpState::Empty => panic!(), + } + } +} + #[cfg(unix)] mod sys { use std::os::unix::prelude::*; diff --git a/tests/udp-line-frames.rs b/tests/udp-line-frames.rs deleted file mode 100644 index 90e8968d6..000000000 --- a/tests/udp-line-frames.rs +++ /dev/null @@ -1,74 +0,0 @@ -extern crate tokio_core; -extern crate env_logger; -extern crate futures; - -use std::io; -use std::net::{SocketAddr}; -use futures::{future, Future, Stream, Sink, IntoFuture}; -use tokio_core::io::{write_all, read, FramedUdp, CodecUdp, Io}; -use tokio_core::net::{UdpSocket}; -use tokio_core::reactor::{Core, Timeout}; -use std::time::Duration; -use std::str; - -pub struct LineCodec { - addr : Option -} - -impl CodecUdp for LineCodec { - type In = Vec; - type Out = Vec; - - fn decode(&mut self, addr : &SocketAddr, buf: &mut Vec) -> Result, io::Error> { - self.addr = Some(*addr); - match buf.as_slice().iter().position(|&b| b == b'\n') { - Some(i) => Ok(Some(buf[.. i + 1].into())), - None => Ok(None), - } - } - - fn encode(&mut self, item: Vec, into: &mut Vec) -> SocketAddr { - into.extend_from_slice(item.as_slice()); - into.push('\n' as u8); - - self.addr.unwrap() - } -} - -#[test] -fn echo() { - drop(env_logger::init()); - - let mut core = Core::new().unwrap(); - let handle = core.handle(); - - let srvcodec = LineCodec { addr : None }; - let clicodec = LineCodec { addr : None }; - - let srvaddr : SocketAddr = "127.0.0.1:31999".parse().unwrap(); - let clientaddr : SocketAddr = "127.0.0.1:32000".parse().unwrap(); - - let server = UdpSocket::bind(&srvaddr, &handle).unwrap(); - let client = UdpSocket::bind(&clientaddr, &handle).unwrap(); - - let job = client.send_to(b"PING", &srvaddr); - let _ = core.run(job.into_future()).unwrap(); - - let (srvstream, srvsink) = server.framed(srvcodec).split(); - let srvloop = srvstream.for_each(move |buf| { - println!("{}", str::from_utf8(buf.as_slice()).unwrap()); - srvsink.send(b"PONG".to_vec()).map(|_| ()).wait() - }); - - let (clistream, clisink) = client.framed(clicodec).split(); - let cliloop = clistream.for_each(move |buf| { - println!("{}", str::from_utf8(buf.as_slice()).unwrap()); - clisink.send(b"PING".to_vec()).map(|_| ()).wait() - }); - - let timeout = Timeout::new(Duration::from_millis(500), &handle).unwrap(); - - let wait = future::select_all(vec![timeout.boxed(), srvloop.boxed(), cliloop.boxed()]); - core.run(wait); - -} From ab3915d47d607a3e3585b8270d8ecb69d7a4249e Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Sun, 20 Nov 2016 11:55:57 -0800 Subject: [PATCH 4/7] forgot a = --- examples/udp-codec.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index fd6d98746..230c98ff3 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -108,7 +108,7 @@ fn main() { //Now we instruct `reactor::Core` to iterate, processing events until its future, `SelectAll` //has completed - if let Err(e) core.run(wait) { + if let Err(e) = core.run(wait) { error!("{}", e.0); } } From b12d32ce1cfc361255e3b938ed660a3e3d437f6d Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Mon, 21 Nov 2016 11:28:25 -0800 Subject: [PATCH 5/7] made send_dgram move self, made FramedUdp::new private, other clean-ups and tweaks --- examples/udp-codec.rs | 10 ++++------ src/io/udp_frame.rs | 11 +++++------ src/net/udp.rs | 39 +++++++++++++-------------------------- 3 files changed, 22 insertions(+), 38 deletions(-) diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index 230c98ff3..f899180a4 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -69,12 +69,10 @@ fn main() { let client = UdpSocket::bind(&clientaddr, &handle).unwrap(); //start things off by sending a ping from the client to the server - //This doesn't go through the codec to encode the message, but rather - //it sends raw data with the send_dgram future - { - let job = client.send_dgram(b"PING\n", &srvaddr); - core.run(job).unwrap(); - } + //This doesn't utilize the codec to encode the message, but rather + //it sends raw data directly to the remote peer with the send_dgram future + let job = client.send_dgram(b"PING\n", srvaddr); + let (client, _buf) = core.run(job).unwrap(); //We create a FramedUdp instance, which associates a socket //with a codec. We then immediate split that into the diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs index 1996ac593..a9b89cb82 100644 --- a/src/io/udp_frame.rs +++ b/src/io/udp_frame.rs @@ -115,7 +115,7 @@ impl Sink for FramedUdp { fn poll_complete(&mut self) -> Poll<(), io::Error> { trace!("flushing framed transport"); - while !self.wr.is_empty() { + if !self.wr.is_empty() { if let Some(outaddr) = self.out_addr { let remaining = self.wr.len(); trace!("writing; remaining={}", remaining); @@ -124,13 +124,12 @@ impl Sink for FramedUdp { self.wr.clear(); self.out_addr = None; if n != remaining { - return Err(io::Error::new(io::ErrorKind::WriteZero, - "failed to write frame datagram to socket")); + return Err(io::Error::new(io::ErrorKind::Other, + "failed to write entire datagram to socket")); } } else { - return Err(io::Error::new(io::ErrorKind::Other, - "outbound stream in invalid state: out_addr is not known")); + panic!("outbound stream in invalid state: out_addr is not known"); } } @@ -153,7 +152,7 @@ impl FramedUdp { /// Creates a new FramedUdp object. It moves the supplied socket, codec /// supplied vecs. - pub fn new(sock : UdpSocket, codec : C, rd_buf : Vec, wr_buf : Vec) -> FramedUdp { + fn new(sock : UdpSocket, codec : C, rd_buf : Vec, wr_buf : Vec) -> FramedUdp { FramedUdp { socket: sock, codec : codec, diff --git a/src/net/udp.rs b/src/net/udp.rs index 109b759a9..7fb6b9a78 100644 --- a/src/net/udp.rs +++ b/src/net/udp.rs @@ -1,7 +1,7 @@ use std::io; use std::net::{self, SocketAddr, Ipv4Addr, Ipv6Addr}; use std::fmt; -use io::FramedUdp; +use io::{FramedUdp, framed_udp}; use futures::{Async, Future, Poll}; use mio; use std::mem; @@ -47,12 +47,7 @@ impl UdpSocket { /// and `DecodeUdp` to implement `Stream` and `Sink` /// This moves the socket into the newly created FramedUdp object pub fn framed(self, codec : C) -> FramedUdp { - FramedUdp::new( - self, - codec, - vec![0; 64 * 1024], - Vec::with_capacity(64 * 1024) - ) + framed_udp(self, codec) } /// Returns the local address that this stream is bound to. @@ -114,7 +109,7 @@ impl UdpSocket { /// be broadly applicable to accepting data which can be converted to a slice. /// The `Window` struct is also available in this crate to provide a different /// window into a slice if necessary. - pub fn send_dgram<'a, T>(&'a self, buf: T, addr : &'a SocketAddr) -> SendDGram + pub fn send_dgram(self, buf: T, addr : SocketAddr) -> SendDGram where T: AsRef<[u8]>, { SendDGram { @@ -294,51 +289,43 @@ impl fmt::Debug for UdpSocket { /// This is created by the [`write_all`] top-level method. /// /// [`write_all`]: fn.write_all.html -pub struct SendDGram<'a, T> { - state: UdpState<'a, T>, +pub struct SendDGram { + state: UdpState, } -enum UdpState<'a, T> { +enum UdpState { Writing { - sock: &'a UdpSocket, + sock: UdpSocket, buf: T, - addr: &'a SocketAddr, + addr: SocketAddr, }, Empty, } - -fn zero_write() -> io::Error { - io::Error::new(io::ErrorKind::WriteZero, "zero-length write") -} - fn incomplete_write(reason : &str) -> io::Error { io::Error::new(io::ErrorKind::Other, reason) } -impl<'a, T> Future for SendDGram<'a, T> +impl Future for SendDGram where T: AsRef<[u8]>, { - type Item = T; + type Item = (UdpSocket, T); type Error = io::Error; - fn poll(&mut self) -> Poll { + fn poll(&mut self) -> Poll<(UdpSocket, T), io::Error> { match self.state { UdpState::Writing { ref sock, ref buf, ref addr} => { let buf = buf.as_ref(); let n = try_nb!(sock.send_to(&buf, addr)); - if n == 0 { - return Err(zero_write()) - } if n != buf.len() { return Err(incomplete_write("Failed to send entire message in datagram")) } } - UdpState::Empty => panic!("poll a SendAllTo after it's done"), + UdpState::Empty => panic!("poll a SendDGram after it's done"), } match mem::replace(&mut self.state, UdpState::Empty) { - UdpState::Writing { buf, .. } => Ok((buf).into()), + UdpState::Writing { sock, buf, .. } => Ok(Async::Ready((sock, (buf).into()))), UdpState::Empty => panic!(), } } From 2cb600bd19fb655c69ded7ac3cb2e86172c6285b Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Tue, 22 Nov 2016 08:29:02 -0800 Subject: [PATCH 6/7] changed CodecUdp::decode to return Self::In instead of Option --- examples/udp-codec.rs | 25 ++++++++++++++++--------- src/io/udp_frame.rs | 15 ++++----------- 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/examples/udp-codec.rs b/examples/udp-codec.rs index f899180a4..afc70d75f 100644 --- a/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -25,20 +25,27 @@ use std::str; /// so that it can respond back. /// In the real world, one would probably /// want an associative of remote peers to their state +/// +/// Note that this takes a pretty draconian stance by returning +/// an error if it can't find a newline in the datagram it received pub struct LineCodec { addr : Option } impl CodecUdp for LineCodec { - type In = Vec; + type In = Vec>; type Out = Vec; - fn decode(&mut self, addr : &SocketAddr, buf: &[u8]) -> Result, io::Error> { + fn decode(&mut self, addr : &SocketAddr, buf: &[u8]) -> Result { trace!("decoding {} - {}", str::from_utf8(buf).unwrap(), addr); self.addr = Some(*addr); - match buf.iter().position(|&b| b == b'\n') { - Some(i) => Ok(Some(buf[.. i].into())), - None => Ok(None), + let res : Vec> = buf.split(|c| *c == b'\n').map(|s| s.into()).collect(); + if res.len() > 0 { + Ok(res) + } + else { + Err(io::Error::new(io::ErrorKind::Other, + "failed to find newline in datagram")) } } @@ -83,8 +90,8 @@ fn main() { //Note that we pass srvsink into fold, so that it can be //supplied to every iteration. The reason for this is //sink.send moves itself into `send` and then returns itself - let srvloop = srvstream.fold(srvsink, move |sink, buf| { - println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + let srvloop = srvstream.fold(srvsink, move |sink, lines| { + println!("{}", str::from_utf8(lines[0].as_slice()).unwrap()); sink.send(b"PONG".to_vec()) }).map(|_| ()); @@ -92,8 +99,8 @@ fn main() { let (clistream, clisink) = client.framed(clicodec).split(); //And another infinite iteration - let cliloop = clistream.fold(clisink, move |sink, buf| { - println!("{}", str::from_utf8(buf.as_slice()).unwrap()); + let cliloop = clistream.fold(clisink, move |sink, lines| { + println!("{}", str::from_utf8(lines[0].as_slice()).unwrap()); sink.send(b"PING".to_vec()) }).map(|_| ()); diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs index a9b89cb82..d09d0a8fa 100644 --- a/src/io/udp_frame.rs +++ b/src/io/udp_frame.rs @@ -45,17 +45,11 @@ pub trait CodecUdp { /// decode, as the next poll_read that occurs will write the next datagram /// into the buffer, without regard for what is already there. /// - /// If the bytes look valid, but a frame isn't fully available yet, then - /// `Ok(None)` is returned. This indicates to the `Framed` instance that - /// it needs to read some more bytes before calling this method again. - /// In such a case, it is decode's responsibility to copy the data - /// into their own internal buffer for future use. - /// /// Finally, if the bytes in the buffer are malformed then an error is /// returned indicating why. This informs `Framed` that the stream is now /// corrupt and should be terminated. /// - fn decode(&mut self, src: &SocketAddr, buf: &[u8]) -> Result, io::Error>; + fn decode(&mut self, src: &SocketAddr, buf: &[u8]) -> Result; } /// A unified `Stream` and `Sink` interface to an underlying `Io` object, using @@ -82,10 +76,9 @@ impl Stream for FramedUdp { Ok((n, addr)) => { trace!("read {} bytes", n); trace!("attempting to decode a frame"); - if let Some(frame) = try!(self.codec.decode(&addr, & self.rd[.. n])) { - trace!("frame decoded from buffer"); - return Ok(Async::Ready(Some(frame))); - } + let frame = try!(self.codec.decode(&addr, & self.rd[.. n])); + trace!("frame decoded from buffer"); + return Ok(Async::Ready(Some(frame))); } Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { return Ok(Async::NotReady) From 1a6753df1ffb84fb76d51446ba22394765b07477 Mon Sep 17 00:00:00 2001 From: Rick Richardson Date: Tue, 22 Nov 2016 08:47:57 -0800 Subject: [PATCH 7/7] added Default Codec for Udp --- src/io/mod.rs | 2 +- src/io/udp_frame.rs | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/src/io/mod.rs b/src/io/mod.rs index 694174892..4c9ac0bf9 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -44,7 +44,7 @@ mod window; mod write_all; pub use self::copy::{copy, Copy}; pub use self::frame::{EasyBuf, EasyBufMut, FramedRead, FramedWrite, Framed, Codec}; -pub use self::udp_frame::{FramedUdp, framed_udp, FramedUdpRead, FramedUdpWrite, CodecUdp}; +pub use self::udp_frame::{FramedUdp, framed_udp, FramedUdpRead, FramedUdpWrite, CodecUdp, VecDGramCodec}; pub use self::flush::{flush, Flush}; pub use self::read_exact::{read_exact, ReadExact}; pub use self::read_to_end::{read_to_end, ReadToEnd}; diff --git a/src/io/udp_frame.rs b/src/io/udp_frame.rs index d09d0a8fa..99cc90a5a 100644 --- a/src/io/udp_frame.rs +++ b/src/io/udp_frame.rs @@ -241,3 +241,22 @@ impl Sink for FramedUdpWrite { } } +/// Default implementation of a DGram "parser" +/// This receives and produces a tuple of ('SocketAddr', `Vec`) +pub struct VecDGramCodec; + +impl CodecUdp for VecDGramCodec { + type In = (SocketAddr, Vec); + type Out = (SocketAddr, Vec); + + fn decode(&mut self, addr : &SocketAddr, buf: &[u8]) -> Result { + Ok((*addr, buf.into())) + } + + fn encode(&mut self, item: &Self::Out, into: &mut Vec) -> SocketAddr { + into.extend_from_slice(item.1.as_slice()); + into.push('\n' as u8); + item.0 + } +} +