5

I am trying to write a simple exmaple of futures-rs mpsc queues usage:

extern crate futures; // v0.1 (old)

use futures::{Sink, Stream};
use futures::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel::<i32>(1000);

    let handle = thread::spawn(move || {
        tx.clone().send(1);
        tx.clone().send(2);
        tx.clone().send(3);
    });

    let mut rx = rx.map(|x| {
        println!("stream: {}", x);
        x * x
    });

    handle.join().unwrap();

    rx.poll().unwrap();
}

But it doesn't output anything to the console (I expect it to print stream: 1, stream: 2 and stream: 3). I also tried to replace rx.poll().unwrap() with rx.wait(), but it still outputs nothing. And I didn't find any usage examples in the futures-rs docs. What am I doing wrong?

1 Answer 1

5

It is highly recommended to read the warning and error messages that the compiler tells you. This is one of the big benefits of a statically typed language with a compiler:

warning: unused result which must be used: futures do nothing unless polled, #[warn(unused_must_use)] on by default
  --> src/main.rs:11:9
   |
11 |         tx.clone().send(1);
   |         ^^^^^^^^^^^^^^^^^^^

warning: unused result which must be used: futures do nothing unless polled, #[warn(unused_must_use)] on by default
  --> src/main.rs:12:9
   |
12 |         tx.clone().send(2);
   |         ^^^^^^^^^^^^^^^^^^^

warning: unused result which must be used: futures do nothing unless polled, #[warn(unused_must_use)] on by default
  --> src/main.rs:13:9
   |
13 |         tx.clone().send(3);
   |         ^^^^^^^^^^^^^^^^^^^

I'm no expert with futures, but this compiles without warnings and prints all three values:

extern crate futures; // 0.1.23

use futures::{sync::mpsc, Async, Future, Sink, Stream};
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel(1000);

    let handle = thread::spawn(move || {
        tx.send(1)
            .and_then(|tx| tx.send(2))
            .and_then(|tx| tx.send(3))
            .wait()
            .expect("Unable to send");
    });

    let mut rx = rx.map(|x| x * x);

    handle.join().unwrap();

    while let Ok(Async::Ready(Some(v))) = rx.poll() {
        println!("stream: {}", v);
    }
}

and_then is used to send each subsequent value after the previous one. wait is used to block the generated thread until everything has successfully sent. The poll method is used to grab values off the queue until it runs out. There are multiple ways that might fail, and I'm ignoring them all and only concentrating on the success case.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.