TAO/orbsvcs/tests/Notify/Ordering/Notify_Sequence_Push_Consumer.cpp

   1 #include "Notify_Sequence_Push_Consumer.h"
   2 #include "Notify_Test_Client.h"
   3 #include "common.h"
   4 #include "orbsvcs/TimeBaseC.h"
   5 #include "tao/debug.h"
   6 #include "ace/OS_NS_unistd.h"
   7
   8 const int BATCH_SIZE = 5;
   9
  10 Notify_Sequence_Push_Consumer::Notify_Sequence_Push_Consumer (
  11   const char* name,
  12   CORBA::Short policy,
  13   bool use_ordering,
  14   int expected,
  15   Notify_Test_Client& client)
  16   : name_ (name),
  17   order_policy_ (policy),
  18   use_ordering_ (use_ordering),
  19   expected_ (expected),
  20   count_ (0),
  21   previous_first_ (0),
  22   client_ (client)
  23 {
  24   this->client_.consumer_start (this);
  25 }
  26
  27
  28 void
  29 Notify_Sequence_Push_Consumer::_connect (
  30   CosNotifyChannelAdmin::ConsumerAdmin_ptr consumer_admin)
  31 {
  32   CosNotifyComm::SequencePushConsumer_var objref =
  33     this->_this ();
  34
  35   CosNotifyChannelAdmin::ProxySupplier_var proxysupplier =
  36     consumer_admin->obtain_notification_push_supplier (
  37     CosNotifyChannelAdmin::SEQUENCE_EVENT,
  38     proxy_id_);
  39
  40   this->proxy_ =
  41     CosNotifyChannelAdmin::SequenceProxyPushSupplier::_narrow (
  42     proxysupplier.in ());
  43
  44   CosNotification::QoSProperties properties (3);
  45   properties.length (3);
  46   properties[0].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
  47   properties[0].value <<= (CORBA::Long) BATCH_SIZE;
  48   properties[1].name = CORBA::string_dup (CosNotification::PacingInterval);
  49   properties[1].value <<= (TimeBase::TimeT) (1000 * 10000); // 1 secs
  50   if (use_ordering_)
  51   {
  52     properties[2].name = CORBA::string_dup (CosNotification::OrderPolicy);
  53     properties[2].value <<= this->order_policy_;
  54   }
  55   else
  56   {
  57     properties.length(2);
  58   }
  59
  60   this->proxy_->set_qos (properties);
  61   this->proxy_->connect_sequence_push_consumer (objref.in ());
  62
  63   // give ownership to POA
  64   this->_remove_ref ();
  65 }
  66
  67
  68 void
  69 Notify_Sequence_Push_Consumer::push_structured_events (
  70   const CosNotification::EventBatch& events)
  71 {
  72   if (count_ == 0)
  73   {
  74     // Sleep long enough to force the channel to back up, otherwise
  75     // there will be no ordering.
  76     ACE_OS::sleep(2);
  77   }
  78
  79   ACE_ASSERT(events.length() == static_cast<CORBA::ULong>(BATCH_SIZE));
  80
  81   count_ += events.length();
  82
  83   if (this->count_ > this->expected_)
  84   {
  85     ACE_ERROR ((LM_ERROR, ACE_TEXT ("ERROR: too many events received.\n")));
  86   }
  87
  88   if (this->count_ >= this->expected_)
  89   {
  90     this->client_.consumer_done (this);
  91   }
  92
  93   ACE_ASSERT(events[0].header.variable_header.length() == 3);
  94   ACE_ASSERT(ACE_OS::strcmp(events[0].header.variable_header[0].name.in(), "id") == 0);
  95   CORBA::Long first_id = 0;
  96   events[0].header.variable_header[0].value >>= first_id;
  97   CORBA::Long last_id = 0;
  98   events[events.length() - 1].header.variable_header[0].value >>= last_id;
  99
 100   ACE_DEBUG((LM_DEBUG, "\n Received id %d-%d\n", first_id, last_id));
 101
 102   int events_length = static_cast<int>(events.length());
 103
 104   CORBA::Long previous_id = first_id;
 105
 106   if (count_ > events_length) // Ignore the very first batch
 107   {
 108     // First check that the sequences are ordered correctly
 109     for (CORBA::ULong idx = 1; idx < events.length(); ++idx)
 110     {
 111       CORBA::Long id = 0;
 112
 113       events[idx].header.variable_header[0].value >>= id;
 114       CORBA::Long expected_id = previous_id + 1;
 115       if (order_policy_ == CosNotification::PriorityOrder
 116         || order_policy_ == CosNotification::DeadlineOrder)
 117       {
 118         expected_id = previous_id - 1;
 119       }
 120       if (id != expected_id)
 121       {
 122         ACE_ERROR((LM_ERROR, "Error: Expected:%d Received:%d\n", expected_id, id));
 123         return;
 124       }
 125       previous_id = id;
 126     }
 127
 128     // Next check that the first id in the sequence is ordered
 129     // relative to the previously retrieved sequence.
 130     if (previous_first_ != 0)
 131     {
 132       CORBA::Long expected_id = previous_first_ + BATCH_SIZE;
 133       if (order_policy_ == CosNotification::PriorityOrder
 134         || order_policy_ == CosNotification::DeadlineOrder)
 135       {
 136         expected_id = previous_first_ - BATCH_SIZE;
 137       }
 138       if (first_id != expected_id)
 139       {
 140         ACE_ERROR((LM_ERROR, "Error: Expected:%d Received:%d\n", expected_id, first_id));
 141         return;
 142       }
 143     }
 144     previous_first_ = first_id;
 145   }
 146 }