/// Listens for messages available from the broker, processes them and adds them to the
/// record file. This process ends once a simulation-end-message was recieved.
/// Fails if the messaging process fails and returns the according error.
Listens for messages available from the broker, processes them and adds them to the
record file. This process ends once a simulation-end-message was recieved.
Fails if the messaging process fails and returns the according error.
Listens for messages available from the broker processes them and adds them to the record file . This process ends once a simulation - end - message was recieved . Fails if the messaging process fails and returns the according error .
pub fn listen_to_broker(&mut self) -> Result<(), PvError> {
        // Setup a consumer and listen to all incomming messages until the simulation ends.
        let mut connection = Connection::insecure_open(&self.broker_url)?;
        let channel = connection.open_channel(None)?;
        let queue = channel.queue_declare(METER_ROUTING_KEY, QueueDeclareOptions::default())?;
        let consumer = queue.consume(ConsumerOptions::default())?;
        for message in consumer.receiver().iter() {
            match message {
                ConsumerMessage::Delivery(delivery) => {
                    let message: BrokerMessage = serde_json::from_slice(&delivery.body)?;
                    consumer.ack(delivery)?;
                    if message.is_simulation_end() {
                        // Cancel the consumer if the simulation ended.
                        consumer.cancel()?;
                    } else {
                        // If the simulation is ongoing add the message to the records.
                        let record = self.message_to_record(message)?;
                        self.records.push(record);
                    }
                }
                // The consumer is cancelled once the simulation ended.
                ConsumerMessage::ClientCancelled => break,
                other => {
                    return Err(PvError::InternalError(format!(
                        "Consumer did not expect: {:?}",
                        other
                    )))
                }
            }
        }
        connection.close()?;
        Ok(())
    }
pub fn listen_to_broker ( & mut self ) -> Result < ( ) , PvError > { // Setup a consumer and listen to all incomming messages until the simulation ends. let mut connection = Connection :: insecure_open ( & self . broker_url ) ? ; let channel = connection . open_channel ( None ) ? ; let queue = channel . queue_declare ( METER_ROUTING_KEY , QueueDeclareOptions :: default ( ) ) ? ; let consumer = queue . consume ( ConsumerOptions :: default ( ) ) ? ; for message in consumer . receiver ( ) . iter ( ) { match message { ConsumerMessage :: Delivery ( delivery ) => { let message : BrokerMessage = serde_json :: from_slice ( & delivery . body ) ? ; consumer . ack ( delivery ) ? ; if message . is_simulation_end ( ) { // Cancel the consumer if the simulation ended. consumer . cancel ( ) ? ; } else { // If the simulation is ongoing add the message to the records. let record = self . message_to_record ( message ) ? ; self . records . push ( record ) ; } } // The consumer is cancelled once the simulation ended. ConsumerMessage :: ClientCancelled => break , other => { return Err ( PvError :: InternalError ( format ! ( " " other ) ) ) } } } connection . close ( ) ? ; Ok ( ( ) ) }
Listens for messages available from the broker, processes them and adds them to the
record file.
Listens for messages available from the broker processes them and adds them to the record file .
&mut self