TAO/orbsvcs/tests/Notify/MC/Notify_Structured_Push_Consumer.cpp

   1 // ******************************************************************
   2 // Include Section
   3 // ******************************************************************
   4
   5 #include "ace/OS_NS_unistd.h"
   6 #include "Notify_Structured_Push_Consumer.h"
   7 #include "Notify_Test_Client.h"
   8 #include "orbsvcs/Notify/Notify_Extensions.h"
   9 #include "common.h"
  10
  11 // ******************************************************************
  12 // Subroutine Section
  13 // ******************************************************************
  14
  15 Notify_Structured_Push_Consumer::Notify_Structured_Push_Consumer (
  16                                             const char* name,
  17                                             unsigned int expected,
  18                                             Notify_Test_Client& client)
  19  : name_ (name),
  20    expected_ (expected),
  21    count_ (0),
  22    delay_count_ (0),
  23    delay_period_ (5),
  24    client_ (client)
  25 {
  26   this->client_.consumer_start (this);
  27 }
  28
  29 void
  30 Notify_Structured_Push_Consumer::set_delay_parameters (unsigned int delay_count, unsigned long delay_period)
  31 {
  32   this->delay_count_ = delay_count;
  33   this->delay_period_ = delay_period;
  34 }
  35
  36 void
  37 Notify_Structured_Push_Consumer::_connect (
  38                 CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
  39 {
  40 #ifdef TEST_QOS_MAX_EVENTS_PER_CONSUMER
  41   CosNotification::QoSProperties properties (8);
  42   properties.length (2);
  43   CORBA::ULong idx = 0;
  44   properties[idx].name = CORBA::string_dup (CosNotification::DiscardPolicy);
  45   properties[idx].value <<= this->CosNotification::FifoOrder;
  46   idx += 1;
  47   properties[idx].name = CORBA::string_dup (CosNotification::MaxEventsPerConsumer);
  48   properties[idx].value <<= 500;
  49   idx += 1;
  50   ACE_OS::printf("Setting %d::%d QoS properties in Admin.\n", (int)idx, (int)properties.length());
  51   consumer_admin->set_qos (properties);
  52 #endif // TEST_QOS_MAX_EVENTS_PER_CONSUMER
  53
  54   CosNotifyComm::StructuredPushConsumer_var objref = this->_this ();
  55
  56   CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
  57     consumer_admin->obtain_notification_push_supplier (
  58       CosNotifyChannelAdmin::STRUCTURED_EVENT,
  59       proxy_id_);
  60
  61   this->proxy_ =
  62     CosNotifyChannelAdmin::StructuredProxyPushSupplier::_narrow (
  63       proxysupplier.in ());
  64
  65   this->proxy_->connect_structured_push_consumer (objref.in ());
  66
  67   CosNotification::EventTypeSeq added (1);
  68   CosNotification::EventTypeSeq removed (1);
  69   added.length (1);
  70   removed.length (1);
  71
  72   added[0].domain_name = CORBA::string_dup ("*");
  73   added[0].type_name = CORBA::string_dup ("examples");
  74
  75   removed[0].domain_name = CORBA::string_dup ("*");
  76   removed[0].type_name = CORBA::string_dup ("*");
  77
  78   this->subscription_change (added, removed);
  79
  80   // give ownership to POA
  81   this->_remove_ref ();
  82 }
  83
  84
  85 void
  86 Notify_Structured_Push_Consumer::push_structured_event (
  87   const CosNotification::StructuredEvent&)
  88 {
  89   ACE_DEBUG ((LM_DEBUG, "-"));
  90   static const ACE_Time_Value sl (0, 2000);
  91
  92   this->count_++;
  93
  94   if (this->delay_count_ != 0 && this->count_ % this->delay_count_ == 0)
  95     {
  96       ACE_OS::sleep (this->delay_period_);
  97     }
  98
  99   if (this->count_ > this->expected_)
 100   {
 101     ACE_ERROR ((LM_ERROR,
 102       ACE_TEXT ("Structured Consumer (%P|%t): ERROR: too ")
 103       ACE_TEXT ("many events received.\n")));
 104   }
 105
 106   if (this->count_ >= this->expected_)
 107   {
 108     ACE_DEBUG ((LM_DEBUG, "\nConsumer received %u events.\n", count_));
 109     this->client_.consumer_done (this);
 110   }
 111   ACE_OS::sleep (sl);
 112 }