#include <TransportSendBuffer.h>

Inheritance diagram for OpenDDS::DCPS::SingleSendBuffer:

Collaboration diagram for OpenDDS::DCPS::SingleSendBuffer:

[legend]

Classes
class	Proxy

Public Member Functions
void	release_all ()

typedef	OPENDDS_VECTOR (BufferType) BufferVec

typedef	OPENDDS_MAP (SequenceNumber, BufferType) BufferMap

void	release_acked (SequenceNumber seq)

void	remove_acked (SequenceNumber seq, BufferVec &removed)

size_t	n_chunks () const

	SingleSendBuffer (size_t capacity, size_t max_samples_per_packet)

	~SingleSendBuffer ()

bool	resend (const SequenceRange &range, DisjointSequence *gaps=0)

void	retain_all (const GUID_t &pub_id)

void	insert (SequenceNumber sequence, TransportSendStrategy::QueueType queue, ACE_Message_Block chain)

void	insert_fragment (SequenceNumber sequence, SequenceNumber fragment, bool is_last_fragment, TransportSendStrategy::QueueType queue, ACE_Message_Block chain)

void	pre_insert (SequenceNumber sequence)

void	pre_clear ()

bool	has_frags (const SequenceNumber &seq) const

Public Member Functions inherited from OpenDDS::DCPS::TransportSendBuffer
size_t	capacity () const

void	bind (TransportSendStrategy *strategy)

LockType &	strategy_lock ()

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

Static Public Attributes
static const size_t	UNLIMITED = 0

Private Member Functions
void	check_capacity_i (BufferVec &removed)

void	release_i (BufferMap::iterator buffer_iter)

void	remove_i (BufferMap::iterator buffer_iter, BufferVec &removed)

RemoveResult	retain_buffer (const GUID_t &pub_id, BufferType &buffer)

void	insert_buffer (BufferType &buffer, TransportSendStrategy::QueueType queue, ACE_Message_Block chain)

bool	resend_i (const SequenceRange &range, DisjointSequence *gaps=0)

bool	resend_i (const SequenceRange &range, DisjointSequence *gaps, const GUID_t &destination)

void	resend_fragments_i (SequenceNumber sequence, const DisjointSequence &fragments, size_t &cumulative_send_count)

typedef	OPENDDS_MAP (SequenceNumber, BufferMap) FragmentMap

typedef	OPENDDS_MAP (SequenceNumber, GUID_t) DestinationMap

typedef	OPENDDS_SET (SequenceNumber) SequenceNumberSet

Private Attributes
size_t	n_chunks_

MessageBlockAllocator	retained_mb_allocator_

DataBlockAllocator	retained_db_allocator_

MessageBlockAllocator	replaced_mb_allocator_

DataBlockAllocator	replaced_db_allocator_

BufferMap	buffers_

FragmentMap	fragments_

DestinationMap	destinations_

SequenceNumberSet	pre_seq_

SequenceNumber	minimum_sn_allowed_

ACE_Thread_Mutex	mutex_

Additional Inherited Members
Public Types inherited from OpenDDS::DCPS::TransportSendBuffer
typedef TransportSendStrategy::LockType	LockType

Protected Types inherited from OpenDDS::DCPS::TransportSendBuffer
typedef TransportSendStrategy::QueueType	QueueType

typedef std::pair< QueueType , ACE_Message_Block >	BufferType

Protected Member Functions inherited from OpenDDS::DCPS::TransportSendBuffer
	TransportSendBuffer (size_t capacity)

virtual	~TransportSendBuffer ()

void	resend_one (const BufferType &buffer)

Protected Member Functions inherited from OpenDDS::DCPS::RcObject
	RcObject ()

Protected Attributes inherited from OpenDDS::DCPS::TransportSendBuffer
TransportSendStrategy *	strategy_

const size_t	capacity_

Detailed Description

