OpenDDS::DCPS::MulticastDataLink Class Reference

#include <MulticastDataLink.h>

Inheritance diagram for OpenDDS::DCPS::MulticastDataLink:
Inheritance graph
[legend]
Collaboration diagram for OpenDDS::DCPS::MulticastDataLink:
Collaboration graph
[legend]

List of all members.

Public Member Functions

 MulticastDataLink (MulticastTransport &transport, const MulticastSessionFactory_rch &session_factory, MulticastPeer local_peer, MulticastInst &config, TransportReactorTask *reactor_task, bool is_active)
virtual ~MulticastDataLink ()
MulticastTransport & transport ()
MulticastPeer local_peer () const
MulticastSendStrategy * send_strategy ()
MulticastReceiveStrategy * receive_strategy ()
SingleSendBuffer * send_buffer ()
MulticastInst & config ()
TransportReactorTask * reactor_task ()
ACE_Reactor * get_reactor ()
ACE_Proactor * get_proactor ()
ACE_SOCK_Dgram_Mcast & socket ()
bool join (const ACE_INET_Addr &group_address)
MulticastSession_rch find_or_create_session (MulticastPeer remote_peer)
MulticastSession_rch find_session (MulticastPeer remote_peer)
bool check_header (const TransportHeader &header)
bool check_header (const DataSampleHeader &header)
void sample_received (ReceivedDataSample &sample)
bool reassemble (ReceivedDataSample &data, const TransportHeader &header)

Private Member Functions

typedef OPENDDS_MAP (MulticastPeer, MulticastSession_rch) MulticastSessionMap
virtual void stop_i ()
void syn_received_no_session (MulticastPeer source, const Message_Block_Ptr &data, bool swap_bytes)
void release_remote_i (const RepoId &remote)
bool ready_to_deliver (const ReceivedDataSample &data)

Private Attributes

MulticastSessionFactory_rch session_factory_
MulticastPeer local_peer_
TransportReactorTask * reactor_task_
MulticastSendStrategy_rch send_strategy_
 The transport send strategy object for this DataLink.
MulticastReceiveStrategy_rch recv_strategy_
unique_ptr< SingleSendBuffer > send_buffer_
ACE_SOCK_Dgram_Mcast socket_
ACE_SYNCH_RECURSIVE_MUTEX session_lock_
MulticastSessionMap sessions_
RepoIdSet readers_selected_
RepoIdSet readers_withheld_

Detailed Description

Definition at line 42 of file MulticastDataLink.h.

Constructor & Destructor Documentation

OpenDDS::DCPS::MulticastDataLink::MulticastDataLink ( MulticastTransport &  transport,
const MulticastSessionFactory_rch &  session_factory,
MulticastPeer  local_peer,
MulticastInst &  config,
TransportReactorTask *  reactor_task,
bool  is_active 
)

Definition at line 37 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), OpenDDS::DCPS::TransportInst::max_samples_per_packet_, OpenDDS::DCPS::MulticastInst::nak_depth_, OpenDDS::DCPS::unique_ptr< T, Deleter >::reset(), send_buffer_, send_strategy_, and session_factory_.

00043 : DataLink(transport, 0 /*priority*/, false /*loopback*/, is_active),
00044   session_factory_(session_factory),
00045   local_peer_(local_peer),
00046   reactor_task_(reactor_task),
00047   send_strategy_(make_rch<MulticastSendStrategy>(this)),
00048   recv_strategy_(make_rch<MulticastReceiveStrategy>(this))
00049 {
00050   // A send buffer may be bound to the send strategy to ensure a
00051   // configured number of most-recent datagrams are retained:
00052   if (this->session_factory_->requires_send_buffer()) {
00053     this->send_buffer_.reset(new SingleSendBuffer(config.nak_depth_,
00054                                               config.max_samples_per_packet_));
00055     this->send_strategy_->send_buffer(this->send_buffer_.get());
00056   }
00057 }

Here is the call graph for this function:

OpenDDS::DCPS::MulticastDataLink::~MulticastDataLink (  )  [virtual]

Definition at line 59 of file MulticastDataLink.cpp.

References send_buffer_, and send_strategy_.

00060 {
00061   if (this->send_buffer_) {
00062     this->send_strategy_->send_buffer(0);
00063   }
00064 }

