Event Handler that services multicast requests for IOR of a bootstrappable service. More...

#include <InfoRepoMulticastResponder.h>

Inheritance diagram for OpenDDS::Federator::InfoRepoMulticastResponder:

Collaboration diagram for OpenDDS::Federator::InfoRepoMulticastResponder:

[legend]

Public Member Functions
	InfoRepoMulticastResponder ()

virtual	~InfoRepoMulticastResponder ()

int	init (CORBA::ORB_ptr orb, u_short port, const char *mcast_addr)
	Initialization method. More...

int	init (CORBA::ORB_ptr orb, const char *mcast_addr)

virtual int	handle_input (ACE_HANDLE n)
	Callback when input is received on the handle. More...

virtual int	handle_timeout (const ACE_Time_Value &tv, const void *arg)
	Callback when a timeout has occurred. More...

virtual ACE_HANDLE	get_handle () const
	Returns the internal handle used to receive multicast. More...

Public Member Functions inherited from ACE_Event_Handler
virtual	~ACE_Event_Handler (void)

virtual void	set_handle (ACE_HANDLE)

virtual int	priority (void) const

virtual void	priority (int priority)

virtual int	handle_output (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exception (ACE_HANDLE fd=ACE_INVALID_HANDLE)

virtual int	handle_exit (ACE_Process *)

virtual int	handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)

virtual int	handle_signal (int signum, siginfo_t =0, ucontext_t =0)

virtual int	resume_handler (void)

virtual int	handle_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual int	handle_group_qos (ACE_HANDLE=ACE_INVALID_HANDLE)

virtual void	reactor (ACE_Reactor *reactor)

virtual ACE_Reactor *	reactor (void) const

virtual ACE_Reactor_Timer_Interface *	reactor_timer_interface (void) const

virtual Reference_Count	add_reference (void)

virtual Reference_Count	remove_reference (void)

Reference_Counting_Policy &	reference_counting_policy (void)

Private Member Functions
int	common_init (CORBA::ORB_ptr orb)
	Factor common functionality from the two init functions. More...

Private Attributes
bool	initialized_
	Are we initialized? More...

CORBA::ORB_var	orb_
	The ORB. More...

ACE_SOCK_Dgram_Mcast	mcast_dgram_
	multicast endpoint of communication More...

ACE_INET_Addr	mcast_addr_
	multicast address More...

ACE_INET_Addr	response_addr_
	address of response. More...

ACE_SOCK_Dgram	response_
	socket for response to the multicast More...

ACE_CString	mcast_nic_

Additional Inherited Members
Public Types inherited from ACE_Event_Handler
typedef long	Reference_Count

Static Public Member Functions inherited from ACE_Event_Handler
static ACE_THR_FUNC_RETURN	read_adapter (void *event_handler)

static int	register_stdin_handler (ACE_Event_Handler eh, ACE_Reactor reactor, ACE_Thread_Manager *thr_mgr, int flags=THR_DETACHED)

static int	remove_stdin_handler (ACE_Reactor reactor, ACE_Thread_Manager thr_mgr)

Public Attributes inherited from ACE_Event_Handler
	LO_PRIORITY

	HI_PRIORITY

	NULL_MASK

	READ_MASK

	WRITE_MASK

	EXCEPT_MASK

	ACCEPT_MASK

	CONNECT_MASK

	TIMER_MASK

	QOS_MASK

	GROUP_QOS_MASK

	SIGNAL_MASK

	ALL_EVENTS_MASK

	RWE_MASK

	DONT_CALL

	ACE_EVENT_HANDLER_NOT_RESUMED

	ACE_REACTOR_RESUMES_HANDLER

	ACE_APPLICATION_RESUMES_HANDLER

Protected Types inherited from ACE_Event_Handler
typedef ACE_Atomic_Op< ACE_SYNCH_MUTEX, Reference_Count >	Atomic_Reference_Count

Protected Member Functions inherited from ACE_Event_Handler
	ACE_Event_Handler (ACE_Reactor *=0, int priority=ACE_Event_Handler::LO_PRIORITY)

Protected Attributes inherited from ACE_Event_Handler
Atomic_Reference_Count	reference_count_

Detailed Description

Event Handler that services multicast requests for IOR of a bootstrappable service.

This class uses the ACE_SOCK_Dgram_Mcast class and should be registered with a reactor and should be initialized with the ior of the service to be multicasted.

Definition at line 42 of file InfoRepoMulticastResponder.h.

Constructor & Destructor Documentation

◆ InfoRepoMulticastResponder()

OpenDDS::Federator::InfoRepoMulticastResponder::InfoRepoMulticastResponder ( )

