ec_default_factory.cpp

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

// EC_Default_Factory.cpp,v 1.45 2003/11/04 05:21:32 dhinton Exp

#include "EC_Default_Factory.h"
#include "EC_Reactive_Dispatching.h"
#include "EC_MT_Dispatching.h"
#include "EC_Basic_Filter_Builder.h"
#include "EC_Prefix_Filter_Builder.h"
#include "EC_ConsumerAdmin.h"
#include "EC_SupplierAdmin.h"
#include "EC_Default_ProxyConsumer.h"
#include "EC_Default_ProxySupplier.h"
#include "EC_Trivial_Supplier_Filter.h"
#include "EC_Per_Supplier_Filter.h"
#include "EC_ObserverStrategy.h"
#include "EC_Null_Scheduling.h"
#include "EC_Group_Scheduling.h"
#include "EC_Reactive_Timeout_Generator.h"
#include "EC_Event_Channel_Base.h"
#include "EC_Reactive_ConsumerControl.h"
#include "EC_Reactive_SupplierControl.h"

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

#include "tao/ORB_Core.h"

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

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

ACE_RCSID(Event, EC_Default_Factory, "EC_Default_Factory.cpp,v 1.45 2003/11/04 05:21:32 dhinton Exp")

TAO_EC_Default_Factory::~TAO_EC_Default_Factory (void)
{
}

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

void
TAO_EC_Default_Factory::unsupported_option_value (const char * option_name,
                                                  const char * option_value)
{
  ACE_ERROR ((LM_ERROR,
              "EC_Default_Factory - "
              "Unsupported <%s> option value: <%s>. "
              "Ignoring this option - using defaults instead.\n",
              option_name,
              option_value));
}

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

  int priority =
    (ACE_Sched_Params::priority_min (ACE_SCHED_FIFO) +
     ACE_Sched_Params::priority_max (ACE_SCHED_FIFO)) / 2;
  this->dispatching_threads_priority_ =
    ACE_Sched_Params::next_priority (ACE_SCHED_FIFO, priority);

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

      if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECDispatching")) == 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
                  this->unsupported_option_value ("-ECDispatching", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECDispatchingThreads")) == 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("-ECFiltering")) == 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->filtering_ = 0;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("basic")) == 0)
                this->filtering_ = 1;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("prefix")) == 0)
                this->filtering_ = 2;
              else
                  this->unsupported_option_value ("-ECFiltering", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECSupplierFilter")) == 0
               // @@ TODO remove, only for backwards compatibility
               || ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECSupplierFiltering")) == 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_filtering_ = 0;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("per-supplier")) == 0)
                this->supplier_filtering_ = 1;
              else
                  this->unsupported_option_value ("-ECSupplierFilter", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECTimeout")) == 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->timeout_ = 0;
              else
                  this->unsupported_option_value ("-ECTimeout", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECObserver")) == 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->observer_ = 0;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("basic")) == 0)
                this->observer_ = 1;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reactive")) == 0)
                this->observer_ = 2;
              else
                  this->unsupported_option_value ("-ECObserver", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECScheduling")) == 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->scheduling_ = 0;
              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("group")) == 0)
                this->scheduling_ = 1;
              else
                  this->unsupported_option_value ("-ECScheduling", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECProxyPushConsumerCollection")) == 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);
              int collection_type = 0;
              int synch_type = 0;
              int iteration_type = 0;

              ACE_TCHAR* aux;
              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,
                                "EC_Default_Factory - "
                                "Unknown consumer collection modifier <%s>.\n", arg));
                }
              ACE_OS::free (opt);
              this->consumer_collection_ =
                 (synch_type << 8) |(collection_type << 4) | iteration_type;
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECProxyPushSupplierCollection")) == 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);
              int collection_type = 0;
              int synch_type = 0;
              int iteration_type = 0;

              ACE_TCHAR* aux;
              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,
                                "EC_Default_Factory - "
                                "Unknown supplier collection modifier <%s>.\n", arg));
                }
              ACE_OS::free(opt);
              this->supplier_collection_ =
                (synch_type << 8) | (collection_type << 4) | iteration_type;
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECProxyConsumerLock")) == 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
                this->unsupported_option_value ("-ECProxyConsumerLock", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECProxySupplierLock")) == 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
                this->unsupported_option_value ("-ECProxySupplierLock", opt);
              arg_shifter.consume_arg ();
            }
        }

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

          if (arg_shifter.is_parameter_next ())
            {
              // Save argument for later use
              this->orbid_ = ACE_TEXT_ALWAYS_CHAR(arg_shifter.get_current ());
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECConsumerControl")) == 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
                this->unsupported_option_value ("-ECConsumerControl", opt);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_LIB_TEXT("-ECSupplierControl")) == 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