Inheritance diagram for OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader:

Collaboration diagram for OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader:

[legend]

Public Member Functions
	RtpsReader (const RtpsUdpDataLink_rch &link, const GUID_t &id)

virtual	~RtpsReader ()

bool	add_writer (const WriterInfo_rch &info)

bool	has_writer (const GUID_t &id) const

bool	remove_writer (const GUID_t &id)

size_t	writer_count () const

bool	should_nack_fragments (const RcHandle< RtpsUdpDataLink > &link, const WriterInfo_rch &info)

void	pre_stop_helper ()

void	process_heartbeat_i (const RTPS::HeartBeatSubmessage &heartbeat, const GUID_t &src, bool directed, MetaSubmessageVec &meta_submessages)

bool	process_data_i (const RTPS::DataSubmessage &data, const GUID_t &src, MetaSubmessageVec &meta_submessages)

void	process_gap_i (const RTPS::GapSubmessage &gap, const GUID_t &src, bool directed, MetaSubmessageVec &meta_submessages)

void	process_heartbeat_frag_i (const RTPS::HeartBeatFragSubmessage &hb_frag, const GUID_t &src, bool directed, MetaSubmessageVec &meta_submessages)

void	deliver_held_data (const GUID_t &src)

const GUID_t &	id () const

void	log_remote_counts (const char *funcname)

Public Member Functions inherited from OpenDDS::DCPS::RcObject
virtual	~RcObject ()

virtual void	_add_ref ()

virtual void	_remove_ref ()

long	ref_count () const

WeakObject *	_get_weak_object () const

Private Member Functions
void	send_preassociation_acknacks (const MonotonicTimePoint &now)

void	gather_preassociation_acknack_i (MetaSubmessageVec &meta_submessages, const WriterInfo_rch &writer)

void	gather_ack_nacks_i (const WriterInfo_rch &writer, const RtpsUdpDataLink_rch &link, bool heartbeat_was_non_final, MetaSubmessageVec &meta_submessages, ACE_CDR::ULong &cumulative_bits_added)

void	generate_nack_frags_i (MetaSubmessageVec &meta_submessages, const WriterInfo_rch &wi, EntityId_t reader_id, EntityId_t writer_id, ACE_CDR::ULong &cumulative_bits_added)

Private Attributes
ACE_Thread_Mutex	mutex_

WeakRcHandle< RtpsUdpDataLink >	link_

const GUID_t	id_

WriterInfoMap	remote_writers_

WriterInfoSet	preassociation_writers_

bool	stopping_

CORBA::Long	nackfrag_count_

RcHandle< SporadicEvent >	preassociation_task_

TimeDuration	heartbeat_period_

Additional Inherited Members
Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Detailed Description

Definition at line 621 of file RtpsUdpDataLink.h.

Constructor & Destructor Documentation

◆ RtpsReader()

OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::RtpsReader	(	const RtpsUdpDataLink_rch &	link,
		const GUID_t &	id
	)

Definition at line 1651 of file RtpsUdpDataLink.cpp.

   : link_(link)
   , id_(id)
   , stopping_(false)
   , nackfrag_count_(0)
   , preassociation_task_(make_rch<SporadicEvent>(link->event_dispatcher(), make_rch<PmfNowEvent<RtpsReader> >(rchandle_from(this), &RtpsUdpDataLink::RtpsReader::send_preassociation_acknacks)))
   , heartbeat_period_(link ? link->config()->heartbeat_period_ : TimeDuration(RtpsUdpInst::DEFAULT_HEARTBEAT_PERIOD_SEC))
 {
 }

◆ ~RtpsReader()

OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::~RtpsReader ( )

virtual

Definition at line 1661 of file RtpsUdpDataLink.cpp.

1662 {

1663 }

Member Function Documentation

◆ add_writer()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::add_writer ( const WriterInfo_rch & info )

Definition at line 2150 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, gather_preassociation_acknack_i(), heartbeat_period_, link_, log_remote_counts(), mutex_, preassociation_task_, preassociation_writers_, OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages(), remote_writers_, and stopping_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::associated().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
 
   if (stopping_) {
     return false;
   }
 
   WriterInfoMap::const_iterator iter = remote_writers_.find(writer->id_);
   if (iter == remote_writers_.end()) {
     remote_writers_[writer->id_] = writer;
     preassociation_writers_.insert(writer);
     log_remote_counts("add_writer");
 
     RtpsUdpDataLink_rch link = link_.lock();
     if (!link) {
       return false;
     }
 
     preassociation_task_->schedule(heartbeat_period_);
     MetaSubmessageVec meta_submessages;
     gather_preassociation_acknack_i(meta_submessages, writer);
     g.release();
     link->queue_submessages(meta_submessages);
 
     return true;
   }
   return false;
 }

◆ deliver_held_data()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::deliver_held_data ( const GUID_t & src )

