How to wait indefinitely on a future stream inside a future


I have a futures::mpsc::unbounded that receives futures, I want to create a future that receives from the channel and executes the futures with a delay in between, and pass that future to a tokio;
Tried receiver.for_each and works but it returns when there are no more items in the stream, I need it to just wait in there till a new item is received.
I am basically looking for a waiting/blocking version of for_each that I can pass to a tokio_core.
This is what i have:

    let dequeuer = self.dequeuer.take().unwrap();
        let mut core = Core::new().unwrap();
        let handle = core.handle();

        let done = dequeuer.for_each(|msg| {


That’s what for_each would do - it only completes when the underlying Stream indicates that no future items will be delivered.

Are you sure you’re not prematurely dropping all the senders? That would cause the stream to end.


Hi vitalyd,
You are right, the problem was a premature close