/// Receive store events from Postgres and send them to all active
/// subscriptions. Detect stale subscriptions in the process and
/// close them.
Receive store events from Postgres and send them to all active
subscriptions. Detect stale subscriptions in the process and
close them.
Receive store events from Postgres and send them to all active subscriptions . Detect stale subscriptions in the process and close them .
fn handle_store_events(
        &self,
        store_events: Box<dyn Stream<Item = StoreEvent, Error = ()> + Send>,
    ) {
        let logger = self.logger.clone();
        let subscriptions = self.subscriptions.clone();
        graph::spawn(
            store_events
                .for_each(move |event| {
                    let senders = subscriptions.read().unwrap().clone();
                    let logger = logger.clone();
                    let subscriptions = subscriptions.clone();
                    // Write change to all matching subscription streams; remove subscriptions
                    // whose receiving end has been dropped
                    stream::iter_ok::<_, ()>(senders).for_each(move |(id, sender)| {
                        let logger = logger.clone();
                        let subscriptions = subscriptions.clone();
                        sender.send(event.clone()).then(move |result| {
                            match result {
                                Err(_send_error) => {
                                    // Receiver was dropped
                                    debug!(logger, "Unsubscribe"; "id" => &id);
                                    subscriptions.write().unwrap().remove(&id);
                                    Ok(())
                                }
                                Ok(_sender) => Ok(()),
                            }
                        })
                    })
                })
                .compat(),
        );
    }
fn handle_store_events ( & self , store_events : Box < dyn Stream < Item = StoreEvent , Error = ( ) > + Send > , ) { let logger = self . logger . clone ( ) ; let subscriptions = self . subscriptions . clone ( ) ; graph :: spawn ( store_events . for_each ( move | event | { let senders = subscriptions . read ( ) . unwrap ( ) . clone ( ) ; let logger = logger . clone ( ) ; let subscriptions = subscriptions . clone ( ) ; // Write change to all matching subscription streams; remove subscriptions // whose receiving end has been dropped stream :: iter_ok :: < _ , ( ) > ( senders ) . for_each ( move | ( id , sender ) | { let logger = logger . clone ( ) ; let subscriptions = subscriptions . clone ( ) ; sender . send ( event . clone ( ) ) . then ( move | result | { match result { Err ( _send_error ) => { // Receiver was dropped debug ! ( logger " " " " id ) ; subscriptions . write ( ) . unwrap ( ) . remove ( & id ) ; Ok ( ( ) ) } Ok ( _sender ) => Ok ( ( ) ) , } } ) } ) } ) . compat ( ) , ) ; }
Receive store events from Postgres and send them to all active
subscriptions.
Receive store events from Postgres and send them to all active subscriptions .
&self
store_events
Box<dyn Stream<Item = StoreEvent, Error = ()> + Send>