Implementation of TransportSendBuffer that manages data for a single domain of SequenceNumbers – for a given SingleSendBuffer object, the sequence numbers passed to insert() must be generated from the same place.

Definition at line 72 of file TransportSendBuffer.h.

Constructor & Destructor Documentation

◆ SingleSendBuffer()

OpenDDS::DCPS::SingleSendBuffer::SingleSendBuffer	(	size_t	capacity,
		size_t	max_samples_per_packet
	)

Definition at line 53 of file TransportSendBuffer.cpp.

   : TransportSendBuffer(capacity),
     n_chunks_(capacity * max_samples_per_packet),
     retained_mb_allocator_(n_chunks_ * 2),
     retained_db_allocator_(n_chunks_ * 2),
     replaced_mb_allocator_(n_chunks_ * 2),
     replaced_db_allocator_(n_chunks_ * 2)
 {
 }

◆ ~SingleSendBuffer()

OpenDDS::DCPS::SingleSendBuffer::~SingleSendBuffer ( )

Definition at line 64 of file TransportSendBuffer.cpp.

References release_all().

 {
   release_all();
 }

Member Function Documentation

◆ check_capacity_i()

void OpenDDS::DCPS::SingleSendBuffer::check_capacity_i ( BufferVec & removed )

private

Definition at line 349 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_TEXT(), buffers_, OpenDDS::DCPS::TransportSendBuffer::capacity_, LM_DEBUG, remove_i(), and OpenDDS::DCPS::Transport_debug_level.

Referenced by insert(), and insert_fragment().

 {
   if (capacity_ == SingleSendBuffer::UNLIMITED) {
     return;
   }
   // Age off oldest sample if we are at capacity:
   if (buffers_.size() == capacity_) {
     BufferMap::iterator it(buffers_.begin());
     if (it == buffers_.end()) return;
 
     if (Transport_debug_level > 5) {
       ACE_DEBUG((LM_DEBUG,
         ACE_TEXT("(%P|%t) SingleSendBuffer::check_capacity() - ")
         ACE_TEXT("aging off PDU: %q as buffer(0x%@,0x%@)\n"),
         it->first.getValue(),
         it->second.first, it->second.second
       ));
     }
 
     remove_i(it, removed);
   }
 }

◆ has_frags()

bool OpenDDS::DCPS::SingleSendBuffer::has_frags ( const SequenceNumber & seq ) const

Definition at line 373 of file TransportSendBuffer.cpp.

References fragments_.

 {
   return fragments_.find(seq) != fragments_.end();
 }

◆ insert()

void OpenDDS::DCPS::SingleSendBuffer::insert	(	SequenceNumber	sequence,
		TransportSendStrategy::QueueType *	queue,
		ACE_Message_Block *	chain
	)

virtual

Implements OpenDDS::DCPS::TransportSendBuffer.

Definition at line 245 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), buffers_, check_capacity_i(), destinations_, OpenDDS::DCPS::SequenceNumber::getValue(), OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::HISTORIC_SAMPLE_FLAG, insert_buffer(), LM_DEBUG, minimum_sn_allowed_, OpenDDS::DCPS::TransportQueueElement::msg(), mutex_, OpenDDS::DCPS::BasicQueue< T >::peek(), pre_seq_, OpenDDS::DCPS::BasicQueue< T >::size(), OpenDDS::DCPS::TransportQueueElement::subscription_id(), OpenDDS::DCPS::DataSampleHeader::test_flag(), and OpenDDS::DCPS::Transport_debug_level.

 {
   BufferVec removed;
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   if (sequence < minimum_sn_allowed_) {
     return;
   }
   check_capacity_i(removed);
 
   BufferType& buffer = buffers_[sequence];
   pre_seq_.erase(sequence);
   insert_buffer(buffer, queue, chain);
 
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG,
       ACE_TEXT("(%P|%t) SingleSendBuffer::insert() - ")
       ACE_TEXT("saved PDU: %q as buffer(0x%@,0x%@)\n"),
       sequence.getValue(),
       buffer.first, buffer.second
     ));
   }
 
   if (queue && queue->size() == 1) {
     const TransportQueueElement* elt = queue->peek();
     const GUID_t subId = elt->subscription_id();
     const ACE_Message_Block* msg = elt->msg();
     if (msg && subId != GUID_UNKNOWN &&
         !DataSampleHeader::test_flag(HISTORIC_SAMPLE_FLAG, msg)) {
       destinations_[sequence] = subId;
     }
   }
   g.release();
   for (size_t i = 0; i < removed.size(); ++i) {
     RemoveAllVisitor visitor;
     removed[i].first->accept_remove_visitor(visitor);
     delete removed[i].first;
     Message_Block_Ptr to_release(removed[i].second);
   }
 }

