futures_util/compat/
compat01as03.rs

1use futures_01::executor::{
2    spawn as spawn01, Notify as Notify01, NotifyHandle as NotifyHandle01, Spawn as Spawn01,
3    UnsafeNotify as UnsafeNotify01,
4};
5use futures_01::{Async as Async01, Future as Future01, Stream as Stream01};
6#[cfg(feature = "sink")]
7use futures_01::{AsyncSink as AsyncSink01, Sink as Sink01};
8use futures_core::{future::Future as Future03, stream::Stream as Stream03, task as task03};
9#[cfg(feature = "sink")]
10use futures_sink::Sink as Sink03;
11use std::boxed::Box;
12use std::cell::UnsafeCell;
13use std::pin::Pin;
14use std::task::Context;
15
16#[cfg(feature = "io-compat")]
17#[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
18#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
19pub use io::{AsyncRead01CompatExt, AsyncWrite01CompatExt};
20
21/// Converts a futures 0.1 Future, Stream, AsyncRead, or AsyncWrite
22/// object to a futures 0.3-compatible version,
23#[derive(Debug)]
24#[must_use = "futures do nothing unless you `.await` or poll them"]
25pub struct Compat01As03<T> {
26    pub(crate) inner: Spawn01<T>,
27}
28
29impl<T> Unpin for Compat01As03<T> {}
30
31impl<T> Compat01As03<T> {
32    /// Wraps a futures 0.1 Future, Stream, AsyncRead, or AsyncWrite
33    /// object in a futures 0.3-compatible wrapper.
34    pub fn new(object: T) -> Self {
35        Self { inner: spawn01(object) }
36    }
37
38    fn in_notify<R>(&mut self, cx: &mut Context<'_>, f: impl FnOnce(&mut T) -> R) -> R {
39        let notify = &WakerToHandle(cx.waker());
40        self.inner.poll_fn_notify(notify, 0, f)
41    }
42
43    /// Get a reference to 0.1 Future, Stream, AsyncRead, or AsyncWrite object contained within.
44    pub fn get_ref(&self) -> &T {
45        self.inner.get_ref()
46    }
47
48    /// Get a mutable reference to 0.1 Future, Stream, AsyncRead or AsyncWrite object contained
49    /// within.
50    pub fn get_mut(&mut self) -> &mut T {
51        self.inner.get_mut()
52    }
53
54    /// Consume this wrapper to return the underlying 0.1 Future, Stream, AsyncRead, or
55    /// AsyncWrite object.
56    pub fn into_inner(self) -> T {
57        self.inner.into_inner()
58    }
59}
60
61/// Extension trait for futures 0.1 [`Future`](futures_01::future::Future)
62pub trait Future01CompatExt: Future01 {
63    /// Converts a futures 0.1
64    /// [`Future<Item = T, Error = E>`](futures_01::future::Future)
65    /// into a futures 0.3
66    /// [`Future<Output = Result<T, E>>`](futures_core::future::Future).
67    ///
68    /// ```
69    /// # futures::executor::block_on(async {
70    /// # // TODO: These should be all using `futures::compat`, but that runs up against Cargo
71    /// # // feature issues
72    /// use futures_util::compat::Future01CompatExt;
73    ///
74    /// let future = futures_01::future::ok::<u32, ()>(1);
75    /// assert_eq!(future.compat().await, Ok(1));
76    /// # });
77    /// ```
78    fn compat(self) -> Compat01As03<Self>
79    where
80        Self: Sized,
81    {
82        Compat01As03::new(self)
83    }
84}
85impl<Fut: Future01> Future01CompatExt for Fut {}
86
87/// Extension trait for futures 0.1 [`Stream`](futures_01::stream::Stream)
88pub trait Stream01CompatExt: Stream01 {
89    /// Converts a futures 0.1
90    /// [`Stream<Item = T, Error = E>`](futures_01::stream::Stream)
91    /// into a futures 0.3
92    /// [`Stream<Item = Result<T, E>>`](futures_core::stream::Stream).
93    ///
94    /// ```
95    /// # futures::executor::block_on(async {
96    /// use futures::stream::StreamExt;
97    /// use futures_util::compat::Stream01CompatExt;
98    ///
99    /// let stream = futures_01::stream::once::<u32, ()>(Ok(1));
100    /// let mut stream = stream.compat();
101    /// assert_eq!(stream.next().await, Some(Ok(1)));
102    /// assert_eq!(stream.next().await, None);
103    /// # });
104    /// ```
105    fn compat(self) -> Compat01As03<Self>
106    where
107        Self: Sized,
108    {
109        Compat01As03::new(self)
110    }
111}
112impl<St: Stream01> Stream01CompatExt for St {}
113
114/// Extension trait for futures 0.1 [`Sink`](futures_01::sink::Sink)
115#[cfg(feature = "sink")]
116#[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
117pub trait Sink01CompatExt: Sink01 {
118    /// Converts a futures 0.1
119    /// [`Sink<SinkItem = T, SinkError = E>`](futures_01::sink::Sink)
120    /// into a futures 0.3
121    /// [`Sink<T, Error = E>`](futures_sink::Sink).
122    ///
123    /// ```
124    /// # futures::executor::block_on(async {
125    /// use futures::{sink::SinkExt, stream::StreamExt};
126    /// use futures_util::compat::{Stream01CompatExt, Sink01CompatExt};
127    ///
128    /// let (tx, rx) = futures_01::unsync::mpsc::channel(1);
129    /// let (mut tx, mut rx) = (tx.sink_compat(), rx.compat());
130    ///
131    /// tx.send(1).await.unwrap();
132    /// drop(tx);
133    /// assert_eq!(rx.next().await, Some(Ok(1)));
134    /// assert_eq!(rx.next().await, None);
135    /// # });
136    /// ```
137    fn sink_compat(self) -> Compat01As03Sink<Self, Self::SinkItem>
138    where
139        Self: Sized,
140    {
141        Compat01As03Sink::new(self)
142    }
143}
144#[cfg(feature = "sink")]
145impl<Si: Sink01> Sink01CompatExt for Si {}
146
147fn poll_01_to_03<T, E>(x: Result<Async01<T>, E>) -> task03::Poll<Result<T, E>> {
148    match x? {
149        Async01::Ready(t) => task03::Poll::Ready(Ok(t)),
150        Async01::NotReady => task03::Poll::Pending,
151    }
152}
153
154impl<Fut: Future01> Future03 for Compat01As03<Fut> {
155    type Output = Result<Fut::Item, Fut::Error>;
156
157    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> task03::Poll<Self::Output> {
158        poll_01_to_03(self.in_notify(cx, Future01::poll))
159    }
160}
161
162impl<St: Stream01> Stream03 for Compat01As03<St> {
163    type Item = Result<St::Item, St::Error>;
164
165    fn poll_next(
166        mut self: Pin<&mut Self>,
167        cx: &mut Context<'_>,
168    ) -> task03::Poll<Option<Self::Item>> {
169        match self.in_notify(cx, Stream01::poll)? {
170            Async01::Ready(Some(t)) => task03::Poll::Ready(Some(Ok(t))),
171            Async01::Ready(None) => task03::Poll::Ready(None),
172            Async01::NotReady => task03::Poll::Pending,
173        }
174    }
175}
176
177/// Converts a futures 0.1 Sink object to a futures 0.3-compatible version
178#[cfg(feature = "sink")]
179#[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
180#[derive(Debug)]
181#[must_use = "sinks do nothing unless polled"]
182pub struct Compat01As03Sink<S, SinkItem> {
183    pub(crate) inner: Spawn01<S>,
184    pub(crate) buffer: Option<SinkItem>,
185    pub(crate) close_started: bool,
186}
187
188#[cfg(feature = "sink")]
189impl<S, SinkItem> Unpin for Compat01As03Sink<S, SinkItem> {}
190
191#[cfg(feature = "sink")]
192impl<S, SinkItem> Compat01As03Sink<S, SinkItem> {
193    /// Wraps a futures 0.1 Sink object in a futures 0.3-compatible wrapper.
194    pub fn new(inner: S) -> Self {
195        Self { inner: spawn01(inner), buffer: None, close_started: false }
196    }
197
198    fn in_notify<R>(&mut self, cx: &mut Context<'_>, f: impl FnOnce(&mut S) -> R) -> R {
199        let notify = &WakerToHandle(cx.waker());
200        self.inner.poll_fn_notify(notify, 0, f)
201    }
202
203    /// Get a reference to 0.1 Sink object contained within.
204    pub fn get_ref(&self) -> &S {
205        self.inner.get_ref()
206    }
207
208    /// Get a mutable reference to 0.1 Sink contained within.
209    pub fn get_mut(&mut self) -> &mut S {
210        self.inner.get_mut()
211    }
212
213    /// Consume this wrapper to return the underlying 0.1 Sink.
214    pub fn into_inner(self) -> S {
215        self.inner.into_inner()
216    }
217}
218
219#[cfg(feature = "sink")]
220impl<S, SinkItem> Stream03 for Compat01As03Sink<S, SinkItem>
221where
222    S: Stream01,
223{
224    type Item = Result<S::Item, S::Error>;
225
226    fn poll_next(
227        mut self: Pin<&mut Self>,
228        cx: &mut Context<'_>,
229    ) -> task03::Poll<Option<Self::Item>> {
230        match self.in_notify(cx, Stream01::poll)? {
231            Async01::Ready(Some(t)) => task03::Poll::Ready(Some(Ok(t))),
232            Async01::Ready(None) => task03::Poll::Ready(None),
233            Async01::NotReady => task03::Poll::Pending,
234        }
235    }
236}
237
238#[cfg(feature = "sink")]
239impl<S, SinkItem> Sink03<SinkItem> for Compat01As03Sink<S, SinkItem>
240where
241    S: Sink01<SinkItem = SinkItem>,
242{
243    type Error = S::SinkError;
244
245    fn start_send(mut self: Pin<&mut Self>, item: SinkItem) -> Result<(), Self::Error> {
246        debug_assert!(self.buffer.is_none());
247        self.buffer = Some(item);
248        Ok(())
249    }
250
251    fn poll_ready(
252        mut self: Pin<&mut Self>,
253        cx: &mut Context<'_>,
254    ) -> task03::Poll<Result<(), Self::Error>> {
255        match self.buffer.take() {
256            Some(item) => match self.in_notify(cx, |f| f.start_send(item))? {
257                AsyncSink01::Ready => task03::Poll::Ready(Ok(())),
258                AsyncSink01::NotReady(i) => {
259                    self.buffer = Some(i);
260                    task03::Poll::Pending
261                }
262            },
263            None => task03::Poll::Ready(Ok(())),
264        }
265    }
266
267    fn poll_flush(
268        mut self: Pin<&mut Self>,
269        cx: &mut Context<'_>,
270    ) -> task03::Poll<Result<(), Self::Error>> {
271        let item = self.buffer.take();
272        match self.in_notify(cx, |f| match item {
273            Some(i) => match f.start_send(i)? {
274                AsyncSink01::Ready => f.poll_complete().map(|i| (i, None)),
275                AsyncSink01::NotReady(t) => Ok((Async01::NotReady, Some(t))),
276            },
277            None => f.poll_complete().map(|i| (i, None)),
278        })? {
279            (Async01::Ready(_), _) => task03::Poll::Ready(Ok(())),
280            (Async01::NotReady, item) => {
281                self.buffer = item;
282                task03::Poll::Pending
283            }
284        }
285    }
286
287    fn poll_close(
288        mut self: Pin<&mut Self>,
289        cx: &mut Context<'_>,
290    ) -> task03::Poll<Result<(), Self::Error>> {
291        let item = self.buffer.take();
292        let close_started = self.close_started;
293
294        let result = self.in_notify(cx, |f| {
295            if !close_started {
296                if let Some(item) = item {
297                    if let AsyncSink01::NotReady(item) = f.start_send(item)? {
298                        return Ok((Async01::NotReady, Some(item), false));
299                    }
300                }
301
302                if let Async01::NotReady = f.poll_complete()? {
303                    return Ok((Async01::NotReady, None, false));
304                }
305            }
306
307            Ok((<S as Sink01>::close(f)?, None, true))
308        });
309
310        match result? {
311            (Async01::Ready(_), _, _) => task03::Poll::Ready(Ok(())),
312            (Async01::NotReady, item, close_started) => {
313                self.buffer = item;
314                self.close_started = close_started;
315                task03::Poll::Pending
316            }
317        }
318    }
319}
320
321#[repr(transparent)]
322struct NotifyWaker(UnsafeCell<task03::Waker>);
323
324#[allow(missing_debug_implementations)] // false positive: this is private type
325#[derive(Clone)]
326struct WakerToHandle<'a>(&'a task03::Waker);
327
328impl From<WakerToHandle<'_>> for NotifyHandle01 {
329    fn from(handle: WakerToHandle<'_>) -> Self {
330        let waker_ptr: Box<task03::Waker> = Box::new(handle.0.clone());
331        // NotifyWaker is a transparent (pointer compatible) wrapper for
332        // task03::Waker (and wrapping in UnsafeCell is fine).
333        let ptr: *mut NotifyWaker = Box::into_raw(waker_ptr) as *mut NotifyWaker;
334
335        unsafe { Self::new(ptr) }
336    }
337}
338
339impl Notify01 for NotifyWaker {
340    fn notify(&self, _: usize) {
341        unsafe { &*self.0.get() }.wake_by_ref();
342    }
343}
344
345unsafe impl Send for NotifyWaker {}
346unsafe impl Sync for NotifyWaker {}
347
348unsafe impl UnsafeNotify01 for NotifyWaker {
349    unsafe fn clone_raw(&self) -> NotifyHandle01 {
350        WakerToHandle(unsafe { &*self.0.get() }).into()
351    }
352
353    unsafe fn drop_raw(&self) {
354        /* UnsafeNotify01::drop_raw says this should receive `*mut Self`,
355         * but that isn't dyn compatible.
356         * miri is unhappy when a `*mut` is created from a `&` reference,
357         * so need to go through `UnsafeCell`.
358         */
359        let waker: *mut task03::Waker = self.0.get();
360        drop(unsafe { Box::from_raw(waker) });
361    }
362}
363
364#[cfg(feature = "io-compat")]
365#[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
366mod io {
367    use super::*;
368    use futures_io::{AsyncRead as AsyncRead03, AsyncWrite as AsyncWrite03};
369    use std::io::Error;
370    use tokio_io::{AsyncRead as AsyncRead01, AsyncWrite as AsyncWrite01};
371
372    /// Extension trait for tokio-io [`AsyncRead`](tokio_io::AsyncRead)
373    #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
374    pub trait AsyncRead01CompatExt: AsyncRead01 {
375        /// Converts a tokio-io [`AsyncRead`](tokio_io::AsyncRead) into a futures-io 0.3
376        /// [`AsyncRead`](futures_io::AsyncRead).
377        ///
378        /// ```
379        /// # if cfg!(miri) { return; } // Miri does not support epoll_create
380        /// # futures::executor::block_on(async {
381        /// use futures::io::AsyncReadExt;
382        /// use futures_util::compat::AsyncRead01CompatExt;
383        ///
384        /// let input = b"Hello World!";
385        /// let reader /* : impl tokio_io::AsyncRead */ = std::io::Cursor::new(input);
386        /// let mut reader /* : impl futures::io::AsyncRead + Unpin */ = reader.compat();
387        ///
388        /// let mut output = Vec::with_capacity(12);
389        /// reader.read_to_end(&mut output).await.unwrap();
390        /// assert_eq!(output, input);
391        /// # });
392        /// ```
393        fn compat(self) -> Compat01As03<Self>
394        where
395            Self: Sized,
396        {
397            Compat01As03::new(self)
398        }
399    }
400    impl<R: AsyncRead01> AsyncRead01CompatExt for R {}
401
402    /// Extension trait for tokio-io [`AsyncWrite`](tokio_io::AsyncWrite)
403    #[cfg_attr(docsrs, doc(cfg(feature = "io-compat")))]
404    pub trait AsyncWrite01CompatExt: AsyncWrite01 {
405        /// Converts a tokio-io [`AsyncWrite`](tokio_io::AsyncWrite) into a futures-io 0.3
406        /// [`AsyncWrite`](futures_io::AsyncWrite).
407        ///
408        /// ```
409        /// # if cfg!(miri) { return; } // Miri does not support epoll_create
410        /// # futures::executor::block_on(async {
411        /// use futures::io::AsyncWriteExt;
412        /// use futures_util::compat::AsyncWrite01CompatExt;
413        ///
414        /// let input = b"Hello World!";
415        /// let mut cursor = std::io::Cursor::new(Vec::with_capacity(12));
416        ///
417        /// let mut writer = (&mut cursor).compat();
418        /// writer.write_all(input).await.unwrap();
419        ///
420        /// assert_eq!(cursor.into_inner(), input);
421        /// # });
422        /// ```
423        fn compat(self) -> Compat01As03<Self>
424        where
425            Self: Sized,
426        {
427            Compat01As03::new(self)
428        }
429    }
430    impl<W: AsyncWrite01> AsyncWrite01CompatExt for W {}
431
432    impl<R: AsyncRead01> AsyncRead03 for Compat01As03<R> {
433        fn poll_read(
434            mut self: Pin<&mut Self>,
435            cx: &mut Context<'_>,
436            buf: &mut [u8],
437        ) -> task03::Poll<Result<usize, Error>> {
438            poll_01_to_03(self.in_notify(cx, |x| x.poll_read(buf)))
439        }
440    }
441
442    impl<W: AsyncWrite01> AsyncWrite03 for Compat01As03<W> {
443        fn poll_write(
444            mut self: Pin<&mut Self>,
445            cx: &mut Context<'_>,
446            buf: &[u8],
447        ) -> task03::Poll<Result<usize, Error>> {
448            poll_01_to_03(self.in_notify(cx, |x| x.poll_write(buf)))
449        }
450
451        fn poll_flush(
452            mut self: Pin<&mut Self>,
453            cx: &mut Context<'_>,
454        ) -> task03::Poll<Result<(), Error>> {
455            poll_01_to_03(self.in_notify(cx, AsyncWrite01::poll_flush))
456        }
457
458        fn poll_close(
459            mut self: Pin<&mut Self>,
460            cx: &mut Context<'_>,
461        ) -> task03::Poll<Result<(), Error>> {
462            poll_01_to_03(self.in_notify(cx, AsyncWrite01::shutdown))
463        }
464    }
465}
futures_util/compat/compat01as03.rs

futures_util/compat/
compat01as03.rs