Definition at line 4526 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::id_, OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::link_, LM_DEBUG, OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::mutex_, OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::stopping_, and OpenDDS::DCPS::Transport_debug_level.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   if (stopping_) {
     return;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   OPENDDS_VECTOR(ReceivedDataSample) to_deliver;
 
   const WriterInfoMap::iterator wi = remote_writers_.find(src);
   if (wi != remote_writers_.end()) {
     const SequenceNumber ca = wi->second->recvd_.cumulative_ack();
     const WriterInfo::HeldMap::iterator end = wi->second->held_.upper_bound(ca);
     for (WriterInfo::HeldMap::iterator it = wi->second->held_.begin(); it != end; /*increment in loop body*/) {
       to_deliver.push_back(it->second);
       wi->second->held_.erase(it++);
     }
   }
 
   const GUID_t dst = id_;
 
   g.release();
 
   for (OPENDDS_VECTOR(ReceivedDataSample)::iterator it = to_deliver.begin(); it != to_deliver.end(); ++it) {
     if (Transport_debug_level > 5) {
       LogGuid reader(dst);
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpDataLink::DeliverHeldData::~DeliverHeldData -")
                  ACE_TEXT(" deliver sequence: %q to %C\n"),
                  it->header_.sequence_.getValue(),
                  reader.c_str()));
     }
     link->data_received(*it, dst);
   }
 }

◆ gather_ack_nacks_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::gather_ack_nacks_i	(	const WriterInfo_rch &	writer,
		const RtpsUdpDataLink_rch &	link,
		bool	heartbeat_was_non_final,
		MetaSubmessageVec &	meta_submessages,
		ACE_CDR::ULong &	cumulative_bits_added
	)

private

Definition at line 2287 of file RtpsUdpDataLink.cpp.

Referenced by process_heartbeat_frag_i(), and process_heartbeat_i().

 {
   const bool should_nack_frags = should_nack_fragments(link, writer);
   if (writer->should_nack() ||
       should_nack_frags) {
     using namespace OpenDDS::RTPS;
     const EntityId_t reader_id = id_.entityId;
     const EntityId_t writer_id = writer->id_.entityId;
     MetaSubmessage meta_submessage(id_, writer->id_);
 
     const DisjointSequence& recvd = writer->recvd_;
     const SequenceNumber& hb_high = writer->hb_last_;
     const SequenceNumber ack = recvd.empty() ? 1 : ++SequenceNumber(recvd.cumulative_ack());
     const SequenceNumber::Value ack_val = ack.getValue();
     CORBA::ULong num_bits = 0;
     LongSeq8 bitmap;
 
     if (recvd.disjoint()) {
       bitmap.length(DisjointSequence::bitmap_num_longs(ack, recvd.last_ack().previous()));
       if (bitmap.length() > 0) {
         (void)recvd.to_bitmap(bitmap.get_buffer(), bitmap.length(),
                               num_bits, cumulative_bits_added, true);
       }
     }
 
     if (!recvd.empty() && hb_high > recvd.high()) {
       const SequenceNumber eff_high =
         (hb_high <= ack_val + 255) ? hb_high : (ack_val + 255);
       const SequenceNumber::Value eff_high_val = eff_high.getValue();
       // Nack the range between the received high and the effective high.
       const CORBA::ULong old_len = bitmap.length(),
         new_len = DisjointSequence::bitmap_num_longs(ack, eff_high);
       if (new_len > old_len) {
         bitmap.length(new_len);
         for (CORBA::ULong i = old_len; i < new_len; ++i) {
           bitmap[i] = 0;
         }
       }
       const CORBA::ULong idx_hb_high = CORBA::ULong(eff_high_val - ack_val),
         idx_recv_high = recvd.disjoint() ?
         CORBA::ULong(recvd.high().getValue() - ack_val) : 0;
       DisjointSequence::fill_bitmap_range(idx_recv_high, idx_hb_high,
                                           bitmap.get_buffer(), new_len,
                                           num_bits, cumulative_bits_added);
     }
 
     // If the receive strategy is holding any fragments, those should
     // not be "nacked" in the ACKNACK reply.  They will be accounted for
     // in the NACK_FRAG(s) instead.
     const bool frags_modified =
       link->receive_strategy()->remove_frags_from_bitmap(bitmap.get_buffer(),
                                                          num_bits, ack, writer->id_, cumulative_bits_added);
     if (frags_modified) {
       for (CORBA::ULong i = 0; i < bitmap.length(); ++i) {
         if ((i + 1) * 32 <= num_bits) {
           if (bitmap[i]) {
             break;
           }
         } else {
           if ((0xffffffff << (32 - (num_bits % 32))) & bitmap[i]) {
             break;
           }
         }
       }
     }
 
     AckNackSubmessage acknack = {
       {ACKNACK,
        CORBA::Octet(FLAG_E),
        0 /*length*/},
       reader_id,
       writer_id,
       { // SequenceNumberSet: acking bitmapBase - 1
         to_rtps_seqnum(ack),
         num_bits, bitmap
       },
       {writer->heartbeat_recvd_count_}
     };
     meta_submessage.sm_.acknack_sm(acknack);
     meta_submessages.push_back(meta_submessage);
 
     if (should_nack_frags) {
       generate_nack_frags_i(meta_submessages, writer, reader_id, writer_id, cumulative_bits_added);
     }
   } else if (heartbeat_was_non_final) {
     using namespace OpenDDS::RTPS;
     const DisjointSequence& recvd = writer->recvd_;
     const CORBA::ULong num_bits = 0;
     const LongSeq8 bitmap;
     const SequenceNumber ack = recvd.empty() ? 1 : ++SequenceNumber(recvd.cumulative_ack());
     const EntityId_t reader_id = id_.entityId;
     const EntityId_t writer_id = writer->id_.entityId;
 
     MetaSubmessage meta_submessage(id_, writer->id_);
 
     AckNackSubmessage acknack = {
       {ACKNACK,
        CORBA::Octet(FLAG_E | FLAG_F),
        0 /*length*/},
       reader_id,
       writer_id,
       { // SequenceNumberSet: acking bitmapBase - 1
         to_rtps_seqnum(ack),
         num_bits, bitmap
       },
       {writer->heartbeat_recvd_count_}
     };
     meta_submessage.sm_.acknack_sm(acknack);
     meta_submessages.push_back(meta_submessage);
   }
 }