◆ insert_buffer()

void OpenDDS::DCPS::SingleSendBuffer::insert_buffer	(	BufferType &	buffer,
		TransportSendStrategy::QueueType *	queue,
		ACE_Message_Block *	chain
	)

private

Definition at line 288 of file TransportSendBuffer.cpp.

References OpenDDS::DCPS::BasicQueue< T >::accept_visitor(), ACE_NEW, ACE_Message_Block::duplicate(), retained_db_allocator_, and retained_mb_allocator_.

Referenced by insert(), and insert_fragment().

 {
   // Copy sample's TransportQueueElements:
   TransportSendStrategy::QueueType*& elems = buffer.first;
   ACE_NEW(elems, TransportSendStrategy::QueueType());
 
   CopyChainVisitor visitor(*elems,
                            &retained_mb_allocator_,
                            &retained_db_allocator_,
                            true);
   queue->accept_visitor(visitor);
 
   buffer.second = chain->duplicate();
 }

◆ insert_fragment()

void OpenDDS::DCPS::SingleSendBuffer::insert_fragment	(	SequenceNumber	sequence,
		SequenceNumber	fragment,
		bool	is_last_fragment,
		TransportSendStrategy::QueueType *	queue,
		ACE_Message_Block *	chain
	)

Definition at line 306 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_GUARD, ACE_TEXT(), buffers_, check_capacity_i(), fragments_, OpenDDS::DCPS::SequenceNumber::getValue(), insert_buffer(), LM_DEBUG, minimum_sn_allowed_, mutex_, pre_seq_, and OpenDDS::DCPS::Transport_debug_level.

 {
   BufferVec removed;
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   if (sequence < minimum_sn_allowed_) {
     return;
   }
   check_capacity_i(removed);
 
   // Insert into buffers_ so that the overall capacity is maintained
   // The entry in buffers_ with two null pointers indicates that the
   // actual data is stored in fragments_[sequence].
   buffers_[sequence] = std::make_pair(static_cast<QueueType*>(0),
                                       static_cast<ACE_Message_Block*>(0));
 
   BufferType& buffer = fragments_[sequence][fragment];
   if (is_last_fragment) {
     pre_seq_.erase(sequence);
   }
   insert_buffer(buffer, queue, chain);
 
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG,
       ACE_TEXT("(%P|%t) SingleSendBuffer::insert_fragment() - ")
       ACE_TEXT("saved PDU: %q,%q as buffer(0x%@,0x%@)\n"),
       sequence.getValue(), fragment.getValue(),
       buffer.first, buffer.second
     ));
   }
   g.release();
   for (size_t i = 0; i < removed.size(); ++i) {
     RemoveAllVisitor visitor;
     removed[i].first->accept_remove_visitor(visitor);
     delete removed[i].first;
     Message_Block_Ptr to_release(removed[i].second);
   }
 }

◆ n_chunks()

ACE_INLINE size_t OpenDDS::DCPS::SingleSendBuffer::n_chunks ( ) const

Definition at line 29 of file TransportSendBuffer.inl.

