TAO/orbsvcs/examples/Notify/Lanes/Consumer_Client.cpp

   1 #include "Consumer_Client.h"
   2 #include "Consumer.h"
   3 #include "ORB_Run_Task.h"
   4 #include "ace/Arg_Shifter.h"
   5 #include "orbsvcs/NotifyExtC.h"
   6 #include "orbsvcs/CosNamingC.h"
   7 #include "tao/ORB_Core.h"
   8 #include "ace/Sched_Params.h"
   9 #include "ace/OS_NS_errno.h"
  10
  11
  12
  13 TAO_Notify_Lanes_Consumer_Client::TAO_Notify_Lanes_Consumer_Client (TAO_Notify_ORB_Objects& orb_objects)
  14   : orb_objects_ (orb_objects)
  15   , lane_priority_ (0)
  16   , consumer_ (0)
  17 {
  18 }
  19
  20 TAO_Notify_Lanes_Consumer_Client::~TAO_Notify_Lanes_Consumer_Client ()
  21 {
  22 }
  23
  24 int
  25 TAO_Notify_Lanes_Consumer_Client::parse_args (int argc, ACE_TCHAR *argv[])
  26 {
  27   ACE_Arg_Shifter arg_shifter (argc, argv);
  28
  29   const ACE_TCHAR *current_arg = 0;
  30
  31   while (arg_shifter.is_anything_left ())
  32     {
  33       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-LanePriority")))) // LanePriority
  34         {
  35           if (current_arg != 0)
  36             {
  37               this->lane_priority_ = ACE_OS::atoi (current_arg);
  38
  39               char type[BUFSIZ];
  40               ACE_OS::sprintf (type, "TEST_TYPE_%d", this->lane_priority_);
  41               this->event_type_ = type;
  42             }
  43
  44           arg_shifter.consume_arg ();
  45         }
  46       else
  47         {
  48           arg_shifter.ignore_arg ();
  49         }
  50     }
  51
  52   return 0;
  53 }
  54
  55 void
  56 TAO_Notify_Lanes_Consumer_Client::initialize (void)
  57 {
  58   ACE_DEBUG ((LM_DEBUG, "(%P, %t)Initializing Consumer Client with lane priority = %d, event type = (%s)\n"
  59               , this->lane_priority_, this->event_type_.c_str ()));
  60
  61   PortableServer::POAManager_var poa_manager =
  62     this->orb_objects_.root_poa_->the_POAManager ();
  63
  64   poa_manager->activate ();
  65
  66   // Resolve the Notification Factory.
  67   CosNotifyChannelAdmin::EventChannelFactory_var ecf = this->orb_objects_.notify_factory ();
  68
  69   // Find the EventChannel created by the supplier.
  70   CosNotifyChannelAdmin::ChannelIDSeq_var channel_seq = ecf->get_all_channels ();
  71
  72   CosNotifyChannelAdmin::EventChannel_var ec;
  73
  74   if (channel_seq->length() > 0)
  75     {
  76       ec = ecf->get_event_channel (channel_seq[0]);
  77     }
  78   else
  79     {
  80       ACE_DEBUG ((LM_DEBUG, "No Event Channel active!\n"));
  81       return;
  82     }
  83
  84   // Create a Consumer Admin
  85   CosNotifyChannelAdmin::AdminID adminid = 0;
  86
  87   CosNotifyChannelAdmin::ConsumerAdmin_var consumer_admin =
  88     ec->new_for_consumers (CosNotifyChannelAdmin::AND_OP, adminid);
  89
  90   ACE_ASSERT (!CORBA::is_nil (consumer_admin.in ()));
  91
  92   PortableServer::POA_var rt_poa = this->create_rt_poa ();
  93
  94   // Create a Consumer
  95   this->consumer_ = new TAO_Notify_Lanes_Consumer (this->orb_objects_);
  96
  97   // Initialize it.
  98   this->consumer_->init (rt_poa, consumer_admin, this->event_type_);
  99 }
 100
 101 PortableServer::POA_ptr
 102 TAO_Notify_Lanes_Consumer_Client::create_rt_poa (void)
 103 {
 104   PortableServer::POA_var rt_poa;
 105
 106   // Create an RT POA with a lane at the given priority.
 107   CORBA::Policy_var priority_model_policy;
 108   CORBA::Policy_var lanes_policy;
 109
 110   CORBA::Policy_var activation_policy =
 111     this->orb_objects_.root_poa_->create_implicit_activation_policy (PortableServer::IMPLICIT_ACTIVATION);
 112
 113   // Create a priority model policy.
 114   priority_model_policy =
 115     this->orb_objects_.rt_orb_->create_priority_model_policy (RTCORBA::CLIENT_PROPAGATED
 116                                                               , 0);
 117
 118   RTCORBA::ThreadpoolLanes lanes (1);
 119   lanes.length (1);
 120
 121   lanes[0].lane_priority = this->lane_priority_;
 122   lanes[0].static_threads = 1;
 123   lanes[0].dynamic_threads = 0;
 124
 125
 126   // Create a thread-pool.
 127   CORBA::ULong stacksize = 0;
 128   CORBA::Boolean allow_request_buffering = 0;
 129   CORBA::ULong max_buffered_requests = 0;
 130   CORBA::ULong max_request_buffer_size = 0;
 131   CORBA::Boolean allow_borrowing = 0;
 132
 133   // Create the thread-pool.
 134   RTCORBA::ThreadpoolId threadpool_id =
 135     this->orb_objects_.rt_orb_->create_threadpool_with_lanes (stacksize,
 136                                                               lanes,
 137                                                               allow_borrowing,
 138                                                               allow_request_buffering,
 139                                                               max_buffered_requests,
 140                                                               max_request_buffer_size);
 141
 142   // Create a thread-pool policy.
 143   lanes_policy =
 144     this->orb_objects_.rt_orb_->create_threadpool_policy (threadpool_id);
 145
 146   CORBA::PolicyList poa_policy_list;
 147
 148   poa_policy_list.length (3);
 149   poa_policy_list[0] = priority_model_policy;
 150   poa_policy_list[1] = activation_policy;
 151   poa_policy_list[2] = lanes_policy;
 152
 153   PortableServer::POAManager_var poa_manager =
 154     this->orb_objects_.root_poa_->the_POAManager ();
 155
 156   rt_poa = this->orb_objects_.root_poa_->create_POA ("RT POA!",
 157                                                      poa_manager.in (),
 158                                                      poa_policy_list);
 159
 160   return rt_poa._retn ();
 161 }
 162
 163 void
 164 TAO_Notify_Lanes_Consumer_Client::run (void)
 165 {
 166   this->consumer_->run ();
 167 }
 168
 169 int
 170 TAO_Notify_Lanes_Consumer_Client::svc (void)
 171 {
 172   try
 173     {
 174       // Initialize this threads priority.
 175       this->orb_objects_.current_->the_priority (0);
 176
 177       this->initialize (); //Init the Client
 178
 179       this->run ();
 180     }
 181   catch (const CORBA::Exception& ex)
 182     {
 183       ex._tao_print_exception (ACE_TEXT ("Supplier error "));
 184
 185     }
 186
 187   return 0;
 188 }
 189
 190 int
 191 ACE_TMAIN (int argc, ACE_TCHAR *argv[])
 192 {
 193   try
 194     {
 195       // Initialize an ORB
 196       CORBA::ORB_var orb = CORBA::ORB_init (argc, argv);
 197
 198       TAO_Notify_ORB_Objects orb_objects;
 199
 200       orb_objects.init (orb);
 201
 202       TAO_Notify_ORB_Run_Task orb_run_task (orb_objects);
 203
 204       TAO_Notify_Lanes_Consumer_Client client (orb_objects);
 205
 206       if (client.parse_args (argc, argv) != 0)
 207         {
 208           ACE_DEBUG ((LM_DEBUG, "Consumer_Client::Error parsing options\n"));
 209           return -1;
 210         }
 211
 212       long flags = THR_NEW_LWP | THR_JOINABLE;
 213
 214       flags |=
 215         orb->orb_core ()->orb_params ()->thread_creation_flags ();
 216
 217
 218       if (orb_run_task.activate (flags) == -1 || client.activate (flags) == -1)
 219         {
 220           if (ACE_OS::last_error () == EPERM)
 221             ACE_ERROR_RETURN ((LM_ERROR,
 222                                ACE_TEXT ("Insufficient privilege to activate ACE_Task.\n")),
 223                               -1);
 224           else
 225             ACE_DEBUG ((LM_ERROR,
 226                         ACE_TEXT ("(%t) Task activation at priority %d failed.\n")));
 227         }
 228
 229       orb_run_task.thr_mgr ()->wait ();
 230       client.thr_mgr ()->wait ();
 231     }
 232   catch (const CORBA::Exception& ex)
 233     {
 234       ex._tao_print_exception (ACE_TEXT ("Consumer Client error "));
 235     }
 236
 237   return 0;
 238 }