Vfrdtyky

I am having a play with async/await by using Tokio with the async/await feature enabled in my Cargo.toml (and the latest Rust nightly with 2018 edition):

tokio = { version = "0.1.11", features = ["async-await-preview"] }

I've run into an error I don't understand, which is reproduced in this minimal example:

#![feature(await_macro, async_await, futures_api)]

use tokio::prelude::*;



pub fn main() {

    tokio::run_async(async {

        let s: Option<Box<dyn Sink<SinkItem = u8, SinkError = ()> + Send + Sync + 'static>> = None;



        if let Some(sink) = s {

            await!(sink.send_async(100));

        }

    });

}

The error is:

error[E0277]: the trait bound `for<'r> (dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r): futures::sink::Sink` is not satisfied

 --> src/main.rs:5:5

  |

5 |     tokio::run_async(async {

  |     ^^^^^^^^^^^^^^^^ the trait `for<'r> futures::sink::Sink` is not implemented for `dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send`

  |

  = note: required because of the requirements on the impl of `futures::sink::Sink` for `std::boxed::Box<dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send>`

  = note: required because it appears within the type `tokio_async_await::sink::send::Send<'_, std::boxed::Box<dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send>>`

  = note: required because it appears within the type `for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}`

  = note: required because it appears within the type `[static generator@src/main.rs:5:28: 11:6 for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}]`

  = note: required because it appears within the type `std::future::GenFuture<[static generator@src/main.rs:5:28: 11:6 for<'r, 's, 't0, 't1> {std::option::Option<std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 'r)>>, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 's)>, tokio_async_await::sink::send::Send<'t0, std::boxed::Box<(dyn futures::sink::Sink<SinkItem=u8, SinkError=()> + std::marker::Sync + std::marker::Send + 't1)>>, ()}]>`

  = note: required because it appears within the type `impl std::future::Future`

  = note: required by `tokio::async_await::run_async`

and it goes away if I remove the line beginning "await!".

When using Tokio without the experimental async/await support, my program is happy with the idea that a Box<dyn Sink> is a Sink, so I'm not really sure why the error crops up using async/await stuff.

What does the error mean? How can I get around this?

My fix was to wrap the Boxed Sinks into a newtype and implement the Sink trait on that. I think that perhaps Box<dyn Sink> does not implement Sink in this nightly, which is basically what the error message implies (I guess the async/await shims redefine Sink and don't implement it on Box).

My wrapper ended up looking like this:

struct BoxedSink<I, E>(Box<dyn Sink<SinkItem = I, SinkError = E> + Send + Sync + 'static>);



impl<I, E> Sink for BoxedSink<I, E> {

    type SinkItem = I;

    type SinkError = E;



    fn start_send(

        &mut self,

        input: Self::SinkItem,

    ) -> Result<AsyncSink<Self::SinkItem>, Self::SinkError> {

        self.0.start_send(input)

    }



    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {

        self.0.poll_complete()

    }

}

You have to wrap your Box<Sinks> in this to make them once again implement Sink.