References ACE_GUARD_RETURN, and ACE_INLINE.

 {
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, this->n_chunks_);
   return this->n_chunks_;
 }

◆ OPENDDS_MAP() [1/3]

typedef OpenDDS::DCPS::SingleSendBuffer::OPENDDS_MAP	(	SequenceNumber	,
		BufferType
	)

◆ OPENDDS_MAP() [2/3]

typedef OpenDDS::DCPS::SingleSendBuffer::OPENDDS_MAP	(	SequenceNumber	,
		BufferMap
	)

private

◆ OPENDDS_MAP() [3/3]

typedef OpenDDS::DCPS::SingleSendBuffer::OPENDDS_MAP	(	SequenceNumber	,
		GUID_t
	)

private

◆ OPENDDS_SET()

typedef OpenDDS::DCPS::SingleSendBuffer::OPENDDS_SET ( SequenceNumber )

private

◆ OPENDDS_VECTOR()

typedef OpenDDS::DCPS::SingleSendBuffer::OPENDDS_VECTOR ( BufferType )

Referenced by resend_fragments_i().

◆ pre_clear()

void OpenDDS::DCPS::SingleSendBuffer::pre_clear ( )

inline

Definition at line 199 of file TransportSendBuffer.h.

   {
     pre_seq_.clear();
   }

◆ pre_insert()

ACE_INLINE void OpenDDS::DCPS::SingleSendBuffer::pre_insert ( SequenceNumber sequence )

Definition at line 36 of file TransportSendBuffer.inl.

References ACE_GUARD, and OPENDDS_END_VERSIONED_NAMESPACE_DECL.

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   pre_seq_.insert(sequence);
 }

◆ release_acked()

void OpenDDS::DCPS::SingleSendBuffer::release_acked ( SequenceNumber seq )

Definition at line 80 of file TransportSendBuffer.cpp.

References ACE_GUARD, buffers_, minimum_sn_allowed_, mutex_, and release_i().

                                                   {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   BufferMap::iterator buffer_iter = buffers_.find(seq);
   if (buffer_iter != buffers_.end()) {
     release_i(buffer_iter);
   }
   minimum_sn_allowed_ = std::max(minimum_sn_allowed_, seq + 1);
 }

◆ release_all()

void OpenDDS::DCPS::SingleSendBuffer::release_all ( )

Definition at line 70 of file TransportSendBuffer.cpp.

References ACE_GUARD, buffers_, mutex_, and release_i().

Referenced by ~SingleSendBuffer().

 {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   for (BufferMap::iterator it = buffers_.begin();
        it != buffers_.end();) {
     release_i(it++);
   }
 }

◆ release_i()

void OpenDDS::DCPS::SingleSendBuffer::release_i ( BufferMap::iterator buffer_iter )

private

Definition at line 100 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_TEXT(), buffers_, destinations_, fragments_, LM_DEBUG, and OpenDDS::DCPS::Transport_debug_level.

Referenced by release_acked(), release_all(), and retain_all().

 {
   BufferType& buffer(buffer_iter->second);
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG,
       ACE_TEXT("(%P|%t) SingleSendBuffer::release() - ")
       ACE_TEXT("releasing buffer at: (0x%@,0x%@)\n"),
       buffer.first, buffer.second
     ));
   }
 
   if (buffer.first && buffer.second) {
     // not a fragment
     RemoveAllVisitor visitor;
     buffer.first->accept_remove_visitor(visitor);
     delete buffer.first;
 
     Message_Block_Ptr to_release(buffer.second);
     buffer.second = 0;
 
   } else {
     // data actually stored in fragments_
     const FragmentMap::iterator fm_it = fragments_.find(buffer_iter->first);
     if (fm_it != fragments_.end()) {
       for (BufferMap::iterator bm_it = fm_it->second.begin();
            bm_it != fm_it->second.end(); ++bm_it) {
         RemoveAllVisitor visitor;
         bm_it->second.first->accept_remove_visitor(visitor);
         delete bm_it->second.first;
 
         Message_Block_Ptr to_release(bm_it->second.second);
         bm_it->second.second = 0;
       }
       fragments_.erase(fm_it);
     }
   }
 
   destinations_.erase(buffer_iter->first);
   buffers_.erase(buffer_iter);
 }