Member Function Documentation

bool OpenDDS::DCPS::MulticastDataLink::check_header ( const DataSampleHeader &  header  ) 

Definition at line 214 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::DataSampleHeader::message_id_, receive_strategy(), session_lock_, sessions_, and OpenDDS::DCPS::TRANSPORT_CONTROL.

00215 {
00216   if (header.message_id_ == TRANSPORT_CONTROL) return true;
00217 
00218   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00219       guard,
00220       this->session_lock_,
00221       false);
00222 
00223   // Skip data sample unless there is a session for it.
00224   return (this->sessions_.count(receive_strategy()->received_header().source_) > 0);
00225 }

Here is the call graph for this function:

bool OpenDDS::DCPS::MulticastDataLink::check_header ( const TransportHeader &  header  ) 

Definition at line 195 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::DataLink::is_active(), session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::check_header().

00196 {
00197   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00198       guard,
00199       this->session_lock_,
00200       false);
00201 
00202   MulticastSessionMap::iterator it(this->sessions_.find(header.source_));
00203   if (it == this->sessions_.end() && is_active()) {
00204     return false;
00205   }
00206   if (it != this->sessions_.end() && it->second->acked()) {
00207     return it->second->check_header(header);
00208   }
00209 
00210   return true;
00211 }

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE MulticastInst & OpenDDS::DCPS::MulticastDataLink::config (  ) 

Definition at line 44 of file MulticastDataLink.inl.

References OpenDDS::DCPS::MulticastTransport::config(), and transport().

Referenced by OpenDDS::DCPS::MulticastSendStrategy::async_send(), OpenDDS::DCPS::ReliableSession::expire_naks(), join(), OpenDDS::DCPS::NakWatchdog::next_interval(), OpenDDS::DCPS::SynWatchdog::next_interval(), OpenDDS::DCPS::SynWatchdog::next_timeout(), OpenDDS::DCPS::SynWatchdog::on_timeout(), OpenDDS::DCPS::MulticastSendStrategy::send_bytes_i(), OpenDDS::DCPS::ReliableSession::send_naks(), and syn_received_no_session().

00045 {
00046   return transport().config();
00047 }

Here is the call graph for this function:

Here is the caller graph for this function:

MulticastSession_rch OpenDDS::DCPS::MulticastDataLink::find_or_create_session ( MulticastPeer  remote_peer  ) 

Definition at line 156 of file MulticastDataLink.cpp.

References ACE_TEXT(), OpenDDS::DCPS::RcHandle< T >::is_nil(), LM_ERROR, ACE_Event_Handler::reactor(), session_factory_, session_lock_, sessions_, and transport().

00157 {
00158   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00159       guard,
00160       this->session_lock_,
00161       MulticastSession_rch());
00162 
00163   MulticastSessionMap::iterator it(this->sessions_.find(remote_peer));
00164   if (it != this->sessions_.end()) {
00165     return it->second;
00166   }
00167 
00168   MulticastSession_rch session =
00169     this->session_factory_->create(transport().reactor(), transport().reactor_owner(), this, remote_peer);
00170   if (session.is_nil()) {
00171     ACE_ERROR_RETURN((LM_ERROR,
00172         ACE_TEXT("(%P|%t) ERROR: ")
00173         ACE_TEXT("MulticastDataLink::find_or_create_session: ")
00174         ACE_TEXT("failed to create session for remote peer: %#08x%08x!\n"),
00175         (unsigned int) (remote_peer >> 32),
00176         (unsigned int) remote_peer),
00177         MulticastSession_rch());
00178   }
00179 
00180   std::pair<MulticastSessionMap::iterator, bool> pair = this->sessions_.insert(
00181       MulticastSessionMap::value_type(remote_peer, session));
00182   if (pair.first == this->sessions_.end()) {
00183     ACE_ERROR_RETURN((LM_ERROR,
00184         ACE_TEXT("(%P|%t) ERROR: ")
00185         ACE_TEXT("MulticastDataLink::find_or_create_session: ")
00186         ACE_TEXT("failed to insert session for remote peer: %#08x%08x!\n"),
00187         (unsigned int) (remote_peer >> 32),
00188         (unsigned int) remote_peer),
00189         MulticastSession_rch());
00190   }
00191   return session;
00192 }

