#include <MulticastSession.h>

Inheritance diagram for OpenDDS::DCPS::MulticastSession:

[legend]

Collaboration diagram for OpenDDS::DCPS::MulticastSession:

[legend]

Public Member Functions
virtual	~MulticastSession ()

MulticastDataLink *	link ()

MulticastPeer	remote_peer () const

bool	acked ()

void	set_acked ()

virtual bool	is_reliable ()

void	syn_received (const Message_Block_Ptr &control)

void	send_all_syn (const MonotonicTimePoint &now)

void	send_syn (const GUID_t &local_writer, const GUID_t &remote_reader)

void	synack_received (const Message_Block_Ptr &control)

void	send_synack (const GUID_t &local_reader, const GUID_t &remote_writer)

virtual void	send_naks ()

virtual bool	check_header (const TransportHeader &header)=0

virtual void	record_header_received (const TransportHeader &header)=0

virtual bool	ready_to_deliver (const TransportHeader &header, const ReceivedDataSample &data)=0

virtual void	release_remote (const GUID_t &)

virtual bool	control_received (char submessage_id, const Message_Block_Ptr &control)

virtual bool	start (bool active, bool acked)=0

virtual void	stop ()

bool	reassemble (ReceivedDataSample &data, const TransportHeader &header)

void	add_remote (const GUID_t &local)

void	add_remote (const GUID_t &local, const GUID_t &remote)

void	remove_remote (const GUID_t &local, const GUID_t &remote)

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

Protected Types
typedef ACE_Reverse_Lock< ACE_Thread_Mutex >	Reverse_Lock_t

Protected Member Functions
	MulticastSession (RcHandle< ReactorInterceptor > interceptor, MulticastDataLink *link, MulticastPeer remote_peer)

void	send_control (char submessage_id, Message_Block_Ptr data)

void	start_syn ()

virtual void	syn_hook (const SequenceNumber &)

typedef	OPENDDS_MAP_CMP (GUID_t, RepoIdSet, GUID_tKeyLessThan) PendingRemoteMap

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

Protected Attributes
MulticastDataLink *	link_

MulticastPeer	remote_peer_

ACE_Thread_Mutex	start_lock_

Reverse_Lock_t	reverse_start_lock_

bool	started_

bool	active_

TransportReassembly	reassembly_

bool	acked_

PendingRemoteMap	pending_remote_map_

Private Types
typedef PmfSporadicTask< MulticastSession >	Sporadic

Private Member Functions
void	remove_remote_i (const GUID_t &local, const GUID_t &remote)

Private Attributes
ACE_Thread_Mutex	ack_lock_

RcHandle< Sporadic >	syn_watchdog_

TimeDuration	syn_delay_

const TimeDuration	initial_syn_delay_

String	config_name

Detailed Description

Definition at line 34 of file MulticastSession.h.

Member Typedef Documentation

◆ Reverse_Lock_t

typedef ACE_Reverse_Lock<ACE_Thread_Mutex> OpenDDS::DCPS::MulticastSession::Reverse_Lock_t

protected

Definition at line 96 of file MulticastSession.h.

◆ Sporadic

typedef PmfSporadicTask<MulticastSession> OpenDDS::DCPS::MulticastSession::Sporadic

private

Definition at line 125 of file MulticastSession.h.

Constructor & Destructor Documentation

◆ ~MulticastSession()

OpenDDS::DCPS::MulticastSession::~MulticastSession ( )

virtual

Definition at line 49 of file MulticastSession.cpp.

References syn_watchdog_.

 {
   syn_watchdog_->cancel();
 }

◆ MulticastSession()

OpenDDS::DCPS::MulticastSession::MulticastSession	(	RcHandle< ReactorInterceptor >	interceptor,
		MulticastDataLink *	link,
		MulticastPeer	remote_peer
	)

protected

Definition at line 31 of file MulticastSession.cpp.

   : link_(link)
   , remote_peer_(remote_peer)
   , reverse_start_lock_(start_lock_)
   , started_(false)
   , active_(true)
   , reassembly_(link->config()->fragment_reassembly_timeout_)
   , acked_(false)
   , syn_watchdog_(make_rch<Sporadic>(TheServiceParticipant->time_source(),
                                      interceptor,
                                      rchandle_from(this),
                                      &MulticastSession::send_all_syn))
   , initial_syn_delay_(link->config()->syn_interval_)
   , config_name(link->config()->name())
 {}