◆ remove_acked()

void OpenDDS::DCPS::SingleSendBuffer::remove_acked	(	SequenceNumber	seq,
		BufferVec &	removed
	)

Definition at line 90 of file TransportSendBuffer.cpp.

References ACE_GUARD, buffers_, minimum_sn_allowed_, mutex_, and remove_i().

                                                                      {
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   BufferMap::iterator buffer_iter = buffers_.find(seq);
   if (buffer_iter != buffers_.end()) {
     remove_i(buffer_iter, removed);
   }
   minimum_sn_allowed_ = std::max(minimum_sn_allowed_, seq + 1);
 }

◆ remove_i()

void OpenDDS::DCPS::SingleSendBuffer::remove_i	(	BufferMap::iterator	buffer_iter,
		BufferVec &	removed
	)

private

Definition at line 142 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_TEXT(), buffers_, destinations_, fragments_, LM_DEBUG, and OpenDDS::DCPS::Transport_debug_level.

Referenced by check_capacity_i(), and remove_acked().

 {
   BufferType& buffer(buffer_iter->second);
   if (Transport_debug_level > 5) {
     ACE_DEBUG((LM_DEBUG,
       ACE_TEXT("(%P|%t) SingleSendBuffer::release() - ")
       ACE_TEXT("releasing buffer at: (0x%@,0x%@)\n"),
       buffer.first, buffer.second
     ));
   }
 
   if (buffer.first && buffer.second) {
     // not a fragment
     removed.push_back(buffer);
   } else {
     // data actually stored in fragments_
     const FragmentMap::iterator fm_it = fragments_.find(buffer_iter->first);
     if (fm_it != fragments_.end()) {
       for (BufferMap::iterator bm_it = fm_it->second.begin();
            bm_it != fm_it->second.end(); ++bm_it) {
         removed.push_back(bm_it->second);
       }
       fragments_.erase(fm_it);
     }
   }
 
   destinations_.erase(buffer_iter->first);
   buffers_.erase(buffer_iter);
 }

◆ resend()

bool OpenDDS::DCPS::SingleSendBuffer::resend	(	const SequenceRange &	range,
		DisjointSequence *	gaps = `0`
	)

Definition at line 379 of file TransportSendBuffer.cpp.

References ACE_GUARD_RETURN, mutex_, resend_i(), and OpenDDS::DCPS::TransportSendBuffer::strategy_lock().

Referenced by OpenDDS::DCPS::ReliableSession::nak_received().

 {
   ACE_GUARD_RETURN(LockType, guard, strategy_lock(), false);
   ACE_GUARD_RETURN(ACE_Thread_Mutex, g, mutex_, false);
   return resend_i(range, gaps);
 }

◆ resend_fragments_i()

void OpenDDS::DCPS::SingleSendBuffer::resend_fragments_i	(	SequenceNumber	sequence,
		const DisjointSequence &	fragments,
		size_t &	cumulative_send_count
	)

private

Definition at line 442 of file TransportSendBuffer.cpp.

