cec_default_factory.cpp

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

// CEC_Default_Factory.cpp,v 1.22 2003/11/04 05:21:32 dhinton Exp

#include "CEC_Default_Factory.h"
#include "CEC_MT_Dispatching.h"
#include "CEC_Reactive_Pulling_Strategy.h"
#include "CEC_ConsumerAdmin.h"
#include "CEC_SupplierAdmin.h"
#include "CEC_ProxyPushConsumer.h"
#include "CEC_ProxyPullConsumer.h"
#include "CEC_ProxyPushSupplier.h"
#include "CEC_ProxyPullSupplier.h"
#include "CEC_EventChannel.h"
#include "CEC_Reactive_ConsumerControl.h"
#include "CEC_Reactive_SupplierControl.h"

#include "orbsvcs/ESF/ESF_Immediate_Changes.h"
#include "orbsvcs/ESF/ESF_Delayed_Changes.h"
#include "orbsvcs/ESF/ESF_Copy_On_Write.h"
#include "orbsvcs/ESF/ESF_Copy_On_Read.h"
#include "orbsvcs/ESF/ESF_Proxy_List.h"
#include "orbsvcs/ESF/ESF_Proxy_RB_Tree.h"

#include "ace/Arg_Shifter.h"
#include "ace/Sched_Params.h"
#include "ace/OS_NS_strings.h"

#if defined (TAO_HAS_TYPED_EVENT_CHANNEL)
#include "CEC_TypedConsumerAdmin.h"
#include "CEC_TypedSupplierAdmin.h"
#include "CEC_TypedEventChannel.h"
#endif /* TAO_HAS_TYPED_EVENT_CHANNEL */

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

ACE_RCSID (CosEvent, 
           CEC_Default_Factory, 
           "CEC_Default_Factory.cpp,v 1.22 2003/11/04 05:21:32 dhinton Exp")

TAO_CEC_Default_Factory::~TAO_CEC_Default_Factory (void)
{
    if (orbid_dupped_ != 0) 
      {
        ACE_OS::free (orbid_);
      }
}

int
TAO_CEC_Default_Factory::init_svcs (void)
{
  return 
    ACE_Service_Config::static_svcs ()->insert (
        &ace_svc_desc_TAO_CEC_Default_Factory
      );
}

int
TAO_CEC_Default_Factory::parse_collection_arg (ACE_TCHAR* opt)
{
  int collection_type = 0;
  int synch_type = 0;
  int iteration_type = 0;

  ACE_TCHAR* aux = 0;
  for (ACE_TCHAR* arg = ACE_OS::strtok_r (opt, ACE_LIB_TEXT(":"), &aux);
       arg != 0;
       arg = ACE_OS::strtok_r (0, ACE_LIB_TEXT(":"), &aux))
    {
      if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("mt")) == 0)
        synch_type = 0;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("st")) == 0)
        synch_type = 1;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("list")) == 0)
        collection_type = 0;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("rb_tree")) == 0)
        collection_type = 1;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("immediate")) == 0)
        iteration_type = 0;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("copy_on_read")) == 0)
        iteration_type = 1;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("copy_on_write")) == 0)
        iteration_type = 2;
      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("delayed")) == 0)
        iteration_type = 3;
      else
        ACE_ERROR ((LM_ERROR,
                    "CEC_Default_Factory - "
                    "unknown collection modifier <%s>\n",
                    arg));
    }
  return (synch_type << 8) |(collection_type << 4) | iteration_type;
}

int
TAO_CEC_Default_Factory::init (int argc, ACE_TCHAR* argv[])
{
  ACE_Arg_Shifter arg_shifter (argc, argv);

  this->dispatching_threads_flags_ =
    THR_SCHED_DEFAULT|THR_BOUND|THR_NEW_LWP;
  this->dispatching_threads_priority_ =
    ACE_THR_PRI_OTHER_DEF;

  while (arg_shifter.is_anything_left ())
    {
      const ACE_TCHAR *arg = arg_shifter.get_current ();

      if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECDispatching")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reactive")) == 0)
                {
                  this->dispatching_ = 0;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("mt")) == 0)
                {
                  this->dispatching_ = 1;
                }
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              "CEC_Default_Factory - "
                              "unsupported dispatching <%s>\n",
                              opt));
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECDispatchingThreads")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              this->dispatching_threads_ = ACE_OS::atoi (opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECProxyConsumerCollection")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR *current_arg = arg_shifter.get_current ();
              ACE_TCHAR *opt = ACE_OS::strdup (current_arg);
              this->consumer_collection_ = this->parse_collection_arg (opt);
              ACE_OS::free (opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECProxySupplierCollection")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR *current_arg = arg_shifter.get_current();
              ACE_TCHAR* opt = ACE_OS::strdup(current_arg);
              this->supplier_collection_ =
                this->parse_collection_arg (opt);
              ACE_OS::free(opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECProxyConsumerLock")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("null")) == 0)
                {
                  this->consumer_lock_ = 0;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("thread")) == 0)
                {
                  this->consumer_lock_ = 1;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("recursive")) == 0)
                {
                  this->consumer_lock_ = 2;
                }
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              "CEC_Default_Factory - "
                              "unsupported consumer lock <%s>\n",
                              opt));
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECProxySupplierLock")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("null")) == 0)
                {
                  this->supplier_lock_ = 0;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("thread")) == 0)
                {
                  this->supplier_lock_ = 1;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("recursive")) == 0)
                {
                  this->supplier_lock_ = 2;
                }
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              "CEC_Default_Factory - "
                              "unsupported supplier lock <%s>\n",
                              opt));
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECReactivePullingPeriod")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              this->reactive_pulling_period_ = ACE_OS::atoi (opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECUseORBId")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              // Copy argument for later use
              this->orbid_ = ACE_OS::strdup(ACE_TEXT_ALWAYS_CHAR(arg_shifter.get_current ()));
              orbid_dupped_ = 1;
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECConsumerControl")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("null")) == 0)
                {
                  this->consumer_control_ = 0;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reactive")) == 0)
                {
                  this->consumer_control_ = 1;
                }
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              "CEC_Default_Factory - "
                              "unsupported consumer control <%s>\n",
                              opt));
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECSupplierControl")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("null")) == 0)
                {
                  this->supplier_control_ = 0;
                }
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reactive")) == 0)
                {
                  this->supplier_control_ = 1;
                }
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              "CEC_Default_Factory - "
                              "unsupported supplier control <%s>\n",
                              opt));
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-CECConsumerControlPeriod")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              this->consumer_control_period_ = ACE_OS::atoi (opt);
              arg_shifter.consume_arg ();
            }