TAO_ECG_Mcast_EH Class Reference

Event Handler for Mcast messages. NOT THREAD-SAFE. More...

#include <ECG_Mcast_EH.h>

Inheritance diagram for TAO_ECG_Mcast_EH:

[legend]Collaboration diagram for TAO_ECG_Mcast_EH:

[legend]List of all members.

Public Member Functions
virtual int	handle_input (ACE_HANDLE fd)
	TAO_ECG_Mcast_EH (TAO_ECG_Dgram_Handler *recv, const ACE_TCHAR *net_if=0, CORBA::ULong buf_sz=0)
	Initialization and termination methods.
virtual	~TAO_ECG_Mcast_EH (void)
	Destructor.
void	open (RtecEventChannelAdmin::EventChannel_ptr ec)
virtual int	shutdown (void)
	TAO_ECG_Handler_Shutdown method.
Private Types
typedef ACE_Unbounded_Set< ACE_INET_Addr >	Address_Set
typedef ACE_Array_Base< Subscription >	Subscriptions
Private Member Functions
void	update_consumer (const RtecEventChannelAdmin::ConsumerQOS &sub)
void	compute_required_subscriptions (const RtecEventChannelAdmin::ConsumerQOS &sub, Address_Set &multicast_addresses)
int	delete_unwanted_subscriptions (Address_Set &multicast_addresses)
void	add_new_subscriptions (Address_Set &multicast_addresses)
Private Attributes
ACE_TCHAR *	net_if_
	The NIC used to subscribe for multicast traffic.
Subscriptions	subscriptions_
	List of multicast addresses we subscribe to and dgrams we use.
TAO_ECG_Dgram_Handler *	receiver_
	We callback to this object when a message arrives.
CORBA::ULong	recvbuf_size_
	SOCKbuf size.
TAO_EC_Servant_Var< Observer >	observer_
TAO_EC_Auto_Command< Observer_Disconnect_Command >	auto_observer_disconnect_
Friends
class	Observer
	Make update_consumer () accessible to Observer.
Classes
class	Observer
	Observes changes in the EC consumer subscriptions and notifies TAO_ECG_Mcast_EH when there are changes. More...
class	Observer_Disconnect_Command
	Disconnects Observer from the Event Channel. More...
struct	Subscription

---

Detailed Description

Event Handler for Mcast messages. NOT THREAD-SAFE.

This object acts as an Observer to Event Channel. It subscribes to multicast groups that carry events matching the EC's subscriptions. This object then receives callbacks from the Reactor when data is available on the mcast sockets and alerts TAO_ECG_Dgram_Handler, which reads the data, transforms it into event and pushes to the Event Channel.

Definition at line 52 of file ECG_Mcast_EH.h.

---

Member Typedef Documentation

typedef ACE_Unbounded_Set<ACE_INET_Addr> TAO_ECG_Mcast_EH::Address_Set [private]

Definition at line 148 of file ECG_Mcast_EH.h.

typedef ACE_Array_Base<Subscription> TAO_ECG_Mcast_EH::Subscriptions [private]

Definition at line 230 of file ECG_Mcast_EH.h.

---

Constructor & Destructor Documentation

TAO_BEGIN_VERSIONED_NAMESPACE_DECL TAO_ECG_Mcast_EH::TAO_ECG_Mcast_EH	(	TAO_ECG_Dgram_Handler *	recv,
		const ACE_TCHAR *	net_if = 0,
		CORBA::ULong	buf_sz = 0
	)

Initialization and termination methods.

Constructor. Messages received by this EH will be forwarded to the recv. net_if can be used to specify NIC where multicast messages are expected. would be used to alter the default buffer size.

See comments for receiver_ data member on why raw pointer is used for the recv argument.

Definition at line 20 of file ECG_Mcast_EH.cpp.

References ACE_ASSERT.

  : net_if_ (net_if ? ACE_OS::strdup (net_if) : 0)
  , subscriptions_ ()
  , receiver_ (recv)
  , recvbuf_size_ (sz)
  , observer_ ()
  , auto_observer_disconnect_ ()
{
  ACE_ASSERT (this->receiver_);
}

TAO_ECG_Mcast_EH::~TAO_ECG_Mcast_EH

(

void

)

[virtual]

Destructor.

Definition at line 33 of file ECG_Mcast_EH.cpp.

References ACE_OS::free().

{
  ACE_OS::free (this->net_if_);
}

---

Member Function Documentation

void TAO_ECG_Mcast_EH::add_new_subscriptions

(

Address_Set &

multicast_addresses

)

[private]

Parameters:

multicast_addresses

