直接使用AsyncRead和AsyncWrite

到目前为止，我们都是在Tokio提供的I/O组合器场景下讨论了AsyncRead和AsyncWrite。通常这些就够了，但有时您需要实现自己的组合器，直接执行异步读写。

用AsyncRead读取数据

AsyncRead的核心是poll_read方法。该方法检查Err类型是否为WouldBlock，如果是，表明I/O read操作可能被阻塞的，则返回NotReady，这就使我们可以与futures互操作。当你写一个内部包含AsyncRead的Future（或类似的东西，例如Stream）时，poll_read 很可能就是你将要与之交互的方法。

要记住一点：poll_read遵循与Future::poll相同的契约。具体而言，你不能返回NotReady，除非你已安排当前任务在取得进展时，会被通知再次被调用。基于此，我们可以在自己futures的poll方法内调用poll_read; 当我们从poll_read中转发一个NotReady的时候，我们知道这是遵循poll合约的，因为poll_read遵循相同的合约。

Tokio用于确保poll_read以后通知当前任务task的确切机制不在本节讨论的范围，但如果您感兴趣，可以在Tokio内部原理的非阻塞I/O中阅读更多相关内容。

有了这一切，让我们看看如何自己实现read_exact 这个方法！

  1. #[macro_use]
  2. extern crate futures;
  3. use std::io;
  4. use tokio::prelude::*;
  6. // This is going to be our Future.
  7. // In the common case, this is set to Some(Reading),
  8. // but we'll set it to None when we return Async::Ready
  9. // so that we can return the reader and the buffer.
  10. struct ReadExact<R, T>(Option<Reading<R, T>>);
  12. struct Reading<R, T> {
  13.     // This is the stream we're reading from.
  14.     reader: R,
  15.     // This is the buffer we're reading into.
  16.     buffer: T,
  17.     // And this is how far into the buffer we've written.
  18.     pos: usize,
  19. }
  21. // We want to be able to construct a ReadExact over anything
  22. // that implements AsyncRead, and any buffer that can be
  23. // thought of as a &mut [u8].
  24. fn read_exact<R, T>(reader: R, buffer: T) -> ReadExact<R, T>
  25. where
  26.     R: AsyncRead,
  27.     T: AsMut<[u8]>,
  28. {
  29.     ReadExact(Some(Reading {
  30.         reader,
  31.         buffer,
  32.         // Initially, we've read no bytes into buffer.
  33.         pos: 0,
  34.     }))
  35. }
  37. impl<R, T> Future for ReadExact<R, T>
  38. where
  39.     R: AsyncRead,
  40.     T: AsMut<[u8]>,
  41. {
  42.     // When we've filled up the buffer, we want to return both the buffer
  43.     // with the data that we read and the reader itself.
  44.     type Item = (R, T);
  45.     type Error = io::Error;
  47.     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
  48.         match self.0 {
  49.             Some(Reading {
  50.                 ref mut reader,
  51.                 ref mut buffer,
  52.                 ref mut pos,
  53.             }) => {
  54.                 let buffer = buffer.as_mut();
  55.                 // Check that we haven't finished
  56.                 while *pos < buffer.len() {
  57.                     // Try to read data into the remainder of the buffer.
  58.                     // Just like read in std::io::Read, poll_read *can* read
  59.                     // fewer bytes than the length of the buffer it is given,
  60.                     // and we need to handle that by looking at its return
  61.                     // value, which is the number of bytes actually read.
  62.                     //
  63.                     // Notice that we are using try_ready! here, so if poll_read
  64.                     // returns NotReady (or an error), we will do the same!
  65.                     // We uphold the contract that we have arranged to be
  66.                     // notified later because poll_read follows that same
  67.                     // contract, and _it_ returned NotReady.
  68.                     let n = try_ready!(reader.poll_read(&mut buffer[*pos..]));
  69.                     *pos += n;
  71.                     // If no bytes were read, but there was no error, this
  72.                     // generally implies that the reader will provide no more
  73.                     // data (for example, because the TCP connection was closed
  74.                     // by the other side).
  75.                     if n == 0 {
  76.                         return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof"));
  77.                     }
  78.                 }
  79.             }
  80.             None => panic!("poll a ReadExact after it's done"),
  81.         }
  83.         // We need to return the reader and the buffer, which we can only
  84.         // do by moving them out of self. We do this by taking our state
  85.         // and leaving `None`. This _should_ be fine, because poll()
  86.         // requires callers to not call poll() again after Ready has been
  87.         // returned, so we should only ever see Some(Reading) when poll()
  88.         // is called.
  89.         let reading = self.0.take().expect("must have seen Some above");
  90.         Ok(Async::Ready((reading.reader, reading.buffer)))
  91.     }
  92. }

用AsyncWrite写数据

就像poll_read是AsyncRead的核心一样，poll_write也是AsyncWrite的核心部分。和poll_read一样，该方法检查Err类型是否为WouldBlock，如果是，则表明write操作将被阻塞，就返回NotReady，这再次让我们与futures互操作。AsyncWrite也有一个poll_flush，它提供了一个Write flush的异步版本。poll_flush确保先前通过poll_write写入的任何字节都被刷到底层I/O资源上（例如，发送网络数据包）。类似于poll_write，它封装了Write::flush，映射WouldBlock错误为NotReady，指示flush仍在进行中。

