cec_typedeventchannel.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

// -*- C++ -*-
//
// CEC_TypedEventChannel.cpp,v 1.5 2003/12/10 09:25:06 jwillemsen Exp

#include "CEC_TypedEventChannel.h"
#include "CEC_Dispatching.h"
#include "CEC_TypedConsumerAdmin.h"
#include "CEC_TypedSupplierAdmin.h"
#include "CEC_ConsumerControl.h"
#include "CEC_SupplierControl.h"
#include "tao/debug.h"
#include "ace/Dynamic_Service.h"

#if ! defined (__ACE_INLINE__)
#include "CEC_TypedEventChannel.i"
#endif /* __ACE_INLINE__ */


// Implementation skeleton constructor
TAO_CEC_TypedEventChannel::
TAO_CEC_TypedEventChannel (const TAO_CEC_TypedEventChannel_Attributes& attr,
                           TAO_CEC_Factory* factory,
                           int own_factory)
  : typed_supplier_poa_ (PortableServer::POA::_duplicate (attr.typed_supplier_poa)),
    typed_consumer_poa_ (PortableServer::POA::_duplicate (attr.typed_consumer_poa)),
    orb_ (CORBA::ORB::_duplicate (attr.orb)),
    interface_repository_ (CORBA::Repository::_duplicate (attr.interface_repository)),
    factory_ (factory),
    own_factory_ (own_factory),
    consumer_reconnect_ (attr.consumer_reconnect),
    supplier_reconnect_ (attr.supplier_reconnect),
    disconnect_callbacks_ (attr.disconnect_callbacks),
    destroy_on_shutdown_ (attr.destroy_on_shutdown),
    destroyed_ (0)
{
  if (this->factory_ == 0)
    {
      this->factory_ =
        ACE_Dynamic_Service<TAO_CEC_Factory>::instance ("CEC_Factory");
      this->own_factory_ = 0;
      ACE_ASSERT (this->factory_ != 0);
    }

  this->dispatching_ =
    this->factory_->create_dispatching (this);
  this->typed_consumer_admin_ =
    this->factory_->create_consumer_admin (this);
  this->typed_supplier_admin_ =
    this->factory_->create_supplier_admin (this);
  this->consumer_control_ =
    this->factory_->create_consumer_control (this);
  this->supplier_control_ =
    this->factory_->create_supplier_control (this);
}

// Implementation skeleton destructor
TAO_CEC_TypedEventChannel::~TAO_CEC_TypedEventChannel (void)
{
  this->clear_ifr_cache ();
  this->interface_description_.close ();

  this->factory_->destroy_dispatching (this->dispatching_);
  this->dispatching_ = 0;

  this->factory_->destroy_consumer_admin (this->typed_consumer_admin_);
  this->typed_consumer_admin_ = 0;
  this->factory_->destroy_supplier_admin (this->typed_supplier_admin_);
  this->typed_supplier_admin_ = 0;

  if (this->own_factory_)
    delete this->factory_;
}

