1

I'd like to be able to do something along these lines in order to close a Receiver stream asynchronously:

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

However, line A runs because I have to move the stream on line B in order to call .for_each on it. If I don't call .for_each (or something like it), I can't execute the AndThen at all, as far as I know. I can't call .for_each without actually moving the object because for_each is a moving method.

Is it possible for me to do what I'm trying to do? This seems like it should definitely be possible, but maybe I'm missing something obvious.

I'm using futures at 0.1 and tokio at 0.1.

  • To add, I think I could have included no code snippet at all, and this would have been a high enough quality question. The code snippet was just to illustrate one possible approach that I know doesn't work as well as an explanation as to why it doesn't work. The question is what approach would work. – River Tam Dec 24 '18 at 17:03
  • I understand that, but I'm confused about why you'd need to compile it to answer the question. I don't want to fix the code snippet: I want to find a code snippet that works. The code snippet is fundamentally flawed from top to bottom. – River Tam Dec 24 '18 at 17:04
  • included no code snippet at all — such a question might be on topic perhaps, but people also tend to look unfavorably on questions without any attempt to solve it, so such a question might be downvoted. – Shepmaster Dec 24 '18 at 17:04
  • Yep, that's exactly why I included a snippet in the first place. I wanted to show that I'd tried something, and what I'd tried, and why it didn't work. Potential answers don't really need to consider what was wrong with the approach, especially when the question clearly states why it doesn't work. – River Tam Dec 24 '18 at 17:05
2

Not going to lie, I'm with @shepmaster on this one, your question is pretty unclear. That said, it feels like you're trying to do something the mpsc part of futures isn't geared to do.

Anyway. Explanation time.

Whenever you combine/compose streams (or futures!), every single composition method takes self, not &self or &mut self as I think you might have hoped.

The moment you get to this code block of yours:

    {
        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);
    }

...The stream is extracted from the Arc<Option<Receiver<T>>> when you take() it, and the content of it is replaced by None. You then spawn it on the Tokio reactor, which starts processing this part. This rx is now on the loop, and no longer available to you. Additionally, your maybe_stream now contains None.

After a delay, you then try to take() the content of the Arc<Option<Receiver<T>>> (line A). Since there's now nothing left, you're left with nothing, and therefore there is nothing left to close. Your code errors out.

Instead of passing around a mpsc::Receiver and hoping to destroy it, use a mechanism to stop the stream itself. You can do so yourself or you can use a crate like stream-cancel to do so for you.

The DIY version is here, modified from your code:

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

The added take_while() allows you to operate on either the stream's content, or an outside predicate, to continue or to stop a stream. Note that even though we're using an AtomicBool, we still need the Arc due to 'static lifetime requirements from Tokio.

Reversing the flow

After some discussion in the comments, this solution may be more suited for your use case. I effectively implemented a fan-out stream covered by a circuit breaker. The magic happens here:

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))
            }
        }
    }
}

If the status indicator is set to false, the above stream is polled; the result is then sent to all listeners. If the result of the poll is Async::Ready(None) (indicating that the stream is finished), all listener channels are closed.

If the status indicator is set to true, all listener channels are closed, and the stream returns Async::Ready(None) (and is dropped from execution by Tokio).

The FanOut object is cloneable, but only the initial instance will do anything.

  • The .take_until approach is interesting. In the meantime, I've implemented a similar approach that actually drops the tx and checks whether the tx has been dropped inside the fold (this makes more sense given the code I'm actually writing outside of the scope of the question). I was really hoping to achieve this from the Receiver side however, rather than from the Sender side. This is because my Senders are organized in a structure, while my Receivers are given to the client which will treat them as "special". – River Tam Dec 24 '18 at 17:54
  • As such, as a direct answer to the question, I think the stream-cancel library is more suitable -- it allows me to do exactly what I want to do. I will need to benchmark, however, as I want to make sure it's not going to harm the performance of the nominal flow. – River Tam Dec 24 '18 at 17:55
  • The return in the take_while is non-idiomatic. – Shepmaster Dec 24 '18 at 17:56
  • @RiverTam the clue was in the name: mpsc - multiple producers, single consumer. You could also send your receiver into a fan-out structure (I've had to write my own on this once after trying to adapt spmc. Not sure if there is a crate fully satisfying your need) and start/stop consumers as a result. The take_while() explicitely does what stream-cancel does under the hood. The source of it uses virtually the same construct :-) – Sébastien Renauld Dec 24 '18 at 18:04
  • @SébastienRenauld Yes, before trying all of this I was looking into mpmc or spmc solutions, but I found nothing that worked exactly how I wanted, so I ended up going with my own solution. multiqueue does what I want, but it has a bug in it (stackoverflow.com/questions/53711246/…). Crossbeam channels also offer a broadcast I think, but they don't work with tokio as far as I could tell. – River Tam Dec 24 '18 at 18:06
1

You can use a crate like stream-cancel to achieve this. Here, I've used the Valved stream wrapper which takes an existing stream and returns a value you can use to cancel the stream later on:

use futures::{
    future::lazy,
    {Future, Sink, Stream},
}; // 0.1.25
use stream_cancel::Valved; // 0.4.4
use tokio::timer::{Delay, Interval}; // 0.1.13

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

        tokio::spawn({
            rx.for_each(|num| {
                println!("{}", num);
                Ok(())
            })
        });

        tokio::spawn({
            Delay::new(std::time::Instant::now() + std::time::Duration::from_secs(1))
                .map_err(|e| eprintln!("Some delay err {:?}", e))
                .map(move |_| trigger.cancel()),
        });

        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)
                        .map_err(|e| eprintln!("Send error?! {:?}", e))
                        .map(move |tx| (tx, i + 1))
                })
                .map(|_| ()),
        });

        Ok(())
    }));
}

The crate has other types suited for subtly different usecases, be sure to check out the documentation.

See Sébastien Renauld's answer for one way of implementing this yourself.

Your Answer

By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

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