Member Function Documentation

◆ acked()

bool OpenDDS::DCPS::MulticastSession::acked ( )

Definition at line 55 of file MulticastSession.cpp.

References ACE_GUARD_RETURN, ACE_SYNCH_MUTEX, ack_lock_, and acked_.

Referenced by OpenDDS::DCPS::ReliableSession::ready_to_deliver(), and OpenDDS::DCPS::ReliableSession::send_naks().

 {
   ACE_GUARD_RETURN(ACE_SYNCH_MUTEX, guard, this->ack_lock_, false);
   return this->acked_;
 }

◆ add_remote() [1/2]

void OpenDDS::DCPS::MulticastSession::add_remote ( const GUID_t & local )

Definition at line 335 of file MulticastSession.cpp.

References OpenDDS::DCPS::GuidConverter::isWriter(), and start_syn().

 {
   const GuidConverter conv(local);
   if (conv.isWriter()) {
     // Active peers schedule a watchdog timer to initiate a 2-way
     // handshake to verify that passive endpoints can send/receive
     // data reliably. This process must be executed using the
     // transport reactor thread to prevent blocking.
     // Only publisher send syn so just schedule for pub role.
     this->start_syn();
   }
 }

◆ add_remote() [2/2]

void OpenDDS::DCPS::MulticastSession::add_remote	(	const GUID_t &	local,
		const GUID_t &	remote
	)

Definition at line 349 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, OpenDDS::DCPS::GuidConverter::isWriter(), pending_remote_map_, and start_syn().

 {
   const GuidConverter conv(local);
 
   {
     ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
     pending_remote_map_[local].insert(remote);
   }
 
   if (conv.isWriter()) {
     // Active peers schedule a watchdog timer to initiate a 2-way
     // handshake to verify that passive endpoints can send/receive
     // data reliably. This process must be executed using the
     // transport reactor thread to prevent blocking.
     // Only publisher send syn so just schedule for pub role.
     this->start_syn();
   }
 }

◆ check_header()

virtual bool OpenDDS::DCPS::MulticastSession::check_header ( const TransportHeader & header )

pure virtual

Implemented in OpenDDS::DCPS::ReliableSession, and OpenDDS::DCPS::BestEffortSession.

◆ control_received()

bool OpenDDS::DCPS::MulticastSession::control_received	(	char	submessage_id,
		const Message_Block_Ptr &	control
	)

virtual

Reimplemented in OpenDDS::DCPS::ReliableSession.

Definition at line 102 of file MulticastSession.cpp.

References OpenDDS::DCPS::MULTICAST_SYN, OpenDDS::DCPS::MULTICAST_SYNACK, syn_received(), and synack_received().

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

 {
   switch (submessage_id) {
   case MULTICAST_SYN:
     syn_received(control);
     break;
 
   case MULTICAST_SYNACK:
     synack_received(control);
     break;
 
   default:
     return false;
   }
 
   return true;
 }

◆ is_reliable()

virtual bool OpenDDS::DCPS::MulticastSession::is_reliable ( )

inlinevirtual

Reimplemented in OpenDDS::DCPS::ReliableSession, and OpenDDS::DCPS::BestEffortSession.

Definition at line 45 of file MulticastSession.h.

45 { return false;}

◆ link()

ACE_INLINE MulticastDataLink * OpenDDS::DCPS::MulticastSession::link ( void )

Definition at line 14 of file MulticastSession.inl.

References ACE_INLINE, and link_.

Referenced by OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), OpenDDS::DCPS::ReliableSession::send_naks(), send_syn(), send_synack(), syn_received(), and synack_received().

 {
   return this->link_;
 }

◆ OPENDDS_MAP_CMP()

typedef OpenDDS::DCPS::MulticastSession::OPENDDS_MAP_CMP	(	GUID_t	,
		RepoIdSet	,
		GUID_tKeyLessThan
	)

protected

◆ ready_to_deliver()

virtual bool OpenDDS::DCPS::MulticastSession::ready_to_deliver	(	const TransportHeader &	header,
		const ReceivedDataSample &	data
	)

pure virtual

