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

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