Definition at line 27 of file InfoRepoMulticastResponder.cpp.

   : initialized_(false)
 {
 }

◆ ~InfoRepoMulticastResponder()

OpenDDS::Federator::InfoRepoMulticastResponder::~InfoRepoMulticastResponder ( )

virtual

Definition at line 32 of file InfoRepoMulticastResponder.cpp.

References ACE_ERROR, ACE_TEXT(), initialized_, ACE_SOCK_Dgram_Mcast::leave(), LM_ERROR, and mcast_dgram_.

 {
   if (
     this->initialized_ &&
     (this->mcast_dgram_.leave(this->mcast_addr_) == -1)) {
     ACE_ERROR((LM_ERROR, "%p\n", ACE_TEXT("~InfoRepoMulticastResponder()")));
   }
 }

Member Function Documentation

◆ common_init()

int OpenDDS::Federator::InfoRepoMulticastResponder::common_init ( CORBA::ORB_ptr orb )

private

Factor common functionality from the two init functions.

Definition at line 109 of file InfoRepoMulticastResponder.cpp.

References CORBA::ORB::_duplicate(), ACE_ERROR_RETURN, ACE_TEXT(), ACE_TEXT_CHAR_TO_TCHAR, ACE_String_Base< char >::c_str(), ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE, initialized_, ACE_SOCK_Dgram_Mcast::join(), ACE_String_Base< char >::length(), LM_ERROR, mcast_dgram_, mcast_nic_, ACE_SOCK_Dgram::open(), ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO, ACE_SOCK_Dgram_Mcast::opts(), orb_, response_, response_addr_, and ACE_INET_Addr::set().

Referenced by init().

 {
   orb_ = CORBA::ORB::_duplicate(orb);
 
   if (this->response_addr_.set((u_short) 0) == -1)
     ACE_ERROR_RETURN((LM_ERROR,
                       "InfoRepoMulticastResponder::common_init() %p\n",
                       ACE_TEXT("set")),
                      -1);
 
   else if (this->response_.open(this->response_addr_) == -1) {
     ACE_ERROR_RETURN((LM_ERROR,
                       "%p\n",
                       ACE_TEXT("set")),
                      -1);
   }
 
   // Use ACE_SOCK_Dgram_Mcast factory to subscribe to multicast group.
 #ifdef ACE_HAS_MAC_OSX
   mcast_dgram_.opts(ACE_SOCK_Dgram_Mcast::OPT_BINDADDR_NO |
                     ACE_SOCK_Dgram_Mcast::DEFOPT_NULLIFACE);
 #endif
   if (this->mcast_nic_.length() != 0) {
     if (this->mcast_dgram_.join(this->mcast_addr_,
                                 1,
                                 ACE_TEXT_CHAR_TO_TCHAR(this->mcast_nic_.c_str())) == -1)
       ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::common_init() %p\n",
                         ACE_TEXT("subscribe")), -1);
 
   } else {
     if (this->mcast_dgram_.join(this->mcast_addr_) == -1)
       ACE_ERROR_RETURN((LM_ERROR,
                         "InfoRepoMulticastResponder::common_init() %p\n",
                         ACE_TEXT("subscribe")),
                        -1);
   }
 
   this->initialized_ = true;
   return 0;
 }

◆ get_handle()

ACE_HANDLE OpenDDS::Federator::InfoRepoMulticastResponder::get_handle ( void ) const

virtual

Returns the internal handle used to receive multicast.

Reimplemented from ACE_Event_Handler.

Definition at line 42 of file InfoRepoMulticastResponder.cpp.

References ACE_IPC_SAP::get_handle(), and mcast_dgram_.

 {
   return this->mcast_dgram_.get_handle();
 }

◆ handle_input()

int OpenDDS::Federator::InfoRepoMulticastResponder::handle_input ( ACE_HANDLE n )

virtual

Callback when input is received on the handle.

Reimplemented from ACE_Event_Handler.