◆ gather_preassociation_acknack_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::gather_preassociation_acknack_i	(	MetaSubmessageVec &	meta_submessages,
		const WriterInfo_rch &	writer
	)

private

Definition at line 2257 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::ACKNACK, OpenDDS::RTPS::Submessage::acknack_sm, OpenDDS::DCPS::GUID_t::entityId, OpenDDS::RTPS::FLAG_E, id_, OPENDDS_ASSERT, and OpenDDS::DCPS::MetaSubmessage::sm_.

Referenced by add_writer(), and send_preassociation_acknacks().

 {
   using namespace OpenDDS::RTPS;
 
   OPENDDS_ASSERT(writer->recvd_.empty());
   const CORBA::ULong num_bits = 0;
   const LongSeq8 bitmap;
   const EntityId_t reader_id = id_.entityId;
   const EntityId_t writer_id = writer->id_.entityId;
 
   MetaSubmessage meta_submessage(id_, writer->id_);
 
   AckNackSubmessage acknack = {
     {ACKNACK,
      CORBA::Octet(FLAG_E),
      0 /*length*/},
     reader_id,
     writer_id,
     { // SequenceNumberSet: acking bitmapBase - 1
       {0, 1},
       num_bits, bitmap
     },
     {writer->heartbeat_recvd_count_}
   };
   meta_submessage.sm_.acknack_sm(acknack);
   meta_submessages.push_back(meta_submessage);
 }

◆ generate_nack_frags_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::generate_nack_frags_i	(	MetaSubmessageVec &	meta_submessages,
		const WriterInfo_rch &	wi,
		EntityId_t	reader_id,
		EntityId_t	writer_id,
		ACE_CDR::ULong &	cumulative_bits_added
	)

private

Definition at line 2882 of file RtpsUdpDataLink.cpp.

References OpenDDS::RTPS::NackFragSubmessage::count, OpenDDS::DCPS::RtpsUdpDataLink::extend_bitmap_range(), OpenDDS::RTPS::FLAG_E, OpenDDS::RTPS::NackFragSubmessage::fragmentNumberState, id_, link_, OpenDDS::RTPS::NACK_FRAG, OpenDDS::RTPS::Submessage::nack_frag_sm, nackfrag_count_, OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_MAP(), OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR(), OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), OpenDDS::DCPS::MetaSubmessage::sm_, OpenDDS::RTPS::to_rtps_seqnum(), OpenDDS::RTPS::Count_t::value, and OpenDDS::RTPS::NackFragSubmessage::writerSN.

Referenced by gather_ack_nacks_i().

 {
   typedef OPENDDS_MAP(SequenceNumber, RTPS::FragmentNumber_t)::iterator iter_t;
   typedef RtpsUdpReceiveStrategy::FragmentInfo::value_type Frag_t;
   RtpsUdpReceiveStrategy::FragmentInfo frag_info;
 
   // This is an internal method, locks already locked,
   // we just need a local handle to the link
   RtpsUdpDataLink_rch link = link_.lock();
 
   // Populate frag_info with two possible sources of NackFrags:
   // 1. sequence #s in the reception gaps that we have partially received
   OPENDDS_VECTOR(SequenceRange) missing = wi->recvd_.missing_sequence_ranges();
   for (size_t i = 0; i < missing.size(); ++i) {
     link->receive_strategy()->has_fragments(missing[i], wi->id_, &frag_info);
   }
   // 1b. larger than the last received seq# but less than the heartbeat.lastSN
   if (!wi->recvd_.empty() && wi->recvd_.high() < wi->hb_last_) {
     const SequenceRange range(wi->recvd_.high() + 1, wi->hb_last_);
     link->receive_strategy()->has_fragments(range, wi->id_, &frag_info);
   }
   for (size_t i = 0; i < frag_info.size(); ++i) {
     // If we've received a HeartbeatFrag, we know the last (available) frag #
     const iter_t heartbeat_frag = wi->frags_.find(frag_info[i].first);
     if (heartbeat_frag != wi->frags_.end()) {
       extend_bitmap_range(frag_info[i].second, heartbeat_frag->second.value, cumulative_bits_added);
     }
   }
 
   // 2. sequence #s outside the recvd_ gaps for which we have a HeartbeatFrag
   const iter_t low = wi->frags_.lower_bound(wi->recvd_.cumulative_ack()),
               high = wi->frags_.upper_bound(wi->recvd_.last_ack()),
                end = wi->frags_.end();
   for (iter_t iter = wi->frags_.begin(); iter != end; ++iter) {
     if (iter == low) {
       // skip over the range covered by step #1 above
       if (high == end) {
         break;
       }
       iter = high;
     }
 
     const SequenceRange range(iter->first, iter->first);
     if (!link->receive_strategy()->has_fragments(range, wi->id_, &frag_info)) {
       // it was not in the recv strategy, so the entire range is "missing"
       frag_info.push_back(Frag_t(iter->first, RTPS::FragmentNumberSet()));
       RTPS::FragmentNumberSet& fnSet = frag_info.back().second;
       fnSet.bitmapBase.value = 1;
       fnSet.numBits = std::min(CORBA::ULong(256), iter->second.value);
       fnSet.bitmap.length((fnSet.numBits + 31) / 32);
       for (CORBA::ULong i = 0; i < fnSet.bitmap.length(); ++i) {
         fnSet.bitmap[i] = 0xFFFFFFFF;
       }
     }
   }
 
   if (frag_info.empty()) {
     return;
   }
 
   const RTPS::NackFragSubmessage nackfrag_prototype = {
     {RTPS::NACK_FRAG, RTPS::FLAG_E, 0 /* length set below */},
     reader_id,
     writer_id,
     {0, 0}, // writerSN set below
     RTPS::FragmentNumberSet(), // fragmentNumberState set below
     {0} // count set below
   };
 
   meta_submessages.reserve(meta_submessages.size() + frag_info.size());
   for (size_t i = 0; i < frag_info.size(); ++i) {
     MetaSubmessage meta_submessage(id_, wi->id_);
     meta_submessage.sm_.nack_frag_sm(nackfrag_prototype);
     RTPS::NackFragSubmessage& nackfrag = meta_submessage.sm_.nack_frag_sm();
     nackfrag.writerSN = to_rtps_seqnum(frag_info[i].first);
     nackfrag.fragmentNumberState = frag_info[i].second;
     nackfrag.count.value = ++nackfrag_count_;
     meta_submessages.push_back(meta_submessage);
   }
 }

