rust - 如何关闭已修改和正在执行的 `futures::sync::mpsc::Receiver` 流？

Question

我希望能够按照这些思路做一些事情，以便异步关闭 Receiver 流:

extern crate futures;
extern crate tokio;

use futures::future::lazy;
use futures::stream::AndThen;
use futures::sync::mpsc::Receiver;
use futures::{Future, Sink, Stream};
use std::sync::{Arc, Mutex};

use tokio::timer::{Delay, Interval};

fn main() {
    tokio::run(lazy(|| {
        let (tx, rx) = futures::sync::mpsc::channel(1000);

        let arc = Arc::new(Mutex::<Option<AndThen<Receiver<u32>, _, _>>>::new(None));

        {
            let mut and_then = arc.lock().unwrap();
            *and_then = Some(rx.and_then(|num| {
                println!("{}", num);
                Ok(())
            }));
        }

        let arc_clone = arc.clone();
        // This is the part I'd like to be able to do
        // After one second, close the `Receiver` so that future
        // calls to the `Sender` don't call the callback above in the
        // closure passed to `rx.and_then`
        tokio::spawn(
            Delay::new(std::time::Instant::now() + std::time::Duration::from_secs(1))
                .map_err(|e| eprintln!("Some delay err {:?}", e))
                .and_then(move |_| {
                    let mut maybe_stream = arc_clone.lock().unwrap();
                    match maybe_stream.take() {
                        Some(stream) => stream.into_inner().close(),
                        None => eprintln!("Can't close non-existent stream"), // line "A"
                    }
                    Ok(())
                }),
        );

        {
            let mut maybe_stream = arc.lock().unwrap();
            let stream = maybe_stream.take().expect("Stream already ripped out"); // line "B"

            let rx = stream.for_each(|_| Ok(()));
            tokio::spawn(rx);
        }

        tokio::spawn(
            Interval::new_interval(std::time::Duration::from_millis(10))
                .take(10)
                .map_err(|e| {
                    eprintln!("Interval error?! {:?}", e);
                })
                .fold((tx, 0), |(tx, i), _| {
                    tx.send(i as u32)
                        .map_err(|e| eprintln!("Send error?! {:?}", e))
                        .map(move |tx| (tx, i + 1))
                })
                .map(|_| ()),
        );

        Ok(())
    }));
}

Playground

但是，A 行运行是因为我必须移动 B 行上的流才能对其调用 .for_each。如果我不调用 .for_each(或类似的东西)，据我所知，我根本无法执行 AndThen。我无法在不实际移动对象的情况下调用 .for_each，因为 for_each 是一种移动方法。

我有可能做我想做的事吗？这似乎绝对是可能的，但也许我遗漏了一些明显的东西。

我使用 0.1 的 future 和 0.1 的 tokio。

Answer 1

不会撒谎，我和@shepmaster 一起解决这个问题，你的问题很不清楚。也就是说，它感觉就像您正在尝试做一些 mpsc 的事情一样的一部分 futures不适合做。

无论如何。讲解时间。

无论何时组合/组合流(或 future !)，每个组合方法都需要 self , 不是 &self或 &mut self正如我认为您可能希望的那样。

当你到达你的这个代码块的那一刻:

    {
        let mut maybe_stream = arc.lock().unwrap();
        let stream = maybe_stream.take().expect("Stream already ripped out"); // line "B"

        let rx = stream.for_each(|_| Ok(()));
        tokio::spawn(rx);
    }

...流是从Arc<Option<Receiver<T>>>中提取的当你take()它，它的内容被替换为None .然后在 Tokio react 器上生成它，它开始处理这部分。这rx现在处于循环中，您不再可用。此外，您的 maybe_stream现在包含 None .

延迟一段时间后，您尝试 take() Arc<Option<Receiver<T>>>的内容(A 行)。因为现在什么都没有了，你也什么都没有了，因此也没有什么可以关闭的了。您的代码出错了。

而不是传递 mpsc::Receiver并希望摧毁它，使用一种机制来停止流本身。您可以自己这样做，也可以使用像 stream-cancel 这样的 crate 为你这样做。

DIY 版本在这里，根据您的代码修改:

extern crate futures;
extern crate tokio;

use futures::future::lazy;
use futures::{future, Future, Sink, Stream};
use std::sync::{Arc, RwLock};
use std::sync::atomic::{Ordering, AtomicBool};
use tokio::timer::{Delay, Interval};

fn main() {
    tokio::run(lazy(|| {
        let (tx, rx) = futures::sync::mpsc::channel(1000);

        let circuit_breaker:Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
        let c_b_copy = Arc::clone(&circuit_breaker);
        tokio::spawn(
            Delay::new(std::time::Instant::now() + std::time::Duration::from_secs(1))
                .map_err(|e| eprintln!("Some delay err {:?}", e))
                .and_then(move |_| {
                    // We set the CB to true in order to stop processing of the stream
                    circuit_breaker.store(true, Ordering::Relaxed);
                    Ok(())
                }),
        );

        {
            let rx2 = rx.for_each(|e| {
                println!("{:?}", e);
                Ok(())
            });
            tokio::spawn(rx2);
        }

        tokio::spawn(
            Interval::new_interval(std::time::Duration::from_millis(100))
                .take(100)
                // take_while causes the stream to continue as long as its argument returns a future resolving to true.
                // In this case, we're checking every time if the circuit-breaker we've introduced is false
                .take_while(move |_| {
                    future::ok(
                        c_b_copy.load(Ordering::Relaxed) == false
                    );
                })
                .map_err(|e| {
                    eprintln!("Interval error?! {:?}", e);
                })
                .fold((tx, 0), |(tx, i), _| {
                    tx.send(i as u32)
                        .map_err(|e| eprintln!("Send error?! {:?}", e))
                        .map(move |tx| (tx, i + 1))
                })
                .map(|_| ()),
        );

        Ok(())
    }));
}

Playground

添加的take_while()允许您对流的内容或外部谓词进行操作以继续或停止流。请注意，即使我们使用的是 AtomicBool , 我们还需要 Arc由于'static Tokio 的生命周期要求。

逆流

经过评论的讨论，this solution可能更适合您的用例。我有效地实现了一个由断路器覆盖的扇出流。奇迹发生在这里:

impl<S> Stream for FanOut<S> where S:Stream, S::Item:Clone {

    type Item = S::Item;

    type Error = S::Error;

    fn poll(&mut self) -> Result<Async<Option<S::Item>>, S::Error> {
        match self.inner.as_mut() {
            Some(ref mut r) => {
                let mut breaker = self.breaker.write().expect("Poisoned lock");
                match breaker.status {
                    false => {
                        let item = r.poll();
                        match &item {
                            &Ok(Async::Ready(Some(ref i))) => {
                                breaker.registry.iter_mut().for_each(|sender| {
                                    sender.try_send(i.clone()).expect("Dead channel");
                                });
                                item
                            },
                            _ => item
                        }
                    },
                    true => Ok(Async::Ready(None))
                }
            }
            _ => {

                let mut breaker = self.breaker.write().expect("Poisoned lock");
                // Stream is over, drop all the senders

                breaker.registry = vec![];
                Ok(Async::Ready(None))
            }
        }
    }
}

如果状态指示器设置为false，则轮询上述流；然后将结果发送给所有听众。如果 poll 的结果是Async::Ready(None) (表示推流结束)，关闭所有监听 channel 。

如果状态指示器设置为真，则关闭所有监听 channel ，流返回Async::Ready(None) (并被 Tokio 取消执行)。

FanOut对象是可克隆的，但只有初始实例才能执行任何操作。