References OpenDDS::DCPS::DisjointSequence::empty(), fragments_, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OPENDDS_VECTOR(), and OpenDDS::DCPS::TransportSendBuffer::resend_one().

 {
   if (fragments_.empty() || requested_frags.empty()) {
     return;
   }
   const FragmentMap::const_iterator fm_it = fragments_.find(seq);
   if (fm_it == fragments_.end()) {
     return;
   }
   const BufferMap& buffers = fm_it->second;
   const OPENDDS_VECTOR(SequenceRange)& psr = requested_frags.present_sequence_ranges();
 
   BufferMap::const_iterator it = buffers.lower_bound(psr.front().first);
   BufferMap::const_iterator end = buffers.lower_bound(psr.back().second);
   if (end != buffers.end()) {
     ++end;
   }
 
   SequenceNumber frag_min;
   size_t i = 0;
 
   // Iterate over both containers simultaneously
   while (i < psr.size() && it != end) {
     if (psr[i].second < frag_min) {
       ++i;
     } else {
       // Once the range max is over our fragment minimum, we either
       // expect overlap (resend fragment) or the range is too high (skip fragment)
       // Either way, we will increment the fragment now to avoid duplicate resends
       if (it->first >= psr[i].first) {
         resend_one(it->second); // overlap - resend fragment buffer
         ++cumulative_send_count;
       }
       frag_min = it->first + 1; // increment fragment buffer
       ++it;
     }
   }
 }

◆ resend_i() [1/2]

bool OpenDDS::DCPS::SingleSendBuffer::resend_i	(	const SequenceRange &	range,
		DisjointSequence *	gaps = `0`
	)

private

Definition at line 387 of file TransportSendBuffer.cpp.

References OpenDDS::DCPS::GUID_UNKNOWN.

Referenced by resend().

 {
   return resend_i(range, gaps, GUID_UNKNOWN);
 }

◆ resend_i() [2/2]

bool OpenDDS::DCPS::SingleSendBuffer::resend_i	(	const SequenceRange &	range,
		DisjointSequence *	gaps,
		const GUID_t &	destination
	)

private

Definition at line 393 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_TEXT(), buffers_, destinations_, fragments_, OpenDDS::DCPS::GUID_UNKNOWN, OpenDDS::DCPS::DisjointSequence::insert(), LM_DEBUG, OpenDDS::DCPS::TransportSendBuffer::resend_one(), and OpenDDS::DCPS::Transport_debug_level.

 {
   //Special case, nak to make sure it has all history
   if (buffers_.empty()) throw std::exception();
   const SequenceNumber lowForAllResent = range.first == SequenceNumber() ? buffers_.begin()->first : range.first;
   const bool has_dest = destination != GUID_UNKNOWN;
 
   for (SequenceNumber sequence(range.first);
        sequence <= range.second; ++sequence) {
     // Re-send requested sample if still buffered; missing samples
     // will be scored against the given DisjointSequence:
     BufferMap::iterator it(buffers_.find(sequence));
     DestinationMap::iterator dest_data;
     if (has_dest) {
       dest_data = destinations_.find(sequence);
     }
     if (it == buffers_.end() || (has_dest && (dest_data == destinations_.end() ||
                                               dest_data->second != destination))) {
       if (gaps) {
         gaps->insert(sequence);
       }
     } else {
       if (Transport_debug_level > 5) {
         ACE_DEBUG((LM_DEBUG,
                    ACE_TEXT("(%P|%t) SingleSendBuffer::resend() - ")
                    ACE_TEXT("resending PDU: %q, (0x%@,0x%@)\n"),
                    sequence.getValue(),
                    it->second.first,
                    it->second.second));
       }
       if (it->second.first && it->second.second) {
         resend_one(it->second);
       } else {
         const FragmentMap::iterator fm_it = fragments_.find(it->first);
         if (fm_it != fragments_.end()) {
           for (BufferMap::iterator bm_it = fm_it->second.begin();
                 bm_it != fm_it->second.end(); ++bm_it) {
             resend_one(bm_it->second);
           }
         }
       }
     }
   }
   // Have we resent all requested data?
   return lowForAllResent >= buffers_.begin()->first && range.second <= buffers_.rbegin()->first;
 }

◆ retain_all()

void OpenDDS::DCPS::SingleSendBuffer::retain_all ( const GUID_t & pub_id )

virtual

Reimplemented from OpenDDS::DCPS::TransportSendBuffer.