◆ has_writer()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::has_writer ( const GUID_t & id ) const

Definition at line 2181 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, mutex_, and remote_writers_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
   return remote_writers_.count(id) != 0;
 }

◆ id()

const GUID_t& OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id ( void ) const

inline

Definition at line 651 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::bundle_mapped_meta_submessages().

651 { return id_; }

OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_

const GUID_t id_

Definition: RtpsUdpDataLink.h:673

◆ log_remote_counts()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::log_remote_counts ( const char * funcname )

Definition at line 4828 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::id_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_remote_counts, OPENDDS_END_VERSIONED_NAMESPACE_DECL, and OpenDDS::DCPS::transport_debug.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_reader(), add_writer(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), process_heartbeat_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::remove_reader(), and remove_writer().

 {
   if (transport_debug.log_remote_counts) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_remote_counts} "
       "RtpsUdpDataLink::RtpsReader::%C: "
       "%C pre: %b assoc: %b\n",
       funcname, LogGuid(id_).c_str(),
       preassociation_writers_.size(), remote_writers_.size()));
   }
 }

◆ pre_stop_helper()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::pre_stop_helper ( )

Definition at line 1617 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, OpenDDS::DCPS::InternalDataReaderListener< NetworkInterfaceAddress >::mutex_, OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), ACE_Guard< ACE_LOCK >::release(), and OpenDDS::DCPS::DataLink::strategy_lock_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::client_stop().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   if (stopping_) {
     return;
   }
 
   stopping_ = true;
 
   preassociation_writers_.clear();
   log_remote_counts("pre_stop_helper");
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   GuardType guard(link->strategy_lock_);
   if (link->receive_strategy() == 0) {
     return;
   }
 
   for (WriterInfoMap::iterator it = remote_writers_.begin(); it != remote_writers_.end(); ++it) {
     it->second->held_.clear();
   }
 
   guard.release();
   g.release();
 
   preassociation_task_->cancel();
 }

