TAO/orbsvcs/tests/Notify/Discarding/Sequence_Supplier.cpp

   1 // ******************************************************************
   2 // Include Section
   3 // ******************************************************************
   4
   5 #include "ace/Get_Opt.h"
   6 #include "ace/Auto_Ptr.h"
   7
   8 #include "tao/ORB_Core.h"
   9
  10 #include "orbsvcs/CosNotifyChannelAdminS.h"
  11 #include "orbsvcs/CosNotifyCommC.h"
  12 #include "orbsvcs/CosNamingC.h"
  13 #include "orbsvcs/TimeBaseC.h"
  14 #include "Notify_SequencePushSupplier.h"
  15 #include "goS.h"
  16
  17 #include "Notify_Test_Client.h"
  18
  19 #include "ace/OS_NS_unistd.h"
  20
  21 // ******************************************************************
  22 // Data Section
  23 // ******************************************************************
  24
  25 static TAO_Notify_Tests_SequencePushSupplier* supplier_1 = 0;
  26
  27 static CORBA::Boolean use_deadline_ordering = 0;
  28
  29 // Must be a multiple of the consumers batch size.
  30 static int num_events = 40;
  31
  32 // Must match the consumers batch size so that we send
  33 // the correct number of each type of event per batch, which
  34 // allows us to validate the discard policy.
  35 static const CORBA::Long BATCH_SIZE = 4;
  36
  37 static const ACE_TCHAR *ior_output_file = ACE_TEXT ("supplier.ior");
  38
  39 // ******************************************************************
  40 // Subroutine Section
  41 // ******************************************************************
  42
  43 // The supplier will not start sending events until the
  44 // go() operation is invoked.
  45 class sig_i : public POA_sig
  46 {
  47 public:
  48   sig_i(CORBA::ORB_ptr orb)
  49     : orb_(orb)
  50     , started_(false)
  51   {
  52   }
  53
  54   void go (void)
  55   {
  56     started_ = true;
  57   }
  58
  59   void done (void)
  60   {
  61     started_ = false;
  62   }
  63
  64   void wait_for_startup()
  65   {
  66     while (! started_) {
  67       ACE_Time_Value tv(0, 100 * 1000); // 100ms
  68       orb_->run(tv);
  69     }
  70   }
  71
  72   void wait_for_completion()
  73   {
  74     while (started_) {
  75       ACE_Time_Value tv(0, 100 * 1000); // 100ms
  76       orb_->run(tv);
  77     }
  78   }
  79
  80 private:
  81   CORBA::ORB_ptr orb_;
  82   bool started_;
  83 };
  84
  85 class Supplier_Client : public Notify_Test_Client
  86 {
  87 public:
  88   virtual int parse_args (int argc, ACE_TCHAR *argv[]);
  89 };
  90
  91
  92 int
  93 Supplier_Client::parse_args (int argc, ACE_TCHAR *argv[])
  94 {
  95   ACE_Get_Opt get_opts (argc, argv, ACE_TEXT("o:e:d"));
  96   int c;
  97
  98   while ((c = get_opts ()) != -1)
  99     switch (c)
 100   {
 101     case 'd':
 102       use_deadline_ordering = 1;
 103 #if !defined (ACE_HAS_TIMED_MESSAGE_BLOCKS)
 104       ACE_ERROR_RETURN ((LM_ERROR,
 105         "This order policy requires timed message "
 106         "blocks.\nPlease #define "
 107         "ACE_HAS_TIMED_MESSAGE_BLOCKS in your "
 108         "config.h\n"), -1);
 109 #endif /* ACE_HAS_TIMED_MESSAGE_BLOCKS */
 110       break;
 111     case 'e':
 112       num_events = ACE_OS::atoi (get_opts.optarg);
 113       num_events -= num_events % BATCH_SIZE; // round down
 114       break;
 115
 116     case 'o':
 117       ior_output_file = get_opts.optarg;
 118       break;
 119
 120     default:
 121       ACE_ERROR_RETURN ((LM_ERROR,
 122         "usage:  %s "
 123         "-o <iorfile> -e <# of events> -d"
 124         "\n",
 125         argv [0]),
 126         -1);
 127   }
 128
 129   // Indicates successful parsing of the command line
 130   return 0;
 131 }
 132
 133
 134 static CosNotifyChannelAdmin::SupplierAdmin_ptr
 135 create_supplieradmin (CosNotifyChannelAdmin::EventChannel_ptr ec)
 136 {
 137   CosNotifyChannelAdmin::AdminID adminid = 0;
 138   CosNotifyChannelAdmin::SupplierAdmin_var admin =
 139     ec->new_for_suppliers (CosNotifyChannelAdmin::AND_OP,
 140     adminid);
 141
 142
 143   return CosNotifyChannelAdmin::SupplierAdmin::_duplicate (admin.in ());
 144 }
 145
 146
 147 static void
 148 SendBatch (int batch_id)
 149 {
 150   CosNotification::EventBatch events;
 151   events.length(BATCH_SIZE);
 152   for (CORBA::Long i = 0; i < BATCH_SIZE; ++i)
 153   {
 154     int id = batch_id * BATCH_SIZE + i + 1;
 155
 156     CosNotification::StructuredEvent event;
 157
 158     event.header.fixed_header.event_type.domain_name = CORBA::string_dup ("a");
 159     event.header.fixed_header.event_type.type_name = CORBA::string_dup ("b");
 160     event.header.fixed_header.event_name = CORBA::string_dup ("c");
 161
 162     event.header.variable_header.length (3);
 163     event.header.variable_header[0].name =
 164       CORBA::string_dup ("Id");
 165     event.header.variable_header[0].value <<= (CORBA::Long)id;
 166
 167     event.header.variable_header[1].name =
 168       CORBA::string_dup (CosNotification::Priority);
 169     event.header.variable_header[1].value <<= (CORBA::Short)(id);
 170
 171     event.header.variable_header[2].name =
 172       CORBA::string_dup (CosNotification::Timeout);
 173     event.header.variable_header[2].value <<= (TimeBase::TimeT) (id * 10000);
 174
 175     events[i] = event;
 176   }
 177   supplier_1->send_events (events);
 178 }
 179
 180 static void
 181 create_suppliers (CosNotifyChannelAdmin::SupplierAdmin_ptr admin,
 182                   PortableServer::POA_ptr poa)
 183 {
 184   // start up the supplier
 185   ACE_NEW_THROW_EX (supplier_1,
 186     TAO_Notify_Tests_SequencePushSupplier (),
 187     CORBA::NO_MEMORY ());
 188
 189   supplier_1->init (poa);
 190
 191   supplier_1->connect (admin);
 192 }
 193
 194
 195 // ******************************************************************
 196 // Main Section
 197 // ******************************************************************
 198
 199 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 200 {
 201   int status = 0;
 202   ACE_Auto_Ptr< sig_i > sig_impl;
 203   try
 204   {
 205     Supplier_Client client;
 206     status = client.init (argc, argv);
 207
 208     if (status == 0)
 209     {
 210       CosNotifyChannelAdmin::EventChannel_var ec =
 211         client.create_event_channel ("MyEventChannel", 0);
 212
 213       if (use_deadline_ordering)
 214       {
 215         CosNotification::QoSProperties qos (1);
 216         qos.length (1);
 217         qos[0].name = CORBA::string_dup (CosNotification::OrderPolicy);
 218         qos[0].value <<= (CORBA::Short)CosNotification::DeadlineOrder;
 219         ec->set_qos (qos);
 220       }
 221
 222       sig_impl.reset( new sig_i( client.orb() ) );
 223       sig_var sig = sig_impl->_this ();
 224
 225       CORBA::String_var ior =
 226         client.orb()->object_to_string (sig.in ());
 227
 228       // If the ior_output_file exists, output the ior to it
 229       if (ior_output_file != 0)
 230       {
 231         FILE *output_file= ACE_OS::fopen (ior_output_file, "w");
 232         ACE_ASSERT(output_file != 0);
 233         ACE_OS::fprintf (output_file, "%s", ior.in ());
 234         ACE_OS::fclose (output_file);
 235       }
 236
 237       CosNotifyChannelAdmin::SupplierAdmin_var admin =
 238         create_supplieradmin (ec.in ());
 239       ACE_ASSERT(!CORBA::is_nil (admin.in ()));
 240       create_suppliers (admin.in (), client.root_poa ());
 241
 242       sig_impl->wait_for_startup();
 243
 244       ACE_DEBUG((LM_DEBUG, "1 supplier sending %d events...\n", num_events));
 245       for (int i = 0; i < num_events / BATCH_SIZE; ++i)
 246       {
 247         ACE_DEBUG((LM_DEBUG, "+"));
 248         SendBatch (i);
 249       }
 250       ACE_DEBUG((LM_DEBUG, "\nSupplier sent %d events.\n", num_events));
 251
 252       sig_impl->wait_for_completion();
 253
 254       ACE_OS::unlink (ior_output_file);
 255
 256       supplier_1->disconnect();
 257
 258       ec->destroy();
 259     }
 260   }
 261   catch (const CORBA::Exception& e)
 262   {
 263     e._tao_print_exception ("Error: ");
 264     status = 1;
 265   }
 266
 267   return status;
 268 }