pub fn start(mut self) -> Result<(TerminateHandle, impl Future<Output = ()>), RouterError> {
        self.validate_routing()?;

        let (terminate_handle, mut stream) =
            EventStream::new(self.external_receiver, self.internal_receiver);
        let mut consumers = self.consumers;
        let mut inspect_logger = self.inspect_logger;

        let fut = async move {
            loop {
                match stream.next().await {
                    None => {
                        debug!("Event ingestion finished");
                        break;
                    }
                    Some(event) => {
                        inspect_logger.log(&event);

                        let event_type = event.ty();
                        let weak_consumers = match consumers.get_mut(&event_type) {
                            Some(c) => c,
                            None => continue,
                        };

                        let event_without_singleton_data = event.clone();
                        let mut event_with_singleton_data =
                            if event.is_singleton() { Some(event) } else { None };

                        // Consumers which weak reference could be upgraded will be stored here.
                        let mut active_consumers = vec![];
                        for consumer in weak_consumers.iter_mut().filter_map(|c| c.upgrade()) {
                            active_consumers.push(Arc::downgrade(&consumer));
                            let e = event_with_singleton_data
                                .take()
                                .unwrap_or_else(|| event_without_singleton_data.clone());
                            consumer.handle(e).await;
                        }

                        // We insert the list of active consumers in the map at the key for this
                        // event type. This leads to dropping the previous list of weak references
                        // which contains consumers which aren't active anymore.
                        consumers.insert(event_type, active_consumers);
                    }
                }
            }
        };
        Ok((terminate_handle, fut))
    }
/// Starts listening for events emitted by the registered producers and dispatching them to
/// registered consumers.
///
/// First, validates that for every event type that will be dispatched, there exists at least
/// one consumer. And that for every event that will be consumed, there exists at least one
/// producer.
///
/// Afterwards, listens to events emitted by producers. When an event arrives it sends it to
/// all consumers of the event. If the event is singleton, the first consumer that was
/// registered will get the singleton data and the rest won't.
Starts listening for events emitted by the registered producers and dispatching them to
registered consumers.

First, validates that for every event type that will be dispatched, there exists at least
one consumer. And that for every event that will be consumed, there exists at least one
producer.

Afterwards, listens to events emitted by producers. When an event arrives it sends it to
all consumers of the event. If the event is singleton, the first consumer that was
registered will get the singleton data and the rest won't.
Starts listening for events emitted by the registered producers and dispatching them to registered consumers . First validates that for every event type that will be dispatched there exists at least one consumer . And that for every event that will be consumed there exists at least one producer . Afterwards listens to events emitted by producers . When an event arrives it sends it to all consumers of the event . If the event is singleton the first consumer that was registered will get the singleton data and the rest won ' t .
pub fn start(mut self) -> Result<(TerminateHandle, impl Future<Output = ()>), RouterError> {
        self.validate_routing()?;
        let (terminate_handle, mut stream) =
            EventStream::new(self.external_receiver, self.internal_receiver);
        let mut consumers = self.consumers;
        let mut inspect_logger = self.inspect_logger;
        let fut = async move {
            loop {
                match stream.next().await {
                    None => {
                        debug!("Event ingestion finished");
                        break;
                    }
                    Some(event) => {
                        inspect_logger.log(&event);
                        let event_type = event.ty();
                        let weak_consumers = match consumers.get_mut(&event_type) {
                            Some(c) => c,
                            None => continue,
                        };
                        let event_without_singleton_data = event.clone();
                        let mut event_with_singleton_data =
                            if event.is_singleton() { Some(event) } else { None };
                        // Consumers which weak reference could be upgraded will be stored here.
                        let mut active_consumers = vec![];
                        for consumer in weak_consumers.iter_mut().filter_map(|c| c.upgrade()) {
                            active_consumers.push(Arc::downgrade(&consumer));
                            let e = event_with_singleton_data
                                .take()
                                .unwrap_or_else(|| event_without_singleton_data.clone());
                            consumer.handle(e).await;
                        }
                        // We insert the list of active consumers in the map at the key for this
                        // event type. This leads to dropping the previous list of weak references
                        // which contains consumers which aren't active anymore.
                        consumers.insert(event_type, active_consumers);
                    }
                }
            }
        };
        Ok((terminate_handle, fut))
    }
pub fn start ( mut self ) -> Result < ( TerminateHandle , impl Future < Output = ( ) > ) , RouterError > { self . validate_routing ( ) ? ; let ( terminate_handle , mut stream ) = EventStream :: new ( self . external_receiver , self . internal_receiver ) ; let mut consumers = self . consumers ; let mut inspect_logger = self . inspect_logger ; let fut = async move { loop { match stream . next ( ) . await { None => { debug ! ( " " ) ; break ; } Some ( event ) => { inspect_logger . log ( & event ) ; let event_type = event . ty ( ) ; let weak_consumers = match consumers . get_mut ( & event_type ) { Some ( c ) => c , None => continue , } ; let event_without_singleton_data = event . clone ( ) ; let mut event_with_singleton_data = if event . is_singleton ( ) { Some ( event ) } else { None } ; // Consumers which weak reference could be upgraded will be stored here. let mut active_consumers = vec ! [ ] ; for consumer in weak_consumers . iter_mut ( ) . filter_map ( | c | c . upgrade ( ) ) { active_consumers . push ( Arc :: downgrade ( & consumer ) ) ; let e = event_with_singleton_data . take ( ) . unwrap_or_else ( | | event_without_singleton_data . clone ( ) ) ; consumer . handle ( e ) . await ; } // We insert the list of active consumers in the map at the key for this // event type. This leads to dropping the previous list of weak references // which contains consumers which aren't active anymore. consumers . insert ( event_type , active_consumers ) ; } } } } ; Ok ( ( terminate_handle , fut ) ) }
Starts listening for events emitted by the registered producers and dispatching them to
registered consumers.
Starts listening for events emitted by the registered producers and dispatching them to registered consumers .
mut self
mut self