Implemented in OpenDDS::DCPS::ReliableSession, and OpenDDS::DCPS::BestEffortSession.

◆ reassemble()

bool OpenDDS::DCPS::MulticastSession::reassemble	(	ReceivedDataSample &	data,
		const TransportHeader &	header
	)

Definition at line 326 of file MulticastSession.cpp.

References OpenDDS::DCPS::TransportHeader::first_fragment_, OpenDDS::DCPS::TransportReassembly::reassemble(), reassembly_, and OpenDDS::DCPS::TransportHeader::sequence_.

 {
   return this->reassembly_.reassemble(header.sequence_,
                                       header.first_fragment_,
                                       data);
 }

◆ record_header_received()

virtual void OpenDDS::DCPS::MulticastSession::record_header_received ( const TransportHeader & header )

pure virtual

Implemented in OpenDDS::DCPS::ReliableSession, and OpenDDS::DCPS::BestEffortSession.

◆ release_remote()

virtual void OpenDDS::DCPS::MulticastSession::release_remote ( const GUID_t & )

inlinevirtual

Reimplemented in OpenDDS::DCPS::ReliableSession.

Definition at line 60 of file MulticastSession.h.

60 {};

◆ remote_peer()

ACE_INLINE MulticastPeer OpenDDS::DCPS::MulticastSession::remote_peer ( ) const

Definition at line 20 of file MulticastSession.inl.

References OPENDDS_END_VERSIONED_NAMESPACE_DECL, and remote_peer_.

 {
   return this->remote_peer_;
 }

◆ remove_remote()

void OpenDDS::DCPS::MulticastSession::remove_remote	(	const GUID_t &	local,
		const GUID_t &	remote
	)

Definition at line 370 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, and remove_remote_i().

 {
   ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
   remove_remote_i(local, remote);
 }

◆ remove_remote_i()

void OpenDDS::DCPS::MulticastSession::remove_remote_i	(	const GUID_t &	local,
		const GUID_t &	remote
	)

private

Definition at line 378 of file MulticastSession.cpp.

References OpenDDS::DCPS::GuidConverter::isWriter(), OPENDDS_END_VERSIONED_NAMESPACE_DECL, pending_remote_map_, and syn_watchdog_.

Referenced by remove_remote(), and synack_received().

 {
   const GuidConverter conv(local);
 
   const bool empty_before = pending_remote_map_.empty();
   pending_remote_map_[local].erase(remote);
   if (pending_remote_map_[local].empty()) {
     pending_remote_map_.erase(local);
   }
   const bool empty = pending_remote_map_.empty() && !empty_before;
 
   if (conv.isWriter() && empty && this->syn_watchdog_) {
     this->syn_watchdog_->cancel();
   }
 }

◆ send_all_syn()

void OpenDDS::DCPS::MulticastSession::send_all_syn ( const MonotonicTimePoint & now )

Definition at line 192 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, pending_remote_map_, send_syn(), syn_delay_, and syn_watchdog_.

 {
   ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
   for (PendingRemoteMap::const_iterator pos1 = pending_remote_map_.begin(), limit = pending_remote_map_.end();
        pos1 != limit; ++pos1) {
     const GUID_t& local_writer = pos1->first;
     for (RepoIdSet::const_iterator pos2 = pos1->second.begin(), limit = pos1->second.end(); pos2 != limit; ++pos2) {
       const GUID_t& remote_reader = *pos2;
       send_syn(local_writer, remote_reader);
     }
   }
 
   // Exponential back-off.
   syn_delay_ *= 2;
   syn_watchdog_->schedule(syn_delay_);
 }

◆ send_control()

void OpenDDS::DCPS::MulticastSession::send_control	(	char	submessage_id,
		Message_Block_Ptr	data
	)

protected

Definition at line 77 of file MulticastSession.cpp.

References ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::DataLink::create_control(), header, link_, LM_ERROR, OpenDDS::DCPS::move(), OpenDDS::DCPS::DataLink::send_control(), and OpenDDS::DCPS::SEND_CONTROL_OK.