◆ process_data_i()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_data_i	(	const RTPS::DataSubmessage &	data,
		const GUID_t &	src,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 1666 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_GUARD_RETURN, ACE_TEXT(), OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::SequenceNumber::getValue(), id_, link_, LM_DEBUG, OpenDDS::DCPS::TransportDebug::log_dropped_messages, mutex_, OpenDDS::DCPS::SequenceNumber::previous(), OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), ACE_Guard< ACE_LOCK >::release(), remote_writers_, stopping_, OpenDDS::DCPS::DataLink::strategy_lock_, OpenDDS::DCPS::swap(), OpenDDS::RTPS::to_opendds_seqnum(), OpenDDS::DCPS::transport_debug, OpenDDS::DCPS::Transport_debug_level, and OpenDDS::RTPS::DataSubmessage::writerSN.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
 
   if (stopping_) {
     return false;
   }
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return false;
   }
 
   GuardType guard(link->strategy_lock_);
   if (link->receive_strategy() == 0) {
     return false;
   }
 
   const SequenceNumber seq = to_opendds_seqnum(data.writerSN);
   DeliverHeldData dhd;
   const WriterInfoMap::iterator wi = remote_writers_.find(src);
   if (wi != remote_writers_.end()) {
     const WriterInfo_rch& writer = wi->second;
 
     DeliverHeldData dhd2(rchandle_from(this), src);
     std::swap(dhd, dhd2);
 
     writer->frags_.erase(seq);
 
     if (writer->recvd_.empty()) {
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) -")
                    ACE_TEXT(" data seq: %q from %C being from %C expecting heartbeat\n"),
                    seq.getValue(),
                    LogGuid(src).c_str(),
                    LogGuid(id_).c_str()));
       }
       const ReceivedDataSample* sample =
         link->receive_strategy()->withhold_data_from(id_);
       writer->held_.insert(std::make_pair(seq, *sample));
 
     } else if (writer->recvd_.contains(seq)) {
       if (transport_debug.log_dropped_messages) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_data_i: %C -> %C duplicate sample\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
       }
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) -")
                              ACE_TEXT(" data seq: %q from %C being DROPPED from %C because it's ALREADY received\n"),
                              seq.getValue(),
                              LogGuid(src).c_str(),
                              LogGuid(id_).c_str()));
       }
       link->receive_strategy()->withhold_data_from(id_);
 
     } else if (!writer->held_.empty()) {
       const ReceivedDataSample* sample =
         link->receive_strategy()->withhold_data_from(id_);
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) WITHHOLD %q\n", seq.getValue()));
         writer->recvd_.dump();
       }
       writer->held_.insert(std::make_pair(seq, *sample));
       writer->recvd_.insert(seq);
 
     } else if (writer->recvd_.disjoint() || writer->recvd_.cumulative_ack() != seq.previous()) {
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) -")
                              ACE_TEXT(" data seq: %q from %C being WITHHELD from %C because it's EXPECTING more data\n"),
                              seq.getValue(),
                              LogGuid(src).c_str(),
                              LogGuid(id_).c_str()));
       }
       const ReceivedDataSample* sample =
         link->receive_strategy()->withhold_data_from(id_);
       writer->held_.insert(std::make_pair(seq, *sample));
       writer->recvd_.insert(seq);
 
     } else {
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) -")
                              ACE_TEXT(" data seq: %q from %C to %C OK to deliver\n"),
                              seq.getValue(),
                              LogGuid(src).c_str(),
                              LogGuid(id_).c_str()));
       }
       writer->recvd_.insert(seq);
       link->receive_strategy()->do_not_withhold_data_from(id_);
     }
 
   } else {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_data_i: %C -> %C unknown remote writer\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     if (Transport_debug_level > 5) {
       ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) RtpsUdpDataLink::process_data_i(DataSubmessage) -")
                            ACE_TEXT(" data seq: %q from %C to %C dropped because of unknown writer\n"),
                            to_opendds_seqnum(data.writerSN).getValue(),
                            LogGuid(src).c_str(),
                            LogGuid(id_).c_str()));
     }
     link->receive_strategy()->withhold_data_from(id_);
   }
 
   // Release for delivering held data.
   guard.release();
   g.release();
 
   return false;
 }

