直接使用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::*;

10. struct ReadExact<R, T>(Option<Reading<R, T>>);

12. struct Reading<R, T> {

14. reader: R,

16. buffer: T,

18. pos: usize,
19. }

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,

33. pos: 0,
34. }))
35. }

37. impl<R, T> Future for ReadExact<R, T>
38. where
39. R: AsyncRead,
40. T: AsMut<[u8]>,
41. {

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();

56. while *pos < buffer.len() {

68. let n = try_ready!(reader.poll_read(&mut buffer[*pos..]));
69. *pos += n;

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. }

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::*;

11. struct WriteAll<W, T>(Option<Writing<W, T>>);

13. struct Writing<W, T> {

15. writer: W,

17. buffer: T,

19. pos: usize,
20. }

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,

34. pos: 0,
35. }))
36. }

38. impl<W, T> Future for WriteAll<W, T>
39. where
40. W: AsyncWrite,
41. T: AsRef<[u8]>,
42. {

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();

57. while *pos < buffer.len() {

69. let n = try_ready!(writer.poll_write(&buffer[*pos..]));
70. *pos += n;

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. }

89. let writing = self.0.take().expect("must have seen Some above");
90. Ok(Async::Ready((writing.writer, writing.buffer)))
91. }
92. }