Definition at line 159 of file InfoRepoMulticastResponder.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_ERROR_RETURN, ACE_HTONS, ACE_IPV6_LOCALHOST, ACE_NTOHS, ACE_TEXT(), OpenDDS::DCPS::LogAddr::c_str(), ACE_SOCK_Stream::close(), ACE_SOCK_Connector::connect(), OpenDDS::DCPS::DCPS_debug_level, ACE::get_ip_interfaces(), header, OpenDDS::DCPS::LogAddr::HostPort, TAO_Pseudo_Var_T< Object >::in(), ACE_INET_Addr::is_linklocal(), CORBA::is_nil(), LM_DEBUG, LM_ERROR, mcast_dgram_, MSG_PEEK, OPENDDS_END_VERSIONED_NAMESPACE_DECL, orb_, ACE_SOCK_Dgram::recv(), ACE_SOCK_Stream::sendv_n(), ACE_INET_Addr::set(), and ACE_INET_Addr::set_port_number().

 {
   if (OpenDDS::DCPS::DCPS_debug_level > 0)
     ACE_DEBUG((LM_DEBUG, "Entered InfoRepoMulticastResponder::handle_input\n"));
 
   // The length of the service name string that follows.
   CORBA::Short header;
   // Port to which to reply.
   ACE_UINT16 remote_port;
   // Name of the service for which the client is looking.
   char object_key[BUFSIZ];
 
   ACE_INET_Addr remote_addr;
 
   // Take a peek at the header to find out how long is the service
   // name string we should receive.
   ssize_t n = this->mcast_dgram_.recv(&header,
                                       sizeof(header),
                                       remote_addr,
                                       MSG_PEEK);
 
   if (n <= 0)
     ACE_ERROR_RETURN((LM_ERROR,
                       "InfoRepoMulticastResponder::handle_input - peek %d\n",
                       n),
                      0);
 
   else if (ACE_NTOHS(header) <= 0)
     ACE_ERROR_RETURN((LM_ERROR,
                       "InfoRepoMulticastResponder::handle_input() Header value < 1\n"),
                      0);
 
   // Receive full client multicast request.
   const int iovcnt = 3;
   iovec iov[iovcnt];
 
   iov[0].iov_base = (char *) &header;
   iov[0].iov_len  = sizeof(header);
   iov[1].iov_base = (char *) &remote_port;
   iov[1].iov_len  = sizeof(ACE_UINT16);
   iov[2].iov_base = (char *) object_key;
   iov[2].iov_len  = ACE_NTOHS(header);
 
   // Read the iovec.
   n = this->mcast_dgram_.recv(iov,
                               iovcnt,
                               remote_addr);
 
   if (n <= 0)
     ACE_ERROR_RETURN((LM_ERROR,
                       "InfoRepoMulticastResponder::handle_input recv = %d\n",
                       n),
                      0);
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) Received multicast from %C.\n"
                "Service Name received : %s\n"
                "Port received : %u\n",
                DCPS::LogAddr(remote_addr, DCPS::LogAddr::HostPort).c_str(),
                object_key,
                ACE_NTOHS(remote_port)));
   }
 
   // Grab the IOR table.
   CORBA::Object_var table_object =
     orb_->resolve_initial_references("IORTable");
 
   IORTable::Locator_var locator =
     IORTable::Locator::_narrow(table_object.in());
 
   if (CORBA::is_nil(locator.in())) {
     ACE_ERROR((LM_ERROR, ACE_TEXT("Nil IORTable\n")));
   }
 
   std::string ior;
 
   {
     CORBA::String_var ior_result;
 
     try {
       ior_result = locator->locate(object_key);
 
     } catch (const IORTable::NotFound&) {
       ACE_ERROR_RETURN((LM_ERROR,
                         "InfoRepoMulticastResponder::handle_input() Object key not found\n"),
                        0);
     }
 
     ior = ior_result;
   }
 
   // Reply to the multicast message.
   ACE_SOCK_Connector connector;
   ACE_INET_Addr peer_addr;
   ACE_SOCK_Stream stream;
 
   peer_addr.set(remote_addr);
   peer_addr.set_port_number(ACE_NTOHS(remote_port));
 
 #if defined (ACE_HAS_IPV6)
 
   if (peer_addr.is_linklocal()) {
     // If this is one of our local linklocal interfaces this is not going
     // to work.
     // Creating a connection using such interface to the client listening
     // at the IPv6 ANY address is not going to work (I'm not quite sure why
     // but it probably has to do with the rather restrictive routing rules
     // for linklocal interfaces).
     // So we see if this is one of our local interfaces and if so create the
     // connection using the IPv6 loopback address instead.
     ACE_INET_Addr  peer_tmp(peer_addr);
     peer_tmp.set_port_number(static_cast<u_short>(0));
     ACE_INET_Addr* tmp = 0;
     size_t cnt = 0;
     int err = ACE::get_ip_interfaces(cnt, tmp);
 
     if (err == 0) {
       for (size_t i = 0; i < cnt; ++i) {
         if (peer_tmp == tmp[i]) {
           peer_addr.set(ACE_NTOHS(remote_port),
                         ACE_IPV6_LOCALHOST);
           break;
         }
       }
 
       delete[] tmp;
     }
   }
 
 #endif /* ACE_HAS_IPV6 */
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0) {
     ACE_DEBUG((LM_DEBUG, "(%P|%t) Replying to peer %C.\n",
                DCPS::LogAddr(peer_addr, DCPS::LogAddr::HostPort).c_str()));
   }
 
   // Connect.
   if (connector.connect(stream, peer_addr) == -1)
     ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::connect failed\n"), 0);
 
   // Send the IOR back to the client.  (Send iovec, which contains ior
   // length as the first element, and ior itself as the second.)
 
   // Length of ior to be sent.
   const CORBA::Short data_len = ACE_HTONS(static_cast<CORBA::Short>(ior.length()) + 1);
 
   // Vector to be sent.
   const int cnt = 2;
   iovec iovp[cnt];
 
   // The length of ior to be sent.
   iovp[0].iov_base = (char *) &data_len;
   iovp[0].iov_len  = sizeof(CORBA::Short);
 
   // The ior.
   iovp[1].iov_base = const_cast<char*>(ior.c_str());
   iovp[1].iov_len  = static_cast<u_long>(ior.length() + 1);
 
   const ssize_t result = stream.sendv_n(iovp, cnt);
   // Close the stream.
   stream.close();
 
   // Check for error.
   if (result == -1)
     ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::send failed\n"), 0);
 
   if (OpenDDS::DCPS::DCPS_debug_level > 0)
     ACE_DEBUG((LM_DEBUG,
                "(%P|%t) InfoRepoMulticastResponder::handle_input() ior: <%C>\n"
                "sent to %C.\n"
                "result from send = %d\n",
                ior.c_str(),
                DCPS::LogAddr(peer_addr).c_str(),
                result));
 
   return 0;
 }

