Delaying a Tokio Stream

user_4275937 0
Viewed 1125

Given a Stream, I want to create a new Stream where elements are yielded with a time delay between them.

I tried to write code that does that using tokio_core::reactor::Timeout and the and_then combinator for Streams, but the delay doesn't work: I get all the elements immediately, without a delay.

Here is a self contained example (playground):

extern crate tokio_core;
extern crate futures;

use std::time::Duration;

use futures::{Future, Stream, stream, Sink};
use self::futures::sync::{mpsc};
use tokio_core::reactor;

const NUM_ITEMS: u32 = 8;

fn main() {
    let mut core = reactor::Core::new().unwrap();
    let handle = core.handle();
    let chandle = handle.clone();

    let (sink, stream) = mpsc::channel::<u32>(0);

    let send_stream = stream::iter_ok(0 .. NUM_ITEMS)
        .and_then(move |i: u32| {
                  let cchandle = chandle.clone();
                  println!("Creating a timeout object...");
                  reactor::Timeout::new(Duration::new(1,0), &cchandle)
                      .map_err(|_| ())
                      .and_then(|_| Ok(i))
        });

    let sink = sink.sink_map_err(|_| ());
    handle.spawn(sink.send_all(send_stream).and_then(|_| Ok(())));

    let mut incoming_items = Vec::new();
    {
        let keep_messages = stream.for_each(|item| {
            incoming_items.push(item);
            println!("item = {}", item);
            Ok(())
        });

        core.run(keep_messages).unwrap();
    }
    assert_eq!(incoming_items, (0 .. NUM_ITEMS).collect::<Vec<u32>>());
}

For completeness, this is the output I get:

Creating a timeout object...
Creating a timeout object...
item = 0
Creating a timeout object...
item = 1
Creating a timeout object...
item = 2
Creating a timeout object...
item = 3
Creating a timeout object...
item = 4
Creating a timeout object...
item = 5
Creating a timeout object...
item = 6
item = 7

I suspect that the problem is somewhere in these lines:

reactor::Timeout::new(Duration::new(1,0), &cchandle)
    .map_err(|_| ())
    .and_then(|_| Ok(i))

It is possible that I don't really wait on the returned Timeout object, though I'm not sure how to solve it.

rust-tokio rust stream
1 Answers

As I suspected, the problem was was the manipulation (using and_then) of the newly created Timeout. We either need to first unwrap the result from the call to reactor::Timeout::new, which could become messy if done manually, or use into_future, to convert the result into a Future, and then work with it using Future combinators.

Code for solving the problem:

extern crate tokio_core;
extern crate futures;

use std::time::Duration;

use futures::{Future, Stream, stream, Sink, IntoFuture};
use self::futures::sync::{mpsc};
use tokio_core::reactor;

const NUM_ITEMS: u32 = 8;

fn main() {
    let mut core = reactor::Core::new().unwrap();
    let handle = core.handle();
    let chandle = handle.clone();

    let (sink, stream) = mpsc::channel::<u32>(0);

    let send_stream = stream::iter_ok(0 .. NUM_ITEMS)
        .and_then(move |i: u32| {
                  let cchandle = chandle.clone();
                  println!("Creating a timeout object...");
                  reactor::Timeout::new(Duration::new(1,0), &cchandle)
                      .into_future()
                      .and_then(move |timeout| timeout.and_then(move |_| Ok(i)))
                      .map_err(|_| ())
        });

    let sink = sink.sink_map_err(|_| ());
    handle.spawn(sink.send_all(send_stream).and_then(|_| Ok(())));

    let mut incoming_items = Vec::new();
    {
        let keep_messages = stream.for_each(|item| {
            incoming_items.push(item);
            println!("item = {}", item);
            Ok(())
        });

        core.run(keep_messages).unwrap();
    }
    assert_eq!(incoming_items, (0 .. NUM_ITEMS).collect::<Vec<u32>>());
}

Note that two and_then are being used. The first one unwraps the Result obtained from calling reactor::Timeout::new. The second one actually waits for the Timeout to fire.

user_42759370
Related