1 | use crate::io::{AsyncBufRead, AsyncWrite}; |
2 | use std::future::Future; |
3 | use std::io; |
4 | use std::pin::Pin; |
5 | use std::task::{Context, Poll}; |
6 | |
7 | cfg_io_util! { |
8 | /// A future that asynchronously copies the entire contents of a reader into a |
9 | /// writer. |
10 | /// |
11 | /// This struct is generally created by calling [`copy_buf`][copy_buf]. Please |
12 | /// see the documentation of `copy_buf()` for more details. |
13 | /// |
14 | /// [copy_buf]: copy_buf() |
15 | #[derive (Debug)] |
16 | #[must_use = "futures do nothing unless you `.await` or poll them" ] |
17 | struct CopyBuf<'a, R: ?Sized, W: ?Sized> { |
18 | reader: &'a mut R, |
19 | writer: &'a mut W, |
20 | amt: u64, |
21 | } |
22 | |
23 | /// Asynchronously copies the entire contents of a reader into a writer. |
24 | /// |
25 | /// This function returns a future that will continuously read data from |
26 | /// `reader` and then write it into `writer` in a streaming fashion until |
27 | /// `reader` returns EOF or fails. |
28 | /// |
29 | /// On success, the total number of bytes that were copied from `reader` to |
30 | /// `writer` is returned. |
31 | /// |
32 | /// This is a [`tokio::io::copy`] alternative for [`AsyncBufRead`] readers |
33 | /// with no extra buffer allocation, since [`AsyncBufRead`] allow access |
34 | /// to the reader's inner buffer. |
35 | /// |
36 | /// [`tokio::io::copy`]: crate::io::copy |
37 | /// [`AsyncBufRead`]: crate::io::AsyncBufRead |
38 | /// |
39 | /// # Errors |
40 | /// |
41 | /// The returned future will finish with an error will return an error |
42 | /// immediately if any call to `poll_fill_buf` or `poll_write` returns an |
43 | /// error. |
44 | /// |
45 | /// # Examples |
46 | /// |
47 | /// ``` |
48 | /// use tokio::io; |
49 | /// |
50 | /// # async fn dox() -> std::io::Result<()> { |
51 | /// let mut reader: &[u8] = b"hello"; |
52 | /// let mut writer: Vec<u8> = vec![]; |
53 | /// |
54 | /// io::copy_buf(&mut reader, &mut writer).await?; |
55 | /// |
56 | /// assert_eq!(b"hello", &writer[..]); |
57 | /// # Ok(()) |
58 | /// # } |
59 | /// ``` |
60 | pub async fn copy_buf<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> io::Result<u64> |
61 | where |
62 | R: AsyncBufRead + Unpin + ?Sized, |
63 | W: AsyncWrite + Unpin + ?Sized, |
64 | { |
65 | CopyBuf { |
66 | reader, |
67 | writer, |
68 | amt: 0, |
69 | }.await |
70 | } |
71 | } |
72 | |
73 | impl<R, W> Future for CopyBuf<'_, R, W> |
74 | where |
75 | R: AsyncBufRead + Unpin + ?Sized, |
76 | W: AsyncWrite + Unpin + ?Sized, |
77 | { |
78 | type Output = io::Result<u64>; |
79 | |
80 | fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
81 | loop { |
82 | let me: &mut CopyBuf<'_, R, W> = &mut *self; |
83 | let buffer: &[u8] = ready!(Pin::new(&mut *me.reader).poll_fill_buf(cx))?; |
84 | if buffer.is_empty() { |
85 | ready!(Pin::new(&mut self.writer).poll_flush(cx))?; |
86 | return Poll::Ready(Ok(self.amt)); |
87 | } |
88 | |
89 | let i: usize = ready!(Pin::new(&mut *me.writer).poll_write(cx, buffer))?; |
90 | if i == 0 { |
91 | return Poll::Ready(Err(std::io::ErrorKind::WriteZero.into())); |
92 | } |
93 | self.amt += i as u64; |
94 | Pin::new(&mut *self.reader).consume(amt:i); |
95 | } |
96 | } |
97 | } |
98 | |
99 | #[cfg (test)] |
100 | mod tests { |
101 | use super::*; |
102 | |
103 | #[test ] |
104 | fn assert_unpin() { |
105 | use std::marker::PhantomPinned; |
106 | crate::is_unpin::<CopyBuf<'_, PhantomPinned, PhantomPinned>>(); |
107 | } |
108 | } |
109 | |