List of multicast addresses to which we need to subscribe to in order to be receiving all event types in the current consumer subscriptions.

Definition at line 231 of file ECG_Mcast_EH.cpp.

References ACE_ERROR, ACE_NEW, ACE_NONBLOCK, ACE_Unbounded_Set_Ex< T, C >::begin(), ACE_Unbounded_Set_Ex< T, C >::end(), ENOTSUP, LM_ERROR, ACE_Event_Handler::READ_MASK, ACE_Array_Base< T >::size(), SO_RCVBUF, socket(), SOL_SOCKET, and subscriptions_.

Referenced by update_consumer().

{
  typedef ACE_Unbounded_Set_Iterator<ACE_INET_Addr> Address_Iterator;
  for (Address_Iterator k = multicast_addresses.begin ();
       k != multicast_addresses.end ();
       ++k)
    {
      Subscription new_subscription;
      new_subscription.mcast_addr = *k;
      ACE_NEW (new_subscription.dgram, ACE_SOCK_Dgram_Mcast);

      size_t const subscriptions_size = this->subscriptions_.size ();
      this->subscriptions_.size (subscriptions_size + 1);
      this->subscriptions_[subscriptions_size] = new_subscription;

      ACE_SOCK_Dgram_Mcast *socket = new_subscription.dgram;

      if (socket->open (new_subscription.mcast_addr, this->net_if_, 1) == -1) {
        ACE_ERROR ((LM_ERROR,
                    "Error: %d - Unable to open multicast socket\n",
                    errno ));
      }

      if ( socket->enable (ACE_NONBLOCK) != 0 ) {
        ACE_ERROR ((LM_ERROR,
                    "Error: %d - Unable to enable nonblocking on mcast_eh\n",
                    errno ));
      }

      if (socket->join (new_subscription.mcast_addr, 1, this->net_if_) == -1) {
        ACE_ERROR ((LM_ERROR,
                    "Error: %d - Unable to join multicast group\n",
                    errno ));
      }

      if (this->recvbuf_size_ != 0
          && (((ACE_SOCK_Dgram *)socket)->set_option(SOL_SOCKET,
                                                     SO_RCVBUF,
                                                     (void *) &this->recvbuf_size_,
                                                     sizeof (this->recvbuf_size_)) == -1)
          && errno != ENOTSUP )
        {
          ACE_ERROR ((LM_ERROR,
                      "Error: %d - Unable to set mcast_eh recvbuf_size:%d\n",
                      errno,
                      this->recvbuf_size_));
        }
      (void) this->reactor ()->register_handler (
                                         socket->get_handle (),
                                         this,
                                         ACE_Event_Handler::READ_MASK);
    }
}

void TAO_ECG_Mcast_EH::compute_required_subscriptions	(	const RtecEventChannelAdmin::ConsumerQOS &	sub,
		Address_Set &	multicast_addresses
	)			[private]

Parameters:

	sub	The list of event types that our event channel consumers are interested in.
	multicast_addresses	This method populates this list with multicast addresses that we need to be subscribed to in order to receive event types specified in /a sub.

Exceptions:

CORBA::SystemException

This method needs to perform several CORBA invocations, and it propagates any exceptions back to the caller.

Definition at line 146 of file ECG_Mcast_EH.cpp.

References ACE_ES_EVENT_UNDEFINED, RtecEventChannelAdmin::ConsumerQOS::dependencies, TAO_ECG_Dgram_Handler::get_addr(), TAO_ECG_Dgram_Handler::get_address(), inet_addr(), ACE_Unbounded_Set_Ex< T, C >::insert(), RtecUDPAdmin::UDP_Addr::ipaddr, RtecUDPAdmin::UDP_Addr::port, receiver_, RtecUDPAdmin::Rtec_inet, and RtecUDPAdmin::Rtec_inet6.

Referenced by update_consumer().

{
  CORBA::ULong count = sub.dependencies.length ();
  for (CORBA::ULong i = 0; i != count; ++i)
    {
      const RtecEventComm::EventHeader& header =
                           sub.dependencies[i].event.header;
      if (0 < header.type && header.type < ACE_ES_EVENT_UNDEFINED)
        {
          continue;
        }

      ACE_INET_Addr inet_addr;
      try
        {
          // Grab the right mcast group for this event...
          RtecUDPAdmin::UDP_Address_var udp_addr;

          this->receiver_->get_address (header, udp_addr.out());
          switch (udp_addr->_d())
            {
            case RtecUDPAdmin::Rtec_inet:
              inet_addr.set(udp_addr->v4_addr().port,
                            udp_addr->v4_addr().ipaddr);
              break;
            case RtecUDPAdmin::Rtec_inet6:
#if defined (ACE_HAS_IPV6)
              inet_addr.set_type(PF_INET6);
#endif
              inet_addr.set_address(udp_addr->v6_addr().ipaddr,16,0);
              inet_addr.set_port_number(udp_addr->v6_addr().port);
              break;
            }
        }
      catch (const ::CORBA::BAD_OPERATION &)
        {
          // server only supports IPv4
           // Grab the right mcast group for this event...
          RtecUDPAdmin::UDP_Addr udp_addr;
          this->receiver_->get_addr (header, udp_addr);
          inet_addr.set (udp_addr.port, udp_addr.ipaddr);
        }

      // Ignore errors, if the element is in the set we simply ignore
      // the problem...
      (void) multicast_addresses.insert (inet_addr);
    }
}