void
TAO_CEC_TypedEventChannel::activate (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
{
  this->dispatching_->activate ();
  this->consumer_control_->activate ();
  this->supplier_control_->activate ();
}

void
TAO_CEC_TypedEventChannel::shutdown (ACE_ENV_SINGLE_ARG_DECL)
{
  this->dispatching_->shutdown ();
  this->supplier_control_->shutdown ();
  this->consumer_control_->shutdown ();

  PortableServer::POA_var typed_consumer_poa =
    this->typed_consumer_admin_->_default_POA (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  PortableServer::ObjectId_var typed_consumer_id =
    typed_consumer_poa->servant_to_id (this->typed_consumer_admin_ ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
  typed_consumer_poa->deactivate_object (typed_consumer_id.in () ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  PortableServer::POA_var typed_supplier_poa =
    this->typed_supplier_admin_->_default_POA (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  PortableServer::ObjectId_var typed_supplier_id =
    typed_supplier_poa->servant_to_id (this->typed_supplier_admin_ ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
  typed_supplier_poa->deactivate_object (typed_supplier_id.in () ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  this->typed_supplier_admin_->shutdown (ACE_ENV_SINGLE_ARG_PARAMETER);

  this->typed_consumer_admin_->shutdown (ACE_ENV_SINGLE_ARG_PARAMETER);

  if (destroy_on_shutdown_)
    {
      // Deactivate the Typed EC
      PortableServer::POA_var t_poa =
        this->_default_POA (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_CHECK;

      PortableServer::ObjectId_var t_id =
        t_poa->servant_to_id (this ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      t_poa->deactivate_object (t_id.in () ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;

      this->orb_->shutdown(0);
      ACE_CHECK;
    }
}

void
TAO_CEC_TypedEventChannel::connected (TAO_CEC_TypedProxyPushConsumer* consumer
                                      ACE_ENV_ARG_DECL)
{
  this->typed_supplier_admin_->connected (consumer ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_CEC_TypedEventChannel::reconnected (TAO_CEC_TypedProxyPushConsumer* consumer
                                        ACE_ENV_ARG_DECL)
{
  this->typed_supplier_admin_->reconnected (consumer ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_CEC_TypedEventChannel::disconnected (TAO_CEC_TypedProxyPushConsumer* consumer
                                         ACE_ENV_ARG_DECL)
{
  this->typed_supplier_admin_->disconnected (consumer ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_CEC_TypedEventChannel::connected (TAO_CEC_ProxyPushSupplier* supplier
                                      ACE_ENV_ARG_DECL)
{
  this->typed_consumer_admin_->connected (supplier ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_CEC_TypedEventChannel::reconnected (TAO_CEC_ProxyPushSupplier* supplier
                                        ACE_ENV_ARG_DECL)
{
  this->typed_consumer_admin_->reconnected (supplier ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

void
TAO_CEC_TypedEventChannel::disconnected (TAO_CEC_ProxyPushSupplier* supplier
                                         ACE_ENV_ARG_DECL)
{
  this->typed_consumer_admin_->disconnected (supplier ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;
}

// Find from the ifr cache the operation and return the parameter array pointer.
TAO_CEC_Operation_Params *
TAO_CEC_TypedEventChannel::find_from_ifr_cache (const char *operation)
{
  TAO_CEC_Operation_Params *found = 0;

  this->interface_description_.find (operation, found);

  return found;
}

// Insert the operation and its parameters into the ifr cache.
int
TAO_CEC_TypedEventChannel::insert_into_ifr_cache (const char *operation_,
                                                  TAO_CEC_Operation_Params *parameters_)
{
  // Make sure that the supplied Object reference is valid,
  // i.e. not nil.
  if (operation_ == 0 || parameters_ == 0)
    {
      errno = EINVAL;
      return -1;
    };

  CORBA::String_var operation = CORBA::string_dup (operation_);

  int result = this->interface_description_.bind (operation.in (), parameters_);

  if (result == 0)
    {
      // Transfer ownership to the Object InterfaceDescription map.
      (void) operation._retn ();
    }

  return result;
}

// Clear the ifr cache, freeing up all its contents.
int
TAO_CEC_TypedEventChannel::clear_ifr_cache (void)
{
  for (Iterator i = this->interface_description_.begin ();
       i != this->interface_description_.end ();
       ++i)
    {
      if (TAO_debug_level >= 10)
        {
          ACE_DEBUG ((LM_DEBUG, "***** Destroying operation %s from ifr cache *****\n",
                      ACE_const_cast (char *, (*i).ext_id_)));
        }

      // Deallocate the operation
      CORBA::string_free (ACE_const_cast (char *, (*i).ext_id_));

      // Destroy the parameter
      delete ((*i).int_id_);
    }

  int result = this->interface_description_.unbind_all ();

  return result;
}


// The function performs a lookup_id of the passed interface in the IFR,
// and then obtains the FullInterfaceDescription.
// The base interfaces for the interface are stored on this class.
// All the operations and their parameters are then inserted in the ifr cache.
// Function returns 0 if successful or -1 on a failure.
int
TAO_CEC_TypedEventChannel::cache_interface_description (const char *interface_
                                                        ACE_ENV_ARG_DECL)
{
  ACE_TRY
    {
      // Lookup the Interface Name in the IFR
      CORBA::Contained_var contained =
        this->interface_repository_->lookup_id (interface_ ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      // Narrow the interface
      CORBA::InterfaceDef_var interface =
        CORBA::InterfaceDef::_narrow (contained.in () ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;

      if (CORBA::is_nil (interface.in () ))
        {
          if (TAO_debug_level >= 10)
            {
              ACE_DEBUG ((LM_DEBUG, "***** CORBA::InterfaceDef::_narrow failed for interface %s *****\n", interface_ ));
            }
          return -1;
        }
      else
        {
          // Obtain the full interface description
          CORBA::InterfaceDef::FullInterfaceDescription_var fid =
            interface->describe_interface (ACE_ENV_SINGLE_ARG_PARAMETER);
          ACE_TRY_CHECK;

          // Obtain the base interfaces
          this->base_interfaces_ = fid->base_interfaces;
          if (TAO_debug_level >= 10)
            {
              for (CORBA::ULong base=0; base<fid->base_interfaces.length(); base++)
                {
                  ACE_DEBUG ((LM_DEBUG, "***** Base interface %s found on interface %s *****\n",
                              fid->base_interfaces[base].in(),
                              interface_ ));
                }