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

futures_util/compat/
compat01as03.rs
