Implement Stream for Listener types (#2275)

The Incoming types currently don't take ownership of the listener, but
in most cases, users who want to use the Listener as a stream will only
want to use the stream from that point on. So, implement Stream directly
on the Listener types.
This commit is contained in:
Akshay Narayan 2020-02-26 13:38:41 -05:00 committed by GitHub
parent 1dadc701c0
commit 0589acc9ff
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 91 additions and 0 deletions

View File

@ -12,8 +12,22 @@ use std::task::{Context, Poll};
cfg_tcp! {
/// A TCP socket server, listening for connections.
///
/// Also implements a stream over the connections being received on this listener.
///
/// The stream will never return `None` and will also not yield the peer's
/// `SocketAddr` structure. Iterating over it is equivalent to calling accept in a loop.
///
/// # Errors
///
/// Note that accepting a connection can lead to various errors and not all
/// of them are necessarily fatal for example having too many open file
/// descriptors or the other side closing the connection while it waits in
/// an accept queue. These would terminate the stream if not handled in any
/// way.
///
/// # Examples
///
/// Using [`TcpListener::accept`]:
/// ```no_run
/// use tokio::net::TcpListener;
///
@ -34,6 +48,24 @@ cfg_tcp! {
/// }
/// }
/// ```
///
/// Using `impl Stream`:
/// ```no_run
/// use tokio::{net::TcpListener, stream::StreamExt};
///
/// #[tokio::main]
/// async fn main() {
/// let mut listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
/// while let Some(stream) = listener.next().await {
/// match stream {
/// Ok(stream) => {
/// println!("new client!");
/// }
/// Err(e) => { /* connection failed */ }
/// }
/// }
/// }
/// ```
pub struct TcpListener {
io: PollEvented<mio::net::TcpListener>,
}
@ -335,6 +367,19 @@ impl TcpListener {
}
}
#[cfg(feature = "stream")]
impl crate::stream::Stream for TcpListener {
type Item = io::Result<TcpStream>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
let (socket, _) = ready!(self.poll_accept(cx))?;
Poll::Ready(Some(Ok(socket)))
}
}
impl TryFrom<TcpListener> for mio::net::TcpListener {
type Error = io::Error;

View File

@ -14,6 +14,39 @@ use std::task::{Context, Poll};
cfg_uds! {
/// A Unix socket which can accept connections from other Unix sockets.
///
/// Also implements a stream over the connections being received on this listener.
///
/// The stream will never return `None` and will also not yield the peer's
/// `SocketAddr` structure. Iterating over it is equivalent to calling accept in a loop.
///
/// # Errors
///
/// Note that accepting a connection can lead to various errors and not all
/// of them are necessarily fatal for example having too many open file
/// descriptors or the other side closing the connection while it waits in
/// an accept queue. These would terminate the stream if not handled in any
/// way.
///
/// # Examples
///
/// ```no_run
/// use tokio::net::UnixListener;
/// use tokio::stream::StreamExt;
///
/// #[tokio::main]
/// async fn main() {
/// let mut listener = UnixListener::bind("/path/to/the/socket").unwrap();
/// while let Some(stream) = listener.next().await {
/// match stream {
/// Ok(stream) => {
/// println!("new client!");
/// }
/// Err(e) => { /* connection failed */ }
/// }
/// }
/// }
/// ```
pub struct UnixListener {
io: PollEvented<mio_uds::UnixListener>,
}
@ -142,6 +175,19 @@ impl UnixListener {
}
}
#[cfg(feature = "stream")]
impl crate::stream::Stream for UnixListener {
type Item = io::Result<UnixStream>;
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
let (socket, _) = ready!(self.poll_accept(cx))?;
Poll::Ready(Some(Ok(socket)))
}
}
impl TryFrom<UnixListener> for mio_uds::UnixListener {
type Error = io::Error;