/// Sends the event to all dispatchers and waits to be unblocked by all
Sends the event to all dispatchers and waits to be unblocked by all
Sends the event to all dispatchers and waits to be unblocked by all
async fn dispatch(&self, event: &ComponentEvent) -> Result<(), ModelError> {
        // Copy the senders so we don't hold onto the sender map lock
        // If we didn't do this, it is possible to deadlock while holding onto this lock.
        // For example,
        // Task A : call dispatch(event1) -> lock on sender map -> send -> wait for responders
        // Task B : call dispatch(event2) -> lock on sender map
        // If task B was required to respond to event1, then this is a deadlock.
        // Neither task can make progress.
        let dispatchers = {
            let mut dispatcher_map = self.dispatcher_map.lock().await;
            if let Some(dispatchers) = dispatcher_map.get_mut(&event.payload.type_().into()) {
                let mut strong_dispatchers = vec![];
                dispatchers.retain(|dispatcher| {
                    if let Some(dispatcher) = dispatcher.upgrade() {
                        strong_dispatchers.push(dispatcher);
                        true
                    } else {
                        false
                    }
                });
                strong_dispatchers
            } else {
                // There were no senders for this event. Do nothing.
                return Ok(());
            }
        };
        let mut responder_channels = vec![];
        for dispatcher in dispatchers {
            let result = dispatcher.dispatch(event.clone()).await;
            match result {
                Ok(Some(responder_channel)) => {
                    // A future can be canceled if the EventStream was dropped after
                    // a send. We don't crash the system when this happens. It is
                    // perfectly valid for a EventStream to be dropped. That simply
                    // means that the EventStream is no longer interested in future
                    // events. So we force each future to return a success. This
                    // ensures that all the futures can be driven to completion.
                    let responder_channel = async move {
                        trace::duration!("component_manager", "events:wait_for_resume");
                        let _ = responder_channel.await;
                        trace::flow_end!("component_manager", "event", event.id);
                    };
                    responder_channels.push(responder_channel);
                }
                // There's nothing to do if event is outside the scope of the given
                // `EventDispatcher`.
                Ok(None) => (),
                Err(_) => {
                    // A send can fail if the EventStream was dropped. We don't
                    // crash the system when this happens. It is perfectly
                    // valid for a EventStream to be dropped. That simply means
                    // that the EventStream is no longer interested in future
                    // events.
                }
            }
        }
        // Wait until all tasks have used the responder to unblock.
        {
            trace::duration!("component_manager", "events:wait_for_all_resume");
            futures::future::join_all(responder_channels).await;
        }
        Ok(())
    }
async fn dispatch ( & self , event : & ComponentEvent ) -> Result < ( ) , ModelError > { // Copy the senders so we don't hold onto the sender map lock // If we didn't do this, it is possible to deadlock while holding onto this lock. // For example, // Task A : call dispatch(event1) -> lock on sender map -> send -> wait for responders // Task B : call dispatch(event2) -> lock on sender map // If task B was required to respond to event1, then this is a deadlock. // Neither task can make progress. let dispatchers = { let mut dispatcher_map = self . dispatcher_map . lock ( ) . await ; if let Some ( dispatchers ) = dispatcher_map . get_mut ( & event . payload . type_ ( ) . into ( ) ) { let mut strong_dispatchers = vec ! [ ] ; dispatchers . retain ( | dispatcher | { if let Some ( dispatcher ) = dispatcher . upgrade ( ) { strong_dispatchers . push ( dispatcher ) ; true } else { false } } ) ; strong_dispatchers } else { // There were no senders for this event. Do nothing. return Ok ( ( ) ) ; } } ; let mut responder_channels = vec ! [ ] ; for dispatcher in dispatchers { let result = dispatcher . dispatch ( event . clone ( ) ) . await ; match result { Ok ( Some ( responder_channel ) ) => { // A future can be canceled if the EventStream was dropped after // a send. We don't crash the system when this happens. It is // perfectly valid for a EventStream to be dropped. That simply // means that the EventStream is no longer interested in future // events. So we force each future to return a success. This // ensures that all the futures can be driven to completion. let responder_channel = async move { trace :: duration ! ( " " " " ) ; let _ = responder_channel . await ; trace :: flow_end ! ( " " " " event id ) ; } ; responder_channels . push ( responder_channel ) ; } // There's nothing to do if event is outside the scope of the given // `EventDispatcher`. Ok ( None ) => ( ) , Err ( _ ) => { // A send can fail if the EventStream was dropped. We don't // crash the system when this happens. It is perfectly // valid for a EventStream to be dropped. That simply means // that the EventStream is no longer interested in future // events. } } } // Wait until all tasks have used the responder to unblock. { trace :: duration ! ( " " " " ) ; futures :: future :: join_all ( responder_channels ) . await ; } Ok ( ( ) ) }
Sends the event to all dispatchers and waits to be unblocked by all
Sends the event to all dispatchers and waits to be unblocked by all
&self
event
&ComponentEvent