TAO/orbsvcs/tests/CosEvent/Basic/Pull_Push_Event.cpp

   1 #include "Counting_Consumer.h"
   2 #include "Counting_Supplier.h"
   3 #include "orbsvcs/CosEvent/CEC_EventChannel.h"
   4 #include "orbsvcs/CosEvent/CEC_Default_Factory.h"
   5 #include "ace/OS_NS_unistd.h"
   6
   7 int
   8 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
   9 {
  10   TAO_CEC_Default_Factory::init_svcs ();
  11
  12   try
  13     {
  14       // ORB initialization boiler plate...
  15       CORBA::ORB_var orb =
  16         CORBA::ORB_init (argc, argv);
  17
  18       CORBA::Object_var object =
  19         orb->resolve_initial_references ("RootPOA");
  20       PortableServer::POA_var poa =
  21         PortableServer::POA::_narrow (object.in ());
  22       PortableServer::POAManager_var poa_manager =
  23         poa->the_POAManager ();
  24       poa_manager->activate ();
  25
  26       // ****************************************************************
  27
  28       TAO_CEC_EventChannel_Attributes attributes (poa.in (),
  29                                                   poa.in ());
  30
  31       TAO_CEC_EventChannel ec_impl (attributes);
  32       ec_impl.activate ();
  33
  34       CosEventChannelAdmin::EventChannel_var event_channel =
  35         ec_impl._this ();
  36
  37       // ****************************************************************
  38
  39       // Obtain the consumer admin..
  40       CosEventChannelAdmin::ConsumerAdmin_var consumer_admin =
  41         event_channel->for_consumers ();
  42
  43       // Obtain the supplier admin..
  44       CosEventChannelAdmin::SupplierAdmin_var supplier_admin =
  45         event_channel->for_suppliers ();
  46
  47       // ****************************************************************
  48
  49       const int milliseconds = 50;
  50
  51       CEC_Counting_Supplier supplier_00;
  52
  53       supplier_00.connect (supplier_admin.in ());
  54
  55       // ****************************************************************
  56
  57       CEC_Counting_Supplier supplier_01;
  58
  59       supplier_01.connect (supplier_admin.in ());
  60
  61       // ****************************************************************
  62
  63       CEC_Pull_Counting_Supplier supplier_10;
  64
  65       supplier_10.connect (supplier_admin.in ());
  66
  67       // ****************************************************************
  68
  69       CEC_Pull_Counting_Supplier supplier_11;
  70
  71       supplier_11.connect (supplier_admin.in ());
  72
  73       // ****************************************************************
  74
  75       // Create a consumer, intialize its RT_Info structures, and
  76       // connnect to the event channel....
  77       CEC_Counting_Consumer consumer_00 ("Consumer/00");
  78       consumer_00.connect (consumer_admin.in ());
  79
  80       // ****************************************************************
  81
  82       CEC_Pull_Counting_Consumer consumer_01 ("Consumer/01");
  83       consumer_01.connect (consumer_admin.in ());
  84
  85       // ****************************************************************
  86
  87       CEC_Pull_Counting_Consumer consumer_10 ("Consumer/10");
  88       consumer_10.connect (consumer_admin.in ());
  89
  90       // ****************************************************************
  91
  92       CEC_Counting_Supplier_Task task_00 (&supplier_00,
  93                                           milliseconds);
  94       CEC_Counting_Supplier_Task task_01 (&supplier_01,
  95                                           3 * milliseconds);
  96
  97       if (task_00.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
  98         {
  99           ACE_ERROR ((LM_ERROR, "Cannot activate task 00\n"));
 100         }
 101       if (task_01.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
 102         {
 103           ACE_ERROR ((LM_ERROR, "Cannot activate task 01\n"));
 104         }
 105
 106       // ****************************************************************
 107
 108       CEC_Counting_Consumer_Task consumer_task_01 (&consumer_01,
 109                                                    5 * milliseconds);
 110       if (consumer_task_01.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
 111         {
 112           ACE_ERROR ((LM_ERROR, "Cannot activate task 01\n"));
 113         }
 114
 115       // ****************************************************************
 116
 117       CEC_Counting_Consumer_Task consumer_task_10 (&consumer_10,
 118                                                    10);
 119       if (consumer_task_10.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
 120         {
 121           ACE_ERROR ((LM_ERROR, "Cannot activate task 10\n"));
 122         }
 123
 124       // ****************************************************************
 125
 126       ACE_Time_Value tv (15, 0);
 127       ACE_OS::sleep (tv);
 128
 129       task_00.stop ();
 130       task_01.stop ();
 131
 132       ACE_DEBUG ((LM_DEBUG,
 133                   "Events pulled (so far) by Consumer/01 = %d\n",
 134                   consumer_task_01.pull_count ()));
 135
 136       tv = ACE_Time_Value (5, 0);
 137       ACE_OS::sleep (tv);
 138
 139       consumer_task_10.stop ();
 140       consumer_task_01.stop ();
 141
 142       // Wait for all the threads to complete and the return
 143       ACE_Thread_Manager::instance ()->wait ();
 144
 145       // ****************************************************************
 146
 147       consumer_10.disconnect ();
 148
 149       consumer_01.disconnect ();
 150
 151       consumer_00.disconnect ();
 152
 153       supplier_11.disconnect ();
 154       supplier_10.disconnect ();
 155       supplier_01.disconnect ();
 156       supplier_00.disconnect ();
 157
 158       // ****************************************************************
 159
 160       event_channel->destroy ();
 161
 162       // ****************************************************************
 163
 164       poa->destroy (true, true);
 165
 166       orb->destroy ();
 167
 168       // ****************************************************************
 169
 170       CORBA::ULong expected =
 171         task_00.push_count ()
 172         + task_01.push_count ()
 173         + supplier_10.event_count
 174         + supplier_11.event_count;
 175
 176       consumer_00.dump_results (expected, 5);
 177
 178       expected =
 179         consumer_task_01.pull_count ();
 180       consumer_01.dump_results (expected, 5);
 181       expected =
 182         consumer_task_10.pull_count ();
 183       consumer_10.dump_results (expected, 5);
 184     }
 185   catch (const CORBA::Exception& ex)
 186     {
 187       ex._tao_print_exception ("Service");
 188       return 1;
 189     }
 190   return 0;
 191 }