TAO/orbsvcs/tests/CosEvent/Basic/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_Counting_Supplier supplier_10;
  64
  65       supplier_10.connect (supplier_admin.in ());
  66
  67       // ****************************************************************
  68
  69       CEC_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_Counting_Consumer consumer_01 ("Consumer/01");
  83       consumer_01.connect (consumer_admin.in ());
  84
  85       // ****************************************************************
  86
  87       CEC_Counting_Supplier_Task task_00 (&supplier_00, milliseconds);
  88       CEC_Counting_Supplier_Task task_01 (&supplier_01, milliseconds);
  89       CEC_Counting_Supplier_Task task_10 (&supplier_10, milliseconds);
  90       CEC_Counting_Supplier_Task task_11 (&supplier_11, milliseconds);
  91
  92       if (task_00.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
  93         {
  94           ACE_ERROR ((LM_ERROR, "Cannot activate task 00\n"));
  95         }
  96       if (task_01.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
  97         {
  98           ACE_ERROR ((LM_ERROR, "Cannot activate task 01\n"));
  99         }
 100       if (task_10.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
 101         {
 102           ACE_ERROR ((LM_ERROR, "Cannot activate task 10\n"));
 103         }
 104       if (task_11.activate (THR_BOUND|THR_NEW_LWP, 1) != 0)
 105         {
 106           ACE_ERROR ((LM_ERROR, "Cannot activate task 11\n"));
 107         }
 108
 109       ACE_Time_Value tv (5, 0);
 110       ACE_OS::sleep (tv);
 111
 112       task_00.stop ();
 113       task_01.stop ();
 114       task_10.stop ();
 115       task_11.stop ();
 116
 117       // Wait for all the threads to complete and the return
 118       ACE_Thread_Manager::instance ()->wait ();
 119
 120       // ****************************************************************
 121
 122       consumer_01.disconnect ();
 123
 124       consumer_00.disconnect ();
 125
 126       supplier_11.disconnect ();
 127       supplier_10.disconnect ();
 128       supplier_01.disconnect ();
 129       supplier_00.disconnect ();
 130
 131       // ****************************************************************
 132
 133       event_channel->destroy ();
 134
 135       // ****************************************************************
 136
 137       poa->destroy (true, true);
 138
 139       orb->destroy ();
 140
 141       // ****************************************************************
 142
 143       CORBA::ULong expected =
 144         task_00.push_count ()
 145         + task_01.push_count ()
 146         + task_10.push_count ()
 147         + task_11.push_count ();
 148
 149       consumer_00.dump_results (expected, 5);
 150       consumer_01.dump_results (expected, 5);
 151     }
 152   catch (const CORBA::Exception& ex)
 153     {
 154       ex._tao_print_exception ("Service");
 155       return 1;
 156     }
 157   return 0;
 158 }