OpenDDS::DCPS::MulticastManager Class Reference

#include <MulticastManager.h>

Public Member Functions
bool	process (InternalDataReader< NetworkInterfaceAddress >::SampleSequence &samples, InternalSampleInfoSequence &infos, const OPENDDS_STRING &multicast_interface, ACE_Reactor *reactor, ACE_Event_Handler *event_handler, const NetworkAddress &multicast_group_address, ACE_SOCK_Dgram_Mcast &multicast_socket)
	Returns true if at least one group was joined. More...

Private Member Functions
size_t	joined_interface_count () const
bool	join (const NetworkInterfaceAddress &nia, ACE_Reactor *reactor, ACE_Event_Handler *event_handler, const NetworkAddress &multicast_group_address, ACE_SOCK_Dgram_Mcast &multicast_socket)
void	leave (const NetworkInterfaceAddress &nia, const NetworkAddress &multicast_group_address, ACE_SOCK_Dgram_Mcast &multicast_socket)
	OPENDDS_SET (OPENDDS_STRING) joined_interfaces_

Detailed Description

Definition at line 24 of file MulticastManager.h.

Member Function Documentation

join()

bool OpenDDS::DCPS::MulticastManager::join ( const NetworkInterfaceAddress &  nia, ACE_Reactor *  reactor, ACE_Event_Handler *  event_handler, const NetworkAddress &  multicast_group_address, ACE_SOCK_Dgram_Mcast &  multicast_socket  )

private

Definition at line 77 of file MulticastManager.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_GUARD_RETURN, ACE_TEXT_CHAR_TO_TCHAR, OpenDDS::DCPS::NetworkInterfaceAddress::address, OpenDDS::DCPS::LogAddr::c_str(), OpenDDS::DCPS::LogLevel::Error, OpenDDS::DCPS::LogLevel::Info, OpenDDS::DCPS::LogAddr::Ip, OpenDDS::DCPS::NetworkInterfaceAddress::is_ipv4(), OpenDDS::DCPS::NetworkInterfaceAddress::is_ipv6(), joined_interface_count(), LM_ERROR, LM_INFO, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::DCPS::NetworkInterfaceAddress::name, ACE_Event_Handler::READ_MASK, ACE_Reactor::register_handler(), OpenDDS::DCPS::NetworkAddress::to_addr(), and OpenDDS::DCPS::LogLevel::Warning.

Referenced by process().