Here is the call graph for this function:

MulticastSession_rch OpenDDS::DCPS::MulticastDataLink::find_session ( MulticastPeer  remote_peer  ) 

Definition at line 141 of file MulticastDataLink.cpp.

References session_lock_, and sessions_.

00142 {
00143   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00144       guard,
00145       this->session_lock_,
00146       MulticastSession_rch());
00147 
00148   MulticastSessionMap::iterator it(this->sessions_.find(remote_peer));
00149   if (it != this->sessions_.end()) {
00150     return it->second;
00151   }
00152   else return MulticastSession_rch();
00153 }

ACE_INLINE ACE_Proactor * OpenDDS::DCPS::MulticastDataLink::get_proactor (  ) 

Definition at line 63 of file MulticastDataLink.inl.

References OpenDDS::DCPS::TransportReactorTask::get_proactor(), and reactor_task_.

Referenced by OpenDDS::DCPS::MulticastSendStrategy::async_send().

00064 {
00065   if (this->reactor_task_ == 0) return 0;
00066   return this->reactor_task_->get_proactor();
00067 }

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE ACE_Reactor * OpenDDS::DCPS::MulticastDataLink::get_reactor ( void   ) 

Definition at line 56 of file MulticastDataLink.inl.

References OpenDDS::DCPS::TransportReactorTask::get_reactor(), and reactor_task_.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::start_i(), and OpenDDS::DCPS::MulticastReceiveStrategy::stop_i().