Definition at line 173 of file TransportSendBuffer.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD, ACE_TEXT(), buffers_, OpenDDS::DCPS::LogGuid::c_str(), fragments_, LM_DEBUG, LM_WARNING, mutex_, release_i(), OpenDDS::DCPS::REMOVE_ERROR, retain_buffer(), and OpenDDS::DCPS::Transport_debug_level.

 {
   if (Transport_debug_level > 5) {
     LogGuid logger(pub_id);
     ACE_DEBUG((LM_DEBUG,
       ACE_TEXT("(%P|%t) SingleSendBuffer::retain_all() - ")
       ACE_TEXT("copying out blocks for publication: %C\n"),
       logger.c_str()
     ));
   }
   ACE_GUARD(ACE_Thread_Mutex, g, mutex_);
   for (BufferMap::iterator it(buffers_.begin());
        it != buffers_.end();) {
     if (it->second.first && it->second.second) {
       if (retain_buffer(pub_id, it->second) == REMOVE_ERROR) {
         LogGuid logger(pub_id);
         ACE_ERROR((LM_WARNING,
                    ACE_TEXT("(%P|%t) WARNING: ")
                    ACE_TEXT("SingleSendBuffer::retain_all: ")
                    ACE_TEXT("failed to retain data from publication: %C!\n"),
                    logger.c_str()));
         release_i(it++);
       } else {
         ++it;
       }
 
     } else {
       const FragmentMap::iterator fm_it = fragments_.find(it->first);
       if (fm_it != fragments_.end()) {
         for (BufferMap::iterator bm_it = fm_it->second.begin();
              bm_it != fm_it->second.end();) {
           if (retain_buffer(pub_id, bm_it->second) == REMOVE_ERROR) {
             LogGuid logger(pub_id);
             ACE_ERROR((LM_WARNING,
                        ACE_TEXT("(%P|%t) WARNING: ")
                        ACE_TEXT("SingleSendBuffer::retain_all: failed to ")
                        ACE_TEXT("retain fragment data from publication: %C!\n"),
                        logger.c_str()));
             release_i(bm_it++);
           } else {
             ++bm_it;
           }
         }
       }
       ++it;
     }
   }
 }

◆ retain_buffer()

RemoveResult OpenDDS::DCPS::SingleSendBuffer::retain_buffer	(	const GUID_t &	pub_id,
		BufferType &	buffer
	)

private

Definition at line 223 of file TransportSendBuffer.cpp.

References OpenDDS::DCPS::TransportQueueElement::clone_mb(), ACE_Message_Block::release(), OpenDDS::DCPS::REMOVE_ERROR, replaced_db_allocator_, replaced_mb_allocator_, retained_db_allocator_, retained_mb_allocator_, and OpenDDS::DCPS::PacketRemoveVisitor::status().

Referenced by retain_all().

 {
   TransportQueueElement::MatchOnPubId match(pub_id);
   PacketRemoveVisitor visitor(match,
                               buffer.second,
                               buffer.second,
                               replaced_mb_allocator_,
                               replaced_db_allocator_);
 
   buffer.first->accept_replace_visitor(visitor);
   if (visitor.status() != REMOVE_ERROR) {
     // Copy sample's message/data block descriptors:
     ACE_Message_Block* data = buffer.second;
     buffer.second = TransportQueueElement::clone_mb(data,
                                            &retained_mb_allocator_,
                                            &retained_db_allocator_);
     data->release();
   }
   return visitor.status();
 }

Member Data Documentation

◆ buffers_

BufferMap OpenDDS::DCPS::SingleSendBuffer::buffers_

private

Definition at line 231 of file TransportSendBuffer.h.

Referenced by check_capacity_i(), insert(), insert_fragment(), release_acked(), release_all(), release_i(), remove_acked(), remove_i(), resend_i(), and retain_all().

◆ destinations_

DestinationMap OpenDDS::DCPS::SingleSendBuffer::destinations_

private