Referenced by OpenDDS::DCPS::ReliableSession::send_nakack(), OpenDDS::DCPS::ReliableSession::send_naks(), send_syn(), and send_synack().

 {
   DataSampleHeader header;
   Message_Block_Ptr control(
     this->link_->create_control(submessage_id, header, move(data)));
   if (!control) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("MulticastSession::send_control: ")
                ACE_TEXT("create_control failed!\n")));
     return;
   }
 
   int error = this->link_->send_control(header, move(control));
   if (error != SEND_CONTROL_OK) {
     ACE_ERROR((LM_ERROR,
                ACE_TEXT("(%P|%t) ERROR: ")
                ACE_TEXT("MulticastSession::send_control: ")
                ACE_TEXT("send_control failed: %d!\n"),
                error));
     return;
   }
 }

◆ send_naks()

virtual void OpenDDS::DCPS::MulticastSession::send_naks ( )

inlinevirtual

Reimplemented in OpenDDS::DCPS::ReliableSession.

Definition at line 54 of file MulticastSession.h.

References header.

Referenced by send_synack().

54 {}

◆ send_syn()

void OpenDDS::DCPS::MulticastSession::send_syn	(	const GUID_t &	local_writer,
		const GUID_t &	remote_reader
	)

Definition at line 210 of file MulticastSession.cpp.

References OpenDDS::DCPS::LogGuid::c_str(), config_name, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), link(), LM_DEBUG, OpenDDS::DCPS::MulticastDataLink::local_peer(), OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_SYN, remote_peer_, send_control(), and VDBG_LVL.

Referenced by send_all_syn().

 {
   size_t len = sizeof(this->remote_peer_) + 2 * sizeof(GUID_t);
 
   Message_Block_Ptr data( new ACE_Message_Block(len));
 
   Serializer serializer(data.get(), encoding_kind);
 
   serializer << this->remote_peer_;
   serializer.write_octet_array(reinterpret_cast<const ACE_CDR::Octet*>(&local_writer), sizeof(local_writer));
   serializer.write_octet_array(reinterpret_cast<const ACE_CDR::Octet*>(&remote_reader), sizeof(remote_reader));
 
   VDBG_LVL((LM_DEBUG,
             "(%P|%t) MulticastSession[%C]::send_syn "
             "local %#08x%08x %C remote %#08x%08x %C\n",
             config_name.c_str(),
             (unsigned int)(this->link()->local_peer() >> 32),
             (unsigned int) this->link()->local_peer(),
             LogGuid(local_writer).c_str(),
             (unsigned int)(this->remote_peer_ >> 32),
             (unsigned int) this->remote_peer_,
             LogGuid(remote_reader).c_str()),
            2);
 
   // Send control sample to remote peer:
   send_control(MULTICAST_SYN, move(data));
 }

◆ send_synack()

void OpenDDS::DCPS::MulticastSession::send_synack	(	const GUID_t &	local_reader,
		const GUID_t &	remote_writer
	)

Definition at line 287 of file MulticastSession.cpp.

References active_, OpenDDS::DCPS::LogGuid::c_str(), config_name, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), link(), LM_DEBUG, OpenDDS::DCPS::MulticastDataLink::local_peer(), OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_SYNACK, remote_peer_, send_control(), send_naks(), and VDBG_LVL.

Referenced by syn_received().

 {
   size_t len = sizeof(this->remote_peer_) + 2 * sizeof(GUID_t);
 
   Message_Block_Ptr data(new ACE_Message_Block(len));
 
   Serializer serializer(data.get(), encoding_kind);
 
   serializer << this->remote_peer_;
   serializer.write_octet_array(reinterpret_cast<const ACE_CDR::Octet*>(&local_reader), sizeof(local_reader));
   serializer.write_octet_array(reinterpret_cast<const ACE_CDR::Octet*>(&remote_writer), sizeof(remote_writer));
 
   VDBG_LVL((LM_DEBUG, "(%P|%t) MulticastSession[%C]::send_synack "
                       "local %#08x%08x %C remote %#08x%08x %C active %d\n",
                       config_name.c_str(),
                       (unsigned int)(this->link()->local_peer() >> 32),
                       (unsigned int) this->link()->local_peer(),
                       LogGuid(local_reader).c_str(),
                       (unsigned int)(this->remote_peer_ >> 32),
                       (unsigned int) this->remote_peer_,
                       LogGuid(remote_writer).c_str(),
                       this->active_ ? 1 : 0), 2);
 
   // Send control sample to remote peer:
   send_control(MULTICAST_SYNACK, move(data));
 
   // Send naks before sending synack to
   // reduce wait time for resends from remote.
   send_naks();
 }