00057 {
00058   if (this->reactor_task_ == 0) return 0;
00059   return this->reactor_task_->get_reactor();
00060 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool OpenDDS::DCPS::MulticastDataLink::join ( const ACE_INET_Addr &  group_address  ) 

Definition at line 68 of file MulticastDataLink.cpp.

References ACE_TEXT(), ACE_TEXT_CHAR_TO_TCHAR, ACE_SOCK::close(), config(), ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE, ACE_IPC_SAP::get_handle(), ACE_SOCK_Dgram_Mcast::join(), LM_DEBUG, LM_ERROR, OpenDDS::DCPS::MulticastInst::local_address_, ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO, ACE_SOCK_Dgram_Mcast::opts(), OpenDDS::DCPS::MulticastInst::rcv_buffer_size_, recv_strategy_, send_strategy_, OpenDDS::DCPS::set_socket_multicast_ttl(), ACE_OS::setsockopt(), socket_, OpenDDS::DCPS::DataLink::start(), ACE_Utils::truncate_cast(), and VDBG_LVL.

00069 {
00070 
00071   const std::string& net_if = this->config().local_address_;
00072 #ifdef ACE_HAS_MAC_OSX
00073   socket_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
00074                ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
00075 #endif
00076   if (this->socket_.join(group_address, 1,
00077       net_if.empty() ? 0 :
00078           ACE_TEXT_CHAR_TO_TCHAR(net_if.c_str())) != 0) {
00079     ACE_ERROR_RETURN((LM_ERROR,
00080         ACE_TEXT("(%P|%t) ERROR: MulticastDataLink::join: ")
00081         ACE_TEXT("ACE_SOCK_Dgram_Mcast::join failed %m.\n")),
00082         false);
00083   }
00084   VDBG_LVL((LM_DEBUG, ACE_TEXT("(%P|%t) MulticastDataLink::join OK\n")), 6);
00085 
00086   ACE_HANDLE handle = this->socket_.get_handle();
00087 
00088   if (!OpenDDS::DCPS::set_socket_multicast_ttl(this->socket_, this->config().ttl_)) {
00089     ACE_ERROR_RETURN((LM_ERROR,
00090         ACE_TEXT("(%P|%t) ERROR: ")
00091         ACE_TEXT("MulticastDataLink::join: ")
00092         ACE_TEXT("OpenDDS::DCPS::set_socket_multicast_ttl failed.\n")),
00093         false);
00094   }
00095 
00096   int rcv_buffer_size = ACE_Utils::truncate_cast<int>(this->config().rcv_buffer_size_);
00097   if (rcv_buffer_size != 0
00098       && ACE_OS::setsockopt(handle, SOL_SOCKET,
00099           SO_RCVBUF,
00100           (char *) &rcv_buffer_size,
00101           sizeof (int)) < 0) {
00102     ACE_ERROR_RETURN((LM_ERROR,
00103         ACE_TEXT("(%P|%t) ERROR: ")
00104         ACE_TEXT("MulticastDataLink::join: ")
00105         ACE_TEXT("ACE_OS::setsockopt RCVBUF failed.\n")),
00106         false);
00107   }
00108 
00109 #if defined (ACE_DEFAULT_MAX_SOCKET_BUFSIZ)
00110   int snd_size = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
00111 
00112   if (ACE_OS::setsockopt(handle, SOL_SOCKET,
00113       SO_SNDBUF,
00114       (char *) &snd_size,
00115       sizeof(snd_size)) < 0
00116       && errno != ENOTSUP) {
00117     ACE_ERROR_RETURN((LM_ERROR,
00118         ACE_TEXT("(%P|%t) ERROR: ")
00119         ACE_TEXT("MulticastDataLink::join: ")
00120         ACE_TEXT("ACE_OS::setsockopt SNDBUF failed to set the send buffer size to %d errno %m\n"),
00121         snd_size),
00122         false);
00123   }
00124 #endif /* ACE_DEFAULT_MAX_SOCKET_BUFSIZ */
00125 
00126   if (start(static_rchandle_cast<TransportSendStrategy>(this->send_strategy_),
00127       static_rchandle_cast<TransportStrategy>(this->recv_strategy_))
00128       != 0) {
00129     this->socket_.close();
00130     ACE_ERROR_RETURN((LM_ERROR,
00131         ACE_TEXT("(%P|%t) ERROR: ")
00132         ACE_TEXT("MulticastDataLink::join: ")
00133         ACE_TEXT("DataLink::start failed!\n")),
00134         false);
00135   }
00136 
00137   return true;
00138 }

Here is the call graph for this function:

ACE_INLINE MulticastPeer OpenDDS::DCPS::MulticastDataLink::local_peer (  )  const

Definition at line 20 of file MulticastDataLink.inl.

References local_peer_.

Referenced by OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::SynWatchdog::on_timeout(), OpenDDS::DCPS::MulticastSendStrategy::prepare_header_i(), OpenDDS::DCPS::MulticastSession::syn_received(), syn_received_no_session(), and OpenDDS::DCPS::MulticastSession::synack_received().

00021 {
00022   return this->local_peer_;
00023 }

Here is the caller graph for this function:

typedef OpenDDS::DCPS::MulticastDataLink::OPENDDS_MAP ( MulticastPeer  ,
MulticastSession_rch   
) [private]
ACE_INLINE TransportReactorTask * OpenDDS::DCPS::MulticastDataLink::reactor_task (  ) 

Definition at line 50 of file MulticastDataLink.inl.

References reactor_task_.

00051 {
00052   return this->reactor_task_;
00053 }

bool OpenDDS::DCPS::MulticastDataLink::ready_to_deliver ( const ReceivedDataSample &  data  )  [private]

Definition at line 308 of file MulticastDataLink.cpp.

References receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by sample_received().

00309 {
00310   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00311       guard,
00312       this->session_lock_, false);
00313 
00314   const TransportHeader& theader = receive_strategy()->received_header();
00315 
00316   MulticastSessionMap::iterator session_it = sessions_.find(theader.source_);
00317   if (session_it != sessions_.end()) {
00318     MulticastSession_rch sess_rch(session_it->second);
00319     guard.release();
00320     return sess_rch->ready_to_deliver(theader, data);
00321   }
00322 
00323   return true;
00324 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

Definition at line 228 of file MulticastDataLink.cpp.

References session_lock_, sessions_, and OpenDDS::DCPS::TransportHeader::source_.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::reassemble().

00230 {
00231   ACE_GUARD_RETURN(ACE_SYNCH_RECURSIVE_MUTEX,
00232       guard,
00233       this->session_lock_,
00234       false);
00235 
00236   MulticastSessionMap::iterator it(this->sessions_.find(header.source_));
00237   if (it == this->sessions_.end()) return false;
00238   if (it->second->acked()) {
00239     return it->second->reassemble(data, header);
00240   }
00241   return false;
00242 }

Here is the caller graph for this function:

ACE_INLINE MulticastReceiveStrategy * OpenDDS::DCPS::MulticastDataLink::receive_strategy (  ) 

Definition at line 32 of file MulticastDataLink.inl.

References OpenDDS::DCPS::RcHandle< T >::in(), and recv_strategy_.

Referenced by check_header(), OpenDDS::DCPS::ReliableSession::nak_received(), OpenDDS::DCPS::ReliableSession::nakack_received(), ready_to_deliver(), sample_received(), OpenDDS::DCPS::MulticastSession::syn_received(), and OpenDDS::DCPS::MulticastSession::synack_received().

00033 {
00034   return this->recv_strategy_.in();
00035 }

Here is the call graph for this function:

Here is the caller graph for this function:

void OpenDDS::DCPS::MulticastDataLink::release_remote_i ( const RepoId &  remote  )  [private, virtual]

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 327 of file MulticastDataLink.cpp.

References session_lock_, and sessions_.

00328 {
00329   ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX, guard, session_lock_);
00330   MulticastPeer remote_source = (ACE_INT64)RepoIdConverter(remote).federationId() << 32
00331                               | RepoIdConverter(remote).participantId();
00332   MulticastSessionMap::iterator session_it = sessions_.find(remote_source);
00333   if (session_it != sessions_.end() && session_it->second->is_reliable()) {
00334     session_it->second->release_remote(remote);
00335   }
00336 }