Definition at line 237 of file TransportSendBuffer.h.

Referenced by insert(), release_i(), remove_i(), and resend_i().

◆ fragments_

FragmentMap OpenDDS::DCPS::SingleSendBuffer::fragments_

private

Definition at line 234 of file TransportSendBuffer.h.

Referenced by has_frags(), insert_fragment(), release_i(), remove_i(), resend_fragments_i(), resend_i(), and retain_all().

◆ minimum_sn_allowed_

SequenceNumber OpenDDS::DCPS::SingleSendBuffer::minimum_sn_allowed_

private

Definition at line 242 of file TransportSendBuffer.h.

Referenced by insert(), insert_fragment(), release_acked(), and remove_acked().

◆ mutex_

ACE_Thread_Mutex OpenDDS::DCPS::SingleSendBuffer::mutex_

mutableprivate

Definition at line 244 of file TransportSendBuffer.h.

Referenced by insert(), insert_fragment(), release_acked(), release_all(), remove_acked(), resend(), and retain_all().

◆ n_chunks_

size_t OpenDDS::DCPS::SingleSendBuffer::n_chunks_

private

Definition at line 224 of file TransportSendBuffer.h.

◆ pre_seq_

SequenceNumberSet OpenDDS::DCPS::SingleSendBuffer::pre_seq_

private

Definition at line 240 of file TransportSendBuffer.h.

Referenced by insert(), and insert_fragment().

◆ replaced_db_allocator_

DataBlockAllocator OpenDDS::DCPS::SingleSendBuffer::replaced_db_allocator_

private

Definition at line 229 of file TransportSendBuffer.h.

Referenced by retain_buffer().

◆ replaced_mb_allocator_

MessageBlockAllocator OpenDDS::DCPS::SingleSendBuffer::replaced_mb_allocator_

private

Definition at line 228 of file TransportSendBuffer.h.

Referenced by retain_buffer().

◆ retained_db_allocator_

DataBlockAllocator OpenDDS::DCPS::SingleSendBuffer::retained_db_allocator_

private

Definition at line 227 of file TransportSendBuffer.h.

Referenced by insert_buffer(), and retain_buffer().

◆ retained_mb_allocator_

MessageBlockAllocator OpenDDS::DCPS::SingleSendBuffer::retained_mb_allocator_

private

Definition at line 226 of file TransportSendBuffer.h.

Referenced by insert_buffer(), and retain_buffer().

◆ UNLIMITED

const size_t OpenDDS::DCPS::SingleSendBuffer::UNLIMITED = 0

static

Definition at line 76 of file TransportSendBuffer.h.

Referenced by OpenDDS::DCPS::TransportSendBuffer::resend_one().

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

Classes

Public Member Functions

Static Public Attributes

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ SingleSendBuffer()

◆ ~SingleSendBuffer()

Member Function Documentation

◆ check_capacity_i()

◆ has_frags()

◆ insert()

◆ insert_buffer()

◆ insert_fragment()

◆ n_chunks()

◆ OPENDDS_MAP() [1/3]

◆ OPENDDS_MAP() [2/3]

◆ OPENDDS_MAP() [3/3]

◆ OPENDDS_SET()

◆ OPENDDS_VECTOR()

◆ pre_clear()

◆ pre_insert()

◆ release_acked()

◆ release_all()

◆ release_i()

◆ remove_acked()

◆ remove_i()

◆ resend()

◆ resend_fragments_i()

◆ resend_i() [1/2]

◆ resend_i() [2/2]

◆ retain_all()

◆ retain_buffer()

Member Data Documentation

◆ buffers_

◆ destinations_

◆ fragments_

◆ minimum_sn_allowed_

◆ mutex_

◆ n_chunks_

◆ pre_seq_

◆ replaced_db_allocator_

◆ replaced_mb_allocator_

◆ retained_db_allocator_

◆ retained_mb_allocator_

◆ UNLIMITED