◆ set_acked()

void OpenDDS::DCPS::MulticastSession::set_acked ( )

Definition at line 62 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, and acked_.

Referenced by OpenDDS::DCPS::BestEffortSession::start(), and OpenDDS::DCPS::ReliableSession::start().

 {
   ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
   this->acked_ = true;
 }

◆ start()

virtual bool OpenDDS::DCPS::MulticastSession::start	(	bool	active,
		bool	acked
	)

pure virtual

Implemented in OpenDDS::DCPS::ReliableSession, and OpenDDS::DCPS::BestEffortSession.

◆ start_syn()

void OpenDDS::DCPS::MulticastSession::start_syn ( )

protected

Definition at line 69 of file MulticastSession.cpp.

References initial_syn_delay_, syn_delay_, and syn_watchdog_.

Referenced by add_remote().

 {
   syn_watchdog_->cancel();
   syn_delay_ = initial_syn_delay_;
   syn_watchdog_->schedule(TimeDuration(0));
 }

◆ stop()

void OpenDDS::DCPS::MulticastSession::stop ( void )

virtual

Reimplemented in OpenDDS::DCPS::ReliableSession.

Definition at line 320 of file MulticastSession.cpp.

References syn_watchdog_.

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

 {
   this->syn_watchdog_->cancel();
 }

◆ syn_hook()

virtual void OpenDDS::DCPS::MulticastSession::syn_hook ( const SequenceNumber & )

inlineprotectedvirtual

Reimplemented in OpenDDS::DCPS::ReliableSession.

Definition at line 93 of file MulticastSession.h.

Referenced by syn_received().

93 {}

◆ syn_received()

void OpenDDS::DCPS::MulticastSession::syn_received ( const Message_Block_Ptr & control )

Definition at line 122 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, acked_, active_, OpenDDS::DCPS::LogGuid::c_str(), config_name, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), header, link(), link_, LM_DEBUG, OpenDDS::DCPS::MulticastDataLink::local_peer(), pending_remote_map_, OpenDDS::DCPS::MulticastDataLink::receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), remote_peer_, send_synack(), OpenDDS::DCPS::TransportHeader::sequence_, OpenDDS::DCPS::TransportHeader::source_, OpenDDS::DCPS::TransportHeader::swap_bytes(), syn_hook(), OpenDDS::DCPS::MulticastDataLink::transport(), and VDBG_LVL.

Referenced by control_received().

 {
   if (this->active_) return; // pub send syn, then doesn't receive them.
 
   const TransportHeader& header =
     this->link_->receive_strategy()->received_header();
 
   // Not from the remote peer for this session.
   if (this->remote_peer_ != header.source_) return;
 
   Serializer serializer(control.get(), encoding_kind, header.swap_bytes());
 
   MulticastPeer local_peer;
   GUID_t remote_writer;
   GUID_t local_reader;
   serializer >> local_peer; // sent as remote_peer
   serializer.read_octet_array(reinterpret_cast<ACE_CDR::Octet*>(&remote_writer), sizeof(remote_writer));
   serializer.read_octet_array(reinterpret_cast<ACE_CDR::Octet*>(&local_reader), sizeof(local_reader));
 
   // Ignore sample if not destined for us:
   if (local_peer != this->link_->local_peer()) return;
 
   bool call_passive_connection = false;
   bool call_send_synack = true;
   {
     ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
     PendingRemoteMap::const_iterator pos1 = pending_remote_map_.find(local_reader);
     if (pos1 == pending_remote_map_.end()) {
       call_send_synack = false;
     } else {
       RepoIdSet::const_iterator pos2 = pos1->second.find(remote_writer);
       if (pos2 == pos1->second.end()) {
         call_send_synack = false;
       }
     }
 
     VDBG_LVL((LM_DEBUG,
               "(%P|%t) MulticastSession[%C]::syn_received "
               "local %#08x%08x %C remote %#08x%08x %C\n",
               config_name.c_str(),
               (unsigned int)(this->link()->local_peer() >> 32),
               (unsigned int) this->link()->local_peer(),
               LogGuid(local_reader).c_str(),
               (unsigned int)(this->remote_peer_ >> 32),
               (unsigned int) this->remote_peer_,
               LogGuid(remote_writer).c_str()),
              2);
 
     if (!this->acked_) {
       this->acked_ = true;
       syn_hook(header.sequence_);
       call_passive_connection = true;
     }
   }
 
   if (call_passive_connection) {
     MulticastTransport_rch transport = link_->transport();
     if (transport) {
       transport->passive_connection(link_->local_peer(), remote_peer_);
     }
   }
 
   // MULTICAST_SYN control samples are always positively
   // acknowledged by a matching remote peer:
   if (call_send_synack) {
     send_synack(local_reader, remote_writer);
   }
 }