void OpenDDS::DCPS::MulticastDataLink::sample_received ( ReceivedDataSample &  sample  ) 

Definition at line 245 of file MulticastDataLink.cpp.

References OpenDDS::DCPS::DataLink::data_received(), OpenDDS::DCPS::ReceivedDataSample::header_, OpenDDS::DCPS::DataLink::is_active(), OpenDDS::DCPS::DataSampleHeader::message_id_, OpenDDS::DCPS::MULTICAST_SYN, ready_to_deliver(), receive_strategy(), OpenDDS::DCPS::TransportReceiveStrategy< TH, DSH >::received_header(), OpenDDS::DCPS::ReceivedDataSample::sample_, session_lock_, sessions_, OpenDDS::DCPS::TransportHeader::source_, OpenDDS::DCPS::DataSampleHeader::submessage_id_, OpenDDS::DCPS::TransportHeader::swap_bytes(), syn_received_no_session(), and OpenDDS::DCPS::TRANSPORT_CONTROL.

Referenced by OpenDDS::DCPS::MulticastReceiveStrategy::deliver_sample().

00246 {
00247   switch (sample.header_.message_id_) {
00248   case TRANSPORT_CONTROL: {
00249     // Transport control samples are delivered to all sessions
00250     // regardless of association status:
00251     {
00252       char* const ptr = sample.sample_ ? sample.sample_->rd_ptr() : 0;
00253 
00254       ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX,
00255           guard,
00256           this->session_lock_);
00257 
00258       const TransportHeader& theader = receive_strategy()->received_header();
00259 
00260       if (!is_active() && sample.header_.submessage_id_ == MULTICAST_SYN &&
00261           sessions_.find(theader.source_) == sessions_.end()) {
00262         // We have received a SYN but there is no session (yet) for this source.
00263         // Depending on the data, we may need to send SYNACK.
00264 
00265         guard.release();
00266         syn_received_no_session(theader.source_, sample.sample_,
00267                                 theader.swap_bytes());
00268 
00269         guard.acquire();
00270         MulticastSessionMap::iterator s_itr = sessions_.find(theader.source_);
00271         if (s_itr != sessions_.end()) {
00272           s_itr->second->record_header_received(theader);
00273         }
00274 
00275         if (ptr) {
00276           sample.sample_->rd_ptr(ptr);
00277         }
00278         return;
00279       }
00280 
00281       MulticastSessionMap temp_sessions(sessions_);
00282       guard.release();
00283 
00284       for (MulticastSessionMap::iterator it(temp_sessions.begin());
00285           it != temp_sessions.end(); ++it) {
00286         it->second->control_received(sample.header_.submessage_id_,
00287                                      sample.sample_);
00288         it->second->record_header_received(theader);
00289 
00290         // reset read pointer
00291         if (ptr) {
00292           sample.sample_->rd_ptr(ptr);
00293         }
00294       }
00295     }
00296   } break;
00297 
00298   default:
00299 
00300     if (ready_to_deliver(sample)) {
00301       data_received(sample);
00302     }
00303     break;
00304   }
00305 }

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE SingleSendBuffer * OpenDDS::DCPS::MulticastDataLink::send_buffer (  ) 

