Use &mut self in NotificationProcessor::init

[mkatychev]

This pr modifies the NotificationProcessor::init method to use &mut self. This allows us to consume resoruces in self and allows the JoinSet to be passed in later:

pub struct StatusReporter {
    input_rx: Option<UnboundedReceiver<StatusInput>>,
    reports: Arc<Mutex<HashMap<Uuid, Status>>>,
}

impl NotificationProcessor<OurMessage> for StatusReporter {
    fn init(&mut self, join_set: &mut JoinSet<()>) -> UnboundedSender<Notification<OurMessage>> {
        // we consume self.input_rx and set it to `None`
        let rx = self.input_rx.take().expect("uninitialized input_rx");
        spawn_some_other_rx(join_set, rx);

        self.spawn_notification_rx(join_set)
    }

    fn get_topics(&self) -> &[OurTopic] {
        &[OurTopic::Event]
    }
}

[jkleinknox]

Since you are working in this area, would you be able to address the timeout cancellation issue? Essentially the timeout cancellation request doesn't reach the timeout manager in a reasonable amount of time causing the timeout manager to emit a timeout when the SM does not expect a timeout to be emitted.

// Currently the timeout
// is being cancelled, and the state is being changed to a state
// that does not expect a timeout to occur. Due to the coherency
// issue, the timeout is not cancelled, causing an unexpected scenario.

[mergify]

[✅] @mkatychev: Use &mut self in NotificationProcessor::init has been merged successfully.