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

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