diff --git a/tokio-uds/Cargo.toml b/tokio-uds/Cargo.toml index 8a4437dcf..9ac6bbb69 100644 --- a/tokio-uds/Cargo.toml +++ b/tokio-uds/Cargo.toml @@ -26,16 +26,18 @@ async-traits = [] [dependencies] bytes = "0.4.8" -futures-core-preview = { version = "0.3.0-alpha.17" } -iovec = "0.1.2" -libc = "0.2.42" -log = "0.4.2" mio = "0.6.14" mio-uds = "0.6.5" tokio-codec = { version = "0.2.0", path = "../tokio-codec" } tokio-reactor = { version = "0.2.0", path = "../tokio-reactor" } tokio-io = { version = "0.2.0", path = "../tokio-io" } +futures-core-preview = { version = "= 0.3.0-alpha.17" } +futures-util-preview = { version = "= 0.3.0-alpha.17" } +iovec = "0.1.2" +libc = "0.2.42" +log = "0.4.2" + [dev-dependencies] tokio = { version = "0.2.0", path = "../tokio" } tempfile = "3" diff --git a/tokio-uds/src/datagram.rs b/tokio-uds/src/datagram.rs index f36299ad8..1ae2df386 100644 --- a/tokio-uds/src/datagram.rs +++ b/tokio-uds/src/datagram.rs @@ -1,6 +1,7 @@ -use crate::{Recv, RecvFrom, Send, SendTo}; +use tokio_reactor::{Handle, PollEvented}; + use futures_core::ready; -use mio::Ready; +use futures_util::future::poll_fn; use mio_uds; use std::convert::TryFrom; use std::fmt; @@ -9,9 +10,7 @@ use std::net::Shutdown; use std::os::unix::io::{AsRawFd, RawFd}; use std::os::unix::net::{self, SocketAddr}; use std::path::Path; -use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_reactor::{Handle, PollEvented}; /// An I/O object representing a Unix datagram socket. pub struct UnixDatagram { @@ -71,37 +70,9 @@ impl UnixDatagram { self.io.get_ref().connect(path) } - /// Returns a future that sends data on the socket to the remote address to which it is connected. - /// On success, the future will resolve to the number of bytes written. - /// - /// The [`connect`] method will connect this socket to a remote address. The future - /// will resolve to an error if the socket is not connected. - /// - /// [`connect`]: #method.connect - pub fn send<'a, 'b>(&'a mut self, buf: &'b [u8]) -> Send<'a, 'b> { - Send::new(self, buf) - } - - /// Sends data on the socket to the remote address to which it is connected. - /// - /// The [`connect`] method will connect this socket to a remote address. This - /// method will fail if the socket is not connected. - /// - /// [`connect`]: #method.connect - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_written))`. - /// - /// If the socket is not ready for writing, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes writable. - pub fn poll_send( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - ) -> Poll> { - self.poll_send_priv(cx, buf) + /// Sends data on the socket to the socket's peer. + pub async fn send(&mut self, buf: &[u8]) -> io::Result { + poll_fn(|cx| self.poll_send_priv(cx, buf)).await } // Poll IO functions that takes `&self` are provided for the split API. @@ -130,47 +101,9 @@ impl UnixDatagram { } } - /// Returns a future that receives a single datagram message on the socket from - /// the remote address to which it is connected. On success, the future will resolve - /// to the number of bytes read. - /// - /// The function must be called with valid byte array `buf` of sufficient size to - /// hold the message bytes. If a message is too long to fit in the supplied buffer, - /// excess bytes may be discarded. - /// - /// The [`connect`] method will connect this socket to a remote address. The future - /// will fail if the socket is not connected. - /// - /// [`connect`]: #method.connect - pub fn recv<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> Recv<'a, 'b> { - Recv::new(self, buf) - } - - /// Receives a single datagram message on the socket from the remote address to - /// which it is connected. On success, returns the number of bytes read. - /// - /// The function must be called with valid byte array `buf` of sufficient size to - /// hold the message bytes. If a message is too long to fit in the supplied buffer, - /// excess bytes may be discarded. - /// - /// The [`connect`] method will connect this socket to a remote address. This - /// method will fail if the socket is not connected. - /// - /// [`connect`]: #method.connect - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_read))`. - /// - /// If no data is available for reading, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes receivable or is closed. - pub fn poll_recv( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.poll_recv_priv(cx, buf) + /// Receives data from the socket. + pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result { + poll_fn(|cx| self.poll_recv_priv(cx, buf)).await } pub(crate) fn poll_recv_priv( @@ -189,38 +122,12 @@ impl UnixDatagram { } } - /// Returns a future that sends data on the socket to the given address. - /// On success, the future will resolve to the number of bytes written. - /// - /// The future will resolve to an error if the IP version of the socket does - /// not match that of `target`. - pub fn send_to<'a, 'b, P>(&'a mut self, buf: &'b [u8], target: P) -> SendTo<'a, 'b, P> + /// Sends data on the socket to the specified address. + pub async fn send_to

