pub fn process_writer_command(&mut self) {
    while let Ok(cc) = self.writer_command_receiver.try_recv() {
      match cc {
        WriterCommand::DDSData { data } => {
          // We have a new sample here. Things to do:
          // 1. Insert it to history cache and get it sequence numbered
          // 2. Send out data. 
          //    If we are pushing data, send the DATA submessage and HEARTBEAT.
          //    If we are not pushing, send out HEARTBEAT only. Readers will then ask the DATA with ACKNACK.
          let timestamp = self.insert_to_history_cache(data);

          self.increase_heartbeat_counter();

          let partial_message = MessageBuilder::new()
            .ts_msg(self.endianness, Some(Timestamp::now()) );
          let data_hb_message_builder = 
            if self.push_mode {
              if let Some(cache_change) = 
                  self.dds_cache.read().unwrap()
                    .from_topic_get_change(&self.my_topic_name, &timestamp) { 
                partial_message.data_msg( cache_change.clone(), // TODO: We should not clone, too much copying
                                          EntityId::ENTITYID_UNKNOWN, // reader
                                          self.my_guid.entityId, // writer
                                          self.endianness ) 
                // TODO: Here we are cloning the entire payload. We need to rewrite the transmit path to avoid copying.
              } else { partial_message }
            } else { partial_message };
          let final_flag = false;
          let liveliness_flag = false;
          let data_hb_message = data_hb_message_builder
               .heartbeat_msg(self, EntityId::ENTITYID_UNKNOWN, final_flag, liveliness_flag)
               .add_header_and_build(self.my_guid.guidPrefix);
          self.send_message_to_readers(DeliveryMode::Multicast, 
            &data_hb_message, &mut self.readers.values() );
        }

        // WriterCommand::ResetOfferedDeadlineMissedStatus { writer_guid: _, } => {
        //   self.reset_offered_deadline_missed_status();
        // }
      }
    }
  }
// Receive new data samples from the DDS DataWriter
Receive new data samples from the DDS DataWriter
Receive new data samples from the DDS DataWriter
pub fn process_writer_command(&mut self) {
    while let Ok(cc) = self.writer_command_receiver.try_recv() {
      match cc {
        WriterCommand::DDSData { data } => {
          // We have a new sample here. Things to do:
          // 1. Insert it to history cache and get it sequence numbered
          // 2. Send out data. 
          //    If we are pushing data, send the DATA submessage and HEARTBEAT.
          //    If we are not pushing, send out HEARTBEAT only. Readers will then ask the DATA with ACKNACK.
          let timestamp = self.insert_to_history_cache(data);
          self.increase_heartbeat_counter();
          let partial_message = MessageBuilder::new()
            .ts_msg(self.endianness, Some(Timestamp::now()) );
          let data_hb_message_builder = 
            if self.push_mode {
              if let Some(cache_change) = 
                  self.dds_cache.read().unwrap()
                    .from_topic_get_change(&self.my_topic_name, &timestamp) { 
                partial_message.data_msg( cache_change.clone(), // TODO: We should not clone, too much copying
                                          EntityId::ENTITYID_UNKNOWN, // reader
                                          self.my_guid.entityId, // writer
                                          self.endianness ) 
                // TODO: Here we are cloning the entire payload. We need to rewrite the transmit path to avoid copying.
              } else { partial_message }
            } else { partial_message };
          let final_flag = false;
          let liveliness_flag = false;
          let data_hb_message = data_hb_message_builder
               .heartbeat_msg(self, EntityId::ENTITYID_UNKNOWN, final_flag, liveliness_flag)
               .add_header_and_build(self.my_guid.guidPrefix);
          self.send_message_to_readers(DeliveryMode::Multicast, 
            &data_hb_message, &mut self.readers.values() );
        }
        // WriterCommand::ResetOfferedDeadlineMissedStatus { writer_guid: _, } => {
        //   self.reset_offered_deadline_missed_status();
        // }
      }
    }
  }
pub fn process_writer_command ( & mut self ) { while let Ok ( cc ) = self . writer_command_receiver . try_recv ( ) { match cc { WriterCommand :: DDSData { data } => { // We have a new sample here. Things to do: // 1. Insert it to history cache and get it sequence numbered // 2. Send out data.  //    If we are pushing data, send the DATA submessage and HEARTBEAT. //    If we are not pushing, send out HEARTBEAT only. Readers will then ask the DATA with ACKNACK. let timestamp = self . insert_to_history_cache ( data ) ; self . increase_heartbeat_counter ( ) ; let partial_message = MessageBuilder :: new ( ) . ts_msg ( self . endianness , Some ( Timestamp :: now ( ) ) ) ; let data_hb_message_builder = if self . push_mode { if let Some ( cache_change ) = self . dds_cache . read ( ) . unwrap ( ) . from_topic_get_change ( & self . my_topic_name , & timestamp ) { partial_message . data_msg ( cache_change . clone ( ) , // TODO: We should not clone, too much copying EntityId :: ENTITYID_UNKNOWN , // reader self . my_guid . entityId , // writer self . endianness ) // TODO: Here we are cloning the entire payload. We need to rewrite the transmit path to avoid copying. } else { partial_message } } else { partial_message } ; let final_flag = false ; let liveliness_flag = false ; let data_hb_message = data_hb_message_builder . heartbeat_msg ( self , EntityId :: ENTITYID_UNKNOWN , final_flag , liveliness_flag ) . add_header_and_build ( self . my_guid . guidPrefix ) ; self . send_message_to_readers ( DeliveryMode :: Multicast , & data_hb_message , & mut self . readers . values ( ) ) ; } // WriterCommand::ResetOfferedDeadlineMissedStatus { writer_guid: _, } => { //   self.reset_offered_deadline_missed_status(); // } } } }
Receive new data samples from the DDS DataWriter
Receive new data samples from the DDS DataWriter
&mut self