Oooo actually I was wrong. The issue is that when using the task the watcher gets dropped immediately, whereas without the task it is kept alive by the while loop.

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.

I had seen this example on axum.rs Chinese website...have u copied some things? Wink wink