use futures_01::{
task as task01, Async as Async01, AsyncSink as AsyncSink01,
Future as Future01, Poll as Poll01, Sink as Sink01,
StartSend as StartSend01, Stream as Stream01,
};
use futures_core::{
task::{
self as task03,
RawWaker,
RawWakerVTable,
},
TryFuture as TryFuture03,
TryStream as TryStream03,
};
use futures_sink::Sink as Sink03;
use crate::task::{ArcWake as ArcWake03, WakerRef};
use std::{
marker::PhantomData,
mem,
pin::Pin,
sync::Arc,
task::Context,
};
#[derive(Debug, Clone, Copy)]
#[must_use = "futures do nothing unless polled"]
pub struct Compat<T> {
pub(crate) inner: T,
}
#[derive(Debug)]
#[must_use = "sinks do nothing unless polled"]
pub struct CompatSink<T, Item> {
inner: T,
_phantom: PhantomData<fn(Item)>,
}
impl<T> Compat<T> {
pub fn into_inner(self) -> T {
self.inner
}
pub fn new(inner: T) -> Compat<T> {
Compat { inner }
}
}
impl<T, Item> CompatSink<T, Item> {
pub fn into_inner(self) -> T {
self.inner
}
pub fn new(inner: T) -> Self {
CompatSink {
inner,
_phantom: PhantomData,
}
}
}
fn poll_03_to_01<T, E>(x: task03::Poll<Result<T, E>>)
-> Result<Async01<T>, E>
{
match x {
task03::Poll::Ready(Ok(t)) => Ok(Async01::Ready(t)),
task03::Poll::Pending => Ok(Async01::NotReady),
task03::Poll::Ready(Err(e)) => Err(e),
}
}
impl<Fut> Future01 for Compat<Fut>
where
Fut: TryFuture03 + Unpin,
{
type Item = Fut::Ok;
type Error = Fut::Error;
fn poll(&mut self) -> Poll01<Self::Item, Self::Error> {
with_context(self, |inner, cx| poll_03_to_01(inner.try_poll(cx)))
}
}
impl<St> Stream01 for Compat<St>
where
St: TryStream03 + Unpin,
{
type Item = St::Ok;
type Error = St::Error;
fn poll(&mut self) -> Poll01<Option<Self::Item>, Self::Error> {
with_context(self, |inner, cx| match inner.try_poll_next(cx) {
task03::Poll::Ready(None) => Ok(Async01::Ready(None)),
task03::Poll::Ready(Some(Ok(t))) => Ok(Async01::Ready(Some(t))),
task03::Poll::Pending => Ok(Async01::NotReady),
task03::Poll::Ready(Some(Err(e))) => Err(e),
})
}
}
impl<T, Item> Sink01 for CompatSink<T, Item>
where
T: Sink03<Item> + Unpin,
{
type SinkItem = Item;
type SinkError = T::SinkError;
fn start_send(
&mut self,
item: Self::SinkItem,
) -> StartSend01<Self::SinkItem, Self::SinkError> {
with_sink_context(self, |mut inner, cx| {
match inner.as_mut().poll_ready(cx) {
task03::Poll::Ready(Ok(())) => {
inner.start_send(item).map(|()| AsyncSink01::Ready)
}
task03::Poll::Pending => Ok(AsyncSink01::NotReady(item)),
task03::Poll::Ready(Err(e)) => Err(e),
}
})
}
fn poll_complete(&mut self) -> Poll01<(), Self::SinkError> {
with_sink_context(self, |inner, cx| poll_03_to_01(inner.poll_flush(cx)))
}
fn close(&mut self) -> Poll01<(), Self::SinkError> {
with_sink_context(self, |inner, cx| poll_03_to_01(inner.poll_close(cx)))
}
}
#[derive(Clone)]
struct Current(task01::Task);
impl Current {
fn new() -> Current {
Current(task01::current())
}
fn as_waker(&self) -> WakerRef<'_> {
unsafe fn ptr_to_current<'a>(ptr: *const ()) -> &'a Current {
#[allow(clippy::cast_ptr_alignment)]
&*(ptr as *const Current)
}
fn current_to_ptr(current: &Current) -> *const () {
current as *const Current as *const ()
}
unsafe fn clone(ptr: *const ()) -> RawWaker {
mem::transmute::<task03::Waker, RawWaker>(
Arc::new(ptr_to_current(ptr).clone()).into_waker()
)
}
unsafe fn drop(_: *const ()) {}
unsafe fn wake(ptr: *const ()) {
ptr_to_current(ptr).0.notify()
}
let ptr = current_to_ptr(self);
let vtable = &RawWakerVTable::new(clone, wake, wake, drop);
unsafe {
WakerRef::new(task03::Waker::from_raw(RawWaker::new(ptr, vtable)))
}
}
}
impl ArcWake03 for Current {
fn wake_by_ref(arc_self: &Arc<Self>) {
arc_self.0.notify();
}
}
fn with_context<T, R, F>(compat: &mut Compat<T>, f: F) -> R
where
T: Unpin,
F: FnOnce(Pin<&mut T>, &mut Context<'_>) -> R,
{
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
f(Pin::new(&mut compat.inner), &mut cx)
}
fn with_sink_context<T, Item, R, F>(compat: &mut CompatSink<T, Item>, f: F) -> R
where
T: Unpin,
F: FnOnce(Pin<&mut T>, &mut Context<'_>) -> R,
{
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
f(Pin::new(&mut compat.inner), &mut cx)
}
#[cfg(feature = "io-compat")]
mod io {
use super::*;
use futures_io::{AsyncRead as AsyncRead03, AsyncWrite as AsyncWrite03};
use tokio_io::{AsyncRead as AsyncRead01, AsyncWrite as AsyncWrite01};
fn poll_03_to_io<T>(x: task03::Poll<Result<T, std::io::Error>>)
-> Result<T, std::io::Error>
{
match x {
task03::Poll::Ready(Ok(t)) => Ok(t),
task03::Poll::Pending => Err(std::io::ErrorKind::WouldBlock.into()),
task03::Poll::Ready(Err(e)) => Err(e),
}
}
impl<R: AsyncRead03 + Unpin> std::io::Read for Compat<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
poll_03_to_io(Pin::new(&mut self.inner).poll_read(&mut cx, buf))
}
}
impl<R: AsyncRead03 + Unpin> AsyncRead01 for Compat<R> {
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
let initializer = self.inner.initializer();
let does_init = initializer.should_initialize();
if does_init {
initializer.initialize(buf);
}
does_init
}
}
impl<W: AsyncWrite03 + Unpin> std::io::Write for Compat<W> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
poll_03_to_io(Pin::new(&mut self.inner).poll_write(&mut cx, buf))
}
fn flush(&mut self) -> std::io::Result<()> {
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
poll_03_to_io(Pin::new(&mut self.inner).poll_flush(&mut cx))
}
}
impl<W: AsyncWrite03 + Unpin> AsyncWrite01 for Compat<W> {
fn shutdown(&mut self) -> std::io::Result<Async01<()>> {
let current = Current::new();
let waker = current.as_waker();
let mut cx = Context::from_waker(&waker);
poll_03_to_01(Pin::new(&mut self.inner).poll_close(&mut cx))
}
}
}