diff --git a/Cargo.toml b/Cargo.toml index 5883131b9..da766769b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,6 @@ members = [ "tokio", - "tokio-codec", "tokio-executor", "tokio-io", "tokio-macros", @@ -10,5 +9,9 @@ members = [ "tokio-sync", "tokio-test", "tokio-tls", + "tokio-util", + + # Internal + "examples", "build-tests", ] diff --git a/README.md b/README.md index 32e160b7e..c2aa98afd 100644 --- a/README.md +++ b/README.md @@ -94,7 +94,7 @@ async fn main() -> Result<(), Box> { ``` -More examples can be found [here](tokio/examples). Note that the `master` branch +More examples can be found [here](examples). Note that the `master` branch is currently being updated to use `async` / `await`. The examples are not fully ported. Examples for stable Tokio can be found [here](https://github.com/tokio-rs/tokio/tree/v0.1.x/tokio/examples). diff --git a/azure-pipelines.yml b/azure-pipelines.yml index b943bfe61..2acdb5224 100644 --- a/azure-pipelines.yml +++ b/azure-pipelines.yml @@ -26,7 +26,6 @@ jobs: cross: true crates: tokio: - - codec - fs - io - net @@ -61,7 +60,6 @@ jobs: displayName: Test sub crates - rust: beta crates: - tokio-codec: [] tokio-executor: - current-thread - thread-pool @@ -71,6 +69,8 @@ jobs: - async-traits tokio-macros: [] tokio-test: [] + tokio-util: [] + examples: [] # Test compilation failure - template: ci/azure-test-stable.yml diff --git a/ci/patch.toml b/ci/patch.toml index e6dc91487..6d739341b 100644 --- a/ci/patch.toml +++ b/ci/patch.toml @@ -2,10 +2,10 @@ # repository. [patch.crates-io] tokio = { path = "tokio" } -tokio-codec = { path = "tokio-codec" } tokio-executor = { path = "tokio-executor" } tokio-io = { path = "tokio-io" } tokio-macros = { path = "tokio-macros" } tokio-net = { path = "tokio-net" } tokio-sync = { path = "tokio-sync" } tokio-tls = { path = "tokio-tls" } +tokio-util = { path = "tokio-util" } diff --git a/examples/Cargo.toml b/examples/Cargo.toml new file mode 100644 index 000000000..84a546f75 --- /dev/null +++ b/examples/Cargo.toml @@ -0,0 +1,52 @@ +[package] +name = "examples" +version = "0.0.0" +publish = false +edition = "2018" + +[dev-dependencies] +tokio = { version = "=0.2.0-alpha.6", path = "../tokio" } +tokio-util = { version = "=0.2.0-alpha.6", path = "../tokio-util" } + +bytes = "0.4.12" +futures-preview = "=0.3.0-alpha.19" + +[[example]] +name = "chat" +path = "chat.rs" + +[[example]] +name = "connect" +path = "connect.rs" + +[[example]] +name = "echo-udp" +path = "echo-udp.rs" + +[[example]] +name = "echo" +path = "echo.rs" + +[[example]] +name = "hello_world" +path = "hello_world.rs" + +[[example]] +name = "print_each_packet" +path = "print_each_packet.rs" + +[[example]] +name = "proxy" +path = "proxy.rs" + +[[example]] +name = "tinydb" +path = "tinydb.rs" + +[[example]] +name = "udp-client" +path = "udp-client.rs" + +[[example]] +name = "udp-codec" +path = "udp-codec.rs" diff --git a/tokio/examples/README.md b/examples/README.md similarity index 100% rename from tokio/examples/README.md rename to examples/README.md diff --git a/tokio/examples/chat.rs b/examples/chat.rs similarity index 96% rename from tokio/examples/chat.rs rename to examples/chat.rs index 7fc317d84..0a3976d5e 100644 --- a/tokio/examples/chat.rs +++ b/examples/chat.rs @@ -26,17 +26,19 @@ #![warn(rust_2018_idioms)] +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{mpsc, Mutex}; +use tokio_util::codec::{Framed, LinesCodec, LinesCodecError}; + use futures::{Poll, SinkExt, Stream, StreamExt}; -use std::{ - collections::HashMap, env, error::Error, io, net::SocketAddr, pin::Pin, sync::Arc, - task::Context, -}; -use tokio::{ - self, - codec::{Framed, LinesCodec, LinesCodecError}, - net::{TcpListener, TcpStream}, - sync::{mpsc, Mutex}, -}; +use std::collections::HashMap; +use std::env; +use std::error::Error; +use std::io; +use std::net::SocketAddr; +use std::pin::Pin; +use std::sync::Arc; +use std::task::Context; #[tokio::main] async fn main() -> Result<(), Box> { diff --git a/tokio/examples/connect.rs b/examples/connect.rs similarity index 95% rename from tokio/examples/connect.rs rename to examples/connect.rs index c53471d89..0dd14ef24 100644 --- a/tokio/examples/connect.rs +++ b/examples/connect.rs @@ -16,13 +16,14 @@ #![warn(rust_2018_idioms)] +use tokio::io; +use tokio::sync::{mpsc, oneshot}; +use tokio_util::codec::{FramedRead, FramedWrite}; + use futures::{SinkExt, Stream}; -use std::{env, error::Error, net::SocketAddr}; -use tokio::{ - codec::{FramedRead, FramedWrite}, - io, - sync::{mpsc, oneshot}, -}; +use std::env; +use std::error::Error; +use std::net::SocketAddr; #[tokio::main] async fn main() -> Result<(), Box> { @@ -83,10 +84,8 @@ mod tcp { use super::codec; use futures::{future, Sink, SinkExt, Stream, StreamExt}; use std::{error::Error, io, net::SocketAddr}; - use tokio::{ - codec::{FramedRead, FramedWrite}, - net::TcpStream, - }; + use tokio::net::TcpStream; + use tokio_util::codec::{FramedRead, FramedWrite}; pub async fn connect( addr: &SocketAddr, @@ -171,7 +170,7 @@ mod udp { mod codec { use bytes::{BufMut, BytesMut}; use std::io; - use tokio::codec::{Decoder, Encoder}; + use tokio_util::codec::{Decoder, Encoder}; /// A simple `Codec` implementation that just ships bytes around. /// diff --git a/tokio/examples/echo-udp.rs b/examples/echo-udp.rs similarity index 100% rename from tokio/examples/echo-udp.rs rename to examples/echo-udp.rs diff --git a/tokio/examples/echo.rs b/examples/echo.rs similarity index 100% rename from tokio/examples/echo.rs rename to examples/echo.rs diff --git a/tokio/examples/hello_world.rs b/examples/hello_world.rs similarity index 100% rename from tokio/examples/hello_world.rs rename to examples/hello_world.rs diff --git a/tokio/examples/print_each_packet.rs b/examples/print_each_packet.rs similarity index 98% rename from tokio/examples/print_each_packet.rs rename to examples/print_each_packet.rs index 3729c3e96..0a2755457 100644 --- a/tokio/examples/print_each_packet.rs +++ b/examples/print_each_packet.rs @@ -54,10 +54,9 @@ #![warn(rust_2018_idioms)] -use tokio; -use tokio::codec::{BytesCodec, Decoder}; use tokio::net::TcpListener; use tokio::prelude::*; +use tokio_util::codec::{BytesCodec, Decoder}; use std::env; diff --git a/tokio/examples/proxy.rs b/examples/proxy.rs similarity index 100% rename from tokio/examples/proxy.rs rename to examples/proxy.rs diff --git a/tokio/examples/tinydb.rs b/examples/tinydb.rs similarity index 99% rename from tokio/examples/tinydb.rs rename to examples/tinydb.rs index be4951dc5..3fc88f6bd 100644 --- a/tokio/examples/tinydb.rs +++ b/examples/tinydb.rs @@ -41,17 +41,15 @@ #![warn(rust_2018_idioms)] +use tokio::net::TcpListener; +use tokio_util::codec::{Framed, LinesCodec}; + +use futures::{SinkExt, StreamExt}; use std::collections::HashMap; use std::env; use std::error::Error; use std::sync::{Arc, Mutex}; -use tokio; -use tokio::codec::{Framed, LinesCodec}; -use tokio::net::TcpListener; - -use futures::{SinkExt, StreamExt}; - /// The in-memory database shared amongst all clients. /// /// This database will be shared via `Arc`, so to mutate the internal map we're diff --git a/tokio/examples/tinyhttp.rs b/examples/tinyhttp.rs similarity index 100% rename from tokio/examples/tinyhttp.rs rename to examples/tinyhttp.rs diff --git a/tokio/examples/udp-client.rs b/examples/udp-client.rs similarity index 100% rename from tokio/examples/udp-client.rs rename to examples/udp-client.rs diff --git a/tokio/examples/udp-codec.rs b/examples/udp-codec.rs similarity index 94% rename from tokio/examples/udp-codec.rs rename to examples/udp-codec.rs index 7d0aaf69d..baf648867 100644 --- a/tokio/examples/udp-codec.rs +++ b/examples/udp-codec.rs @@ -6,26 +6,23 @@ //! new message with a new destination. Overall, we then use this to construct a //! "ping pong" pair where two sockets are sending messages back and forth. -#![cfg(feature = "rt-full")] #![warn(rust_2018_idioms)] +use tokio::future::FutureExt as TokioFutureExt; +use tokio::io; +use tokio::net::UdpSocket; +use tokio_util::codec::BytesCodec; +use tokio_util::udp::UdpFramed; + +use bytes::Bytes; +use futures::{FutureExt, SinkExt, StreamExt}; use std::env; use std::error::Error; use std::net::SocketAddr; use std::time::Duration; -use bytes::Bytes; - -use futures::{FutureExt, SinkExt, StreamExt}; -use tokio::codec::BytesCodec; -use tokio::future::FutureExt as TokioFutureExt; -use tokio::io; -use tokio::net::{UdpFramed, UdpSocket}; - #[tokio::main] async fn main() -> Result<(), Box> { - let _ = env_logger::init(); - let addr = env::args().nth(1).unwrap_or("127.0.0.1:0".to_string()); // Bind both our sockets and then figure out what ports we got. diff --git a/tokio-codec/CHANGELOG.md b/tokio-codec/CHANGELOG.md deleted file mode 100644 index 2f0ca5f59..000000000 --- a/tokio-codec/CHANGELOG.md +++ /dev/null @@ -1,35 +0,0 @@ -# 0.2.0-alpha.6 (September 30, 2019) - -- Move to `futures-*-preview 0.3.0-alpha.19` -- Move to `pin-project 0.4` - -# 0.2.0-alpha.5 (September 19, 2019) - -- Track tokio release - -# 0.2.0-alpha.4 (August 29, 2019) - -- Track tokio release. - -# 0.2.0-alpha.3 (August 28, 2019) - -### Fix -- Infinite loop in `LinesCodec` (#1489). - -# 0.2.0-alpha.2 (August 17, 2019) - -### Changed -- Update `futures` dependency to 0.3.0-alpha.18. - -# 0.2.0-alpha.1 (August 8, 2019) - -### Changed -- Switch to `async`, `await`, and `std::future`. - -# 0.1.1 (September 26, 2018) - -* Allow setting max line length with `LinesCodec` (#632) - -# 0.1.0 (June 13, 2018) - -* Initial release (#353) diff --git a/tokio-net/Cargo.toml b/tokio-net/Cargo.toml index d7df33973..f19fd7598 100644 --- a/tokio-net/Cargo.toml +++ b/tokio-net/Cargo.toml @@ -62,7 +62,6 @@ uds = [ log = ["tracing/log"] [dependencies] -tokio-codec = { version = "=0.2.0-alpha.6", path = "../tokio-codec" } tokio-executor = { version = "=0.2.0-alpha.6", features = ["blocking"], path = "../tokio-executor" } tokio-io = { version = "=0.2.0-alpha.6", path = "../tokio-io" } tokio-sync = { version = "=0.2.0-alpha.6", path = "../tokio-sync" } diff --git a/tokio-net/src/process/mod.rs b/tokio-net/src/process/mod.rs index 140daa5da..fbe905ade 100644 --- a/tokio-net/src/process/mod.rs +++ b/tokio-net/src/process/mod.rs @@ -55,10 +55,11 @@ //! We can also read input line by line. //! //! ```no_run +//! use tokio::io::{BufReader, AsyncBufReadExt}; +//! use tokio::process::Command; +//! //! use futures_util::stream::StreamExt; -//! use std::process::{Stdio}; -//! use tokio::codec::{FramedRead, LinesCodec}; -//! use tokio_net::process::Command; +//! use std::process::Stdio; //! //! #[tokio::main] //! async fn main() -> Result<(), Box> { @@ -77,7 +78,7 @@ //! let stdout = child.stdout().take() //! .expect("child did not have a handle to stdout"); //! -//! let mut reader = FramedRead::new(stdout, LinesCodec::new()); +//! let mut reader = BufReader::new(stdout).lines(); //! //! // Ensure the child process is spawned in the runtime so it can //! // make progress on its own while we await for any output. diff --git a/tokio-net/src/udp/mod.rs b/tokio-net/src/udp/mod.rs index e5c065854..45656773c 100644 --- a/tokio-net/src/udp/mod.rs +++ b/tokio-net/src/udp/mod.rs @@ -7,9 +7,7 @@ //! //! [`UdpSocket`]: struct.UdpSocket -mod frame; mod socket; pub mod split; -pub use self::frame::UdpFramed; pub use self::socket::UdpSocket; diff --git a/tokio-net/src/udp/socket.rs b/tokio-net/src/udp/socket.rs index c59b564b9..90f212a25 100644 --- a/tokio-net/src/udp/socket.rs +++ b/tokio-net/src/udp/socket.rs @@ -108,7 +108,7 @@ impl UdpSocket { /// /// [`connect`]: #method.connect pub async fn send(&mut self, buf: &[u8]) -> io::Result { - poll_fn(|cx| self.poll_send_priv(cx, buf)).await + poll_fn(|cx| self.poll_send(cx, buf)).await } // Poll IO functions that takes `&self` are provided for the split API. @@ -121,11 +121,8 @@ impl UdpSocket { // While violating this requirement is "safe" from a Rust memory model point // of view, it will result in unexpected behavior in the form of lost // notifications and tasks hanging. - pub(crate) fn poll_send_priv( - &self, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { + #[doc(hidden)] + pub fn poll_send(&self, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { ready!(self.io.poll_write_ready(cx))?; match self.io.get_ref().send(buf) { @@ -150,14 +147,11 @@ impl UdpSocket { /// /// [`connect`]: #method.connect pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result { - poll_fn(|cx| self.poll_recv_priv(cx, buf)).await + poll_fn(|cx| self.poll_recv(cx, buf)).await } - pub(crate) fn poll_recv_priv( - &self, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { + #[doc(hidden)] + pub fn poll_recv(&self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll> { ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?; match self.io.get_ref().recv(buf) { @@ -178,7 +172,7 @@ impl UdpSocket { let mut addrs = target.to_socket_addrs().await?; match addrs.next() { - Some(target) => poll_fn(|cx| self.poll_send_to_priv(cx, buf, &target)).await, + Some(target) => poll_fn(|cx| self.poll_send_to(cx, buf, &target)).await, None => Err(io::Error::new( io::ErrorKind::InvalidInput, "no addresses to send data to", @@ -186,7 +180,9 @@ impl UdpSocket { } } - pub(crate) fn poll_send_to_priv( + // TODO: Public or not? + #[doc(hidden)] + pub fn poll_send_to( &self, cx: &mut Context<'_>, buf: &[u8], @@ -210,10 +206,11 @@ impl UdpSocket { /// to hold the message bytes. If a message is too long to fit in the supplied /// buffer, excess bytes may be discarded. pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { - poll_fn(|cx| self.poll_recv_from_priv(cx, buf)).await + poll_fn(|cx| self.poll_recv_from(cx, buf)).await } - pub(crate) fn poll_recv_from_priv( + #[doc(hidden)] + pub fn poll_recv_from( &self, cx: &mut Context<'_>, buf: &mut [u8], diff --git a/tokio-net/src/udp/split.rs b/tokio-net/src/udp/split.rs index e58f92769..ad8ce0614 100644 --- a/tokio-net/src/udp/split.rs +++ b/tokio-net/src/udp/split.rs @@ -86,7 +86,7 @@ impl UdpSocketRecvHalf { /// to hold the message bytes. If a message is too long to fit in the supplied /// buffer, excess bytes may be discarded. pub async fn recv_from(&mut self, buf: &mut [u8]) -> io::Result<(usize, SocketAddr)> { - poll_fn(|cx| self.0.poll_recv_from_priv(cx, buf)).await + poll_fn(|cx| self.0.poll_recv_from(cx, buf)).await } /// Returns a future that receives a single datagram message on the socket from @@ -102,7 +102,7 @@ impl UdpSocketRecvHalf { /// /// [`connect`]: super::UdpSocket::connect pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result { - poll_fn(|cx| self.0.poll_recv_priv(cx, buf)).await + poll_fn(|cx| self.0.poll_recv(cx, buf)).await } } @@ -120,7 +120,7 @@ impl UdpSocketSendHalf { /// The future will resolve to an error if the IP version of the socket does /// not match that of `target`. pub async fn send_to(&mut self, buf: &[u8], target: &SocketAddr) -> io::Result { - poll_fn(|cx| self.0.poll_send_to_priv(cx, buf, target)).await + poll_fn(|cx| self.0.poll_send_to(cx, buf, target)).await } /// Returns a future that sends data on the socket to the remote address to which it is connected. @@ -131,7 +131,7 @@ impl UdpSocketSendHalf { /// /// [`connect`]: super::UdpSocket::connect pub async fn send(&mut self, buf: &[u8]) -> io::Result { - poll_fn(|cx| self.0.poll_send_priv(cx, buf)).await + poll_fn(|cx| self.0.poll_send(cx, buf)).await } } diff --git a/tokio-net/src/uds/frame.rs b/tokio-net/src/uds/frame.rs deleted file mode 100644 index 584da8ae7..000000000 --- a/tokio-net/src/uds/frame.rs +++ /dev/null @@ -1,175 +0,0 @@ -use super::UnixDatagram; -use bytes::{BufMut, BytesMut}; -use futures::{try_ready, Async, AsyncSink, Poll, Sink, StartSend, Stream}; -use std::io; -use std::os::unix::net::SocketAddr; -use std::path::Path; -use tokio_codec::{Decoder, Encoder}; - -/// A unified `Stream` and `Sink` interface to an underlying `UnixDatagram`, using -/// the `Encoder` and `Decoder` traits to encode and decode frames. -/// -/// Unix datagram sockets work with datagrams, but higher-level code may wants to -/// batch these into meaningful chunks, called "frames". This method layers -/// framing on top of this socket by using the `Encoder` and `Decoder` traits to -/// handle encoding and decoding of messages frames. Note that the incoming and -/// outgoing frame types may be distinct. -/// -/// This function returns a *single* object that is both `Stream` and `Sink`; -/// grouping this into a single object is often useful for layering things which -/// require both read and write access to the underlying object. -/// -/// If you want to work more directly with the streams and sink, consider -/// calling `split` on the `UnixDatagramFramed` returned by this method, which will break -/// them into separate objects, allowing them to interact more easily. -#[must_use = "sinks do nothing unless polled"] -#[derive(Debug)] -pub struct UnixDatagramFramed { - socket: UnixDatagram, - codec: C, - rd: BytesMut, - wr: BytesMut, - out_addr: Option, - flushed: bool, -} - -impl Stream for UnixDatagramFramed { - type Item = (C::Item, SocketAddr); - type Error = C::Error; - - fn poll(&mut self) -> Poll, Self::Error> { - self.rd.reserve(INITIAL_RD_CAPACITY); - - let (_n, addr) = unsafe { - let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut())); - self.rd.advance_mut(n); - (n, addr) - }; - - let span = trace_span!("decoding", from.addr = %addr, dgram.length = _n); - let _e = span.enter(); - trace!("trying to decode a frame..."); - - let frame_res = self.codec.decode(&mut self.rd); - self.rd.clear(); - let frame = frame_res?; - let result = frame.map(|frame| (frame, addr)); - trace!("frame decoded from buffer"); - Ok(Async::Ready(result)) - } -} - -impl, C: Encoder> Sink for UnixDatagramFramed { - type SinkItem = (C::Item, A); - type SinkError = C::Error; - - fn start_send(&mut self, item: Self::SinkItem) -> StartSend { - let span = trace_span!("sending", to.addr = %item.0, flushed = self.flushed); - let _e = span.enter(); - - trace!("sending frame..."); - - if !self.flushed { - match self.poll_complete()? { - Async::Ready(()) => {} - Async::NotReady => return Ok(AsyncSink::NotReady(item)), - } - } - - let (frame, out_addr) = item; - self.codec.encode(frame, &mut self.wr)?; - self.out_addr = Some(out_addr); - self.flushed = false; - trace!(message = "frame encoded", frame.length = pin.wr.len()); - - Ok(AsyncSink::Ready) - } - - fn poll_complete(&mut self) -> Poll<(), C::Error> { - if self.flushed { - return Ok(Async::Ready(())); - } - - let span = trace_span!("flushing", to.addr = %self.out_addr); - let _e = span.enter(); - - let n = { - let out_path = match self.out_addr { - Some(ref out_path) => out_path.as_ref(), - None => { - return Err(io::Error::new( - io::ErrorKind::Other, - "internal error: addr not available while data not flushed", - ) - .into()); - } - }; - - trace!(message = "flushing frame", frame.length = self.wr.len()); - try_ready!(self.socket.poll_send_to(&self.wr, out_path)) - }; - - let wrote_all = n == self.wr.len(); - self.wr.clear(); - self.flushed = true; - - trace!(written.length = n, written.complete = wrote_all); - - if wrote_all { - self.out_addr = None; - Ok(Async::Ready(())) - } else { - Err(io::Error::new( - io::ErrorKind::Other, - "failed to write entire datagram to socket", - ) - .into()) - } - } - - fn close(&mut self) -> Poll<(), C::Error> { - self.poll_complete() - } -} - -const INITIAL_RD_CAPACITY: usize = 64 * 1024; -const INITIAL_WR_CAPACITY: usize = 8 * 1024; - -impl UnixDatagramFramed { - /// Create a new `UnixDatagramFramed` backed by the given socket and codec. - /// - /// See struct level documentation for more details. - pub fn new(socket: UnixDatagram, codec: C) -> UnixDatagramFramed { - UnixDatagramFramed { - socket: socket, - codec: codec, - out_addr: None, - rd: BytesMut::with_capacity(INITIAL_RD_CAPACITY), - wr: BytesMut::with_capacity(INITIAL_WR_CAPACITY), - flushed: true, - } - } - - /// Returns a reference to the underlying I/O stream wrapped by `Framed`. - /// - /// # Note - /// - /// 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) -> &UnixDatagram { - &self.socket - } - - /// Returns a mutable reference to the underlying I/O stream wrapped by - /// `Framed`. - /// - /// # Note - /// - /// 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 UnixDatagram { - &mut self.socket - } -} diff --git a/tokio-net/src/uds/mod.rs b/tokio-net/src/uds/mod.rs index 3b6849925..4447ca5ce 100644 --- a/tokio-net/src/uds/mod.rs +++ b/tokio-net/src/uds/mod.rs @@ -3,16 +3,19 @@ //! This crate provides APIs for using Unix Domain Sockets with Tokio. mod datagram; -// mod frame; -mod incoming; -mod listener; -pub mod split; -mod stream; -mod ucred; - pub use self::datagram::UnixDatagram; + +mod incoming; #[cfg(feature = "async-traits")] pub use self::incoming::Incoming; + +mod listener; pub use self::listener::UnixListener; + +pub mod split; + +mod stream; pub use self::stream::UnixStream; + +mod ucred; pub use self::ucred::UCred; diff --git a/tokio-net/tests/process_stdio.rs b/tokio-net/tests/process_stdio.rs index b7f7c2aff..e98162d4d 100644 --- a/tokio-net/tests/process_stdio.rs +++ b/tokio-net/tests/process_stdio.rs @@ -4,16 +4,15 @@ #[macro_use] extern crate tracing; -use std::env; -use std::io; -use std::process::{ExitStatus, Stdio}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio_net::process::{Child, Command}; use futures_util::future; use futures_util::future::FutureExt; use futures_util::stream::StreamExt; -use tokio::codec::{FramedRead, LinesCodec}; -use tokio::io::AsyncWriteExt; -use tokio_net::process::{Child, Command}; +use std::env; +use std::io; +use std::process::{ExitStatus, Stdio}; mod support; use support::*; @@ -51,7 +50,7 @@ async fn feed_cat(mut cat: Child, n: usize) -> io::Result { }; let read = async { - let mut reader = FramedRead::new(stdout, LinesCodec::new()); + let mut reader = BufReader::new(stdout).lines(); let mut num_lines = 0; // Try to read `n + 1` lines, ensuring the last one is empty diff --git a/tokio-net/tests/udp.rs b/tokio-net/tests/udp.rs index fb8f068a2..124d13879 100644 --- a/tokio-net/tests/udp.rs +++ b/tokio-net/tests/udp.rs @@ -1,11 +1,6 @@ #![warn(rust_2018_idioms)] -use tokio_codec::{Decoder, Encoder}; -use tokio_net::udp::{UdpFramed, UdpSocket}; - -use bytes::{BufMut, BytesMut}; -use futures_util::{future::FutureExt, sink::SinkExt, stream::StreamExt, try_future::try_join}; -use std::io; +use tokio_net::udp::UdpSocket; #[tokio::test] async fn send_recv() -> std::io::Result<()> { @@ -75,72 +70,3 @@ async fn reunite_error() -> std::io::Result<()> { assert!(s.reunite(r1).is_err()); Ok(()) } - -pub struct ByteCodec; - -impl Decoder for ByteCodec { - type Item = Vec; - type Error = io::Error; - - fn decode(&mut self, buf: &mut BytesMut) -> Result>, io::Error> { - let len = buf.len(); - Ok(Some(buf.split_to(len).to_vec())) - } -} - -impl Encoder for ByteCodec { - type Item = Vec; - type Error = io::Error; - - fn encode(&mut self, data: Vec, buf: &mut BytesMut) -> Result<(), io::Error> { - buf.reserve(data.len()); - buf.put(data); - Ok(()) - } -} - -#[tokio::test] -async fn send_framed() -> std::io::Result<()> { - let mut a_soc = UdpSocket::bind("127.0.0.1:0").await?; - let mut b_soc = UdpSocket::bind("127.0.0.1:0").await?; - - let a_addr = a_soc.local_addr()?; - let b_addr = b_soc.local_addr()?; - - // test sending & receiving bytes - { - let mut a = UdpFramed::new(a_soc, ByteCodec); - let mut b = UdpFramed::new(b_soc, ByteCodec); - - let msg = b"4567".to_vec(); - - let send = a.send((msg.clone(), b_addr)); - let recv = b.next().map(|e| e.unwrap()); - let (_, received) = try_join(send, recv).await.unwrap(); - - let (data, addr) = received; - assert_eq!(msg, data); - assert_eq!(a_addr, addr); - - a_soc = a.into_inner(); - b_soc = b.into_inner(); - } - - // test sending & receiving an empty message - { - let mut a = UdpFramed::new(a_soc, ByteCodec); - let mut b = UdpFramed::new(b_soc, ByteCodec); - - let msg = b"".to_vec(); - - let send = a.send((msg.clone(), b_addr)); - let recv = b.next().map(|e| e.unwrap()); - let (_, received) = try_join(send, recv).await.unwrap(); - - let (data, addr) = received; - assert_eq!(msg, data); - assert_eq!(a_addr, addr); - } - - Ok(()) -} diff --git a/tokio-util/CHANGELOG.md b/tokio-util/CHANGELOG.md new file mode 100644 index 000000000..e69de29bb diff --git a/tokio-codec/Cargo.toml b/tokio-util/Cargo.toml similarity index 81% rename from tokio-codec/Cargo.toml rename to tokio-util/Cargo.toml index 6c0d61f90..1ac3a1c87 100644 --- a/tokio-codec/Cargo.toml +++ b/tokio-util/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "tokio-codec" +name = "tokio-util" # When releasing to crates.io: # - Remove path dependencies # - Update html_root_url. @@ -13,14 +13,15 @@ authors = ["Tokio Contributors "] license = "MIT" repository = "https://github.com/tokio-rs/tokio" homepage = "https://tokio.rs" -documentation = "https://docs.rs/tokio-codec/0.2.0-alpha.6/tokio_codec" +documentation = "https://docs.rs/tokio-util/0.2.0-alpha.6/tokio_util" description = """ -Utilities for encoding and decoding frames. +Additional utilities for working with Tokio. """ categories = ["asynchronous"] [dependencies] tokio-io = { version = "=0.2.0-alpha.6", path = "../tokio-io" } +tokio = { version = "=0.2.0-alpha.6", path = "../tokio" } bytes = "0.4.7" futures-core-preview = "=0.3.0-alpha.19" diff --git a/tokio-codec/LICENSE b/tokio-util/LICENSE similarity index 100% rename from tokio-codec/LICENSE rename to tokio-util/LICENSE diff --git a/tokio-codec/README.md b/tokio-util/README.md similarity index 95% rename from tokio-codec/README.md rename to tokio-util/README.md index 359070597..11b2b1841 100644 --- a/tokio-codec/README.md +++ b/tokio-util/README.md @@ -1,4 +1,4 @@ -# tokio-codec +# tokio-util Utilities for encoding and decoding frames. diff --git a/tokio-codec/src/bytes_codec.rs b/tokio-util/src/codec/bytes_codec.rs similarity index 92% rename from tokio-codec/src/bytes_codec.rs rename to tokio-util/src/codec/bytes_codec.rs index c5bf62b80..a7d424e9e 100644 --- a/tokio-codec/src/bytes_codec.rs +++ b/tokio-util/src/codec/bytes_codec.rs @@ -1,5 +1,6 @@ -use crate::decoder::Decoder; -use crate::encoder::Encoder; +use crate::codec::decoder::Decoder; +use crate::codec::encoder::Encoder; + use bytes::{BufMut, Bytes, BytesMut}; use std::io; diff --git a/tokio-codec/src/decoder.rs b/tokio-util/src/codec/decoder.rs similarity index 98% rename from tokio-codec/src/decoder.rs rename to tokio-util/src/codec/decoder.rs index 41b950790..720e0b6ef 100644 --- a/tokio-codec/src/decoder.rs +++ b/tokio-util/src/codec/decoder.rs @@ -1,10 +1,10 @@ -use bytes::BytesMut; -use std::io; +use crate::codec::encoder::Encoder; +use crate::codec::Framed; + use tokio_io::{AsyncRead, AsyncWrite}; -use super::encoder::Encoder; - -use super::Framed; +use bytes::BytesMut; +use std::io; /// Decoding of frames via buffers. /// @@ -75,7 +75,7 @@ pub trait Decoder { /// # use std::io; /// # /// # use bytes::BytesMut; - /// # use tokio_codec::Decoder; + /// # use tokio_util::codec::Decoder; /// # /// # struct MyCodec; /// # diff --git a/tokio-codec/src/encoder.rs b/tokio-util/src/codec/encoder.rs similarity index 100% rename from tokio-codec/src/encoder.rs rename to tokio-util/src/codec/encoder.rs diff --git a/tokio-codec/src/framed.rs b/tokio-util/src/codec/framed.rs similarity index 97% rename from tokio-codec/src/framed.rs rename to tokio-util/src/codec/framed.rs index ac733d59f..e2eb82cb7 100644 --- a/tokio-codec/src/framed.rs +++ b/tokio-util/src/codec/framed.rs @@ -1,9 +1,7 @@ -#![allow(deprecated)] - -use crate::decoder::Decoder; -use crate::encoder::Encoder; -use crate::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2}; -use crate::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2}; +use crate::codec::decoder::Decoder; +use crate::codec::encoder::Encoder; +use crate::codec::framed_read::{framed_read2, framed_read2_with_buffer, FramedRead2}; +use crate::codec::framed_write::{framed_write2, framed_write2_with_buffer, FramedWrite2}; use tokio_io::{AsyncBufRead, AsyncRead, AsyncWrite}; diff --git a/tokio-codec/src/framed_read.rs b/tokio-util/src/codec/framed_read.rs similarity index 99% rename from tokio-codec/src/framed_read.rs rename to tokio-util/src/codec/framed_read.rs index 85f221b9a..71f22150d 100644 --- a/tokio-codec/src/framed_read.rs +++ b/tokio-util/src/codec/framed_read.rs @@ -1,5 +1,5 @@ -use super::framed::Fuse; -use super::Decoder; +use crate::codec::framed::Fuse; +use crate::codec::Decoder; use tokio_io::AsyncRead; diff --git a/tokio-codec/src/framed_write.rs b/tokio-util/src/codec/framed_write.rs similarity index 98% rename from tokio-codec/src/framed_write.rs rename to tokio-util/src/codec/framed_write.rs index f89e46112..fa8aba175 100644 --- a/tokio-codec/src/framed_write.rs +++ b/tokio-util/src/codec/framed_write.rs @@ -1,8 +1,6 @@ -#![allow(deprecated)] - -use super::framed::Fuse; -use crate::decoder::Decoder; -use crate::encoder::Encoder; +use crate::codec::decoder::Decoder; +use crate::codec::encoder::Encoder; +use crate::codec::framed::Fuse; use tokio_io::{AsyncBufRead, AsyncRead, AsyncWrite}; diff --git a/tokio-codec/src/length_delimited.rs b/tokio-util/src/codec/length_delimited.rs similarity index 95% rename from tokio-codec/src/length_delimited.rs rename to tokio-util/src/codec/length_delimited.rs index ff3400de3..579417a2e 100644 --- a/tokio-codec/src/length_delimited.rs +++ b/tokio-util/src/codec/length_delimited.rs @@ -16,7 +16,7 @@ //! //! ``` //! use tokio_io::{AsyncRead, AsyncWrite}; -//! use tokio_codec::{Framed, LengthDelimitedCodec}; +//! use tokio_util::codec::{Framed, LengthDelimitedCodec}; //! //! fn bind_transport(io: T) //! -> Framed @@ -39,8 +39,8 @@ //! Specifically, given the following: //! //! ``` -//! use tokio::codec::{Framed, LengthDelimitedCodec}; //! use tokio::prelude::*; +//! use tokio_util::codec::{Framed, LengthDelimitedCodec}; //! //! use bytes::Bytes; //! @@ -79,7 +79,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(0) // default value @@ -113,7 +113,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(0) // default value @@ -145,7 +145,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(0) // default value @@ -179,7 +179,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(0) // default value @@ -223,7 +223,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(1) // length of hdr1 @@ -269,7 +269,7 @@ //! //! ``` //! # use tokio_io::AsyncRead; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn bind_read(io: T) { //! LengthDelimitedCodec::builder() //! .length_field_offset(1) // length of hdr1 @@ -314,7 +314,7 @@ //! //! ``` //! # use tokio_io::AsyncWrite; -//! # use tokio_codec::LengthDelimitedCodec; +//! # use tokio_util::codec::LengthDelimitedCodec; //! # fn write_frame(io: T) { //! # let _ = //! LengthDelimitedCodec::builder() @@ -340,12 +340,14 @@ //! [`Encoder`]: ../trait.Encoder.html //! [`BytesMut`]: https://docs.rs/bytes/0.4/bytes/struct.BytesMut.html -use crate::{Decoder, Encoder, Framed, FramedRead, FramedWrite}; +use crate::codec::{Decoder, Encoder, Framed, FramedRead, FramedWrite}; + +use tokio_io::{AsyncRead, AsyncWrite}; + use bytes::{Buf, BufMut, Bytes, BytesMut, IntoBuf}; use std::error::Error as StdError; use std::io::{self, Cursor}; use std::{cmp, fmt}; -use tokio_io::{AsyncRead, AsyncWrite}; /// Configure length delimited `LengthDelimitedCodec`s. /// @@ -604,7 +606,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -648,7 +650,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -672,7 +674,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -696,7 +698,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -730,7 +732,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -754,7 +756,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -777,7 +779,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -798,7 +800,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -822,7 +824,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -841,7 +843,7 @@ impl Builder { /// # Examples /// /// ``` - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// # pub fn main() { /// LengthDelimitedCodec::builder() /// .length_field_offset(0) @@ -864,7 +866,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncRead; - /// use tokio_codec::LengthDelimitedCodec; + /// use tokio_util::codec::LengthDelimitedCodec; /// /// # fn bind_read(io: T) { /// LengthDelimitedCodec::builder() @@ -889,7 +891,7 @@ impl Builder { /// /// ``` /// # use tokio_io::AsyncWrite; - /// # use tokio_codec::LengthDelimitedCodec; + /// # use tokio_util::codec::LengthDelimitedCodec; /// # fn write_frame(io: T) { /// LengthDelimitedCodec::builder() /// .length_field_length(2) @@ -910,7 +912,7 @@ impl Builder { /// /// ``` /// # use tokio_io::{AsyncRead, AsyncWrite}; - /// # use tokio_codec::LengthDelimitedCodec; + /// # use tokio_util::codec::LengthDelimitedCodec; /// # fn write_frame(io: T) { /// # let _ = /// LengthDelimitedCodec::builder() diff --git a/tokio-codec/src/lines_codec.rs b/tokio-util/src/codec/lines_codec.rs similarity index 97% rename from tokio-codec/src/lines_codec.rs rename to tokio-util/src/codec/lines_codec.rs index 19741c184..bca697237 100644 --- a/tokio-codec/src/lines_codec.rs +++ b/tokio-util/src/codec/lines_codec.rs @@ -1,5 +1,6 @@ -use crate::decoder::Decoder; -use crate::encoder::Encoder; +use crate::codec::decoder::Decoder; +use crate::codec::encoder::Encoder; + use bytes::{BufMut, BytesMut}; use std::{cmp, fmt, io, str, usize}; @@ -70,13 +71,13 @@ impl LinesCodec { /// /// ``` /// use std::usize; - /// use tokio_codec::LinesCodec; + /// use tokio_util::codec::LinesCodec; /// /// let codec = LinesCodec::new(); /// assert_eq!(codec.max_length(), usize::MAX); /// ``` /// ``` - /// use tokio_codec::LinesCodec; + /// use tokio_util::codec::LinesCodec; /// /// let codec = LinesCodec::new_with_max_length(256); /// assert_eq!(codec.max_length(), 256); diff --git a/tokio-codec/src/macros.rs b/tokio-util/src/codec/macros.rs similarity index 100% rename from tokio-codec/src/macros.rs rename to tokio-util/src/codec/macros.rs diff --git a/tokio-codec/src/lib.rs b/tokio-util/src/codec/mod.rs similarity index 50% rename from tokio-codec/src/lib.rs rename to tokio-util/src/codec/mod.rs index 6aaa15a2b..fd080dc18 100644 --- a/tokio-codec/src/lib.rs +++ b/tokio-util/src/codec/mod.rs @@ -1,16 +1,3 @@ -#![doc(html_root_url = "https://docs.rs/tokio-codec/0.2.0-alpha.6")] -#![warn( - missing_debug_implementations, - missing_docs, - rust_2018_idioms, - unreachable_pub -)] -#![deny(intra_doc_link_resolution_failure)] -#![doc(test( - no_crate_inject, - attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) -))] - //! Utilities for encoding and decoding frames. //! //! Contains adapters to go from streams of bytes, [`AsyncRead`] and @@ -26,19 +13,25 @@ mod macros; mod bytes_codec; -mod decoder; -mod encoder; -mod framed; -mod framed_read; -mod framed_write; -pub mod length_delimited; -mod lines_codec; +pub use self::bytes_codec::BytesCodec; -pub use crate::bytes_codec::BytesCodec; -pub use crate::decoder::Decoder; -pub use crate::encoder::Encoder; -pub use crate::framed::{Framed, FramedParts}; -pub use crate::framed_read::FramedRead; -pub use crate::framed_write::FramedWrite; -pub use crate::length_delimited::{LengthDelimitedCodec, LengthDelimitedCodecError}; -pub use crate::lines_codec::{LinesCodec, LinesCodecError}; +mod decoder; +pub use self::decoder::Decoder; + +mod encoder; +pub use self::encoder::Encoder; + +mod framed; +pub use self::framed::{Framed, FramedParts}; + +mod framed_read; +pub use self::framed_read::FramedRead; + +mod framed_write; +pub use self::framed_write::FramedWrite; + +pub mod length_delimited; +pub use self::length_delimited::{LengthDelimitedCodec, LengthDelimitedCodecError}; + +mod lines_codec; +pub use self::lines_codec::{LinesCodec, LinesCodecError}; diff --git a/tokio-util/src/lib.rs b/tokio-util/src/lib.rs new file mode 100644 index 000000000..5a64673c8 --- /dev/null +++ b/tokio-util/src/lib.rs @@ -0,0 +1,17 @@ +#![doc(html_root_url = "https://docs.rs/tokio-util/0.2.0-alpha.6")] +#![warn( + missing_debug_implementations, + missing_docs, + rust_2018_idioms, + unreachable_pub +)] +#![deny(intra_doc_link_resolution_failure)] +#![doc(test( + no_crate_inject, + attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables)) +))] + +//! Utilities for working with Tokio. + +pub mod codec; +pub mod udp; diff --git a/tokio-net/src/udp/frame.rs b/tokio-util/src/udp/frame.rs similarity index 86% rename from tokio-net/src/udp/frame.rs rename to tokio-util/src/udp/frame.rs index d694e2fed..4a8c3c0b1 100644 --- a/tokio-net/src/udp/frame.rs +++ b/tokio-util/src/udp/frame.rs @@ -1,14 +1,14 @@ -use super::UdpSocket; +use crate::codec::{Decoder, Encoder}; -use tokio_codec::{Decoder, Encoder}; +use tokio::net::UdpSocket; use bytes::{BufMut, BytesMut}; -use core::task::{Context, Poll}; use futures_core::{ready, Stream}; use futures_sink::Sink; use std::io; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::pin::Pin; +use std::task::{Context, Poll}; /// A unified `Stream` and `Sink` interface to an underlying `UdpSocket`, using /// the `Encoder` and `Decoder` traits to encode and decode frames. @@ -47,22 +47,17 @@ impl Stream for UdpFramed { let (_n, addr) = unsafe { // Read into the buffer without having to initialize the memory. - let res = ready!(Pin::new(&mut pin.socket).poll_recv_from_priv(cx, pin.rd.bytes_mut())); + let res = ready!(Pin::new(&mut pin.socket).poll_recv_from(cx, pin.rd.bytes_mut())); let (n, addr) = res?; pin.rd.advance_mut(n); (n, addr) }; - let span = trace_span!("decoding", from.addr = %addr, dgram.length = _n); - let _e = span.enter(); - trace!("trying to decode a frame..."); - let frame_res = pin.codec.decode(&mut pin.rd); pin.rd.clear(); let frame = frame_res?; let result = frame.map(|frame| Ok((frame, addr))); // frame -> (frame, addr) - trace!("frame decoded from buffer"); Poll::Ready(result) } } @@ -84,16 +79,11 @@ impl Sink<(C::Item, SocketAddr)> for UdpFramed { fn start_send(self: Pin<&mut Self>, item: (C::Item, SocketAddr)) -> Result<(), Self::Error> { let (frame, out_addr) = item; - let span = trace_span!("sending", to.addr = %out_addr); - let _e = span.enter(); - trace!("encoding frame..."); - let pin = self.get_mut(); pin.codec.encode(frame, &mut pin.wr)?; pin.out_addr = out_addr; pin.flushed = false; - trace!(message = "frame encoded", frame.length = pin.wr.len()); Ok(()) } @@ -110,18 +100,12 @@ impl Sink<(C::Item, SocketAddr)> for UdpFramed { .. } = *self; - let span = trace_span!("flushing", to.addr = %out_addr, frame.length = wr.len()); - let _e = span.enter(); - trace!("flushing frame..."); - - let n = ready!(socket.poll_send_to_priv(cx, &wr, &out_addr))?; + let n = ready!(socket.poll_send_to(cx, &wr, &out_addr))?; let wrote_all = n == self.wr.len(); self.wr.clear(); self.flushed = true; - trace!(written.length = n, written.complete = wrote_all); - let res = if wrote_all { Ok(()) } else { diff --git a/tokio-util/src/udp/mod.rs b/tokio-util/src/udp/mod.rs new file mode 100644 index 000000000..7c4bb2b3c --- /dev/null +++ b/tokio-util/src/udp/mod.rs @@ -0,0 +1,4 @@ +//! UDP framing + +mod frame; +pub use self::frame::UdpFramed; diff --git a/tokio-codec/tests/codecs.rs b/tokio-util/tests/codecs.rs similarity index 98% rename from tokio-codec/tests/codecs.rs rename to tokio-util/tests/codecs.rs index 2c227a181..4709854ab 100644 --- a/tokio-codec/tests/codecs.rs +++ b/tokio-util/tests/codecs.rs @@ -1,7 +1,8 @@ #![warn(rust_2018_idioms)] +use tokio_util::codec::{BytesCodec, Decoder, Encoder, LinesCodec}; + use bytes::{BufMut, Bytes, BytesMut}; -use tokio_codec::{BytesCodec, Decoder, Encoder, LinesCodec}; #[test] fn bytes_decoder() { diff --git a/tokio-codec/tests/framed.rs b/tokio-util/tests/framed.rs similarity index 97% rename from tokio-codec/tests/framed.rs rename to tokio-util/tests/framed.rs index 53582fb3d..7aa40d881 100644 --- a/tokio-codec/tests/framed.rs +++ b/tokio-util/tests/framed.rs @@ -1,8 +1,8 @@ #![warn(rust_2018_idioms)] use tokio::prelude::*; -use tokio_codec::{Decoder, Encoder, Framed, FramedParts}; use tokio_test::assert_ok; +use tokio_util::codec::{Decoder, Encoder, Framed, FramedParts}; use bytes::{Buf, BufMut, BytesMut, IntoBuf}; use std::io::{self, Read}; diff --git a/tokio-codec/tests/framed_read.rs b/tokio-util/tests/framed_read.rs similarity index 99% rename from tokio-codec/tests/framed_read.rs rename to tokio-util/tests/framed_read.rs index 72c910ba4..2064b9fb1 100644 --- a/tokio-codec/tests/framed_read.rs +++ b/tokio-util/tests/framed_read.rs @@ -1,9 +1,9 @@ #![warn(rust_2018_idioms)] use tokio::prelude::*; -use tokio_codec::{Decoder, FramedRead}; use tokio_test::assert_ready; use tokio_test::task::MockTask; +use tokio_util::codec::{Decoder, FramedRead}; use bytes::{Buf, BytesMut, IntoBuf}; use std::collections::VecDeque; diff --git a/tokio-codec/tests/framed_write.rs b/tokio-util/tests/framed_write.rs similarity index 99% rename from tokio-codec/tests/framed_write.rs rename to tokio-util/tests/framed_write.rs index 9d85c4d83..90a54af16 100644 --- a/tokio-codec/tests/framed_write.rs +++ b/tokio-util/tests/framed_write.rs @@ -1,9 +1,9 @@ #![warn(rust_2018_idioms)] -use tokio_codec::{Encoder, FramedWrite}; use tokio_io::AsyncWrite; use tokio_test::assert_ready; use tokio_test::task::MockTask; +use tokio_util::codec::{Encoder, FramedWrite}; use bytes::{BufMut, BytesMut}; use futures_sink::Sink; diff --git a/tokio-codec/tests/length_delimited.rs b/tokio-util/tests/length_delimited.rs similarity index 99% rename from tokio-codec/tests/length_delimited.rs rename to tokio-util/tests/length_delimited.rs index 404786e31..c78dfac8d 100644 --- a/tokio-codec/tests/length_delimited.rs +++ b/tokio-util/tests/length_delimited.rs @@ -1,12 +1,12 @@ #![warn(rust_2018_idioms)] -use tokio::codec::*; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::prelude::*; use tokio_test::task::MockTask; use tokio_test::{ assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok, }; +use tokio_util::codec::*; use bytes::{BufMut, Bytes, BytesMut}; use futures_util::pin_mut; diff --git a/tokio-util/tests/udp.rs b/tokio-util/tests/udp.rs new file mode 100644 index 000000000..89f5c6a95 --- /dev/null +++ b/tokio-util/tests/udp.rs @@ -0,0 +1,79 @@ +use tokio::net::UdpSocket; +use tokio_util::codec::{Decoder, Encoder}; +use tokio_util::udp::UdpFramed; + +use bytes::{BufMut, BytesMut}; +use futures_util::future::FutureExt; +use futures_util::sink::SinkExt; +use futures_util::stream::StreamExt; +use futures_util::try_future::try_join; +use std::io; + +#[tokio::test] +async fn send_framed() -> std::io::Result<()> { + let mut a_soc = UdpSocket::bind("127.0.0.1:0").await?; + let mut b_soc = UdpSocket::bind("127.0.0.1:0").await?; + + let a_addr = a_soc.local_addr()?; + let b_addr = b_soc.local_addr()?; + + // test sending & receiving bytes + { + let mut a = UdpFramed::new(a_soc, ByteCodec); + let mut b = UdpFramed::new(b_soc, ByteCodec); + + let msg = b"4567".to_vec(); + + let send = a.send((msg.clone(), b_addr)); + let recv = b.next().map(|e| e.unwrap()); + let (_, received) = try_join(send, recv).await.unwrap(); + + let (data, addr) = received; + assert_eq!(msg, data); + assert_eq!(a_addr, addr); + + a_soc = a.into_inner(); + b_soc = b.into_inner(); + } + + // test sending & receiving an empty message + { + let mut a = UdpFramed::new(a_soc, ByteCodec); + let mut b = UdpFramed::new(b_soc, ByteCodec); + + let msg = b"".to_vec(); + + let send = a.send((msg.clone(), b_addr)); + let recv = b.next().map(|e| e.unwrap()); + let (_, received) = try_join(send, recv).await.unwrap(); + + let (data, addr) = received; + assert_eq!(msg, data); + assert_eq!(a_addr, addr); + } + + Ok(()) +} + +pub struct ByteCodec; + +impl Decoder for ByteCodec { + type Item = Vec; + type Error = io::Error; + + fn decode(&mut self, buf: &mut BytesMut) -> Result>, io::Error> { + let len = buf.len(); + Ok(Some(buf.split_to(len).to_vec())) + } +} + +impl Encoder for ByteCodec { + type Item = Vec; + type Error = io::Error; + + fn encode(&mut self, data: Vec, buf: &mut BytesMut) -> Result<(), io::Error> { + buf.reserve(data.len()); + buf.put(data); + Ok(()) + } +} diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index bbd9e75b5..ac78708e2 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -25,7 +25,6 @@ keywords = ["io", "async", "non-blocking", "futures"] [features] default = [ - "codec", "fs", "io", "net", @@ -36,7 +35,6 @@ default = [ "timer", ] -codec = ["io", "tokio-codec", "bytes"] fs = [] io = ["tokio-io"] macros = ["tokio-macros"] @@ -71,12 +69,10 @@ futures-sink-preview = "=0.3.0-alpha.19" futures-util-preview = { version = "=0.3.0-alpha.19", features = ["sink"] } # Everything else is optional... -bytes = { version = "0.4", optional = true } crossbeam-utils = { version = "0.6.0", optional = true } num_cpus = { version = "1.8.0", optional = true } # Backs `DelayQueue` slab = { version = "0.4.1", optional = true } -tokio-codec = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-codec" } tokio-io = { version = "=0.2.0-alpha.6", optional = true, features = ["util"], path = "../tokio-io" } tokio-executor = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-executor" } tokio-macros = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-macros" } @@ -90,6 +86,7 @@ tokio-executor = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio [dev-dependencies] tokio-test = { version = "=0.2.0-alpha.6", path = "../tokio-test" } +tokio-util = { version = "=0.2.0-alpha.6", path = "../tokio-util" } futures-preview = "=0.3.0-alpha.19" futures-util-preview = "=0.3.0-alpha.19" diff --git a/tokio/README.md b/tokio/README.md index 14b8b238d..087ab0258 100644 --- a/tokio/README.md +++ b/tokio/README.md @@ -93,7 +93,7 @@ async fn main() -> Result<(), Box> { } ``` -More examples can be found [here](tokio/examples). Note that the `master` branch +More examples can be found [here](../examples). Note that the `master` branch is currently being updated to use `async` / `await`. The examples are not fully ported. Examples for stable Tokio can be found [here](https://github.com/tokio-rs/tokio/tree/v0.1.x/tokio/examples). diff --git a/tokio/src/net.rs b/tokio/src/net.rs index d412b465e..6aa1e499c 100644 --- a/tokio/src/net.rs +++ b/tokio/src/net.rs @@ -52,10 +52,10 @@ pub mod udp { //! The main struct for UDP is the [`UdpSocket`], which represents a UDP socket. //! //! [`UdpSocket`]: struct.UdpSocket.html - pub use tokio_net::udp::{split, UdpFramed, UdpSocket}; + pub use tokio_net::udp::{split, UdpSocket}; } #[cfg(feature = "udp")] -pub use self::udp::{UdpFramed, UdpSocket}; +pub use self::udp::UdpSocket; #[cfg(all(unix, feature = "uds"))] pub mod unix {