◆ handle_timeout()

int OpenDDS::Federator::InfoRepoMulticastResponder::handle_timeout	(	const ACE_Time_Value &	tv,
		const void *	arg
	)

virtual

Callback when a timeout has occurred.

Reimplemented from ACE_Event_Handler.

Definition at line 152 of file InfoRepoMulticastResponder.cpp.

 {
   return 0;
 }

◆ init() [1/2]

int OpenDDS::Federator::InfoRepoMulticastResponder::init	(	CORBA::ORB_ptr	orb,
		u_short	port,
		const char *	mcast_addr
	)

Initialization method.

Definition at line 48 of file InfoRepoMulticastResponder.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), common_init(), initialized_, LM_ERROR, mcast_addr_, and ACE_INET_Addr::set().

Referenced by OpenDDS::Federator::ManagerImpl::initialize().

 {
   if (this->initialized_) {
     ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::init() already initialized\n"), -1);
   }
 
   if (this->mcast_addr_.set(port, mcast_addr) == -1)
     ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::init() %p\n", ACE_TEXT("set")), -1);
 
   return common_init(orb);
 }

◆ init() [2/2]

int OpenDDS::Federator::InfoRepoMulticastResponder::init	(	CORBA::ORB_ptr	orb,
		const char *	mcast_addr
	)

Initialization method. Takes in "address:port" string as a parameter.

Save for use later.

Definition at line 64 of file InfoRepoMulticastResponder.cpp.

References ACE_ERROR_RETURN, ACE_TEXT(), common_init(), TAO::String_var< charT >::in(), initialized_, TAO::String_var< charT >::inout(), LM_ERROR, mcast_addr_, mcast_nic_, ACE_INET_Addr::set(), ACE_OS::strchr(), CORBA::string_alloc(), ACE_OS::strlen(), and ACE_OS::strncpy().

 {
   if (this->initialized_) {
     ACE_ERROR_RETURN((LM_ERROR, "InfoRepoMulticastResponder::init() already initialized\n"), -1);
   }
 
   // Look for a '@' incase a nic is specified.
   const char* tmpnic = ACE_OS::strchr(mcast_addr, '@');
 
   CORBA::String_var actual_mcast_addr;
   CORBA::ULong length_addr;
 
   if (tmpnic != 0) {
     // i.e. a nic name has been specified
     length_addr = static_cast<CORBA::ULong>(tmpnic - mcast_addr + 1);
     actual_mcast_addr = CORBA::string_alloc(length_addr);
 
     ACE_OS::strncpy(actual_mcast_addr.inout(),
                     mcast_addr,
                     length_addr - 1);
 
     actual_mcast_addr[length_addr - 1] = '\0';
 
     /// Save for use later.
     this->mcast_nic_ = tmpnic + 1;
 
   } else {
     actual_mcast_addr =
       CORBA::string_alloc(static_cast<CORBA::ULong>(ACE_OS::strlen(mcast_addr)));
 
     actual_mcast_addr = mcast_addr;
   }
 
   if (this->mcast_addr_.set(actual_mcast_addr.in()) == -1)
     ACE_ERROR_RETURN((LM_ERROR,
                       "%p\n",
                       ACE_TEXT("set")),
                      -1);
 
   return common_init(orb);
 }