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