AsyncWrite的poll_write，以及poll_flush都遵循与Future::poll和AsyncRead::poll_read相同的合约，即如果你想返回NotReady，则必须保证当前任务能够被在可以进行下去的时候被通知。和poll_read一样，这意味着我们可以安全地在我们自己的futures中调用这些方法，因为我们知道我们也在遵守合同。

Tokio使用和poll_read相同的通知机制来通知poll_write和poll_flush，你可以在Tokio内部原理的非阻塞I/O中阅读更多相关内容。

关闭

AsyncWrite还添加了一个不属于Write的方法：shutdown。从它的文档：

启动或尝试关闭此writer，在I/O连接完全关闭时返回成功。

此方法旨在用于I/O连接的异步关闭。例如，这适用于实现TLS连接的关闭或调用TcpStream::shutdown来关闭代理连接proxied connection。一些协议有时需要清除最终的数据，或者发起优雅关闭握手，适当地读写更多数据。此方法就是实现这些协议所需的优雅关闭握手逻辑的钩子方法（扩展点）。

总结shutdown：它是一种告诉写一方不再有新数据产生的方法，并且它应该以底层I/O协议所需的任何方式指示。例如，对于TCP连接，这通常需要关闭TCP通道channel的写入端，这样，另一端就可以读到0字节，表明已到文件尾。通常，你可以将shutdown视为你要实现Drop时你需要同步地执行的方法; 只是在异步世界中，你不能在Drop简单地处理，因为你需要有一个执行器executor轮询你的writer！

请注意，在一个实现了AsyncWrite和AsyncRead的写半部分调用shutdown不会关闭读半部分。您通常可以继续随意读取数据，直到另一方关闭相应的写半。

一个使用AsyncWrite的例子

废话少说，让我们来看看我们如何实现：

  1. #[macro_use]
  2. extern crate futures;
  3. use std::io;
  4. use tokio::prelude::*;
  6. // This is going to be our Future.
  7. // It'll seem awfully familiar to ReadExact above!
  8. // In the common case, this is set to Some(Writing),
  9. // but we'll set it to None when we return Async::Ready
  10. // so that we can return the writer and the buffer.
  11. struct WriteAll<W, T>(Option<Writing<W, T>>);
  13. struct Writing<W, T> {
  14.     // This is the stream we're writing into.
  15.     writer: W,
  16.     // This is the buffer we're writing from.
  17.     buffer: T,
  18.     // And this is much of the buffer we've written.
  19.     pos: usize,
  20. }
  22. // We want to be able to construct a WriteAll over anything
  23. // that implements AsyncWrite, and any buffer that can be
  24. // thought of as a &[u8].
  25. fn write_all<W, T>(writer: W, buffer: T) -> WriteAll<W, T>
  26. where
  27.     W: AsyncWrite,
  28.     T: AsRef<[u8]>,
  29. {
  30.     WriteAll(Some(Writing {
  31.         writer,
  32.         buffer,
  33.         // Initially, we've written none of the bytes from buffer.
  34.         pos: 0,
  35.     }))
  36. }
  38. impl<W, T> Future for WriteAll<W, T>
  39. where
  40.     W: AsyncWrite,
  41.     T: AsRef<[u8]>,
  42. {
  43.     // When we've written out the entire buffer, we want to return
  44.     // both the buffer and the writer so that the user can re-use them.
  45.     type Item = (W, T);
  46.     type Error = io::Error;
  48.     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
  49.         match self.0 {
  50.             Some(Writing {
  51.                 ref mut writer,
  52.                 ref buffer,
  53.                 ref mut pos,
  54.             }) => {
  55.                 let buffer = buffer.as_ref();
  56.                 // Check that we haven't finished
  57.                 while *pos < buffer.len() {
  58.                     // Try to write the remainder of the buffer into the writer.
  59.                     // Just like write in std::io::Write, poll_write *can* write
  60.                     // fewer bytes than the length of the buffer it is given,
  61.                     // and we need to handle that by looking at its return
  62.                     // value, which is the number of bytes actually written.
  63.                     //
  64.                     // We are using try_ready! here, just like in poll_read in
  65.                     // ReadExact, so that if poll_write returns NotReady (or an
  66.                     // error), we will do the same! We uphold the contract that
  67.                     // we have arranged to be notified later because poll_write
  68.                     // follows that same contract, and _it_ returned NotReady.
  69.                     let n = try_ready!(writer.poll_write(&buffer[*pos..]));
  70.                     *pos += n;
  72.                     // If no bytes were written, but there was no error, this
  73.                     // generally implies that something weird happened under us.
  74.                     // We make sure to turn this into an error for the caller to
  75.                     // deal with.
  76.                     if n == 0 {
  77.                         return Err(io::Error::new(
  78.                             io::ErrorKind::WriteZero,
  79.                             "zero-length write",
  80.                         ));
  81.                     }
  82.                 }
  83.             }
  84.             None => panic!("poll a WriteAll after it's done"),
  85.         }
  87.         // We use the same trick as in ReadExact to ensure that we can return
  88.         // the buffer and the writer once the entire buffer has been written out.
  89.         let writing = self.0.take().expect("must have seen Some above");
  90.         Ok(Async::Ready((writing.writer, writing.buffer)))
  91.     }
  92. }