Context in futures 0.2


#1

I used to be able to write something like this:

pub fn interval() -> UnboundedReceiver<()> {

    let (mut tx, rx) = unbounded();

    spawn(move || {
        loop {
            tx.start_send(()).unwrap();
            tx.poll_ready(context).unwrap(); // <- missing Context
            sleep(Duration::from_millis(100));
        }
    });

    rx
}

How can I get a hold of the Context?


#2

When were you able to write that? Context being exposed is new in 0.2. Can you expand a bit on what you’re trying to do? It’s unclear why you need a start_send/poll_ready sequence on an unbounded channel.


#3

I was able to write it pre 0.2 release where I’d use tx.poll_complete().
Now that you mentioned it, I don’t think I need unbounded channel at all.

Would executor::block_on(tx.send(())).unwrap(); be equivalent?


#4

Ah ok - I thought you were showing a snippet of what you used to be able to do, including the context.

What are you trying to do? Send some message on a timer?


#5

I’m making a stream of current size of terminal using some utils from termion crate.

Full code (that I stole from somewhere :speak_no_evil:):

pub fn size_stream() -> UnboundedReceiver<(u16, u16)> {

    let (mut tx, rx) = unbounded();

    spawn(move || {
        let mut current_size = (0, 0);
        loop {
            match terminal_size() {
                Ok(new_size) => if new_size != current_size {
                    current_size = new_size;
                    tx.start_send(current_size).unwrap();
                    tx.poll_ready().unwrap();
                },
                Err(e) => {}
            }
            sleep(Duration::from_millis(20));
        }
    });

    rx
}

#6

In the old code (or new one, for that matter) you don’t need the start_send/poll_ready duo - you just use tx.unbounded_send(...) and that’s it.

Are you actually using (need) tokio here? Was the old code?


#7

Yes I’m using tokio because of bunch of http requests I’m doing in a background.


#8

Ok I see.

Any idea why the old code uses a background thread to send messages? Is it because the old tokio runtime has a single thread that did IO + execution? Reason I ask is because I don’t think that part should be necessary if you use the 0.2 tokio, where the default execution model is a dedicated IO thread + threadpool for execution. You should be able to use tokio-timer crate to create an Interval that defines your send period. Interval is a Stream, so you can spawn that processing onto the executor, and have each tick send a message on the channel. Something to explore/consider perhaps.


#9

I don’t have a requirement for more that just IO thread. I tried to rewrite it using tokio-timer, but it doesn’t support v0.2 of futures yet so temporarily I need to use background thread with sleep.


#10

Here’s a small example I whipped up using tokio 0.2 and futures 0.1.21 (0.2 is a bit of a mess right now in terms of types moving around, and doesn’t play too nicely).

extern crate futures;
extern crate tokio;

use tokio::prelude::*;
use tokio::timer::Interval;
use std::time::{Duration, Instant};

fn main() {
    let (tx, rx) = futures::sync::mpsc::unbounded();
    let tx_fut = Interval::new(Instant::now(), Duration::from_secs(1))
        .map_err(|_| ())
        .for_each(move |_| {
            tx.unbounded_send(std::time::SystemTime::now()).unwrap();
            Ok(())
        });

    let fut = tx_fut.select(rx.for_each(|time| {
        println!("{:?}", time);
        Ok(())
    }).map_err(|_| ())).then(|_| Ok(()));
    tokio::run(fut);
}

#11

I guess I’ll have to wait for tokio-timer to play nicely with futures 0.2. Then I could probably remove channels all together and just do something like:

let time_stream = Interval::new(Instant::now(), Duration::from_secs(1))
    .map(|_| std::time::SystemTime::now());

I found the code that I “borrowed” from:

I’m not sure how to convert io::stdin().events() iterator into a stream without using a seperate thread.


#12

Since UnboundedSender implements Sink you can also do:

    let tx_fut = Interval::new(Instant::now(), Duration::from_secs(1))
            .map(|_| std::time::SystemTime::now())
            .map_err(|_| ())
            .forward(tx.sink_map_err(|_| ()))
            .map(|_| ());

#13

Nice - is sink_map_err new? I don’t recall seeing it before.


#14

I have no idea. I only found it recently.