pub fn pipe_routes(&mut self) -> anyhow::Result<()> {
        info!("starting communication thread");
        let mut ipc_receiver_set = IpcReceiverSet::new().unwrap();
        let mut names: HashMap<u64, String> = HashMap::new();
        let bridge_names: Vec<String> = self.bridges.keys().cloned().collect();
        for name in bridge_names {
            if let Ok(recv) = self.take_bridge_rx(&name) {
                info!("setting up receiver {}", name);
                let id = ipc_receiver_set.add(recv)?;
                names.insert(id, name.to_string());
            }
        }
        let routes = self
            .routes
            .take()
            .ok_or_else(|| anyhow::anyhow!("routes were not configured"))?;

        let handle = tokio::task::spawn_blocking(move || loop {
            let results = match ipc_receiver_set.select() {
                Ok(results) => results,
                Err(err) => todo!("receiving message failed: {}", err),
            };
            for event in results {
                match event {
                    IpcSelectionResult::MessageReceived(id, message) => {
                        let msg: Message = message.to().unwrap_or_else(|err| {
                            todo!(
                                "receiving message from {:?} failed: {}",
                                names.get(&id),
                                err
                            )
                        });
                        let senders = routes.get(&msg.topic).unwrap_or_else(|| {
                            todo!(
                                "received message from {:?} to topic {} without recepients",
                                names.get(&id),
                                msg.topic
                            )
                        });
                        let except_last = senders.len() - 1;
                        trace!(
                            "sending message from topic {} to {} senders",
                            msg.topic,
                            senders.len()
                        );
                        for (i, tx) in senders[0..except_last].iter().enumerate() {
                            tx.send(msg.clone()).unwrap_or_else(|err| {
                                todo!(
                                    "sending message from topic {} to {} failed: {}",
                                    msg.topic,
                                    i,
                                    err
                                )
                            });
                        }
                        let topic = msg.topic.clone();
                        senders.last().unwrap().send(msg).unwrap_or_else(|err| {
                            todo!(
                                "sending message from topic {} to last sender failed: {}",
                                topic,
                                err
                            )
                        });
                    }
                    IpcSelectionResult::ChannelClosed(id) => {
                        error!("Channel from {:?} closed...", names.get(&id));
                    }
                }
            }
        });
        self.pipes.push(handle);
        Ok(())
    }
/// Spawn router to process messages subscriptions built with `route_topic_to_bridge`
/// Spawns handler in a tokio blocking task thread.
///
/// # Might block
/// This method is not using crossbeam as delivery buffer, hence should use less memory,
/// though it might block if one of channels is not being processed
Spawn router to process messages subscriptions built with `route_topic_to_bridge`
Spawns handler in a tokio blocking task thread.
Spawn router to process messages subscriptions built with ` route_topic_to_bridge ` Spawns handler in a tokio blocking task thread .
pub fn pipe_routes(&mut self) -> anyhow::Result<()> {
        info!("starting communication thread");
        let mut ipc_receiver_set = IpcReceiverSet::new().unwrap();
        let mut names: HashMap<u64, String> = HashMap::new();
        let bridge_names: Vec<String> = self.bridges.keys().cloned().collect();
        for name in bridge_names {
            if let Ok(recv) = self.take_bridge_rx(&name) {
                info!("setting up receiver {}", name);
                let id = ipc_receiver_set.add(recv)?;
                names.insert(id, name.to_string());
            }
        }
        let routes = self
            .routes
            .take()
            .ok_or_else(|| anyhow::anyhow!("routes were not configured"))?;
        let handle = tokio::task::spawn_blocking(move || loop {
            let results = match ipc_receiver_set.select() {
                Ok(results) => results,
                Err(err) => todo!("receiving message failed: {}", err),
            };
            for event in results {
                match event {
                    IpcSelectionResult::MessageReceived(id, message) => {
                        let msg: Message = message.to().unwrap_or_else(|err| {
                            todo!(
                                "receiving message from {:?} failed: {}",
                                names.get(&id),
                                err
                            )
                        });
                        let senders = routes.get(&msg.topic).unwrap_or_else(|| {
                            todo!(
                                "received message from {:?} to topic {} without recepients",
                                names.get(&id),
                                msg.topic
                            )
                        });
                        let except_last = senders.len() - 1;
                        trace!(
                            "sending message from topic {} to {} senders",
                            msg.topic,
                            senders.len()
                        );
                        for (i, tx) in senders[0..except_last].iter().enumerate() {
                            tx.send(msg.clone()).unwrap_or_else(|err| {
                                todo!(
                                    "sending message from topic {} to {} failed: {}",
                                    msg.topic,
                                    i,
                                    err
                                )
                            });
                        }
                        let topic = msg.topic.clone();
                        senders.last().unwrap().send(msg).unwrap_or_else(|err| {
                            todo!(
                                "sending message from topic {} to last sender failed: {}",
                                topic,
                                err
                            )
                        });
                    }
                    IpcSelectionResult::ChannelClosed(id) => {
                        error!("Channel from {:?} closed...", names.get(&id));
                    }
                }
            }
        });
        self.pipes.push(handle);
        Ok(())
    }
pub fn pipe_routes ( & mut self ) -> anyhow :: Result < ( ) > { info ! ( " " ) ; let mut ipc_receiver_set = IpcReceiverSet :: new ( ) . unwrap ( ) ; let mut names : HashMap < u64 , String > = HashMap :: new ( ) ; let bridge_names : Vec < String > = self . bridges . keys ( ) . cloned ( ) . collect ( ) ; for name in bridge_names { if let Ok ( recv ) = self . take_bridge_rx ( & name ) { info ! ( " " name ) ; let id = ipc_receiver_set . add ( recv ) ? ; names . insert ( id , name . to_string ( ) ) ; } } let routes = self . routes . take ( ) . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ; let handle = tokio :: task :: spawn_blocking ( move | | loop { let results = match ipc_receiver_set . select ( ) { Ok ( results ) => results , Err ( err ) => todo ! ( " " err ) , } ; for event in results { match event { IpcSelectionResult :: MessageReceived ( id , message ) => { let msg : Message = message . to ( ) . unwrap_or_else ( | err | { todo ! ( " " names get ( id ) err ) } ) ; let senders = routes . get ( & msg . topic ) . unwrap_or_else ( | | { todo ! ( " " names get ( id ) msg topic ) } ) ; let except_last = senders . len ( ) - 1 ; trace ! ( " " msg topic senders len ( ) ) ; for ( i , tx ) in senders [ 0 .. except_last ] . iter ( ) . enumerate ( ) { tx . send ( msg . clone ( ) ) . unwrap_or_else ( | err | { todo ! ( " " msg topic i err ) } ) ; } let topic = msg . topic . clone ( ) ; senders . last ( ) . unwrap ( ) . send ( msg ) . unwrap_or_else ( | err | { todo ! ( " " topic err ) } ) ; } IpcSelectionResult :: ChannelClosed ( id ) => { error ! ( " " names get ( id ) ) ; } } } } ) ; self . pipes . push ( handle ) ; Ok ( ( ) ) }
Spawn router to process messages subscriptions built with `route_topic_to_bridge`
Spawns handler in a tokio blocking task thread.
Spawn router to process messages subscriptions built with ` route_topic_to_bridge ` Spawns handler in a tokio blocking task thread .
&mut self
&mut self
might block
This method is not using crossbeam as delivery buffer, hence should use less memory,
though it might block if one of channels is not being processed
This method is not using crossbeam as delivery buffer hence should use less memory though it might block if one of channels is not being processed