int TAO_ECG_Mcast_EH::delete_unwanted_subscriptions

(

Address_Set &

multicast_addresses

)

[private]

Parameters:

multicast_addresses

List of multicast addresses we need to be subscribed to in order receive all event types in the current consumer subscriptions.

Definition at line 198 of file ECG_Mcast_EH.cpp.

References ACE_Unbounded_Set_Ex< T, C >::find(), ACE_Event_Handler::READ_MASK, ACE_Unbounded_Set_Ex< T, C >::remove(), ACE_Array_Base< T >::size(), socket(), and subscriptions_.

Referenced by update_consumer().

{
  for (size_t i = 0; i < this->subscriptions_.size (); ++i)
    {
      ACE_INET_Addr multicast_group = this->subscriptions_[i].mcast_addr;
      if (multicast_addresses.find (multicast_group))
        {
          // Remove from the list of subscriptions to be added,
          // because we already subscribe to it...
          (void) multicast_addresses.remove (multicast_group);
          continue;
        }

      // This subscription is no longer needed - remove from reactor,
      // close and delete the socket.
      ACE_SOCK_Dgram_Mcast *socket = this->subscriptions_[i].dgram;
      (void) this->reactor ()->remove_handler (socket->get_handle (),
                                               ACE_Event_Handler::READ_MASK);
      (void) socket->close();
      delete socket;
      // Move the deleted subscription out of the <subscriptions_>
      // array by moving the last subscription in array into its place.
      this->subscriptions_[i] =
        this->subscriptions_[this->subscriptions_.size () - 1];
      this->subscriptions_.size (this->subscriptions_.size () - 1);
      --i;
    }

    return 0;
}

int TAO_ECG_Mcast_EH::handle_input

(

ACE_HANDLE

fd

)

[virtual]

Reactor callback. Notify receiver_ that a dgram corresponding to fd is ready for reading.

Reimplemented from ACE_Event_Handler.

Definition at line 117 of file ECG_Mcast_EH.cpp.

References ACE_Array_Base< T >::size(), socket(), and subscriptions_.

{
  size_t const subscriptions_size = this->subscriptions_.size ();
  for (size_t i = 0; i != subscriptions_size; ++i)
    {
      ACE_SOCK_Dgram_Mcast *socket = this->subscriptions_[i].dgram;
      if (socket->get_handle () == fd)
        {
          return this->receiver_->handle_input (*socket);
        }
    }
  return -1;
}

void TAO_ECG_Mcast_EH::open

(

RtecEventChannelAdmin::EventChannel_ptr

ec

)

Register for changes in the EC subscription list. When the subscription list becomes non-empty we join the proper multicast groups (using Dgram_Handler to translate between event types and mcast groups) and the class registers itself with the reactor.

To insure proper resource clean up, if open () is successful, the user MUST call shutdown () when handler is no longer needed (and its reactor still exists).

Definition at line 39 of file ECG_Mcast_EH.cpp.

References ACE_ERROR, ACE_NEW, activate(), auto_observer_disconnect_, TAO_Objref_Var_T< T >::in(), CORBA::is_nil(), LM_ERROR, and observer_.

Referenced by TAO_ECG_Mcast_Gateway::init_handler().

{
  if (!this->receiver_)
    {
      // We are shut down.
      throw CORBA::INTERNAL();
    }

  if (CORBA::is_nil (ec))
    {
      ACE_ERROR ((LM_ERROR,  "TAO_ECG_Mcast_EH::open(): "
                             "nil ec argument"));
      throw CORBA::INTERNAL ();
    }

  // Create and activate Event Channel Observer.
  ACE_NEW (this->observer_,
           Observer (this));

  if (!this->observer_.in ())
    {
      throw CORBA::NO_MEMORY ();
    }

  TAO_EC_Object_Deactivator observer_deactivator;
  RtecEventChannelAdmin::Observer_var observer_ref;
  PortableServer::POA_var poa =
    this->observer_->_default_POA ();

  activate (observer_ref,
            poa.in (),
            this->observer_.in (),
            observer_deactivator);

  RtecEventChannelAdmin::Observer_Handle handle =
    ec->append_observer (observer_ref.in ());

  this->observer_->set_deactivator (observer_deactivator);
  this->auto_observer_disconnect_.set_command
    (Observer_Disconnect_Command (handle, ec));
}