◆ process_gap_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_gap_i	(	const RTPS::GapSubmessage &	gap,
		const GUID_t &	src,
		bool	directed,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 1791 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, OpenDDS::DCPS::DisjointSequence::contains(), OpenDDS::DCPS::DisjointSequence::empty(), OpenDDS::RTPS::GapSubmessage::gapList, OpenDDS::RTPS::GapSubmessage::gapStart, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::DisjointSequence::high(), id_, OpenDDS::DCPS::DisjointSequence::insert(), link_, LM_DEBUG, LM_ERROR, OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::DisjointSequence::low(), mutex_, OpenDDS::DCPS::RtpsUdpDataLink::OPENDDS_VECTOR(), OpenDDS::DCPS::SequenceNumber::previous(), OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), ACE_Guard< ACE_LOCK >::release(), remote_writers_, OpenDDS::DCPS::DataLink::start(), OpenDDS::DCPS::DataLink::strategy_lock_, OpenDDS::RTPS::to_opendds_seqnum(), and OpenDDS::DCPS::transport_debug.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::received().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   GuardType guard(link->strategy_lock_);
   if (link->receive_strategy() == 0) {
     return;
   }
 
   const WriterInfoMap::iterator wi = remote_writers_.find(src);
   if (wi == remote_writers_.end()) {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_gap_i: %C -> %C unknown remote writer\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const WriterInfo_rch& writer = wi->second;
 
   if (writer->recvd_.empty()) {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_gap_i: %C -> %C preassociation writer\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const SequenceNumber start = to_opendds_seqnum(gap.gapStart);
   const SequenceNumber base = to_opendds_seqnum(gap.gapList.bitmapBase);
 
   if (start < base) {
     writer->recvd_.insert(SequenceRange(start, base.previous()));
   } else if (start != base) {
     ACE_ERROR((LM_ERROR, "(%P|%t) RtpsUdpDataLink::RtpsReader::process_gap_i: ERROR - Incoming GAP has inverted start (%q) & base (%q) values, ignoring start value\n", start.getValue(), base.getValue()));
   }
   writer->recvd_.insert(base, gap.gapList.numBits, gap.gapList.bitmap.get_buffer());
 
   DisjointSequence gaps;
   if (start < base) {
     gaps.insert(SequenceRange(start, base.previous()));
   }
   gaps.insert(base, gap.gapList.numBits, gap.gapList.bitmap.get_buffer());
 
   if (!gaps.empty()) {
     for (WriterInfo::HeldMap::iterator pos = writer->held_.lower_bound(gaps.low()),
            limit = writer->held_.upper_bound(gaps.high()); pos != limit;) {
       if (gaps.contains(pos->first)) {
         writer->held_.erase(pos++);
       } else {
         ++pos;
       }
     }
   }
 
   const OPENDDS_VECTOR(SequenceRange) psr = gaps.present_sequence_ranges();
   for (OPENDDS_VECTOR(SequenceRange)::const_iterator pos = psr.begin(), limit = psr.end(); pos != limit; ++pos) {
     link->receive_strategy()->remove_fragments(*pos, writer->id_);
   }
 
   guard.release();
   g.release();
 
   DeliverHeldData dhd(rchandle_from(this), src);
 }

◆ process_heartbeat_frag_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i	(	const RTPS::HeartBeatFragSubmessage &	hb_frag,
		const GUID_t &	src,
		bool	directed,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 3000 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_GUARD, OpenDDS::DCPS::RtpsUdpDataLink::config(), OpenDDS::RTPS::HeartBeatFragSubmessage::count, OpenDDS::RTPS::FLAG_F, OpenDDS::RTPS::SubmessageHeader::flags, gather_ack_nacks_i(), id_, OpenDDS::RTPS::HeartBeatFragSubmessage::lastFragmentNum, link_, LM_DEBUG, LM_WARNING, OpenDDS::DCPS::TransportDebug::log_dropped_messages, mutex_, OpenDDS::DCPS::InternalTransportStatistics::reader_nack_count, OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), remote_writers_, OpenDDS::RTPS::HeartBeatFragSubmessage::smHeader, OpenDDS::DCPS::DataLink::strategy_lock_, OpenDDS::RTPS::to_opendds_seqnum(), OpenDDS::DCPS::transport_debug, OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_, OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_mutex_, OpenDDS::RTPS::Count_t::value, VDBG, and OpenDDS::RTPS::HeartBeatFragSubmessage::writerSN.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::received().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   GuardType guard(link->strategy_lock_);
   if (link->receive_strategy() == 0) {
     return;
   }
 
   const WriterInfoMap::iterator wi = remote_writers_.find(src);
   if (wi == remote_writers_.end()) {
     // we may not be associated yet, even if the writer thinks we are
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i: %C -> %C unknown remote writer\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const WriterInfo_rch& writer = wi->second;
 
   if (!compare_and_update_counts(hb_frag.count.value, writer->hb_frag_recvd_count_)) {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_heartbeat_frag_i: %C -> %C stale/duplicate message\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     VDBG((LM_WARNING, "(%P|%t) RtpsUdpDataLink::process_heartbeat_frag_i "
           "WARNING Count indicates duplicate, dropping\n"));
     return;
   }
 
   // If seq is outside the heartbeat range or we haven't completely received
   // it yet, send a NackFrag along with the AckNack.  The heartbeat range needs
   // to be checked first because recvd_ contains the numbers below the
   // heartbeat range (so that we don't NACK those).
   const SequenceNumber seq = to_opendds_seqnum(hb_frag.writerSN);
   if (seq > writer->hb_last_ || !writer->recvd_.contains(seq)) {
     writer->frags_[seq] = hb_frag.lastFragmentNum;
     ACE_CDR::ULong cumulative_bits_added = 0;
     gather_ack_nacks_i(writer, link, !(hb_frag.smHeader.flags & RTPS::FLAG_F), meta_submessages, cumulative_bits_added);
     if (cumulative_bits_added) {
       RtpsUdpInst_rch cfg = link->config();
       if (cfg && cfg->count_messages()) {
         ACE_GUARD(ACE_Thread_Mutex, g, link->transport_statistics_mutex_);
         link->transport_statistics_.reader_nack_count[id_] += cumulative_bits_added;
       }
     }
 
   }
 }

◆ process_heartbeat_i()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::process_heartbeat_i	(	const RTPS::HeartBeatSubmessage &	heartbeat,
		const GUID_t &	src,
		bool	directed,
		MetaSubmessageVec &	meta_submessages
	)

Definition at line 1904 of file RtpsUdpDataLink.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, OpenDDS::DCPS::LogGuid::c_str(), OpenDDS::DCPS::RtpsUdpDataLink::config(), OpenDDS::RTPS::HeartBeatSubmessage::count, OpenDDS::DCPS::ENTITYID_UNKNOWN, OpenDDS::RTPS::HeartBeatSubmessage::firstSN, OpenDDS::RTPS::FLAG_F, OpenDDS::RTPS::SubmessageHeader::flags, gather_ack_nacks_i(), OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::GUID_UNKNOWN, id_, OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), OpenDDS::RTPS::HeartBeatSubmessage::lastSN, link_, LM_DEBUG, LM_ERROR, LM_WARNING, OpenDDS::DCPS::TransportDebug::log_dropped_messages, OpenDDS::DCPS::TransportDebug::log_nonfinal_messages, OpenDDS::DCPS::log_progress(), OpenDDS::DCPS::TransportDebug::log_progress, log_remote_counts(), mutex_, OPENDDS_ASSERT, preassociation_writers_, OpenDDS::DCPS::SequenceNumber::previous(), OpenDDS::DCPS::rchandle_from(), OpenDDS::DCPS::InternalTransportStatistics::reader_nack_count, OpenDDS::RTPS::HeartBeatSubmessage::readerId, OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy(), ACE_Guard< ACE_LOCK >::release(), remote_writers_, OpenDDS::RTPS::HeartBeatSubmessage::smHeader, OpenDDS::DCPS::DataLink::strategy_lock_, OpenDDS::RTPS::to_opendds_seqnum(), OpenDDS::DCPS::transport_debug, OpenDDS::DCPS::Transport_debug_level, OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_, OpenDDS::DCPS::RtpsUdpDataLink::transport_statistics_mutex_, OpenDDS::RTPS::Count_t::value, VDBG, and OpenDDS::DCPS::SequenceNumber::ZERO().

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::received().

 {
   // TODO: Delay responses by heartbeat_response_delay_.
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
   RtpsUdpDataLink_rch link = link_.lock();
 
   if (!link) {
     return;
   }
 
   GuardType guard(link->strategy_lock_);
   if (link->receive_strategy() == 0) {
     return;
   }
 
   // Heartbeat Sequence Range
   const SequenceNumber hb_first = to_opendds_seqnum(heartbeat.firstSN);
   const SequenceNumber hb_last = to_opendds_seqnum(heartbeat.lastSN);
 
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) RtpsUdpDataLink::RtpsReader::process_heartbeat_i - %C -> %C first %q last %q count %d\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str(), hb_first.getValue(), hb_last.getValue(), heartbeat.count.value));
   }
 
   const WriterInfoMap::iterator wi = remote_writers_.find(src);
   if (wi == remote_writers_.end()) {
     if (transport_debug.log_dropped_messages) {
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_heartbeat_i: %C -> %C unknown remote writer\n", LogGuid(src).c_str(), LogGuid(id_).c_str()));
     }
     return;
   }
 
   const WriterInfo_rch& writer = wi->second;
 
   if (!compare_and_update_counts(heartbeat.count.value, writer->heartbeat_recvd_count_)) {
     if (transport_debug.log_dropped_messages) {
       const GUID_t dst = heartbeat.readerId == DCPS::ENTITYID_UNKNOWN ? GUID_UNKNOWN : id_;
       ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_dropped_messages} RtpsUdpDataLink::RtpsReader::process_heartbeat_i: %C -> %C stale/duplicate message (%d vs %d)\n",
         LogGuid(src).c_str(), LogGuid(dst).c_str(), heartbeat.count.value, writer->heartbeat_recvd_count_));
     }
     VDBG((LM_WARNING, "(%P|%t) RtpsUdpDataLink::process_heartbeat_i "
           "WARNING Count indicates duplicate, dropping\n"));
     return;
   }
 
   const bool is_final = heartbeat.smHeader.flags & RTPS::FLAG_F;
 
   static const SequenceNumber one, zero = SequenceNumber::ZERO();
 
   bool first_ever_hb = false;
 
   if (!is_final && transport_debug.log_nonfinal_messages) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) {transport_debug.log_nonfinal_messages} RtpsUdpDataLink::RtpsReader::process_heartbeat_i - %C -> %C first %q last %q count %d\n",
       LogGuid(src).c_str(), LogGuid(id_).c_str(), hb_first.getValue(), hb_last.getValue(), heartbeat.count.value));
   }
 
   // Only valid heartbeats (see spec) will be "fully" applied to writer info
   if (!(hb_first < 1 || hb_last < 0 || hb_last < hb_first.previous())) {
     if (writer->recvd_.empty() && (directed || !writer->sends_directed_hb())) {
       OPENDDS_ASSERT(preassociation_writers_.count(writer));
       preassociation_writers_.erase(writer);
       if (transport_debug.log_progress) {
         log_progress("RTPS reader/writer association complete", id_, writer->id_, writer->participant_discovered_at_);
       }
       log_remote_counts("process_heartbeat_i");
 
       const SequenceRange sr(zero, hb_first.previous());
       writer->recvd_.insert(sr);
       while (!writer->held_.empty() && writer->held_.begin()->first <= sr.second) {
         writer->held_.erase(writer->held_.begin());
       }
       for (WriterInfo::HeldMap::const_iterator it = writer->held_.begin(); it != writer->held_.end(); ++it) {
         writer->recvd_.insert(it->first);
       }
       link->receive_strategy()->remove_fragments(sr, writer->id_);
       first_ever_hb = true;
     }
 
     ACE_CDR::ULong cumulative_bits_added = 0;
     if (!writer->recvd_.empty()) {
       writer->hb_last_ = std::max(writer->hb_last_, hb_last);
       gather_ack_nacks_i(writer, link, !is_final, meta_submessages, cumulative_bits_added);
     }
     if (cumulative_bits_added) {
       RtpsUdpInst_rch cfg = link->config();
       if (cfg && cfg->count_messages()) {
         ACE_Guard<ACE_Thread_Mutex> tsg(link->transport_statistics_mutex_);
         link->transport_statistics_.reader_nack_count[id_] += cumulative_bits_added;
       }
     }
   } else {
     ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: RtpsUdpDataLink::RtpsReader::process_heartbeat_i: %C -> %C - INVALID - first %q last %q count %d\n", LogGuid(writer->id_).c_str(), LogGuid(id_).c_str(), hb_first.getValue(), hb_last.getValue(), heartbeat.count.value));
   }
 
   guard.release();
   g.release();
 
   if (first_ever_hb) {
     link->invoke_on_start_callbacks(id_, src, true);
   }
 
   DeliverHeldData dhd(rchandle_from(this), src);
 
   //FUTURE: support assertion of liveliness for MANUAL_BY_TOPIC
   return;
 }

