TAO/examples/CSD_Strategy/ThreadPool4/ServerApp.cpp

   1 #include "ServerApp.h"
   2 #include "Foo_i.h"
   3 #include "Callback_i.h"
   4 #include "ClientTask.h"
   5 #include "OrbShutdownTask.h"
   6 #include "ace/Get_Opt.h"
   7 #include "tao/CSD_ThreadPool/CSD_TP_Strategy.h"
   8 #include "tao/Intrusive_Ref_Count_Handle_T.h"
   9 #include "tao/Messaging/Messaging.h"
  10 #include "tao/AnyTypeCode/Any.h"
  11 // To force static load the service.
  12 #include "tao/PI/PI.h"
  13 #include "tao/CSD_ThreadPool/CSD_ThreadPool.h"
  14
  15
  16 ServerApp::ServerApp()
  17 : synch_with_server_ (0)
  18 {
  19 }
  20
  21
  22 ServerApp::~ServerApp()
  23 {
  24 }
  25
  26
  27 int
  28 ServerApp::run (int argc, ACE_TCHAR* argv[])
  29 {
  30   this->orb_ = CORBA::ORB_init (argc, argv);
  31
  32   // Parse the command-line args for this application.
  33   // * Raises -1 if problems are encountered.
  34   // * Returns 1 if the usage statement was explicitly requested.
  35   // * Returns 0 otherwise.
  36   int result = this->parse_args (argc, argv);
  37   if (result != 0)
  38     {
  39       return result;
  40     }
  41
  42   TheOrbShutdownTask::instance()->orb (this->orb_.in ());
  43
  44   if (synch_with_server_)
  45   {
  46     CORBA::Object_var manager_object =
  47       orb_->resolve_initial_references("ORBPolicyManager");
  48
  49     CORBA::PolicyManager_var policy_manager
  50       = CORBA::PolicyManager::_narrow(manager_object.in());
  51
  52     if (CORBA::is_nil (policy_manager.in ()))
  53     {
  54       ACE_ERROR ((LM_ERROR,
  55                   " (%P|%t) Panic: nil PolicyManager\n"));
  56       throw TestException();
  57     }
  58
  59     CORBA::Any policy_value;
  60     policy_value <<= Messaging::SYNC_WITH_SERVER;
  61     CORBA::PolicyList policies(1);
  62     policies.length(1);
  63
  64     policies[0] =
  65       orb_->create_policy (Messaging::SYNC_SCOPE_POLICY_TYPE,
  66                           policy_value);
  67
  68     policy_manager->set_policy_overrides (policies,
  69                                           CORBA::ADD_OVERRIDE);
  70
  71     policies[0]->destroy ();
  72   }
  73
  74   CORBA::Object_var obj
  75     = orb_->resolve_initial_references("RootPOA");
  76
  77   if (CORBA::is_nil(obj.in()))
  78     {
  79       ACE_ERROR((LM_ERROR,
  80                  "(%P|%t) Failed to resolve initial ref for 'RootPOA'.\n"));
  81       throw TestException();
  82     }
  83
  84   PortableServer::POA_var root_poa
  85     = PortableServer::POA::_narrow(obj.in());
  86
  87   if (CORBA::is_nil(root_poa.in()))
  88     {
  89       ACE_ERROR((LM_ERROR,
  90                  "(%P|%t) Failed to narrow obj ref to POA interface.\n"));
  91       throw TestException();
  92     }
  93
  94   PortableServer::POAManager_var poa_manager
  95     = root_poa->the_POAManager();
  96
  97   // Create the child POA.
  98   CORBA::PolicyList policies(1);
  99   policies.length(1);
 100
 101   policies[0] = root_poa->create_id_assignment_policy(PortableServer::USER_ID);
 102
 103   PortableServer::POA_var child_poa_1 = root_poa->create_POA("ChildPoa_1",
 104                                                            poa_manager.in(),
 105                                                            policies);
 106
 107   if (CORBA::is_nil(child_poa_1.in()))
 108     {
 109       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR [ServerApp::run()]: "
 110                  "Failed to create the ChildPoa_1.\n"));
 111       throw TestException();
 112     }
 113
 114   PortableServer::POA_var child_poa_2 = root_poa->create_POA("ChildPoa_2",
 115                                                            poa_manager.in(),
 116                                                            policies);
 117
 118   if (CORBA::is_nil(child_poa_2.in()))
 119     {
 120       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR [ServerApp::run()]: "
 121                  "Failed to create the ChildPoa_2.\n"));
 122       throw TestException();
 123     }
 124
 125   policies[0]->destroy ();
 126
 127   // Create the thread pool servant dispatching strategy object, and
 128   // hold it in a (local) smart pointer variable.
 129   TAO_Intrusive_Ref_Count_Handle<TAO::CSD::TP_Strategy> csd_tp_strategy =
 130                                                  new TAO::CSD::TP_Strategy();
 131
 132   // We need create multiple working threads otherwise it would deadlock
 133   // with the callback test.
 134   csd_tp_strategy->set_num_threads(2);
 135
 136   // Tell the strategy to apply itself to the child poa.
 137   if (csd_tp_strategy->apply_to(child_poa_1.in()) == false)
 138     {
 139       ACE_ERROR((LM_ERROR, "(%P|%t) ERROR [ServerApp::run()]: "
 140                  "Failed to apply custom dispatching strategy to child poa 1.\n"));
 141       throw TestException();
 142     }
 143
 144   Foo_var foo1 = this->create_foo(child_poa_1.in(),
 145                                   "foo_applied_strategy");
 146
 147   Foo_var foo2 = this->create_foo(child_poa_2.in(),
 148                                   "foo_not_applied_strategy");
 149
 150   Callback_var callback1
 151     = this->create_callback(child_poa_1.in(),
 152                             "callback_applied_strategy");
 153
 154   Callback_var callback2
 155     = this->create_callback(child_poa_2.in(),
 156                             "callback_not_applied_strategy");
 157
 158   // Activate the POA Manager
 159   poa_manager->activate();
 160
 161   ACE_DEBUG((LM_DEBUG,
 162              "(%P|%t) ServerApp is ready.  Running the ORB event loop.\n"));
 163
 164   // Wait for the custom stretegy launch the dispatching threads.
 165   ACE_OS::sleep (2);
 166
 167   ClientTask client1 (foo1.in (), callback1.in ());
 168   if (client1.open() != 0)
 169     {
 170       ACE_ERROR((LM_ERROR,
 171         "(%P|%t) Failed to open the collocated client1.\n"));
 172       throw TestException();
 173     }
 174
 175   ClientTask client2 (foo2.in (), callback2.in ());
 176   if (client2.open() != 0)
 177     {
 178       ACE_ERROR((LM_ERROR,
 179         "(%P|%t) Failed to open the collocated client2.\n"));
 180       throw TestException();
 181     }
 182
 183   // Run the ORB event loop.
 184   orb_->run();
 185
 186   ACE_DEBUG((LM_DEBUG,
 187              "(%P|%t) ServerApp ORB has stopped running.\n"));
 188
 189   ACE_DEBUG((LM_DEBUG,
 190              "(%P|%t) Wait for the OrbShutdownTask.\n"));
 191
 192   TheOrbShutdownTask::instance()->wait ();
 193
 194   ACE_DEBUG((LM_DEBUG,
 195              "(%P|%t) Wait for the collocated client task.\n"));
 196
 197   client1.wait ();
 198   client2.wait ();
 199
 200   ACE_DEBUG((LM_DEBUG,
 201              "(%P|%t) Stop the CSD strategy.\n"));
 202
 203   ACE_DEBUG((LM_DEBUG,
 204              "(%P|%t) ServerApp is destroying the Root POA.\n"));
 205
 206   // Tear-down the root poa and orb_.
 207   root_poa->destroy(1, 1);
 208
 209   ACE_DEBUG((LM_DEBUG,
 210              "(%P|%t) ServerApp is destroying the ORB.\n"));
 211
 212   orb_->destroy();
 213
 214   ACE_DEBUG((LM_DEBUG,
 215              "(%P|%t) ServerApp has completed running successfully.\n"));
 216
 217   return 0;
 218 }
 219
 220
 221 int
 222 ServerApp::parse_args(int argc, ACE_TCHAR* argv[])
 223 {
 224   this->exe_name_ = argv[0];
 225
 226   ACE_Get_Opt get_opts(argc, argv, ACE_TEXT("s"));
 227
 228   int c;
 229
 230   while ((c = get_opts()) != -1)
 231     {
 232       switch (c)
 233       {
 234         case 's':
 235           this->synch_with_server_ = 1;
 236           break;
 237
 238         case '?':
 239           this->usage_statement();
 240           return 1;
 241
 242         default:
 243           this->usage_statement();
 244           return -1;
 245       }
 246     }
 247
 248   return 0;
 249 }
 250
 251
 252 Foo_ptr
 253 ServerApp::create_foo (
 254   PortableServer::POA_ptr poa,
 255   const char* servant_name)
 256 {
 257   PortableServer::ServantBase_var servant
 258     = new Foo_i(servant_name);
 259
 260   PortableServer::ObjectId_var id =
 261                 PortableServer::string_to_ObjectId(servant_name);
 262
 263   poa->activate_object_with_id(id.in(), servant.in());
 264
 265   CORBA::Object_var obj = poa->id_to_reference(id.in());
 266
 267   if (CORBA::is_nil(obj.in()))
 268     {
 269       ACE_ERROR((LM_ERROR,
 270                   "(%P|%t) Failed to activate servant (%s).\n",
 271                   servant_name));
 272       throw TestException();
 273     }
 274
 275   Foo_var foo = Foo::_narrow (obj.in ());
 276
 277   return foo._retn ();
 278 }
 279
 280 Callback_ptr
 281 ServerApp::create_callback(
 282   PortableServer::POA_ptr poa,
 283   const char* servant_name)
 284 {
 285   PortableServer::ServantBase_var servant
 286     = new Callback_i();
 287
 288   PortableServer::ObjectId_var id =
 289     PortableServer::string_to_ObjectId("callback");
 290
 291   poa->activate_object_with_id(id.in(), servant.in());
 292
 293   CORBA::Object_var obj = poa->id_to_reference(id.in());
 294
 295   if (CORBA::is_nil(obj.in()))
 296   {
 297     ACE_ERROR((LM_ERROR,
 298       "(%P|%t) Failed to activate servant (%s).\n",
 299       servant_name));
 300     throw TestException();
 301   }
 302
 303   Callback_var callback = Callback::_narrow (obj.in ());
 304
 305   return callback._retn ();
 306 }
 307
 308 void
 309 ServerApp::usage_statement()
 310 {
 311   ACE_ERROR((LM_ERROR,
 312              "Usage:  %s [options]\n\n"
 313              "OPTIONS:\n\n"
 314              "\t[-s <synch_with_server>]\n"
 315              "\t[-?]\n\n",
 316              this->exe_name_.c_str()));
 317 }
 318