◆ synack_received()

void OpenDDS::DCPS::MulticastSession::synack_received ( const Message_Block_Ptr & control )

Definition at line 240 of file MulticastSession.cpp.

References ACE_GUARD, ACE_SYNCH_MUTEX, ack_lock_, acked_, active_, OpenDDS::DCPS::LogGuid::c_str(), config_name, OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), header, OpenDDS::DCPS::DataLink::invoke_on_start_callbacks(), link(), link_, LM_DEBUG, OpenDDS::DCPS::MulticastDataLink::local_peer(), OpenDDS::DCPS::MulticastDataLink::receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), remote_peer_, remove_remote_i(), OpenDDS::DCPS::TransportHeader::source_, OpenDDS::DCPS::TransportHeader::swap_bytes(), and VDBG_LVL.

Referenced by control_received().

 {
   if (!this->active_) return; // sub send synack, then doesn't receive them.
 
   // Already received ack.
   //if (this->acked()) return;
 
   const TransportHeader& header =
     this->link_->receive_strategy()->received_header();
 
   // Not from the remote peer for this session.
   if (this->remote_peer_ != header.source_) return;
 
   Serializer serializer(control.get(), encoding_kind, header.swap_bytes());
 
   MulticastPeer local_peer;
   GUID_t remote_reader;
   GUID_t local_writer;
   serializer >> local_peer; // sent as remote_peer
   serializer.read_octet_array(reinterpret_cast<ACE_CDR::Octet*>(&remote_reader), sizeof(remote_reader));
   serializer.read_octet_array(reinterpret_cast<ACE_CDR::Octet*>(&local_writer), sizeof(local_writer));
 
   // Ignore sample if not destined for us:
   if (local_peer != this->link_->local_peer()) return;
 
   VDBG_LVL((LM_DEBUG,
             "(%P|%t) MulticastSession[%C]::synack_received "
             "local %#08x%08x %C remote %#08x%08x %C\n",
             config_name.c_str(),
             (unsigned int)(this->link()->local_peer() >> 32),
             (unsigned int) this->link()->local_peer(),
             LogGuid(local_writer).c_str(),
             (unsigned int)(this->remote_peer_ >> 32),
             (unsigned int) this->remote_peer_,
             LogGuid(remote_reader).c_str()),
            2);
 
   {
     ACE_GUARD(ACE_SYNCH_MUTEX, guard, this->ack_lock_);
     this->acked_ = true;
     remove_remote_i(local_writer, remote_reader);
   }
 
   this->link_->invoke_on_start_callbacks(local_writer, remote_reader, true);
 }

Member Data Documentation

◆ ack_lock_

ACE_Thread_Mutex OpenDDS::DCPS::MulticastSession::ack_lock_

private

Definition at line 123 of file MulticastSession.h.

Referenced by acked(), add_remote(), remove_remote(), send_all_syn(), set_acked(), syn_received(), and synack_received().

◆ acked_

bool OpenDDS::DCPS::MulticastSession::acked_

protected

Definition at line 110 of file MulticastSession.h.

Referenced by acked(), set_acked(), syn_received(), and synack_received().

◆ active_

bool OpenDDS::DCPS::MulticastSession::active_

protected

Definition at line 106 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::ReliableSession::check_header(), OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), OpenDDS::DCPS::ReliableSession::record_header_received(), send_synack(), OpenDDS::DCPS::BestEffortSession::start(), OpenDDS::DCPS::ReliableSession::start(), syn_received(), and synack_received().

◆ config_name

String OpenDDS::DCPS::MulticastSession::config_name

private

Definition at line 129 of file MulticastSession.h.