◆ remove_writer()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::remove_writer ( const GUID_t & id )

Definition at line 2188 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, log_remote_counts(), mutex_, preassociation_writers_, and remote_writers_.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::release_reservations_i().

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
   WriterInfoMap::iterator pos = remote_writers_.find(id);
   if (pos != remote_writers_.end()) {
     preassociation_writers_.erase(pos->second);
     remote_writers_.erase(pos);
     log_remote_counts("remove_writer");
     return true;
   }
 
   return false;
 }

◆ send_preassociation_acknacks()

void OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::send_preassociation_acknacks ( const MonotonicTimePoint & now )

private

Definition at line 2228 of file RtpsUdpDataLink.cpp.

References ACE_GUARD, gather_preassociation_acknack_i(), heartbeat_period_, link_, mutex_, preassociation_task_, preassociation_writers_, OpenDDS::DCPS::RtpsUdpDataLink::queue_submessages(), and stopping_.

 {
   RtpsUdpDataLink_rch link = link_.lock();
   if (!link) {
     return;
   }
 
   MetaSubmessageVec meta_submessages;
   {
     ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
 
     if (stopping_ || preassociation_writers_.empty()) {
       return;
     }
 
     // We want a heartbeat from these writers.
     meta_submessages.reserve(preassociation_writers_.size());
     for (WriterInfoSet::const_iterator pos = preassociation_writers_.begin(), limit = preassociation_writers_.end();
          pos != limit; ++pos) {
       gather_preassociation_acknack_i(meta_submessages, *pos);
     }
   }
 
   link->queue_submessages(meta_submessages);
 
   preassociation_task_->schedule(heartbeat_period_);
 }

