tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-09-06 00:00:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

tcp: update API documentation (#1392)

Browse Source
This commit is contained in:
Carl Lerche
2019-08-05 11:50:55 -07:00
committed by GitHub
parent 6cbe3d4f82
commit 6d8cc4e475
5 changed files with 300 additions and 468 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio-io/Cargo.toml
+3 -3
View File
@@ -21,6 +21,9 @@ Core I/O primitives for asynchronous I/O in Rust.
categories = ["asynchronous"] categories = ["asynchronous"]
publish = false publish = false
[features]
util = ["memchr"]
[dependencies] [dependencies]
bytes = "0.4.7" bytes = "0.4.7"
log = "0.4" log = "0.4"
@@ -32,6 +35,3 @@ pin-utils = "0.1.0-alpha.4"
tokio = { version = "0.2.0", path = "../tokio" } tokio = { version = "0.2.0", path = "../tokio" }
futures-util-preview = "0.3.0-alpha.17" futures-util-preview = "0.3.0-alpha.17"
tokio-test = { version = "0.2.0", path = "../tokio-test" } tokio-test = { version = "0.2.0", path = "../tokio-test" }
[features]
util = ["memchr"]
tokio-reactor/src/poll_evented.rs
+28 -14
View File
@@ -52,19 +52,33 @@ use tokio_io::{AsyncRead, AsyncWrite};
/// [`TcpListener`] implements poll_accept by using [`poll_read_ready`] and /// [`TcpListener`] implements poll_accept by using [`poll_read_ready`] and
/// [`clear_read_ready`]. /// [`clear_read_ready`].
/// ///
/// ```rust,ignore /// ```rust
/// pub fn poll_accept(&mut self) -> Poll<(net::TcpStream, SocketAddr), io::Error> { /// use tokio_reactor::PollEvented;
/// let ready = Ready::readable();
/// ///
/// try_ready!(self.poll_evented.poll_read_ready(ready)); /// use futures_core::ready;
/// use mio::Ready;
/// use mio::net::{TcpStream, TcpListener};
/// use std::io;
/// use std::task::{Context, Poll};
/// ///
/// match self.poll_evented.get_ref().accept_std() { /// struct MyListener {
/// Ok(pair) => Ok(Async::Ready(pair)), /// poll_evented: PollEvented<TcpListener>,
/// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { /// }
/// self.poll_evented.clear_read_ready(ready); ///
/// Ok(Async::NotReady) /// impl MyListener {
/// pub fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<TcpStream, io::Error>> {
/// let ready = Ready::readable();
///
/// ready!(self.poll_evented.poll_read_ready(cx, ready))?;
///
/// match self.poll_evented.get_ref().accept() {
/// Ok((socket, _)) => Poll::Ready(Ok(socket)),
/// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
/// self.poll_evented.clear_read_ready(cx, ready);
/// Poll::Pending
/// }
/// Err(e) => Poll::Ready(Err(e)),
/// } /// }
/// Err(e) => Err(e),
/// } /// }
/// } /// }
/// ``` /// ```
@@ -209,8 +223,8 @@ where
/// `writable`. HUP is always implicitly included on platforms that support /// `writable`. HUP is always implicitly included on platforms that support
/// it. /// it.
/// ///
/// If the resource is not ready for a read then `Async::NotReady` is /// If the resource is not ready for a read then `Poll::Pending` is returned
/// returned and the current task is notified once a new event is received. /// and the current task is notified once a new event is received.
/// ///
/// The I/O resource will remain in a read-ready state until readiness is /// The I/O resource will remain in a read-ready state until readiness is
/// cleared by calling [`clear_read_ready`]. /// cleared by calling [`clear_read_ready`].
@@ -241,8 +255,8 @@ where
/// Clears the I/O resource's read readiness state and registers the current /// Clears the I/O resource's read readiness state and registers the current
/// task to be notified once a read readiness event is received. /// task to be notified once a read readiness event is received.
/// ///
/// After calling this function, `poll_read_ready` will return `NotReady` /// After calling this function, `poll_read_ready` will return
/// until a new read readiness event has been received. /// `Poll::Pending` until a new read readiness event has been received.
/// ///
/// The `mask` argument specifies the readiness bits to clear. This may not /// The `mask` argument specifies the readiness bits to clear. This may not
/// include `writable` or `hup`. /// include `writable` or `hup`.
tokio-reactor/src/registration.rs
+10 -10
View File
@@ -263,15 +263,15 @@ impl Registration {
/// ///
/// There are several possible return values: /// There are several possible return values:
/// ///
/// * `Ok(Async::Ready(readiness))` means that the I/O resource has received /// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
/// a new readiness event. The readiness value is included. /// a new readiness event. The readiness value is included.
/// ///
/// * `Ok(NotReady)` means that no new readiness events have been received /// * `Poll::Pending` means that no new readiness events have been received
/// since the last call to `poll_read_ready`. /// since the last call to `poll_read_ready`.
/// ///
/// * `Err(err)` means that the registration has encountered an error. This /// * `Poll::Ready(Err(err))` means that the registration has encountered an
/// error either represents a permanent internal error **or** the fact /// error. This error either represents a permanent internal error **or**
/// that [`register`] was not called first. /// the fact that [`register`] was not called first.
/// ///
/// [`register`]: #method.register /// [`register`]: #method.register
/// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered
@@ -314,15 +314,15 @@ impl Registration {
/// ///
/// There are several possible return values: /// There are several possible return values:
/// ///
/// * `Ok(Async::Ready(readiness))` means that the I/O resource has received /// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
/// a new readiness event. The readiness value is included. /// a new readiness event. The readiness value is included.
/// ///
/// * `Ok(NotReady)` means that no new readiness events have been received /// * `Poll::Pending` means that no new readiness events have been received
/// since the last call to `poll_write_ready`. /// since the last call to `poll_write_ready`.
/// ///
/// * `Err(err)` means that the registration has encountered an error. This /// * `Poll::Ready(Err(err))` means that the registration has encountered an
/// error either represents a permanent internal error **or** the fact /// error. This error either represents a permanent internal error **or**
/// that [`register`] was not called first. /// the fact that [`register`] was not called first.
/// ///
/// [`register`]: #method.register /// [`register`]: #method.register
/// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered
tokio-tcp/src/listener.rs
+38 -123
View File
@@ -18,26 +18,22 @@ use tokio_reactor::{Handle, PollEvented};
/// # Examples /// # Examples
/// ///
/// ```no_run /// ```no_run
/// use futures::stream::Stream; /// #![feature(async_await)]
/// use std::net::SocketAddr;
/// use tokio::net::{TcpListener, TcpStream};
/// ///
/// fn process_socket(socket: TcpStream) { /// use tokio::net::TcpListener;
/// // ... /// use std::error::Error;
/// } /// # async fn process_socket<T>(socket: T) {}
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// #[tokio::main]
/// let listener = TcpListener::bind(&addr)?; /// async fn main() -> Result<(), Box<dyn Error>> {
/// let addr = "127.0.0.1:8080".parse()?;
/// let mut listener = TcpListener::bind(&addr)?;
/// ///
/// // accept connections and process them /// loop {
/// tokio::run(listener.incoming() /// let (socket, _) = listener.accept().await?;
/// .map_err(|e| eprintln!("failed to accept socket; error = {:?}", e))
/// .for_each(|socket| {
/// process_socket(socket); /// process_socket(socket);
/// Ok(()) /// }
/// }) /// }
/// );
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub struct TcpListener { pub struct TcpListener {
io: PollEvented<mio::net::TcpListener>, io: PollEvented<mio::net::TcpListener>,
@@ -64,53 +60,6 @@ impl TcpListener {
Ok(TcpListener::new(l)) Ok(TcpListener::new(l))
} }
/// Attempt to accept a connection and create a new connected `TcpStream` if
/// successful.
///
/// Note that typically for simple usage it's easier to treat incoming
/// connections as a `Stream` of `TcpStream`s with the `incoming` method
/// below.
///
/// # Return
///
/// On success, returns `Ok(Async::Ready((socket, addr)))`.
///
/// If the listener is not ready to accept, the method returns
/// `Ok(Async::NotReady)` and arranges for the current task to receive a
/// notification when the listener becomes ready to accept.
///
/// # Panics
///
/// This function will panic if called from outside of a task context.
///
/// # Examples
///
/// ```no_run
/// use std::net::SocketAddr;
/// use tokio::net::TcpListener;
/// use futures::Async;
///
/// let addr = "127.0.0.1:0".parse::<SocketAddr>()?;
/// let mut listener = TcpListener::bind(&addr)?;
/// match listener.poll_accept() {
/// Ok(Async::Ready((_socket, addr))) => println!("listener ready to accept: {:?}", addr),
/// Ok(Async::NotReady) => println!("listener not ready to accept!"),
/// Err(e) => eprintln!("got an error: {}", e),
/// }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
pub fn poll_accept(
&mut self,
cx: &mut Context<'_>,
) -> Poll<io::Result<(TcpStream, SocketAddr)>> {
let (io, addr) = ready!(self.poll_accept_std(cx))?;
let io = mio::net::TcpStream::from_stream(io)?;
let io = TcpStream::new(io);
Poll::Ready(Ok((io, addr)))
}
/// Accept a new incoming connection from this listener. /// Accept a new incoming connection from this listener.
/// ///
/// This function will yield once a new TCP connection is established. When /// This function will yield once a new TCP connection is established. When
@@ -122,7 +71,19 @@ impl TcpListener {
/// # Examples /// # Examples
/// ///
/// ``` /// ```
/// unimplemented!(); /// #![feature(async_await)]
///
/// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// use tokio::net::TcpListener;
///
/// let addr = "127.0.0.1:8080".parse()?;
/// let mut listener = TcpListener::bind(&addr)?;
/// match listener.accept().await {
/// Ok((_socket, addr)) => println!("new client: {:?}", addr),
/// Err(e) => println!("couldn't get client: {:?}", e),
/// }
/// # Ok(())
/// # }
/// ``` /// ```
#[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988 #[allow(clippy::needless_lifetimes)] // false positive: https://github.com/rust-lang/rust-clippy/issues/3988
pub async fn accept(&mut self) -> io::Result<(TcpStream, SocketAddr)> { pub async fn accept(&mut self) -> io::Result<(TcpStream, SocketAddr)> {
@@ -130,43 +91,19 @@ impl TcpListener {
poll_fn(|cx| self.poll_accept(cx)).await poll_fn(|cx| self.poll_accept(cx)).await
} }
/// Attempt to accept a connection and create a new connected `TcpStream` if pub(crate) fn poll_accept(
/// successful. &mut self,
/// cx: &mut Context<'_>,
/// This function is the same as `accept` above except that it returns a ) -> Poll<io::Result<(TcpStream, SocketAddr)>> {
/// `std::net::TcpStream` instead of a `tokio::net::TcpStream`. This in turn let (io, addr) = ready!(self.poll_accept_std(cx))?;
/// can then allow for the TCP stream to be associated with a different
/// reactor than the one this `TcpListener` is associated with. let io = mio::net::TcpStream::from_stream(io)?;
/// let io = TcpStream::new(io);
/// # Return
/// Poll::Ready(Ok((io, addr)))
/// On success, returns `Ok(Async::Ready((socket, addr)))`. }
///
/// If the listener is not ready to accept, the method returns fn poll_accept_std(
/// `Ok(Async::NotReady)` and arranges for the current task to receive a
/// notification when the listener becomes ready to accept.
///
/// # Panics
///
/// This function will panic if called from outside of a task context.
///
/// # Examples
///
/// ```no_run
/// use std::net::SocketAddr;
/// use tokio::net::TcpListener;
/// use futures::Async;
///
/// let addr = "127.0.0.1:0".parse::<SocketAddr>()?;
/// let mut listener = TcpListener::bind(&addr)?;
/// match listener.poll_accept_std() {
/// Ok(Async::Ready((_socket, addr))) => println!("listener ready to accept: {:?}", addr),
/// Ok(Async::NotReady) => println!("listener not ready to accept!"),
/// Err(e) => eprintln!("got an error: {}", e),
/// }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
pub fn poll_accept_std(
&mut self, &mut self,
cx: &mut Context<'_>, cx: &mut Context<'_>,
) -> Poll<io::Result<(net::TcpStream, SocketAddr)>> { ) -> Poll<io::Result<(net::TcpStream, SocketAddr)>> {
@@ -267,28 +204,6 @@ impl TcpListener {
/// necessarily fatal ‒ for example having too many open file descriptors or the other side /// 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 /// closing the connection while it waits in an accept queue. These would terminate the stream
/// if not handled in any way. /// if not handled in any way.
///
/// If aiming for production, decision what to do about them must be made. The
/// [`tk-listen`](https://crates.io/crates/tk-listen) crate might be of some help.
///
/// # Examples
///
/// ```
/// use tokio::net::TcpListener;
/// use futures::stream::Stream;
/// use std::net::SocketAddr;
///
/// let addr = "127.0.0.1:0".parse::<SocketAddr>()?;
/// let listener = TcpListener::bind(&addr)?;
///
/// listener.incoming()
/// .map_err(|e| eprintln!("failed to accept stream; error = {:?}", e))
/// .for_each(|_socket| {
/// println!("new socket!");
/// Ok(())
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
#[cfg(feature = "async-traits")] #[cfg(feature = "async-traits")]
pub fn incoming(self) -> Incoming { pub fn incoming(self) -> Incoming {
Incoming::new(self) Incoming::new(self)
tokio-tcp/src/stream.rs
+221 -318
View File
@@ -30,19 +30,25 @@ use tokio_reactor::{Handle, PollEvented};
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// use futures::Future; /// #![feature(async_await)]
/// use tokio::io::AsyncWrite;
/// use tokio::net::TcpStream;
/// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:34254".parse::<SocketAddr>()?; /// use tokio::net::TcpStream;
/// let stream = TcpStream::connect(&addr); /// use tokio::prelude::*;
/// stream.map(|mut stream| { /// use std::error::Error;
/// // Attempt to write bytes asynchronously to the stream ///
/// stream.poll_write(&[1]); /// #[tokio::main]
/// }); /// async fn main() -> Result<(), Box<dyn Error>> {
/// # Ok::<_, Box<dyn std::error::Error>>(()) /// let addr = "127.0.0.1:8080".parse()?;
///
/// // Connect to a peer
/// let mut stream = TcpStream::connect(&addr).await?;
///
/// // Write some data.
/// stream.write_all(b"hello world!").await?;
///
/// Ok(())
/// }
/// ``` /// ```
pub struct TcpStream { pub struct TcpStream {
io: PollEvented<mio::net::TcpStream>, io: PollEvented<mio::net::TcpStream>,
@@ -71,16 +77,25 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// use futures::Future; /// #![feature(async_await)]
/// use tokio::net::TcpStream;
/// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:34254".parse::<SocketAddr>()?; /// use tokio::net::TcpStream;
/// let stream = TcpStream::connect(&addr) /// use tokio::prelude::*;
/// .map(|stream| /// use std::error::Error;
/// println!("successfully connected to {}", stream.local_addr().unwrap())); ///
/// # Ok::<_, Box<dyn std::error::Error>>(()) /// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn Error>> {
/// let addr = "127.0.0.1:8080".parse()?;
///
/// // Connect to a peer
/// let mut stream = TcpStream::connect(&addr).await?;
///
/// // Write some data.
/// stream.write_all(b"hello world!").await?;
///
/// Ok(())
/// }
/// ``` /// ```
pub fn connect(addr: &SocketAddr) -> impl Future<Output = io::Result<TcpStream>> { pub fn connect(addr: &SocketAddr) -> impl Future<Output = io::Result<TcpStream>> {
use self::ConnectFutureState::*; use self::ConnectFutureState::*;
@@ -108,12 +123,13 @@ impl TcpStream {
/// ///
/// ```no_run /// ```no_run
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use std::net::TcpStream as StdTcpStream;
/// use tokio_reactor::Handle; /// use tokio_reactor::Handle;
/// ///
/// let std_stream = StdTcpStream::connect("127.0.0.1:34254")?; /// # fn dox() -> std::io::Result<()> {
/// let std_stream = std::net::TcpStream::connect("127.0.0.1:34254")?;
/// let stream = TcpStream::from_std(std_stream, &Handle::default())?; /// let stream = TcpStream::from_std(std_stream, &Handle::default())?;
/// # Ok::<_, Box<dyn std::error::Error>>(()) /// # Ok(())
/// # }
/// ``` /// ```
pub fn from_std(stream: net::TcpStream, handle: &Handle) -> io::Result<TcpStream> { pub fn from_std(stream: net::TcpStream, handle: &Handle) -> io::Result<TcpStream> {
let io = mio::net::TcpStream::from_stream(stream)?; let io = mio::net::TcpStream::from_stream(stream)?;
@@ -158,111 +174,23 @@ impl TcpStream {
ConnectFuture { inner } ConnectFuture { inner }
} }
/// Check the TCP stream's read readiness state.
///
/// The mask argument allows specifying what readiness to notify on. This
/// can be any value, including platform specific readiness, **except**
/// `writable`. HUP is always implicitly included on platforms that support
/// it.
///
/// If the resource is not ready for a read then `Async::NotReady` is
/// returned and the current task is notified once a new event is received.
///
/// The stream will remain in a read-ready state until calls to `poll_read`
/// return `NotReady`.
///
/// # Panics
///
/// This function panics if:
///
/// * `ready` includes writable.
/// * called from outside of a task context.
///
/// # Examples
///
/// ```
/// use mio::Ready;
/// use futures::Async;
/// use futures::Future;
/// use tokio::net::TcpStream;
/// use std::net::SocketAddr;
///
/// let addr = "127.0.0.1:34254".parse::<SocketAddr>()?;
/// let stream = TcpStream::connect(&addr);
///
/// stream.map(|stream| {
/// match stream.poll_read_ready(Ready::readable()) {
/// Ok(Async::Ready(_)) => println!("read ready"),
/// Ok(Async::NotReady) => println!("not read ready"),
/// Err(e) => eprintln!("got error: {}", e),
/// }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
pub fn poll_read_ready(
&self,
cx: &mut Context<'_>,
mask: mio::Ready,
) -> Poll<io::Result<mio::Ready>> {
self.io.poll_read_ready(cx, mask)
}
/// Check the TCP stream's write readiness state.
///
/// This always checks for writable readiness and also checks for HUP
/// readiness on platforms that support it.
///
/// If the resource is not ready for a write then `Async::NotReady` is
/// returned and the current task is notified once a new event is received.
///
/// The I/O resource will remain in a write-ready state until calls to
/// `poll_write` return `NotReady`.
///
/// # Panics
///
/// This function panics if called from outside of a task context.
///
/// # Examples
///
/// ```
/// use futures::Async;
/// use futures::Future;
/// use tokio::net::TcpStream;
/// use std::net::SocketAddr;
///
/// let addr = "127.0.0.1:34254".parse::<SocketAddr>()?;
/// let stream = TcpStream::connect(&addr);
///
/// stream.map(|stream| {
/// match stream.poll_write_ready() {
/// Ok(Async::Ready(_)) => println!("write ready"),
/// Ok(Async::NotReady) => println!("not write ready"),
/// Err(e) => eprintln!("got error: {}", e),
/// }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
self.io.poll_write_ready(cx)
}
/// Returns the local address that this stream is bound to. /// Returns the local address that this stream is bound to.
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future; /// use std::net::SocketAddr;
/// use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.local_addr()?);
/// assert_eq!(stream.local_addr().unwrap(), /// # Ok(())
/// SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 8080))); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn local_addr(&self) -> io::Result<SocketAddr> { pub fn local_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().local_addr() self.io.get_ref().local_addr()
@@ -271,65 +199,25 @@ impl TcpStream {
/// Returns the remote address that this stream is connected to. /// Returns the remote address that this stream is connected to.
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future; /// use std::net::SocketAddr;
/// use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.peer_addr()?);
/// assert_eq!(stream.peer_addr().unwrap(), /// # Ok(())
/// SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 8080))); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn peer_addr(&self) -> io::Result<SocketAddr> { pub fn peer_addr(&self) -> io::Result<SocketAddr> {
self.io.get_ref().peer_addr() self.io.get_ref().peer_addr()
} }
/// Receives data on the socket from the remote address to which it is fn poll_peek(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
/// connected, without removing that data from the queue. On success,
/// returns the number of bytes peeked.
///
/// Successive calls return the same data. This is accomplished by passing
/// `MSG_PEEK` as a flag to the underlying recv system call.
///
/// # Return
///
/// On success, returns `Ok(Async::Ready(num_bytes_read))`.
///
/// If no data is available for reading, the method returns
/// `Ok(Async::NotReady)` and arranges for the current task to receive a
/// notification when the socket becomes readable or is closed.
///
/// # Panics
///
/// This function will panic if called from outside of a task context.
///
/// # Examples
///
/// ```
/// use tokio::net::TcpStream;
/// use futures::Async;
/// use futures::Future;
/// use std::net::SocketAddr;
///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?;
/// let stream = TcpStream::connect(&addr);
///
/// stream.map(|mut stream| {
/// let mut buf = [0; 10];
/// match stream.poll_peek(&mut buf) {
/// Ok(Async::Ready(len)) => println!("read {} bytes", len),
/// Ok(Async::NotReady) => println!("no data available"),
/// Err(e) => eprintln!("got error: {}", e),
/// }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
pub fn poll_peek(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?; ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
match self.io.get_ref().peek(buf) { match self.io.get_ref().peek(buf) {
@@ -351,8 +239,32 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// unimplemented!(); /// #![feature(async_await)]
///
/// use tokio::net::TcpStream;
/// use tokio::prelude::*;
/// use std::error::Error;
///
/// #[tokio::main]
/// async fn main() -> Result<(), Box<dyn Error>> {
/// let addr = "127.0.0.1:8080".parse()?;
///
/// // Connect to a peer
/// let mut stream = TcpStream::connect(&addr).await?;
///
/// let mut b1 = [0; 10];
/// let mut b2 = [0; 10];
///
/// // Peek at the data
/// let n = stream.peek(&mut b1).await?;
///
/// // Read the data
/// assert_eq!(n, stream.read(&mut b2[..n]).await?);
/// assert_eq!(&b1[..n], &b2[..n]);
///
/// Ok(())
/// }
/// ``` /// ```
pub async fn peek(&mut self, buf: &mut [u8]) -> io::Result<usize> { pub async fn peek(&mut self, buf: &mut [u8]) -> io::Result<usize> {
poll_fn(|cx| self.poll_peek(cx, buf)).await poll_fn(|cx| self.poll_peek(cx, buf)).await
@@ -366,18 +278,26 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future; /// use tokio::prelude::*;
/// use std::net::{Shutdown, SocketAddr}; /// use std::error::Error;
/// use std::net::Shutdown;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// #[tokio::main]
/// let stream = TcpStream::connect(&addr); /// async fn main() -> Result<(), Box<dyn Error>> {
/// let addr = "127.0.0.1:8080".parse()?;
/// ///
/// stream.map(|stream| { /// // Connect to a peer
/// stream.shutdown(Shutdown::Both) /// let mut stream = TcpStream::connect(&addr).await?;
/// }); ///
/// # Ok::<_, Box<dyn std::error::Error>>(()) /// // Shutdown the stream
/// stream.shutdown(Shutdown::Write)?;
///
/// Ok(())
/// }
/// ``` /// ```
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> { pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
self.io.get_ref().shutdown(how) self.io.get_ref().shutdown(how)
@@ -391,19 +311,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.nodelay()?);
/// stream.set_nodelay(true).expect("set_nodelay call failed");; /// # Ok(())
/// assert_eq!(stream.nodelay().unwrap_or(false), true); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn nodelay(&self) -> io::Result<bool> { pub fn nodelay(&self) -> io::Result<bool> {
self.io.get_ref().nodelay() self.io.get_ref().nodelay()
@@ -419,18 +339,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_nodelay(true)?;
/// stream.set_nodelay(true).expect("set_nodelay call failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> { pub fn set_nodelay(&self, nodelay: bool) -> io::Result<()> {
self.io.get_ref().set_nodelay(nodelay) self.io.get_ref().set_nodelay(nodelay)
@@ -444,19 +365,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.recv_buffer_size()?);
/// stream.set_recv_buffer_size(100).expect("set_recv_buffer_size failed"); /// # Ok(())
/// assert_eq!(stream.recv_buffer_size().unwrap_or(0), 100); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn recv_buffer_size(&self) -> io::Result<usize> { pub fn recv_buffer_size(&self) -> io::Result<usize> {
self.io.get_ref().recv_buffer_size() self.io.get_ref().recv_buffer_size()
@@ -469,18 +390,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_recv_buffer_size(100)?;
/// stream.set_recv_buffer_size(100).expect("set_recv_buffer_size failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_recv_buffer_size(&self, size: usize) -> io::Result<()> { pub fn set_recv_buffer_size(&self, size: usize) -> io::Result<()> {
self.io.get_ref().set_recv_buffer_size(size) self.io.get_ref().set_recv_buffer_size(size)
@@ -494,19 +416,28 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// Returns whether keepalive messages are enabled on this socket, and if so
/// the duration of time between them.
///
/// For more information about this option, see [`set_keepalive`].
///
/// [`set_keepalive`]: #tymethod.set_keepalive
///
/// # Examples
///
/// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.send_buffer_size()?);
/// stream.set_send_buffer_size(100).expect("set_send_buffer_size failed"); /// # Ok(())
/// assert_eq!(stream.send_buffer_size().unwrap_or(0), 100); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn send_buffer_size(&self) -> io::Result<usize> { pub fn send_buffer_size(&self) -> io::Result<usize> {
self.io.get_ref().send_buffer_size() self.io.get_ref().send_buffer_size()
@@ -519,18 +450,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_send_buffer_size(100)?;
/// stream.set_send_buffer_size(100).expect("set_send_buffer_size failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_send_buffer_size(&self, size: usize) -> io::Result<()> { pub fn set_send_buffer_size(&self, size: usize) -> io::Result<()> {
self.io.get_ref().set_send_buffer_size(size) self.io.get_ref().set_send_buffer_size(size)
@@ -545,19 +477,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.keepalive()?);
/// stream.set_keepalive(None).expect("set_keepalive failed"); /// # Ok(())
/// assert_eq!(stream.keepalive().unwrap(), None); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn keepalive(&self) -> io::Result<Option<Duration>> { pub fn keepalive(&self) -> io::Result<Option<Duration>> {
self.io.get_ref().keepalive() self.io.get_ref().keepalive()
@@ -578,18 +510,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_keepalive(None)?;
/// stream.set_keepalive(None).expect("set_keepalive failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_keepalive(&self, keepalive: Option<Duration>) -> io::Result<()> { pub fn set_keepalive(&self, keepalive: Option<Duration>) -> io::Result<()> {
self.io.get_ref().set_keepalive(keepalive) self.io.get_ref().set_keepalive(keepalive)
@@ -603,19 +536,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.ttl()?);
/// stream.set_ttl(100).expect("set_ttl failed"); /// # Ok(())
/// assert_eq!(stream.ttl().unwrap_or(0), 100); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn ttl(&self) -> io::Result<u32> { pub fn ttl(&self) -> io::Result<u32> {
self.io.get_ref().ttl() self.io.get_ref().ttl()
@@ -628,18 +561,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_ttl(123)?;
/// stream.set_ttl(100).expect("set_ttl failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_ttl(&self, ttl: u32) -> io::Result<()> { pub fn set_ttl(&self, ttl: u32) -> io::Result<()> {
self.io.get_ref().set_ttl(ttl) self.io.get_ref().set_ttl(ttl)
@@ -654,19 +588,19 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// println!("{:?}", stream.linger()?);
/// stream.set_linger(None).expect("set_linger failed"); /// # Ok(())
/// assert_eq!(stream.linger().unwrap(), None); /// # }
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn linger(&self) -> io::Result<Option<Duration>> { pub fn linger(&self) -> io::Result<Option<Duration>> {
self.io.get_ref().linger() self.io.get_ref().linger()
@@ -686,55 +620,24 @@ impl TcpStream {
/// ///
/// # Examples /// # Examples
/// ///
/// ``` /// ```no_run
/// #![feature(async_await)]
///
/// use tokio::net::TcpStream; /// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr; /// use std::net::SocketAddr;
/// ///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?; /// # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
/// let stream = TcpStream::connect(&addr); /// let addr = "127.0.0.1:8080".parse()?;
/// let stream = TcpStream::connect(&addr).await?;
/// ///
/// stream.map(|stream| { /// stream.set_linger(None)?;
/// stream.set_linger(None).expect("set_linger failed"); /// # Ok(())
/// }); /// # }
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ``` /// ```
pub fn set_linger(&self, dur: Option<Duration>) -> io::Result<()> { pub fn set_linger(&self, dur: Option<Duration>) -> io::Result<()> {
self.io.get_ref().set_linger(dur) self.io.get_ref().set_linger(dur)
} }
/// Creates a new independently owned handle to the underlying socket.
///
/// The returned `TcpStream` is a reference to the same stream that this
/// object references. Both handles will read and write the same stream of
/// data, and options set on one stream will be propagated to the other
/// stream.
///
/// # Examples
///
/// ```
/// use tokio::net::TcpStream;
/// use futures::Future;
/// use std::net::SocketAddr;
///
/// let addr = "127.0.0.1:8080".parse::<SocketAddr>()?;
/// let stream = TcpStream::connect(&addr);
///
/// stream.map(|stream| {
/// let clone = stream.try_clone().unwrap();
/// });
/// # Ok::<_, Box<dyn std::error::Error>>(())
/// ```
#[deprecated(since = "0.1.14", note = "use `split()` instead")]
#[doc(hidden)]
pub fn try_clone(&self) -> io::Result<TcpStream> {
// Rationale for deprecation:
// - https://github.com/tokio-rs/tokio/pull/824
// - https://github.com/tokio-rs/tokio/issues/774#issuecomment-451059317
let msg = "`TcpStream::try_clone()` is deprecated because it doesn't work as intended";
Err(io::Error::new(io::ErrorKind::Other, msg))
}
/// Split a `TcpStream` into a read half and a write half, which can be used /// Split a `TcpStream` into a read half and a write half, which can be used
/// to read and write the stream concurrently. /// to read and write the stream concurrently.
/// ///