int TAO_ECG_Mcast_EH::shutdown

(

void

)

[virtual]

TAO_ECG_Handler_Shutdown method.

Remove ourselves from the event channel, unsubscribe from the multicast groups, close the sockets and deregister from the reactor.

Implements TAO_ECG_Handler_Shutdown.

Definition at line 82 of file ECG_Mcast_EH.cpp.

References auto_observer_disconnect_, observer_, ACE_Event_Handler::READ_MASK, receiver_, ACE_Array_Base< T >::size(), and subscriptions_.

{
  // Already shut down.
  if (!this->receiver_)
    return -1;

  // Disconnect Observer from EC.
  this->auto_observer_disconnect_.execute ();

  // Shutdown the observer.
  if (this->observer_.in ())
    {
      this->observer_->shutdown ();
      this->observer_ = 0;
    }

  // Indicates that we are in a shutdown state.
  this->receiver_ = 0;

  // Deregister from reactor, close and clean up sockets.
  size_t const subscriptions_size = this->subscriptions_.size ();
  for (size_t i = 0; i != subscriptions_size; ++i)
    {
      (void) this->reactor ()->remove_handler (
                               this->subscriptions_[i].dgram->get_handle (),
                               ACE_Event_Handler::READ_MASK);
      (void) this->subscriptions_[i].dgram->close();
      delete this->subscriptions_[i].dgram;
    }
  this->subscriptions_.size (0);

  return 0;
}

void TAO_ECG_Mcast_EH::update_consumer

(

const RtecEventChannelAdmin::ConsumerQOS &

sub

)

[private]

The Observer method. Subscribe/unsubscribe to multicast groups according to changes in consumer subscriptions.

Definition at line 132 of file ECG_Mcast_EH.cpp.

References add_new_subscriptions(), compute_required_subscriptions(), and delete_unwanted_subscriptions().

Referenced by TAO_ECG_Mcast_EH::Observer::update_consumer().

{
  Address_Set multicast_addresses;

  this->compute_required_subscriptions (sub,
                                        multicast_addresses);

  this->delete_unwanted_subscriptions (multicast_addresses);

  this->add_new_subscriptions (multicast_addresses);
}

---

Friends And Related Function Documentation

friend class Observer [friend]

Make update_consumer () accessible to Observer.

Definition at line 141 of file ECG_Mcast_EH.h.

---

Member Data Documentation

TAO_EC_Auto_Command<Observer_Disconnect_Command> TAO_ECG_Mcast_EH::auto_observer_disconnect_ [private]

Manages connection of our observer to the Event Channel. ORDER DEPENDENCY: this member should be declared AFTER <observer_>.

Definition at line 274 of file ECG_Mcast_EH.h.

Referenced by open(), and shutdown().

ACE_TCHAR* TAO_ECG_Mcast_EH::net_if_ [private]

The NIC used to subscribe for multicast traffic.

Definition at line 224 of file ECG_Mcast_EH.h.

TAO_EC_Servant_Var<Observer> TAO_ECG_Mcast_EH::observer_ [private]

Event Channel Observer. Detects changes in EC consumer subscriptions. ORDER DEPENDENCY: this member should be declared before <auto_observer_disconnect_>.

Definition at line 270 of file ECG_Mcast_EH.h.

Referenced by open(), and shutdown().

TAO_ECG_Dgram_Handler* TAO_ECG_Mcast_EH::receiver_ [private]

We callback to this object when a message arrives.

Definition at line 262 of file ECG_Mcast_EH.h.

Referenced by compute_required_subscriptions(), and shutdown().

CORBA::ULong TAO_ECG_Mcast_EH::recvbuf_size_ [private]

SOCKbuf size.

Definition at line 265 of file ECG_Mcast_EH.h.

Subscriptions TAO_ECG_Mcast_EH::subscriptions_ [private]

List of multicast addresses we subscribe to and dgrams we use.

Definition at line 250 of file ECG_Mcast_EH.h.

Referenced by add_new_subscriptions(), delete_unwanted_subscriptions(), handle_input(), and shutdown().

---

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

• ECG_Mcast_EH.h
• ECG_Mcast_EH.cpp

---