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

   1 #include "Notify_StructuredPushSupplier.h"
   2 #include "goS.h"
   3
   4 #include "Notify_Test_Client.h"
   5
   6 #include "orbsvcs/CosNotifyChannelAdminS.h"
   7 #include "orbsvcs/CosNotifyCommC.h"
   8 #include "orbsvcs/CosNamingC.h"
   9 #include "orbsvcs/TimeBaseC.h"
  10
  11 #include "tao/ORB_Core.h"
  12
  13 #include "ace/Get_Opt.h"
  14 #include "ace/OS_NS_unistd.h"
  15 #include <memory>
  16
  17 static TAO_Notify_Tests_StructuredPushSupplier* 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
  22 class sig_i : public POA_sig
  23 {
  24 public:
  25   sig_i(CORBA::ORB_ptr orb)
  26     : orb_(orb)
  27     , started_(false)
  28   {
  29   }
  30
  31   void go ()
  32   {
  33     started_ = true;
  34   }
  35
  36   void done ()
  37   {
  38     started_ = false;
  39   }
  40
  41   void wait_for_startup()
  42   {
  43     while (! started_) {
  44       ACE_Time_Value tv(0, 100 * 1000); // 100ms
  45       orb_->run(tv);
  46     }
  47   }
  48
  49   void wait_for_completion()
  50   {
  51     while (started_) {
  52       ACE_Time_Value tv(0, 100 * 1000); // 100ms
  53       orb_->run(tv);
  54     }
  55   }
  56
  57 private:
  58   CORBA::ORB_ptr orb_;
  59   bool started_;
  60 };
  61
  62 class Supplier_Client : public Notify_Test_Client
  63 {
  64 public:
  65   virtual int parse_args (int argc, ACE_TCHAR *argv[]);
  66 };
  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::strcmp (order, ACE_TEXT ("fifo")) == 0)
  82         {
  83           order_policy = CosNotification::FifoOrder;
  84         }
  85         else if (ACE_OS::strcmp (order, ACE_TEXT ("priority")) == 0)
  86         {
  87           order_policy = CosNotification::PriorityOrder;
  88         }
  89         else if (ACE_OS::strcmp (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> "
 122         "-d <fifo|priority|deadline>"
 123         "\n",
 124         argv [0]),
 125         -1);
 126   }
 127
 128   // Indicates successful parsing of the command line
 129   return 0;
 130 }
 131
 132
 133 static CosNotifyChannelAdmin::SupplierAdmin_ptr
 134 create_supplieradmin (CosNotifyChannelAdmin::EventChannel_ptr ec)
 135 {
 136   CosNotifyChannelAdmin::AdminID adminid = 0;
 137   CosNotifyChannelAdmin::SupplierAdmin_var admin =
 138     ec->new_for_suppliers (CosNotifyChannelAdmin::AND_OP,
 139     adminid);
 140
 141
 142   return CosNotifyChannelAdmin::SupplierAdmin::_duplicate (admin.in ());
 143 }
 144
 145
 146 static void
 147 SendEvent (int id)
 148 {
 149   CosNotification::StructuredEvent event;
 150
 151   event.header.fixed_header.event_type.domain_name = CORBA::string_dup ("a");
 152   event.header.fixed_header.event_type.type_name = CORBA::string_dup ("b");
 153   event.header.fixed_header.event_name = CORBA::string_dup ("test");
 154
 155   event.header.variable_header.length (3);
 156   event.header.variable_header[0].name = CORBA::string_dup ("id");
 157   event.header.variable_header[0].value <<= (CORBA::Long) id;
 158   event.header.variable_header[1].name =
 159     CORBA::string_dup (CosNotification::Priority);
 160   event.header.variable_header[1].value <<= (CORBA::Short) id;
 161   event.header.variable_header[2].name =
 162     CORBA::string_dup (CosNotification::Timeout);
 163   event.header.variable_header[2].value <<= (TimeBase::TimeT) (num_events - id);
 164
 165   try
 166   {
 167     supplier_1->send_event (event);
 168   }
 169   catch (const CORBA::Exception& e)
 170   {
 171     e._tao_print_exception ("Error: ");
 172   }
 173 }
 174
 175 static void
 176 create_suppliers (CosNotifyChannelAdmin::SupplierAdmin_ptr admin,
 177                   PortableServer::POA_ptr poa)
 178 {
 179   // startup the supplier
 180   ACE_NEW_THROW_EX (supplier_1,
 181     TAO_Notify_Tests_StructuredPushSupplier (),
 182     CORBA::NO_MEMORY ());
 183
 184   supplier_1->init (poa);
 185
 186   supplier_1->connect (admin);
 187 }
 188
 189 int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 190 {
 191   std::unique_ptr<sig_i> sig_impl;
 192   try
 193   {
 194     Supplier_Client client;
 195     int status = client.init (argc, argv);
 196     ACE_UNUSED_ARG(status);
 197     ACE_ASSERT(status == 0);
 198
 199     CosNotifyChannelAdmin::EventChannel_var ec =
 200       client.create_event_channel ("MyEventChannel", 0);
 201
 202     CosNotification::QoSProperties qos (1);
 203     qos.length (1);
 204     qos[0].name = CORBA::string_dup (CosNotification::OrderPolicy);
 205     qos[0].value <<= order_policy;
 206     ec->set_qos (qos);
 207
 208     CORBA::ORB_ptr orb = client.orb ();
 209
 210     sig_impl.reset( new sig_i(orb));
 211     sig_var sig = sig_impl->_this ();
 212
 213     CORBA::String_var ior =
 214       orb->object_to_string (sig.in ());
 215
 216     if (ior_output_file != 0)
 217     {
 218       FILE *output_file= ACE_OS::fopen (ior_output_file, "w");
 219       if (output_file == 0)
 220         ACE_ERROR_RETURN ((LM_ERROR,
 221         "Cannot open output file %s for "
 222         "writing IOR: %C",
 223         ior_output_file,
 224         ior.in ()),
 225         1);
 226       ACE_OS::fprintf (output_file, "%s", ior.in ());
 227       ACE_OS::fclose (output_file);
 228     }
 229
 230     CosNotifyChannelAdmin::SupplierAdmin_var admin =
 231       create_supplieradmin (ec.in ());
 232     ACE_ASSERT(!CORBA::is_nil (admin.in ()));
 233     create_suppliers (admin.in (), client.root_poa ());
 234
 235     sig_impl->wait_for_startup();
 236
 237     ACE_DEBUG((LM_DEBUG, "1 supplier sending %d events...\n", num_events));
 238     for (int i = 0; i < num_events; ++i)
 239     {
 240       ACE_DEBUG((LM_DEBUG, "+"));
 241       SendEvent (i);
 242     }
 243     ACE_DEBUG((LM_DEBUG, "\nSupplier sent %d events.\n", num_events));
 244
 245     sig_impl->wait_for_completion();
 246
 247     ACE_OS::unlink (ior_output_file);
 248
 249     supplier_1->disconnect();
 250
 251     ec->destroy();
 252
 253     return 0;
 254   }
 255   catch (const CORBA::Exception& e)
 256   {
 257     e._tao_print_exception ("Error: ");
 258   }
 259
 260   return 1;
 261 }