{
  bool joined = false;

  if (joined_interfaces_.count(nia.name) == 0 && nia.is_ipv4()) {
    if (0 == multicast_socket.join(multicast_group_address.to_addr(), 1, nia.name.empty() ? 0 : ACE_TEXT_CHAR_TO_TCHAR(nia.name.c_str()))) {
      joined_interfaces_.insert(nia.name);
      if (log_level >= LogLevel::Info) {
        ACE_DEBUG((LM_INFO,
                   "(%P|%t) INFO: MulticastManager::join: joined group %C on %C/%C (%@ joined count %B)\n",
                   LogAddr(multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
      joined = true;

      if (reactor) {
        if (reactor->register_handler(multicast_socket.get_handle(),
                                      event_handler,
                                      ACE_Event_Handler::READ_MASK) != 0) {
          if (log_level >= LogLevel::Error) {
            ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: MulticastManager::join: failed to register multicast input handler\n"));
          }
        }
      } else if (log_level >= LogLevel::Error) {
        ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: MulticastManager::join: reactor is NULL\n"));
      }
    } else {
      if (log_level >= LogLevel::Warning) {
        ACE_ERROR((LM_WARNING,
                   "(%P|%t) WARNING: MulticastManager::join: failed to join group %C on %C/%C (%@ joined count %B): %m\n",
                   LogAddr(multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    }
  }

#ifdef ACE_HAS_IPV6
  if (ipv6_joined_interfaces_.count(nia.name) == 0 && nia.is_ipv6()) {
    // Windows 7 has an issue with different threads concurrently calling join for ipv6
    static ACE_Thread_Mutex ipv6_static_lock;
    ACE_GUARD_RETURN(ACE_Thread_Mutex, g3, ipv6_static_lock, false);

    if (0 == ipv6_multicast_socket.join(ipv6_multicast_group_address.to_addr(), 1, nia.name.empty() ? 0 : ACE_TEXT_CHAR_TO_TCHAR(nia.name.c_str()))) {
      ipv6_joined_interfaces_.insert(nia.name);
      if (log_level >= LogLevel::Info) {
        ACE_DEBUG((LM_INFO,
                   "(%P|%t) INFO: MulticastManager::join: joined group %C on %C/%C (%@ joined count %B)\n",
                   LogAddr(ipv6_multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
      joined = true;

      if (reactor) {
        if (reactor->register_handler(ipv6_multicast_socket.get_handle(),
                                      event_handler,
                                      ACE_Event_Handler::READ_MASK) != 0) {
          if (log_level >= LogLevel::Error) {
            ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: MulticastManager::join: ipv6 failed to register multicast input handler\n"));
          }
        }
      } else if (log_level >= LogLevel::Error) {
        ACE_ERROR((LM_ERROR, "(%P|%t) ERROR: MulticastManager::join: ipv6 reactor is NULL\n"));
      }
    } else {
      if (log_level >= LogLevel::Warning) {
        ACE_ERROR((LM_WARNING,
                   "(%P|%t) WARNING: MulticastManager::join: failed to join group %C on %C/%C (%@ joined count %B): %m\n",
                   LogAddr(ipv6_multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    }
  }
#endif

  return joined;
}

joined_interface_count()

size_t OpenDDS::DCPS::MulticastManager::joined_interface_count ( ) const

private

Definition at line 68 of file MulticastManager.cpp.

Referenced by join(), and leave().

{
  return joined_interfaces_.size()
#ifdef ACE_HAS_IPV6
    + ipv6_joined_interfaces_.size()
#endif
    ;
}

leave()

void OpenDDS::DCPS::MulticastManager::leave ( const NetworkInterfaceAddress &  nia, const NetworkAddress &  multicast_group_address, ACE_SOCK_Dgram_Mcast &  multicast_socket  )

private

Definition at line 175 of file MulticastManager.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT_CHAR_TO_TCHAR, OpenDDS::DCPS::NetworkInterfaceAddress::address, OpenDDS::DCPS::LogAddr::c_str(), OpenDDS::DCPS::LogLevel::Info, OpenDDS::DCPS::LogAddr::Ip, OpenDDS::DCPS::NetworkInterfaceAddress::is_ipv4(), OpenDDS::DCPS::NetworkInterfaceAddress::is_ipv6(), joined_interface_count(), LM_INFO, LM_WARNING, OpenDDS::DCPS::log_level, OpenDDS::DCPS::NetworkInterfaceAddress::name, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OpenDDS::DCPS::NetworkAddress::to_addr(), and OpenDDS::DCPS::LogLevel::Warning.

Referenced by process().

{
  if (joined_interfaces_.count(nia.name) != 0 && nia.is_ipv4()) {
    if (0 == multicast_socket.leave(multicast_group_address.to_addr(), nia.name.empty() ? 0 : ACE_TEXT_CHAR_TO_TCHAR(nia.name.c_str()))) {
      joined_interfaces_.erase(nia.name);
      if (log_level >= LogLevel::Info) {
        ACE_DEBUG((LM_INFO,
                   "(%P|%t) INFO: MulticastManager::leave: left group %C on %C/%C (%@ joined count %B)\n",
                   LogAddr(multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    } else {
      if (log_level >= LogLevel::Warning) {
        ACE_ERROR((LM_WARNING,
                   "(%P|%t) WARNING: MulticastManager::leave: failed to leave group %C on %C/%C (%@ joined count %B): %m\n",
                   LogAddr(multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    }
  }

#ifdef ACE_HAS_IPV6
  if (ipv6_joined_interfaces_.count(nia.name) != 0 && nia.is_ipv6()) {
    if (0 == ipv6_multicast_socket.leave(ipv6_multicast_group_address.to_addr(), nia.name.empty() ? 0 : ACE_TEXT_CHAR_TO_TCHAR(nia.name.c_str()))) {
      ipv6_joined_interfaces_.erase(nia.name);
      if (log_level >= LogLevel::Info) {
        ACE_DEBUG((LM_INFO,
                   "(%P|%t) INFO: MulticastManager::leave: left group %C on %C/%C (%@ joined count %B)\n",
                   LogAddr(ipv6_multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    } else {
      if (log_level >= LogLevel::Warning) {
        ACE_ERROR((LM_WARNING,
                   "(%P|%t) WARNING: MulticastManager::leave: failed to leave group %C on %C/%C (%@ joined count %B): %m\n",
                   LogAddr(ipv6_multicast_group_address).c_str(),
                   nia.name.empty() ? "all interfaces" : nia.name.c_str(),
                   LogAddr(nia.address, LogAddr::Ip).c_str(),
                   this,
                   joined_interface_count()));
      }
    }
  }
#endif
}

OPENDDS_SET()

OpenDDS::DCPS::MulticastManager::OPENDDS_SET ( OPENDDS_STRING  )

private

process()

bool OpenDDS::DCPS::MulticastManager::process	(	InternalDataReader< NetworkInterfaceAddress >::SampleSequence &	samples,
		InternalSampleInfoSequence &	infos,
		const OPENDDS_STRING &	multicast_interface,
		ACE_Reactor *	reactor,
		ACE_Event_Handler *	event_handler,
		const NetworkAddress &	multicast_group_address,
		ACE_SOCK_Dgram_Mcast &	multicast_socket
	)

Returns true if at least one group was joined.

Definition at line 18 of file MulticastManager.cpp.

References DDS::ALIVE_INSTANCE_STATE, OpenDDS::DCPS::NetworkInterfaceAddress::exclude_from_multicast(), DDS::SampleInfo::instance_state, join(), leave(), and OpenDDS::DCPS::NetworkInterfaceAddress::name.

Referenced by OpenDDS::DCPS::RtpsUdpDataLink::on_data_available(), and OpenDDS::RTPS::Spdp::SpdpTransport::on_data_available().

{
  bool any_joined = false;

  for (size_t idx = 0; idx != samples.size(); ++idx) {
    NetworkInterfaceAddress& nia = samples[idx];
    const DDS::SampleInfo& info = infos[idx];

    if (nia.name.empty()) {
      nia.name = multicast_interface;
    }

    if (info.instance_state == DDS::ALIVE_INSTANCE_STATE) {
      leave(nia, multicast_group_address, multicast_socket
#ifdef ACE_HAS_IPV6
            , ipv6_multicast_group_address, ipv6_multicast_socket
#endif
);
      if (nia.exclude_from_multicast(multicast_interface.c_str())) {
        continue;
      }
      const bool j = join(nia, reactor, event_handler, multicast_group_address, multicast_socket
#ifdef ACE_HAS_IPV6
           , ipv6_multicast_group_address, ipv6_multicast_socket
#endif
           );
      any_joined = any_joined || j;
    } else {
      leave(nia, multicast_group_address, multicast_socket
#ifdef ACE_HAS_IPV6
            , ipv6_multicast_group_address, ipv6_multicast_socket
#endif
            );
    }
  }

  return any_joined;
}

---

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

• MulticastManager.h
• MulticastManager.cpp