Definition at line 38 of file MulticastDataLink.inl.

References OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), and send_buffer_.

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

00039 {
00040   return this->send_buffer_.get();
00041 }

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE MulticastSendStrategy * OpenDDS::DCPS::MulticastDataLink::send_strategy (  ) 

Definition at line 26 of file MulticastDataLink.inl.

References OpenDDS::DCPS::RcHandle< T >::in(), and send_strategy_.

00027 {
00028   return this->send_strategy_.in();
00029 }

Here is the call graph for this function:

ACE_INLINE ACE_SOCK_Dgram_Mcast & OpenDDS::DCPS::MulticastDataLink::socket ( void   ) 

Definition at line 70 of file MulticastDataLink.inl.

References socket_.

Referenced by OpenDDS::DCPS::MulticastSendStrategy::async_send(), OpenDDS::DCPS::MulticastReceiveStrategy::get_handle(), OpenDDS::DCPS::MulticastReceiveStrategy::receive_bytes(), and OpenDDS::DCPS::MulticastSendStrategy::sync_send().

00071 {
00072   return this->socket_;
00073 }

Here is the caller graph for this function:

void OpenDDS::DCPS::MulticastDataLink::stop_i (  )  [private, virtual]

This announces the "stop" event to our subclass. The "stop" event will occur when this DataLink is handling a release_reservations() call and determines that it has just released all of the remaining reservations on this DataLink. The "stop" event will also occur when the TransportImpl is being shutdown() - we call stop_i() from our transport_shutdown() method to handle this case.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 388 of file MulticastDataLink.cpp.

References ACE_SOCK::close(), session_lock_, sessions_, and socket_.

00389 {
00390   ACE_GUARD(ACE_SYNCH_RECURSIVE_MUTEX,
00391       guard,
00392       this->session_lock_);
00393 
00394   for (MulticastSessionMap::iterator it(this->sessions_.begin());
00395       it != this->sessions_.end(); ++it) {
00396     it->second->stop();
00397   }
00398   this->sessions_.clear();
00399 
00400   this->socket_.close();
00401 }

Here is the call graph for this function:

void OpenDDS::DCPS::MulticastDataLink::syn_received_no_session ( MulticastPeer  source,
const Message_Block_Ptr &  data,
bool  swap_bytes 
) [private]

Definition at line 339 of file MulticastDataLink.cpp.

References ACE_TEXT(), config(), OpenDDS::DCPS::DataLink::create_control(), OpenDDS::DCPS::unique_ptr< T, Deleter >::get(), header, LM_DEBUG, LM_ERROR, local_peer(), local_peer_, OpenDDS::DCPS::move(), OpenDDS::DCPS::MULTICAST_SYNACK, OpenDDS::DCPS::MulticastTransport::passive_connection(), OpenDDS::DCPS::DataLink::send_control(), OpenDDS::DCPS::SEND_CONTROL_OK, transport(), and VDBG_LVL.

Referenced by sample_received().

00342 {
00343   Serializer serializer_read(data.get(), swap_bytes);
00344 
00345   MulticastPeer local_peer;
00346   serializer_read >> local_peer;
00347 
00348   if (local_peer != local_peer_) {
00349     return;
00350   }
00351 
00352   VDBG_LVL((LM_DEBUG, "(%P|%t) MulticastDataLink[%C]::syn_received_no_session "
00353       "send_synack local %#08x%08x remote %#08x%08x\n",
00354       this->config().name().c_str(),
00355       (unsigned int) (local_peer >> 32),
00356       (unsigned int) local_peer,
00357       (unsigned int) (source >> 32),
00358       (unsigned int) source), 2);
00359 
00360   Message_Block_Ptr synack_data(new ACE_Message_Block(sizeof(MulticastPeer)));
00361 
00362   Serializer serializer_write(synack_data.get());
00363   serializer_write << source;
00364 
00365   DataSampleHeader header;
00366   Message_Block_Ptr control(
00367       create_control(MULTICAST_SYNACK, header, move(synack_data)));
00368 
00369   if (control == 0) {
00370     ACE_ERROR((LM_ERROR,
00371                ACE_TEXT("(%P|%t) ERROR: ")
00372                ACE_TEXT("MulticastDataLink::syn_received_no_session: ")
00373                ACE_TEXT("create_control failed!\n")));
00374     return;
00375   }
00376 
00377   const int error = send_control(header, move(control));
00378   if (error != SEND_CONTROL_OK) {
00379     ACE_ERROR((LM_ERROR, "(%P|%t) MulticastDataLink::syn_received_no_session: "
00380         "ERROR: send_control failed: %d!\n", error));
00381     return;
00382   }
00383 
00384   transport().passive_connection(local_peer, source);
00385 }

