pub async fn supervisor(
    config: config::Twitter,
    mut rx_requested_follows: mpsc::Receiver<(SocketAddr, Follows)>,
    tx_tweet: broadcast::Sender<Tweet>,
) -> Result<()> {
    // The follows requested and their subscribers
    let mut requested_follows = RequestedFollows::new();

    // Whether we are currently backing off
    let mut backing_off = false;
    // Backoff exponent
    let mut backoff = 0;

    // This is this the initial state, restart and twitter_stream are terminated,
    // the only live future is rx_requested_follows
    // This means that the only way for this select to pick up is for a client to subscribe
    // This state is reached again when no subscriptions remain
    let restart = Fuse::terminated();
    let twitter_stream = Fuse::terminated();
    // See https://docs.rs/tokio/1.0.1/tokio/stream/index.html
    let rx_requested_follows = async_stream::stream! {
        while let Some(item) = rx_requested_follows.recv().await {
            yield item;
        }
    }
    .fuse();

    // pin to stack
    futures::pin_mut!(restart, twitter_stream, rx_requested_follows);

    loop {
        futures::select! {
            // We were scheduled to restart, we verify that anyone is subscribed
            // and start a new stream
            () = restart => {
                backing_off = false;

                if requested_follows.is_empty() {
                    if twitter_stream.is_terminated().not() {
                        log::warn!("closing existing stream");
                        twitter_stream.set(Fuse::terminated());
                    }

                    log::info!("no follows were requested, let's wait some more");
                    continue;
                }

                let follows = HashSet::from_iter(requested_follows.keys().copied());
                twitter_stream
                    .set(stream_consumer(config.token(), follows, tx_tweet.clone()).fuse());
            }

            // The stream has ended, we inspect the given error to know how much we should be
            // delaying the restart, and schedule said restart
            res = twitter_stream => {
                let error = res.err().context("infinite loop cannot return Ok(())")?;
                log::error!("twitter stream error: {:#}", error);

                let delay = inspect_error(error, &mut backoff);
                backing_off = true;

                log::info!("restarting in {:?}", delay);
                restart.set(sleep(delay).fuse());
            }

            // A client has requested new follows, if any are new, we schedule a restart.
            // If a normal (not backing off) restart was already scheduled, we ignore it and
            // re-schedule to 10 seconds.
            msg = rx_requested_follows.next() => {
                let (addr, new_follows) = msg.context("no tx_requested_follows remaining")?;

                // We remove addr from subscriptions it doesn't want anymore
                for (follow, subscribers) in &mut requested_follows {
                    if new_follows.contains(&follow) {
                        continue;
                    }

                    subscribers.retain(|s| s != &addr);
                }

                let num_follows_before = requested_follows.len();

                // We drop follows that nobody wants anymore
                requested_follows.retain(|_, s| s.is_empty().not());

                let mut requires_restart =
                    config.always_restart && num_follows_before < requested_follows.len();
                for follow in new_follows {
                    requires_restart = requires_restart || requested_follows.contains_key(&follow).not();

                    requested_follows
                        .entry(follow)
                        .or_insert_with(HashSet::new)
                        .insert(addr);
                }

                requires_restart = requires_restart
                    || (twitter_stream.is_terminated().not() && requested_follows.is_empty());

                if requires_restart && backing_off.not() {
                    log::info!("found new follows, restarting in {:?}", NEW_FOLLOWS_RESTART_DELAY);
                    if restart.is_terminated().not() {
                        log::info!("intercepted an existing scheduled restart");
                    }
                    restart.set(sleep(NEW_FOLLOWS_RESTART_DELAY).fuse());
                }
            }

            complete => {
                anyhow::bail!("twitter::supervisor should never complete");
            }
        }
    }
}
// starts the twitter stream
// restarts it when it goes down
// restarts it when there are new users to follow
starts the twitter stream
restarts it when it goes down
restarts it when there are new users to follow
starts the twitter stream restarts it when it goes down restarts it when there are new users to follow
pub async fn supervisor(
    config: config::Twitter,
    mut rx_requested_follows: mpsc::Receiver<(SocketAddr, Follows)>,
    tx_tweet: broadcast::Sender<Tweet>,
) -> Result<()> {
    // The follows requested and their subscribers
    let mut requested_follows = RequestedFollows::new();
    // Whether we are currently backing off
    let mut backing_off = false;
    // Backoff exponent
    let mut backoff = 0;
    // This is this the initial state, restart and twitter_stream are terminated,
    // the only live future is rx_requested_follows
    // This means that the only way for this select to pick up is for a client to subscribe
    // This state is reached again when no subscriptions remain
    let restart = Fuse::terminated();
    let twitter_stream = Fuse::terminated();
    // See https://docs.rs/tokio/1.0.1/tokio/stream/index.html
    let rx_requested_follows = async_stream::stream! {
        while let Some(item) = rx_requested_follows.recv().await {
            yield item;
        }
    }
    .fuse();
    // pin to stack
    futures::pin_mut!(restart, twitter_stream, rx_requested_follows);
    loop {
        futures::select! {
            // We were scheduled to restart, we verify that anyone is subscribed
            // and start a new stream
            () = restart => {
                backing_off = false;
                if requested_follows.is_empty() {
                    if twitter_stream.is_terminated().not() {
                        log::warn!("closing existing stream");
                        twitter_stream.set(Fuse::terminated());
                    }
                    log::info!("no follows were requested, let's wait some more");
                    continue;
                }
                let follows = HashSet::from_iter(requested_follows.keys().copied());
                twitter_stream
                    .set(stream_consumer(config.token(), follows, tx_tweet.clone()).fuse());
            }
            // The stream has ended, we inspect the given error to know how much we should be
            // delaying the restart, and schedule said restart
            res = twitter_stream => {
                let error = res.err().context("infinite loop cannot return Ok(())")?;
                log::error!("twitter stream error: {:#}", error);
                let delay = inspect_error(error, &mut backoff);
                backing_off = true;
                log::info!("restarting in {:?}", delay);
                restart.set(sleep(delay).fuse());
            }
            // A client has requested new follows, if any are new, we schedule a restart.
            // If a normal (not backing off) restart was already scheduled, we ignore it and
            // re-schedule to 10 seconds.
            msg = rx_requested_follows.next() => {
                let (addr, new_follows) = msg.context("no tx_requested_follows remaining")?;
                // We remove addr from subscriptions it doesn't want anymore
                for (follow, subscribers) in &mut requested_follows {
                    if new_follows.contains(&follow) {
                        continue;
                    }
                    subscribers.retain(|s| s != &addr);
                }
                let num_follows_before = requested_follows.len();
                // We drop follows that nobody wants anymore
                requested_follows.retain(|_, s| s.is_empty().not());
                let mut requires_restart =
                    config.always_restart && num_follows_before < requested_follows.len();
                for follow in new_follows {
                    requires_restart = requires_restart || requested_follows.contains_key(&follow).not();
                    requested_follows
                        .entry(follow)
                        .or_insert_with(HashSet::new)
                        .insert(addr);
                }
                requires_restart = requires_restart
                    || (twitter_stream.is_terminated().not() && requested_follows.is_empty());
                if requires_restart && backing_off.not() {
                    log::info!("found new follows, restarting in {:?}", NEW_FOLLOWS_RESTART_DELAY);
                    if restart.is_terminated().not() {
                        log::info!("intercepted an existing scheduled restart");
                    }
                    restart.set(sleep(NEW_FOLLOWS_RESTART_DELAY).fuse());
                }
            }
            complete => {
                anyhow::bail!("twitter::supervisor should never complete");
            }
        }
    }
}
pub async fn supervisor ( config : config :: Twitter , mut rx_requested_follows : mpsc :: Receiver < ( SocketAddr , Follows ) > , tx_tweet : broadcast :: Sender < Tweet > , ) -> Result < ( ) > { // The follows requested and their subscribers let mut requested_follows = RequestedFollows :: new ( ) ; // Whether we are currently backing off let mut backing_off = false ; // Backoff exponent let mut backoff = 0 ; // This is this the initial state, restart and twitter_stream are terminated, // the only live future is rx_requested_follows // This means that the only way for this select to pick up is for a client to subscribe // This state is reached again when no subscriptions remain let restart = Fuse :: terminated ( ) ; let twitter_stream = Fuse :: terminated ( ) ; // See https://docs.rs/tokio/1.0.1/tokio/stream/index.html let rx_requested_follows = async_stream :: stream ! { while let Some ( item ) rx_requested_follows recv ( ) await { yield item } } . fuse ( ) ; // pin to stack futures :: pin_mut ! ( restart twitter_stream rx_requested_follows ) ; loop { futures :: select ! { // We were scheduled to restart, we verify that anyone is subscribed // and start a new stream ( ) restart { backing_off false if requested_follows is_empty ( ) { if twitter_stream is_terminated ( ) not ( ) { log warn ( " " ) twitter_stream set ( Fuse terminated ( ) ) } log info ( " " ) continue } let follows HashSet from_iter ( requested_follows keys ( ) copied ( ) ) twitter_stream set ( stream_consumer ( config token ( ) follows tx_tweet clone ( ) ) fuse ( ) ) } // The stream has ended, we inspect the given error to know how much we should be // delaying the restart, and schedule said restart res twitter_stream { let error res err ( ) context ( " " ) log error ( " " error ) let delay inspect_error ( error mut backoff ) backing_off true log info ( " " delay ) restart set ( sleep ( delay ) fuse ( ) ) } // A client has requested new follows, if any are new, we schedule a restart. // If a normal (not backing off) restart was already scheduled, we ignore it and // re-schedule to 10 seconds. msg rx_requested_follows next ( ) { let ( addr new_follows ) msg context ( " " ) // We remove addr from subscriptions it doesn't want anymore for ( follow subscribers ) in mut requested_follows { if new_follows contains ( follow ) { continue } subscribers retain ( s s addr ) } let num_follows_before requested_follows len ( ) // We drop follows that nobody wants anymore requested_follows retain ( s s is_empty ( ) not ( ) ) let mut requires_restart config always_restart num_follows_before requested_follows len ( ) for follow in new_follows { requires_restart requires_restart requested_follows contains_key ( follow ) not ( ) requested_follows entry ( follow ) or_insert_with ( HashSet new ) insert ( addr ) } requires_restart requires_restart ( twitter_stream is_terminated ( ) not ( ) requested_follows is_empty ( ) ) if requires_restart backing_off not ( ) { log info ( " " NEW_FOLLOWS_RESTART_DELAY ) if restart is_terminated ( ) not ( ) { log info ( " " ) } restart set ( sleep ( NEW_FOLLOWS_RESTART_DELAY ) fuse ( ) ) } } complete { anyhow bail ( " " ) } } } }
starts the twitter stream
restarts it when it goes down
restarts it when there are new users to follow
starts the twitter stream restarts it when it goes down restarts it when there are new users to follow
config
config::Twitter
rx_requested_follows
mpsc::Receiver<(SocketAddr, Follows)>
tx_tweet
broadcast::Sender<Tweet>
config
config::Twitter
rx_requested_follows
mpsc::Receiver<(SocketAddr, Follows)>
tx_tweet
broadcast::Sender<Tweet>