Referenced by send_syn(), send_synack(), syn_received(), and synack_received().

◆ initial_syn_delay_

const TimeDuration OpenDDS::DCPS::MulticastSession::initial_syn_delay_

private

Definition at line 128 of file MulticastSession.h.

Referenced by start_syn().

◆ link_

MulticastDataLink* OpenDDS::DCPS::MulticastSession::link_

protected

Definition at line 80 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::ReliableSession::deliver_held_data(), OpenDDS::DCPS::ReliableSession::expire_naks(), link(), OpenDDS::DCPS::ReliableSession::nak_delay(), OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), send_control(), OpenDDS::DCPS::ReliableSession::send_naks(), syn_received(), and synack_received().

◆ pending_remote_map_

PendingRemoteMap OpenDDS::DCPS::MulticastSession::pending_remote_map_

protected

Definition at line 116 of file MulticastSession.h.

Referenced by add_remote(), remove_remote_i(), send_all_syn(), and syn_received().

◆ reassembly_

TransportReassembly OpenDDS::DCPS::MulticastSession::reassembly_

protected

Definition at line 108 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::BestEffortSession::check_header(), OpenDDS::DCPS::ReliableSession::expire_naks(), OpenDDS::DCPS::ReliableSession::nakack_received(), and reassemble().

◆ remote_peer_

MulticastPeer OpenDDS::DCPS::MulticastSession::remote_peer_

protected

Definition at line 82 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::ReliableSession::check_header(), OpenDDS::DCPS::ReliableSession::expire_naks(), OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), OpenDDS::DCPS::BestEffortSession::record_header_received(), OpenDDS::DCPS::ReliableSession::record_header_received(), remote_peer(), OpenDDS::DCPS::ReliableSession::send_naks(), send_syn(), send_synack(), syn_received(), and synack_received().

◆ reverse_start_lock_

Reverse_Lock_t OpenDDS::DCPS::MulticastSession::reverse_start_lock_

protected

Definition at line 97 of file MulticastSession.h.

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

◆ start_lock_

ACE_Thread_Mutex OpenDDS::DCPS::MulticastSession::start_lock_

protected

Definition at line 95 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::BestEffortSession::start(), and OpenDDS::DCPS::ReliableSession::start().

◆ started_

bool OpenDDS::DCPS::MulticastSession::started_

protected

Definition at line 98 of file MulticastSession.h.

Referenced by OpenDDS::DCPS::BestEffortSession::start(), and OpenDDS::DCPS::ReliableSession::start().

◆ syn_delay_

TimeDuration OpenDDS::DCPS::MulticastSession::syn_delay_

private

Definition at line 127 of file MulticastSession.h.

Referenced by send_all_syn(), and start_syn().

◆ syn_watchdog_

RcHandle<Sporadic> OpenDDS::DCPS::MulticastSession::syn_watchdog_

private

Definition at line 126 of file MulticastSession.h.

Referenced by remove_remote_i(), send_all_syn(), start_syn(), stop(), and ~MulticastSession().

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

Public Member Functions

Protected Types

Protected Member Functions

Protected Attributes

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Typedef Documentation

◆ Reverse_Lock_t

◆ Sporadic

Constructor & Destructor Documentation

◆ ~MulticastSession()

◆ MulticastSession()

Member Function Documentation

◆ acked()

◆ add_remote() [1/2]

◆ add_remote() [2/2]

◆ check_header()

◆ control_received()

◆ is_reliable()

◆ link()

◆ OPENDDS_MAP_CMP()

◆ ready_to_deliver()

◆ reassemble()

◆ record_header_received()

◆ release_remote()

◆ remote_peer()

◆ remove_remote()

◆ remove_remote_i()

◆ send_all_syn()

◆ send_control()

◆ send_naks()

◆ send_syn()

◆ send_synack()

◆ set_acked()

◆ start()

◆ start_syn()

◆ stop()

◆ syn_hook()

◆ syn_received()

◆ synack_received()

Member Data Documentation

◆ ack_lock_

◆ acked_

◆ active_

◆ config_name

◆ initial_syn_delay_

◆ link_

◆ pending_remote_map_

◆ reassembly_

◆ remote_peer_

◆ reverse_start_lock_

◆ start_lock_

◆ started_

◆ syn_delay_

◆ syn_watchdog_