(&mut self, buf: &[u8], target: P) -> io::Result where P: AsRef + Unpin, { - SendTo::new(self, buf, target) - } - - /// Sends data on the socket to the given address. On success, returns the - /// number of bytes written. - /// - /// This will return an error when the IP version of the local socket - /// does not match that of `target`. - /// - /// # Return - /// - /// On success, returns `Poll::Ready(Ok(num_bytes_written))`. - /// - /// If the socket is not ready for writing, the method returns - /// `Poll::Pending` and arranges for the current task to receive a - /// notification when the socket becomes writable. - pub fn poll_send_to>( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &[u8], - target: P, - ) -> Poll> { - self.poll_send_to_priv(cx, buf, target.as_ref()) + poll_fn(|cx| self.poll_send_to_priv(cx, buf, target.as_ref())).await } pub(crate) fn poll_send_to_priv( @@ -240,24 +147,9 @@ impl UnixDatagram { } } - /// Returns a future that receives a single datagram on the socket. On success, - /// the future resolves to the number of bytes read and the origin. - /// - /// The function must be called with valid byte array `buf` of sufficient size - /// to hold the message bytes. If a message is too long to fit in the supplied - /// buffer, excess bytes may be discarded. - pub fn recv_from<'a, 'b>(&'a mut self, buf: &'b mut [u8]) -> RecvFrom<'a, 'b> { - RecvFrom::new(self, buf) - } - - /// Receives data from the socket. On success, returns the number of bytes - /// read and the address from whence the data came. - pub fn poll_recv_from( - self: Pin<&mut Self>, - cx: &mut Context<'_>, - buf: &mut [u8], - ) -> Poll> { - self.poll_recv_from_priv(cx, buf) + /// Receives data from the socket. + 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 } pub(crate) fn poll_recv_from_priv( @@ -276,16 +168,6 @@ impl UnixDatagram { } } - /// Test whether this socket is ready to be read or not. - pub fn poll_read_ready(&self, cx: &mut Context<'_>, ready: Ready) -> Poll> { - self.io.poll_read_ready(cx, ready) - } - - /// Test whether this socket is ready to be written to or not. - pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll> { - self.io.poll_write_ready(cx) - } - /// Returns the local address that this socket is bound to. pub fn local_addr(&self) -> io::Result { self.io.get_ref().local_addr() diff --git a/tokio-uds/src/lib.rs b/tokio-uds/src/lib.rs index 9ea1135b4..4c74952f5 100644 --- a/tokio-uds/src/lib.rs +++ b/tokio-uds/src/lib.rs @@ -3,6 +3,7 @@ #![deny(missing_docs, missing_debug_implementations, rust_2018_idioms)] #![cfg_attr(test, deny(warnings))] #![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))] +#![feature(async_await)] //! Unix Domain Sockets for Tokio. //! @@ -12,21 +13,12 @@ mod datagram; // mod frame; mod incoming; mod listener; -mod recv; -mod recv_from; -mod send; -mod send_to; mod stream; mod ucred; pub use crate::datagram::UnixDatagram; -pub use crate::recv::Recv; -pub use crate::recv_from::RecvFrom; -pub use crate::send::Send; -pub use crate::send_to::SendTo; -// pub use crate::frame::UnixDatagramFramed; #[cfg(feature = "async-traits")] pub use crate::incoming::Incoming; -pub use crate::listener::{Accept, UnixListener}; -pub use crate::stream::{ConnectFuture, UnixStream}; +pub use crate::listener::UnixListener; +pub use crate::stream::UnixStream; pub use crate::ucred::UCred; diff --git a/tokio-uds/src/listener.rs b/tokio-uds/src/listener.rs index 22ce198d1..9e9f7c2ab 100644 --- a/tokio-uds/src/listener.rs +++ b/tokio-uds/src/listener.rs @@ -1,17 +1,18 @@ use crate::UnixStream; + +use tokio_reactor::{Handle, PollEvented}; + use futures_core::ready; +use futures_util::future::poll_fn; use mio::Ready; use mio_uds; use std::convert::TryFrom; use std::fmt; -use std::future::Future; use std::io; use std::os::unix::io::{AsRawFd, RawFd}; use std::os::unix::net::{self, SocketAddr}; use std::path::Path; -use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_reactor::{Handle, PollEvented}; /// A Unix socket which can accept connections from other Unix sockets. pub struct UnixListener { @@ -45,42 +46,19 @@ impl UnixListener { self.io.get_ref().local_addr() } - /// Test whether this socket is ready to be read or not. - pub fn poll_read_ready(&self, cx: &mut Context<'_>, ready: Ready) -> Poll> { - self.io.poll_read_ready(cx, ready) - } - /// Returns the value of the `SO_ERROR` option. pub fn take_error(&self) -> io::Result> { self.io.get_ref().take_error() } - /// Returns a future that attempts to accept a connection and creates a new - /// connected `UnixStream` if successful. - pub fn accept(&mut self) -> Accept<'_> { - Accept { listener: self } + /// Accepts a new incoming connection to this listener. + #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 + pub async fn accept(&mut self) -> io::Result<(UnixStream, SocketAddr)> { + poll_fn(|cx| self.poll_accept(cx)).await } - /// Attempt to accept a connection and create a new connected `UnixStream` - /// if successful. - /// - /// This function will attempt an accept operation, but will not block - /// waiting for it to complete. If the operation would block then a "would - /// block" error is returned. Additionally, if this method would block, it - /// registers the current task to receive a notification when it would - /// otherwise not block. - /// - /// Note that typically for simple usage it's easier to treat incoming - /// connections as a `Stream` of `UnixStream`s with the `incoming` method - /// below. - /// - /// # Panics - /// - /// This function will panic if it is called outside the context of a - /// future's task. It's recommended to only call this from the - /// implementation of a `Future::poll`, if necessary. - pub fn poll_accept( - self: Pin<&mut Self>, + pub(crate) fn poll_accept( + &mut self, cx: &mut Context<'_>, ) -> Poll> { let (io, addr) = ready!(self.poll_accept_std(cx))?; @@ -89,31 +67,8 @@ impl UnixListener { Ok((UnixStream::new(io), addr)).into() } - /// Attempt to accept a connection and create a new connected `UnixStream` - /// if successful. - /// - /// This function is the same as `poll_accept` above except that it returns a - /// `mio_uds::UnixStream` instead of a `tokio_udp::UnixStream`. This in turn - /// can then allow for the stream to be associated with a different reactor - /// than the one this `UnixListener` is associated with. - /// - /// This function will attempt an accept operation, but will not block - /// waiting for it to complete. If the operation would block then a "would - /// block" error is returned. Additionally, if this method would block, it - /// registers the current task to receive a notification when it would - /// otherwise not block. - /// - /// Note that typically for simple usage it's easier to treat incoming - /// connections as a `Stream` of `UnixStream`s with the `incoming` method - /// below. - /// - /// # Panics - /// - /// This function will panic if it is called outside the context of a - /// future's task. It's recommended to only call this from the - /// implementation of a `Future::poll`, if necessary. - pub fn poll_accept_std( - self: Pin<&mut Self>, + fn poll_accept_std( + &mut self, cx: &mut Context<'_>, ) -> Poll> { ready!(self.io.poll_read_ready(cx, Ready::readable()))?; @@ -178,18 +133,3 @@ impl AsRawFd for UnixListener { self.io.get_ref().as_raw_fd() } } - -/// Future type returned by [`UnixListener::accept`]. -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct Accept<'a> { - listener: &'a mut UnixListener, -} - -impl<'a> Future for Accept<'a> { - type Output = io::Result<(UnixStream, SocketAddr)>; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - Pin::new(&mut *self.listener).poll_accept(cx) - } -} diff --git a/tokio-uds/src/recv.rs b/tokio-uds/src/recv.rs deleted file mode 100644 index 04e6060fd..000000000 --- a/tokio-uds/src/recv.rs +++ /dev/null @@ -1,30 +0,0 @@ -use crate::UnixDatagram; -use std::future::Future; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that receives a datagram from the connected address. -/// -/// This `struct` is created by [`recv`](crate::UnixDatagram::recv). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct Recv<'a, 'b> { - socket: &'a mut UnixDatagram, - buf: &'b mut [u8], -} - -impl<'a, 'b> Recv<'a, 'b> { - pub(crate) fn new(socket: &'a mut UnixDatagram, buf: &'b mut [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for Recv<'a, 'b> { - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Recv { socket, buf } = self.get_mut(); - socket.poll_recv_priv(cx, buf) - } -} diff --git a/tokio-uds/src/recv_from.rs b/tokio-uds/src/recv_from.rs deleted file mode 100644 index c2cb34bff..000000000 --- a/tokio-uds/src/recv_from.rs +++ /dev/null @@ -1,31 +0,0 @@ -use crate::UnixDatagram; -use std::future::Future; -use std::io; -use std::os::unix::net::SocketAddr; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that receives a datagram. -/// -/// This `struct` is created by [`recv_from`](crate::UnixDatagram::recv_from). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct RecvFrom<'a, 'b> { - socket: &'a UnixDatagram, - buf: &'b mut [u8], -} - -impl<'a, 'b> RecvFrom<'a, 'b> { - pub(crate) fn new(socket: &'a UnixDatagram, buf: &'b mut [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for RecvFrom<'a, 'b> { - type Output = io::Result<(usize, SocketAddr)>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let RecvFrom { socket, buf } = self.get_mut(); - socket.poll_recv_from_priv(cx, buf) - } -} diff --git a/tokio-uds/src/send.rs b/tokio-uds/src/send.rs deleted file mode 100644 index 2abc5adeb..000000000 --- a/tokio-uds/src/send.rs +++ /dev/null @@ -1,30 +0,0 @@ -use crate::UnixDatagram; -use std::future::Future; -use std::io; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that sends a datagram to the connected address. -/// -/// This `struct` is created by [`send`](crate::UnixDatagram::send). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct Send<'a, 'b> { - socket: &'a UnixDatagram, - buf: &'b [u8], -} - -impl<'a, 'b> Send<'a, 'b> { - pub(crate) fn new(socket: &'a UnixDatagram, buf: &'b [u8]) -> Self { - Self { socket, buf } - } -} - -impl<'a, 'b> Future for Send<'a, 'b> { - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let Send { socket, buf } = self.get_mut(); - socket.poll_send_priv(cx, buf) - } -} diff --git a/tokio-uds/src/send_to.rs b/tokio-uds/src/send_to.rs deleted file mode 100644 index 5aa83e3ec..000000000 --- a/tokio-uds/src/send_to.rs +++ /dev/null @@ -1,43 +0,0 @@ -use crate::UnixDatagram; -use std::future::Future; -use std::io; -use std::path::Path; -use std::pin::Pin; -use std::task::{Context, Poll}; - -/// A future that sends a datagram to a given address. -/// -/// This `struct` is created by [`send_to`](crate::UnixDatagram::send_to). -#[must_use = "futures do nothing unless you `.await` or poll them"] -#[derive(Debug)] -pub struct SendTo<'a, 'b, P> { - socket: &'a UnixDatagram, - buf: &'b [u8], - target: P, -} - -impl<'a, 'b, P> SendTo<'a, 'b, P> { - pub(crate) fn new(socket: &'a UnixDatagram, buf: &'b [u8], target: P) -> Self { - Self { - socket, - buf, - target, - } - } -} - -impl<'a, 'b, P> Future for SendTo<'a, 'b, P> -where - P: AsRef + Unpin, -{ - type Output = io::Result; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let SendTo { - socket, - buf, - target, - } = self.get_mut(); - socket.poll_send_to_priv(cx, buf, target.as_ref()) - } -} diff --git a/tokio-uds/src/stream.rs b/tokio-uds/src/stream.rs index 0a8d33ad0..9ac35a928 100644 --- a/tokio-uds/src/stream.rs +++ b/tokio-uds/src/stream.rs @@ -1,12 +1,15 @@ use crate::ucred::{self, UCred}; + +use tokio_io::{AsyncRead, AsyncWrite}; +use tokio_reactor::{Handle, PollEvented}; + use bytes::{Buf, BufMut}; use futures_core::ready; +use futures_util::future::poll_fn; use iovec::IoVec; -use mio::Ready; use mio_uds; use std::convert::TryFrom; use std::fmt; -use std::future::Future; use std::io::{self, Read, Write}; use std::net::Shutdown; use std::os::unix::io::{AsRawFd, RawFd}; @@ -14,8 +17,6 @@ use std::os::unix::net::{self, SocketAddr}; use std::path::Path; use std::pin::Pin; use std::task::{Context, Poll}; -use tokio_io::{AsyncRead, AsyncWrite}; -use tokio_reactor::{Handle, PollEvented}; /// A structure representing a connected Unix socket. /// @@ -26,26 +27,21 @@ pub struct UnixStream { io: PollEvented, } -/// Future returned by `UnixStream::connect` which will resolve to a -/// `UnixStream` when the stream is connected. -#[derive(Debug)] -#[must_use = "futures do nothing unless you `.await` or poll them"] -pub struct ConnectFuture { - stream: Option>, -} - impl UnixStream { /// Connects to the socket named by `path`. /// /// This function will create a new Unix socket and connect to the path /// specified, associating the returned stream with the default event loop's /// handle. - pub fn connect

