adminproperties.cpp

来自「这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用」· C++ 代码 · 共 525 行 · 第 1/2 页

// AdminProperties.cpp,v 1.7 2003/07/06 21:41:57 pradeep Exp

#include "AdminProperties.h"
#include "ace/Arg_Shifter.h"
#include "ace/Get_Opt.h"
#include "tao/debug.h"

/***************************************************************************/

AdminProperties_Task::AdminProperties_Task (void)
  :supplier_ (0), client_ (0)
{
}

void
AdminProperties_Task::init (TAO_Notify_Tests_StructuredPushSupplier *supplier, AdminProperties* client)
{
  supplier_  = supplier;
  client_ = client;
}

int
AdminProperties_Task::init (int argc, ACE_TCHAR *argv[])
{
  return ACE_Task_Base::init (argc,
                                  argv);
}

int
AdminProperties_Task::svc (void)
{
  // operations:
  CosNotification::StructuredEvent event;

  // EventHeader

  // FixedEventHeader
  // EventType
  // string
  event.header.fixed_header.event_type.domain_name = CORBA::string_dup("*");
  // string
  event.header.fixed_header.event_type.type_name = CORBA::string_dup("*");
  // string
  event.header.fixed_header.event_name = CORBA::string_dup("myevent");

  // OptionalHeaderFields
  // PropertySeq
  // sequence<Property>: string name, any value
  event.header.variable_header.length (0); // put nothing here

  // FilterableEventBody
  // PropertySeq
  // sequence<Property>: string name, any value
  event.filterable_data.length (3);
  event.filterable_data[0].name = CORBA::string_dup("threshold");

  event.filterable_data[1].name = CORBA::string_dup("temperature");
  event.filterable_data[1].value <<= (CORBA::Long)70;

  event.filterable_data[2].name = CORBA::string_dup("pressure");
  event.filterable_data[2].value <<= (CORBA::Long)80;

  // @@ CORBA::Short prio = CosNotification::LowestPriority;

 int event_count = this->client_->event_count_;

 ACE_DEBUG ((LM_DEBUG, "Supplier sending %d events...\n", event_count));

 ACE_DECLARE_NEW_CORBA_ENV;

 for (int i = 0 ; i < event_count; ++i)
   {
     event.filterable_data[0].value <<= (CORBA::Long)i;

     // Any.
     event.remainder_of_body <<= (CORBA::Long)i;

     ACE_TRY
        {
          this->supplier_->send_event (event ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
      ACE_CATCH (CORBA::IMP_LIMIT, impl_limit)
        {
          if (TAO_debug_level)
            ACE_DEBUG ((LM_DEBUG, "Event %d was not send due to Impl Limit reached\n", i));

          this->client_->was_rejected_ = 1;
        }
      ACE_CATCHANY
        {
          ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "Exception sending event\n");
        }
      ACE_ENDTRY;
   }

return 0;
}
/***************************************************************************/

AdminProperties_StructuredPushConsumer::AdminProperties_StructuredPushConsumer (AdminProperties* client)
  :client_ (client), events_received_ (0)
{
}

void
AdminProperties_StructuredPushConsumer::push_structured_event (const CosNotification::StructuredEvent & /*notification*/
                                                               ACE_ENV_ARG_DECL_NOT_USED
                                                               )
  ACE_THROW_SPEC ((
                     CORBA::SystemException,
                     CosEventComm::Disconnected
                     ))
{
  ++events_received_;

  if (TAO_debug_level)
    ACE_DEBUG ((LM_DEBUG, "Consumer %x received event %d\n", this, events_received_.value ()));

  ACE_OS::sleep (this->client_->consumer_delay_);
}

/***************************************************************************/

AdminProperties::AdminProperties (void)
  : max_queue_length_ (10),
    max_consumers_ (3),
    max_suppliers_ (3),
    reject_new_events_ (0),
    consumer_delay_ (0, 0),
    initial_delay_ (5, 0),
    consumers_ (4),
    suppliers_ (4),
    event_count_ (30),
    suppliers_connected_count_ (0),
    consumers_connected_count_ (0),
    was_rejected_ (0)
{
}

AdminProperties::~AdminProperties (void)
{
}

int
AdminProperties::parse_args(int argc, char *argv[])
{
  ACE_Arg_Shifter arg_shifter (argc, argv);

  const char *current_arg = 0;

  while (arg_shifter.is_anything_left ())
    {
      if ((current_arg = arg_shifter.get_the_parameter ("-max_queue_length")))
        {
          this->max_queue_length_ = ACE_OS::atoi (current_arg);
          // Max. queue length.

          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-max_consumers")))
        {
          this->max_consumers_ = ACE_OS::atoi (current_arg);
          // Max consumers allowed to connect.
          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-max_suppliers")))
        {
          this->max_suppliers_ = ACE_OS::atoi (current_arg);
          // Max. number of suppliers allowed to connect.
          arg_shifter.consume_arg ();
        }
      else if (arg_shifter.cur_arg_strncasecmp ("-reject_new_events") == 0)
        {
          this->reject_new_events_ = 1;
          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-consumers")))
        {
          this->consumers_ = ACE_OS::atoi (current_arg);
          // Number of consumers to create.
          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-suppliers")))
        {
          this->suppliers_ = ACE_OS::atoi (current_arg);
          // Number of suppliers to create.
          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-ConsumerDelay")))
        {
          this->consumer_delay_ = ACE_Time_Value (ACE_OS::atoi (current_arg), 0); // Consumer delay in secs.

          arg_shifter.consume_arg ();
        }
      else if ((current_arg = arg_shifter.get_the_parameter ("-InitialDelay")))
        {
          this->initial_delay_ = ACE_Time_Value (ACE_OS::atoi (current_arg), 0); // Initial delay in secs.

          arg_shifter.consume_arg ();
        }
      else if (arg_shifter.cur_arg_strncasecmp ("-?") == 0)
        {
          ACE_DEBUG((LM_DEBUG,
                     "usage: %s "
                     "-max_queue_length [max_queue_length] "
                     "-max_consumers [max_consumers] "
                     "-max_suppliers [max_suppliers] "
                     "-reject_new_events [reject_new_events] "
                     "-consumers [consumers] "
                     "-suppliers [suppliers] "
                     "-event_count [event_count] "
                     "-ConsumerDelay [delay_in_sec] "
                     "-InitialDelay [delay_in_secs]\n",
                     argv[0],
                     argv[0]));

          arg_shifter.consume_arg ();

          return -1;
        }
      else
        {
          arg_shifter.ignore_arg ();
        }
    }
    return 0;
}

int
AdminProperties::init (int argc,
                            char *argv []
                            ACE_ENV_ARG_DECL)
{
  // Initialize base class.
  Notify_Test_Client::init (argc,
                            argv
                            ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  CosNotifyChannelAdmin::ChannelID id;

  // Initialize the admin object.
  initial_admin_.length (4);

  this->initial_admin_[0].name =
    CORBA::string_dup (CosNotification::MaxQueueLength);
  this->initial_admin_[0].value <<= this->max_queue_length_;


  this->initial_admin_[1].name =
    CORBA::string_dup (CosNotification::MaxSuppliers);
  this->initial_admin_[1].value <<= this->max_suppliers_;

  this->initial_admin_[2].name =
    CORBA::string_dup (CosNotification::MaxConsumers);
  this->initial_admin_[2].value <<= this->max_consumers_;


  this->initial_admin_[3].name =
    CORBA::string_dup (CosNotification::RejectNewEvents);
  this->initial_admin_[3].value <<= CORBA::Any::from_boolean (
                                        this->reject_new_events_