async fn handle_msg(&mut self, msg: ContextMessage<A>) -> bool {
        match msg {
            ContextMessage::Instant(msg) => self.handle_instant_msg(msg).await,
            ContextMessage::Delayed(msg) => self.handle_delayed_msg(msg),
            ContextMessage::Interval(msg) => self.handle_interval_msg(msg).await,
            ContextMessage::ManualShutDown(tx) => {
                if tx.send(self.state()).is_ok() {
                    self.manual_shutdown = true;
                    return true;
                }
            }
        }

        false
    }
// return true if we want to break the streaming loop
return true if we want to break the streaming loop
return true if we want to break the streaming loop
async fn handle_msg(&mut self, msg: ContextMessage<A>) -> bool {
        match msg {
            ContextMessage::Instant(msg) => self.handle_instant_msg(msg).await,
            ContextMessage::Delayed(msg) => self.handle_delayed_msg(msg),
            ContextMessage::Interval(msg) => self.handle_interval_msg(msg).await,
            ContextMessage::ManualShutDown(tx) => {
                if tx.send(self.state()).is_ok() {
                    self.manual_shutdown = true;
                    return true;
                }
            }
        }
        false
    }
async fn handle_msg ( & mut self , msg : ContextMessage < A > ) -> bool { match msg { ContextMessage :: Instant ( msg ) => self . handle_instant_msg ( msg ) . await , ContextMessage :: Delayed ( msg ) => self . handle_delayed_msg ( msg ) , ContextMessage :: Interval ( msg ) => self . handle_interval_msg ( msg ) . await , ContextMessage :: ManualShutDown ( tx ) => { if tx . send ( self . state ( ) ) . is_ok ( ) { self . manual_shutdown = true ; return true ; } } } false }
return true if we want to break the streaming loop
return true if we want to break the streaming loop
&mut self
msg
ContextMessage<A>