(path: P) -> ConnectFuture + pub async fn connect

(path: P) -> io::Result where P: AsRef, { - let res = mio_uds::UnixStream::connect(path).map(UnixStream::new); - ConnectFuture { stream: Some(res) } + let stream = mio_uds::UnixStream::connect(path)?; + let stream = UnixStream::new(stream); + + poll_fn(|cx| stream.io.poll_write_ready(cx)).await?; + Ok(stream) } /// Consumes a `UnixStream` in the standard library and returns a @@ -78,16 +74,6 @@ impl UnixStream { UnixStream { io } } - /// Test whether this socket is ready to be read or not. - pub fn poll_read_ready(&self, cx: &mut Context<'_>, ready: Ready) -> Poll> { - self.io.poll_read_ready(cx, ready) - } - - /// Test whether this socket is ready to be written to or not. - pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll> { - self.io.poll_write_ready(cx) - } - /// Returns the socket address of the local half of this connection. pub fn local_addr(&self) -> io::Result { self.io.get_ref().local_addr() @@ -340,29 +326,3 @@ impl AsRawFd for UnixStream { self.io.get_ref().as_raw_fd() } } - -impl Future for ConnectFuture { - type Output = io::Result; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let stream = self - .stream - .take() - .expect("ConnectFuture polled after completion")?; - - match stream.io.poll_write_ready(cx) { - Poll::Pending => { - self.stream = Some(Ok(stream)); - return Poll::Pending; - } - Poll::Ready(Err(e)) => return Err(e).into(), - _ => (), - } - - if let Some(e) = stream.io.get_ref().take_error()? { - return Err(e).into(); - } - - Ok(stream).into() - } -}