Here is the call graph for this function:

Here is the caller graph for this function:

ACE_INLINE MulticastTransport & OpenDDS::DCPS::MulticastDataLink::transport ( void   ) 

Definition at line 14 of file MulticastDataLink.inl.

References OpenDDS::DCPS::DataLink::impl().

Referenced by config(), find_or_create_session(), OpenDDS::DCPS::NakWatchdog::reactor_is_shut_down(), OpenDDS::DCPS::SynWatchdog::reactor_is_shut_down(), OpenDDS::DCPS::MulticastSession::syn_received(), and syn_received_no_session().

00015 {
00016   return static_cast<MulticastTransport&>(DataLink::impl());
00017 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

MulticastPeer OpenDDS::DCPS::MulticastDataLink::local_peer_ [private]

Definition at line 86 of file MulticastDataLink.h.

Referenced by local_peer(), and syn_received_no_session().

TransportReactorTask* OpenDDS::DCPS::MulticastDataLink::reactor_task_ [private]

Definition at line 88 of file MulticastDataLink.h.

Referenced by get_proactor(), get_reactor(), and reactor_task().

RepoIdSet OpenDDS::DCPS::MulticastDataLink::readers_selected_ [private]

Definition at line 108 of file MulticastDataLink.h.

RepoIdSet OpenDDS::DCPS::MulticastDataLink::readers_withheld_ [private]

Definition at line 108 of file MulticastDataLink.h.

MulticastReceiveStrategy_rch OpenDDS::DCPS::MulticastDataLink::recv_strategy_ [private]

Definition at line 91 of file MulticastDataLink.h.

Referenced by join(), and receive_strategy().

unique_ptr<SingleSendBuffer> OpenDDS::DCPS::MulticastDataLink::send_buffer_ [private]

Definition at line 93 of file MulticastDataLink.h.

Referenced by MulticastDataLink(), send_buffer(), and ~MulticastDataLink().

MulticastSendStrategy_rch OpenDDS::DCPS::MulticastDataLink::send_strategy_ [private]

The transport send strategy object for this DataLink.

Reimplemented from OpenDDS::DCPS::DataLink.

Definition at line 90 of file MulticastDataLink.h.

Referenced by join(), MulticastDataLink(), send_strategy(), and ~MulticastDataLink().

MulticastSessionFactory_rch OpenDDS::DCPS::MulticastDataLink::session_factory_ [private]

Definition at line 84 of file MulticastDataLink.h.

Referenced by find_or_create_session(), and MulticastDataLink().

ACE_SYNCH_RECURSIVE_MUTEX OpenDDS::DCPS::MulticastDataLink::session_lock_ [private]

Definition at line 97 of file MulticastDataLink.h.

Referenced by check_header(), find_or_create_session(), find_session(), ready_to_deliver(), reassemble(), release_remote_i(), sample_received(), and stop_i().

MulticastSessionMap OpenDDS::DCPS::MulticastDataLink::sessions_ [private]

Definition at line 100 of file MulticastDataLink.h.

Referenced by check_header(), find_or_create_session(), find_session(), ready_to_deliver(), reassemble(), release_remote_i(), sample_received(), and stop_i().

ACE_SOCK_Dgram_Mcast OpenDDS::DCPS::MulticastDataLink::socket_ [private]

Definition at line 95 of file MulticastDataLink.h.

Referenced by join(), socket(), and stop_i().

The documentation for this class was generated from the following files:
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Defines
Generated on 10 Aug 2018 for OpenDDS by  doxygen 1.6.1