Почему я получаю ошибку несоответствующего типа (ожидаемый mpsc :: Receiver, found ()) при использовании локального исполнителя Futures? - PullRequest
Новая
       42

Почему я получаю ошибку несоответствующего типа (ожидаемый mpsc :: Receiver, found ()) при использовании локального исполнителя Futures?

0 голосов
/ 03 мая 2018

У меня есть следующий код: 

extern crate futures;

use futures::channel::mpsc;
use futures::executor::LocalPool;
use futures::prelude::*;

struct Ping(usize);

fn main() {
    let (last_tx, mut prev_rx) = mpsc::channel::<Ping>(1);
    let mut pool = LocalPool::new();
    let mut executor = pool.executor();
    let (tx_1, rx_1) = mpsc::channel::<Ping>(1);
    let (tx_2, rx_2) = mpsc::channel::<Ping>(1);
    executor.spawn_local(rx_1.for_each(move |Ping(size)| {
        if size == 10 {
            tx_2.close();
            println!("Done 2");
        } else {
            let tx = tx_2.clone();
            tx.send(Ping(size + 1));
        }
        Ok(())
    }));

    executor.spawn_local(rx_2.for_each(move |Ping(size)| {
        if size == 10 {
            tx_1.close();
            println!("Done 1");
        } else {
            let tx = tx_1.clone();
            tx.send(Ping(size + 1));
        }
        Ok(())
    }));
}

Не компилируется: 

error[E0271]: type mismatch resolving `<futures::stream::ForEach<futures::channel::mpsc::Receiver<Ping>, std::result::Result<(), futures::Never>, [closure@src/bin/futures_ring_poc.rs:16:40: 25:6 tx_2:_]> as futures::Future>::Item == ()`
  --> src/bin/futures_ring_poc.rs:16:14
   |
16 |     executor.spawn_local(rx_1.for_each(move |Ping(size)| {
   |              ^^^^^^^^^^^ expected struct `futures::channel::mpsc::Receiver`, found ()
   |
   = note: expected type `futures::channel::mpsc::Receiver<Ping>`
              found type `()`

error[E0271]: type mismatch resolving `<futures::stream::ForEach<futures::channel::mpsc::Receiver<Ping>, std::result::Result<(), futures::Never>, [closure@src/bin/futures_ring_poc.rs:27:40: 36:6 tx_1:_]> as futures::Future>::Item == ()`
  --> src/bin/futures_ring_poc.rs:27:14
   |
27 |     executor.spawn_local(rx_2.for_each(move |Ping(size)| {
   |              ^^^^^^^^^^^ expected struct `futures::channel::mpsc::Receiver`, found ()
   |
   = note: expected type `futures::channel::mpsc::Receiver<Ping>`
              found type `()`

Почему? От документы spawn_local должны занять () будущее, которое я передаю ему.

1 Ответ

0 голосов
/ 03 мая 2018

Как говорится в сообщении об ошибке:

разрешение несоответствия типов <ForEach<Receiver<Ping>, Result<(), futures::Never>, [closure]> как <futures::Future>::Item == ()

spawn_local требует, чтобы переданное ему будущее имело связанный с тип Item тип единицы / пустой кортеж / (). Ваше будущее не вернет этот тип.

должно занять () будущее, которое я передаю ему

Я не знаю, почему вы считаете, что это правда. for_each реализует Future путем , возвращая поток как Item.

Это можно исправить с помощью map для сброса потока: 

executor.spawn_local(rx_1.for_each(move |Ping(size)| {
    // ...
}).map(|_| ()));

Это не позволяет вашему коду компилироваться, но исправляет вашу ошибку.

Заставить код работать с большими изменениями. Вот одна возможность, представленная без комментариев, поскольку она не имеет ничего общего с вашим вопросом: 

extern crate futures;

use futures::{
    channel::mpsc::{self, Receiver, Sender},
    executor::LocalPool,
    prelude::*,
};

struct Ping(usize);

fn pinger(
    rx: Receiver<Ping>,
    tx: Sender<Ping>,
    id: &'static str,
) -> impl Future<Item = (), Error = Never> {
    rx.map_err(Never::never_into::<Box<std::error::Error>>)
        .fold(tx, move |tx, Ping(size)| {
            println!("{}: {}", id, size);

            if size >= 10 {
                println!("{}: Done", id);
                tx.close().err_into().right_future()
            } else {
                tx.send(Ping(size + 1)).err_into().left_future()
            }
        })
        .map(|_| ())
        .map_err(move |e| panic!("Task {} failed: {}", id, e))
}

fn main() {
    let mut pool = LocalPool::new();
    let mut executor = pool.executor();

    let (tx_1, rx_1) = mpsc::channel(1);
    let (tx_2, rx_2) = mpsc::channel(1);
    let tx_ignite = tx_1.clone();

    executor.spawn_local(pinger(rx_1, tx_2, "Rx 1")).unwrap();
    executor.spawn_local(pinger(rx_2, tx_1, "Rx 2")).unwrap();

    executor
        .spawn_local({
            tx_ignite
                .send(Ping(0))
                .map(drop)
                .map_err(|e| panic!("{:?}", e))
        })
        .unwrap();

    pool.run(&mut executor);
}
...