pub fn pipe_routes_via_crossbeam(&mut self) -> anyhow::Result<()> {
        info!("starting communication thread");
        let routes = self
            .routes
            .take()
            .ok_or_else(|| anyhow::anyhow!("routes were not configured"))?;
        let (tx, rx) = crossbeam::channel::unbounded();

        let bridge_names: Vec<String> = self.bridges.keys().cloned().collect();
        for name in bridge_names {
            let tx = tx.clone();
            if let Ok(ipc) = self.take_bridge_rx(&name) {
                info!("setting up receiver {}", name);
                let handle = tokio::task::spawn_blocking(move || loop {
                    let msg: Message = ipc.recv().unwrap_or_else(|err| {
                        todo!("receiving message from {} failed: {}", name, err)
                    });
                    tx.send(msg).unwrap_or_else(|err| {
                        todo!("sending message from {} failed: {}", name, err)
                    });
                });
                self.pipes.push(handle);
            }
        }

        // Spawn thread receiving messages from processes to topics
        /*let handle1 = 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 topic = msg.topic.clone();
                        tx.send(msg).unwrap_or_else(|err| {
                            todo!(
                                "sending internal message from {:?} to topic {} failed: {}",
                                names.get(&id),
                                topic,
                                err
                            )
                        });
                    }
                    IpcSelectionResult::ChannelClosed(id) => {
                        error!("Channel from {:?} closed...", names.get(&id));
                    }
                }
            }
        });
        self.pipes.push(handle1);*/

        // Spawn thread sending messages from topics to processes
        let handle2 = tokio::task::spawn_blocking(move || loop {
            let msg = rx.recv()?;
            let senders = routes.get(&msg.topic).unwrap_or_else(|| {
                todo!("received message to topic {} without recepients", 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(); // TODO - see if it impacting perf
            senders.last().unwrap().send(msg).unwrap_or_else(|err| {
                todo!(
                    "sending message from topic {} to last sender failed: {}",
                    topic,
                    err
                )
            });
        });
        self.pipes.push(handle2);
        Ok(())
    }
/// Spawn router to process messages subscriptions built with `route_topic_to_bridge`
/// Spawns handler in 2 tokio blocking task threads using intermediate unbound crossbeam channel.
///
/// Benefits:
/// - Does not block if recipient blocking accepting messages.
/// - Since it is not blocking it alows to increase performance during high load.
///
/// Downsides:
/// - Might grow on memory
/// - When error happen on recipient's side debugging is more complicated as information about sender is not preserved in the log.
///
/// # Might grow on memory
/// This method is using crossbeam as delivery buffer, hence it won't block,
/// though it might use excessive memory to store not processed messages.
Spawn router to process messages subscriptions built with `route_topic_to_bridge`
Spawns handler in 2 tokio blocking task threads using intermediate unbound crossbeam channel.

Does not block if recipient blocking accepting messages.
Since it is not blocking it alows to increase performance during high load.

Might grow on memory
When error happen on recipient's side debugging is more complicated as information about sender is not preserved in the log.
Spawn router to process messages subscriptions built with ` route_topic_to_bridge ` Spawns handler in 2 tokio blocking task threads using intermediate unbound crossbeam channel . Does not block if recipient blocking accepting messages . Since it is not blocking it alows to increase performance during high load . Might grow on memory When error happen on recipient ' s side debugging is more complicated as information about sender is not preserved in the log .
pub fn pipe_routes_via_crossbeam(&mut self) -> anyhow::Result<()> {
        info!("starting communication thread");
        let routes = self
            .routes
            .take()
            .ok_or_else(|| anyhow::anyhow!("routes were not configured"))?;
        let (tx, rx) = crossbeam::channel::unbounded();
        let bridge_names: Vec<String> = self.bridges.keys().cloned().collect();
        for name in bridge_names {
            let tx = tx.clone();
            if let Ok(ipc) = self.take_bridge_rx(&name) {
                info!("setting up receiver {}", name);
                let handle = tokio::task::spawn_blocking(move || loop {
                    let msg: Message = ipc.recv().unwrap_or_else(|err| {
                        todo!("receiving message from {} failed: {}", name, err)
                    });
                    tx.send(msg).unwrap_or_else(|err| {
                        todo!("sending message from {} failed: {}", name, err)
                    });
                });
                self.pipes.push(handle);
            }
        }
        // Spawn thread receiving messages from processes to topics
        /*let handle1 = 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 topic = msg.topic.clone();
                        tx.send(msg).unwrap_or_else(|err| {
                            todo!(
                                "sending internal message from {:?} to topic {} failed: {}",
                                names.get(&id),
                                topic,
                                err
                            )
                        });
                    }
                    IpcSelectionResult::ChannelClosed(id) => {
                        error!("Channel from {:?} closed...", names.get(&id));
                    }
                }
            }
        });
        self.pipes.push(handle1);*/
        // Spawn thread sending messages from topics to processes
        let handle2 = tokio::task::spawn_blocking(move || loop {
            let msg = rx.recv()?;
            let senders = routes.get(&msg.topic).unwrap_or_else(|| {
                todo!("received message to topic {} without recepients", 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(); // TODO - see if it impacting perf
            senders.last().unwrap().send(msg).unwrap_or_else(|err| {
                todo!(
                    "sending message from topic {} to last sender failed: {}",
                    topic,
                    err
                )
            });
        });
        self.pipes.push(handle2);
        Ok(())
    }
pub fn pipe_routes_via_crossbeam ( & mut self ) -> anyhow :: Result < ( ) > { info ! ( " " ) ; let routes = self . routes . take ( ) . ok_or_else ( | | anyhow :: anyhow ! ( " " ) ) ? ; let ( tx , rx ) = crossbeam :: channel :: unbounded ( ) ; let bridge_names : Vec < String > = self . bridges . keys ( ) . cloned ( ) . collect ( ) ; for name in bridge_names { let tx = tx . clone ( ) ; if let Ok ( ipc ) = self . take_bridge_rx ( & name ) { info ! ( " " name ) ; let handle = tokio :: task :: spawn_blocking ( move | | loop { let msg : Message = ipc . recv ( ) . unwrap_or_else ( | err | { todo ! ( " " name err ) } ) ; tx . send ( msg ) . unwrap_or_else ( | err | { todo ! ( " " name err ) } ) ; } ) ; self . pipes . push ( handle ) ; } } // Spawn thread receiving messages from processes to topics /*let handle1 = 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 topic = msg.topic.clone();
                        tx.send(msg).unwrap_or_else(|err| {
                            todo!(
                                "sending internal message from {:?} to topic {} failed: {}",
                                names.get(&id),
                                topic,
                                err
                            )
                        });
                    }
                    IpcSelectionResult::ChannelClosed(id) => {
                        error!("Channel from {:?} closed...", names.get(&id));
                    }
                }
            }
        });
        self.pipes.push(handle1);*/ // Spawn thread sending messages from topics to processes let handle2 = tokio :: task :: spawn_blocking ( move | | loop { let msg = rx . recv ( ) ? ; let senders = routes . get ( & msg . topic ) . unwrap_or_else ( | | { todo ! ( " " 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 ( ) ; // TODO - see if it impacting perf senders . last ( ) . unwrap ( ) . send ( msg ) . unwrap_or_else ( | err | { todo ! ( " " topic err ) } ) ; } ) ; self . pipes . push ( handle2 ) ; Ok ( ( ) ) }
Spawn router to process messages subscriptions built with `route_topic_to_bridge`
Spawns handler in 2 tokio blocking task threads using intermediate unbound crossbeam channel.
Spawn router to process messages subscriptions built with ` route_topic_to_bridge ` Spawns handler in 2 tokio blocking task threads using intermediate unbound crossbeam channel .
&mut self
&mut self
might grow on memory
This method is using crossbeam as delivery buffer, hence it won't block,
though it might use excessive memory to store not processed messages.
This method is using crossbeam as delivery buffer hence it won ' t block though it might use excessive memory to store not processed messages .