1 | use crate::abortable::{AbortHandle, AbortInner, Aborted}; |
2 | use futures_core::future::Future; |
3 | use futures_core::task::{Context, Poll}; |
4 | use futures_io::{AsyncBufRead, AsyncWrite}; |
5 | use pin_project_lite::pin_project; |
6 | use std::io; |
7 | use std::pin::Pin; |
8 | use std::sync::atomic::Ordering; |
9 | use std::sync::Arc; |
10 | |
11 | /// Creates a future which copies all the bytes from one object to another, with its `AbortHandle`. |
12 | /// |
13 | /// The returned future will copy all the bytes read from this `AsyncBufRead` into the |
14 | /// `writer` specified. This future will only complete once abort has been requested or the `reader` has hit |
15 | /// EOF and all bytes have been written to and flushed from the `writer` |
16 | /// provided. |
17 | /// |
18 | /// On success the number of bytes is returned. If aborted, `Aborted` is returned. Otherwise, the underlying error is returned. |
19 | /// |
20 | /// # Examples |
21 | /// |
22 | /// ``` |
23 | /// # futures::executor::block_on(async { |
24 | /// use futures::io::{self, AsyncWriteExt, Cursor}; |
25 | /// use futures::future::Aborted; |
26 | /// |
27 | /// let reader = Cursor::new([1, 2, 3, 4]); |
28 | /// let mut writer = Cursor::new(vec![0u8; 5]); |
29 | /// |
30 | /// let (fut, abort_handle) = io::copy_buf_abortable(reader, &mut writer); |
31 | /// let bytes = fut.await; |
32 | /// abort_handle.abort(); |
33 | /// writer.close().await.unwrap(); |
34 | /// match bytes { |
35 | /// Ok(Ok(n)) => { |
36 | /// assert_eq!(n, 4); |
37 | /// assert_eq!(writer.into_inner(), [1, 2, 3, 4, 0]); |
38 | /// Ok(n) |
39 | /// }, |
40 | /// Ok(Err(a)) => { |
41 | /// Err::<u64, Aborted>(a) |
42 | /// } |
43 | /// Err(e) => panic!("{}" , e) |
44 | /// } |
45 | /// # }).unwrap(); |
46 | /// ``` |
47 | pub fn copy_buf_abortable<R, W>( |
48 | reader: R, |
49 | writer: &mut W, |
50 | ) -> (CopyBufAbortable<'_, R, W>, AbortHandle) |
51 | where |
52 | R: AsyncBufRead, |
53 | W: AsyncWrite + Unpin + ?Sized, |
54 | { |
55 | let (handle: AbortHandle, reg: AbortRegistration) = AbortHandle::new_pair(); |
56 | (CopyBufAbortable { reader, writer, amt: 0, inner: reg.inner }, handle) |
57 | } |
58 | |
59 | pin_project! { |
60 | /// Future for the [`copy_buf()`] function. |
61 | #[derive(Debug)] |
62 | #[must_use = "futures do nothing unless you `.await` or poll them" ] |
63 | pub struct CopyBufAbortable<'a, R, W: ?Sized> { |
64 | #[pin] |
65 | reader: R, |
66 | writer: &'a mut W, |
67 | amt: u64, |
68 | inner: Arc<AbortInner> |
69 | } |
70 | } |
71 | |
72 | macro_rules! ready_or_break { |
73 | ($e:expr $(,)?) => { |
74 | match $e { |
75 | $crate::task::Poll::Ready(t) => t, |
76 | $crate::task::Poll::Pending => break, |
77 | } |
78 | }; |
79 | } |
80 | |
81 | impl<R, W> Future for CopyBufAbortable<'_, R, W> |
82 | where |
83 | R: AsyncBufRead, |
84 | W: AsyncWrite + Unpin + Sized, |
85 | { |
86 | type Output = Result<Result<u64, Aborted>, io::Error>; |
87 | |
88 | fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { |
89 | let mut this = self.project(); |
90 | loop { |
91 | // Check if the task has been aborted |
92 | if this.inner.aborted.load(Ordering::Relaxed) { |
93 | return Poll::Ready(Ok(Err(Aborted))); |
94 | } |
95 | |
96 | // Read some bytes from the reader, and if we have reached EOF, return total bytes read |
97 | let buffer = ready_or_break!(this.reader.as_mut().poll_fill_buf(cx))?; |
98 | if buffer.is_empty() { |
99 | ready_or_break!(Pin::new(&mut this.writer).poll_flush(cx))?; |
100 | return Poll::Ready(Ok(Ok(*this.amt))); |
101 | } |
102 | |
103 | // Pass the buffer to the writer, and update the amount written |
104 | let i = ready_or_break!(Pin::new(&mut this.writer).poll_write(cx, buffer))?; |
105 | if i == 0 { |
106 | return Poll::Ready(Err(io::ErrorKind::WriteZero.into())); |
107 | } |
108 | *this.amt += i as u64; |
109 | this.reader.as_mut().consume(i); |
110 | } |
111 | // Schedule the task to be woken up again. |
112 | // Never called unless Poll::Pending is returned from io objects. |
113 | this.inner.waker.register(cx.waker()); |
114 | |
115 | // Check to see if the task was aborted between the first check and |
116 | // registration. |
117 | // Checking with `Relaxed` is sufficient because |
118 | // `register` introduces an `AcqRel` barrier. |
119 | if this.inner.aborted.load(Ordering::Relaxed) { |
120 | return Poll::Ready(Ok(Err(Aborted))); |
121 | } |
122 | Poll::Pending |
123 | } |
124 | } |
125 | |