◆ should_nack_fragments()

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::should_nack_fragments	(	const RcHandle< RtpsUdpDataLink > &	link,
		const WriterInfo_rch &	info
	)

Definition at line 2210 of file RtpsUdpDataLink.cpp.

References OpenDDS::DCPS::RtpsUdpDataLink::receive_strategy().

Referenced by gather_ack_nacks_i().

 {
   if (!info->frags_.empty()) {
     return true;
   }
 
   if (!info->recvd_.empty()) {
     const SequenceRange range(info->recvd_.cumulative_ack() + 1, info->hb_last_);
     if (link->receive_strategy()->has_fragments(range, info->id_)) {
       return true;
     }
   }
 
   return false;
 }

◆ writer_count()

size_t OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::writer_count ( ) const

Definition at line 2203 of file RtpsUdpDataLink.cpp.

References ACE_GUARD_RETURN, mutex_, and remote_writers_.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, 0);
   return remote_writers_.size();
 }

Member Data Documentation

◆ heartbeat_period_

TimeDuration OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::heartbeat_period_

private

Definition at line 679 of file RtpsUdpDataLink.h.

Referenced by add_writer(), and send_preassociation_acknacks().

◆ id_

const GUID_t OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::id_

private

Definition at line 673 of file RtpsUdpDataLink.h.

◆ link_

WeakRcHandle<RtpsUdpDataLink> OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::link_

private

Definition at line 672 of file RtpsUdpDataLink.h.

◆ mutex_

ACE_Thread_Mutex OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::mutex_

mutableprivate

Definition at line 671 of file RtpsUdpDataLink.h.

◆ nackfrag_count_

CORBA::Long OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::nackfrag_count_

private

Definition at line 677 of file RtpsUdpDataLink.h.

Referenced by generate_nack_frags_i().

◆ preassociation_task_

RcHandle<SporadicEvent> OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::preassociation_task_

private

Definition at line 678 of file RtpsUdpDataLink.h.

Referenced by add_writer(), and send_preassociation_acknacks().

◆ preassociation_writers_

WriterInfoSet OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::preassociation_writers_

private

Definition at line 675 of file RtpsUdpDataLink.h.

Referenced by add_writer(), process_heartbeat_i(), remove_writer(), and send_preassociation_acknacks().

◆ remote_writers_

WriterInfoMap OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::remote_writers_

private

Definition at line 674 of file RtpsUdpDataLink.h.

Referenced by add_writer(), has_writer(), process_data_i(), process_gap_i(), process_heartbeat_frag_i(), process_heartbeat_i(), remove_writer(), and writer_count().

◆ stopping_

bool OpenDDS::DCPS::RtpsUdpDataLink::RtpsReader::stopping_

private

Definition at line 676 of file RtpsUdpDataLink.h.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::add_reader(), add_writer(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::make_leader_lagger(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_acknack(), process_data_i(), OpenDDS::DCPS::RtpsUdpDataLink::RtpsWriter::process_nackfrag(), and send_preassociation_acknacks().

The documentation for this class was generated from the following files:

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ RtpsReader()

◆ ~RtpsReader()

Member Function Documentation

◆ add_writer()

◆ deliver_held_data()

◆ gather_ack_nacks_i()

◆ gather_preassociation_acknack_i()

◆ generate_nack_frags_i()

◆ has_writer()

◆ id()

◆ log_remote_counts()

◆ pre_stop_helper()

◆ process_data_i()

◆ process_gap_i()

◆ process_heartbeat_frag_i()

◆ process_heartbeat_i()

◆ remove_writer()

◆ send_preassociation_acknacks()

◆ should_nack_fragments()

◆ writer_count()

Member Data Documentation

◆ heartbeat_period_

◆ id_

◆ link_

◆ mutex_

◆ nackfrag_count_

◆ preassociation_task_

◆ preassociation_writers_

◆ remote_writers_

◆ stopping_