src/plpbus/OrbServerImpl.cc

   1 /*
   2  * OrbServerImpl.cc
   3  *
   4  *  Created on: Aug 11, 2010
   5  *      Author: lamikr
   6  */
   7
   8 #include <plp/log.h>
   9
  10 #include "ServerEventThread.hh"
  11 #include "OrbServerImpl.hh"
  12 #include "ClientServerCommon.hh"
  13 #include "BusMessageInternal.hh"
  14
  15 using namespace std;
  16 using namespace plpbus;
  17 using namespace plpbus_orb;
  18
  19 static omni_mutex       server_thread_mutex;
  20 static omni_condition   server_thread_shutdown_signal(&server_thread_mutex);
  21
  22 OrbServerImpl::OrbServerImpl()
  23 {
  24         _orb                    = NULL;
  25         _poa                    = NULL;
  26         _server_thread_count    = 0;
  27         _shutdown_pending       = 0;
  28         _listener               = NULL;
  29 }
  30
  31 OrbServerImpl::~OrbServerImpl()
  32 {
  33         log_info("OrbServerImpl destroyed.\n");
  34 }
  35
  36 void OrbServerImpl::send_message_and_request_response(OrbClient_ptr response_listener_param,
  37                                                 const char* msg_req)
  38 {
  39         char    *msg_rsp;
  40
  41         msg_rsp = NULL;
  42         if (CORBA::is_nil(response_listener_param) == false) {
  43                 if (_listener != NULL) {
  44                         log_debug("send_message_and_request_response(): %s, server callback != NULL\n", msg_req);
  45                         _listener->request_received(msg_req, &msg_rsp);
  46                         response_listener_param->receive_response_message(msg_rsp);
  47                 }
  48                 else {
  49                         log_error("send_message_and_request_response() error, server callback == NULL\n");
  50                 }
  51         }
  52         else {
  53                 log_error("invalid callback object received.\n");
  54         }
  55 }
  56
  57 char *OrbServerImpl::send_message_and_wait_response(const char* msg_req_param, ::CORBA::Long& err_flg) {
  58         char    *msg_rsp;
  59         char    *ret_val;
  60
  61         err_flg = 0;
  62         msg_rsp = NULL;
  63         if (_listener != NULL) {
  64                 log_debug("send_message_and_wait_response(): %s, server_callback != NULL\n", msg_req_param);
  65                 _listener->request_received(msg_req_param, &msg_rsp);
  66         }
  67         else {
  68                 log_error("send_message_and_wait_response() error, server callback == NULL\n");
  69                 msg_rsp = strdup(msg_req_param);
  70                 err_flg = -1;
  71         }
  72         ret_val = CORBA::string_dup(msg_rsp);
  73         return ret_val;
  74 }
  75
  76 void OrbServerImpl::send_dataitem_message_and_request_response(OrbClient_ptr response_listener_param,
  77                                                         const ::DataItemSequence& msg_req_param) {
  78         BusMessage      *msg_req;
  79         BusMessage      *msg_rsp;
  80
  81         if (CORBA::is_nil(response_listener_param) == false) {
  82                 if (_listener != NULL) {
  83                         msg_req = new BusMessageInternal(msg_req_param);
  84                         log_debug("send_dataitem_message_and_request_response(), server_callback != NULL\n");
  85                         log_debug("msg_req:\n");
  86                         msg_req->printout();
  87                         msg_rsp = NULL;
  88                         _listener->request_received(msg_req, &msg_rsp);
  89                         log_debug("msg_rsp length: %d\n", msg_rsp->_dataItemSeq.length());
  90                         msg_rsp->printout();
  91                         response_listener_param->receive_response_dataitem_sequence(msg_rsp->_dataItemSeq);
  92                 }
  93                 else {
  94                         log_error("send_dataitem_message_and_request_response() error, server_callback == NULL\n");
  95                 }
  96         }
  97 }
  98
  99 CORBA::Long OrbServerImpl::send_dataitem_message_and_wait_response(const DataItemSequence& req_seq_param,
 100                 DataItemSequence_out rsp_seq_param) {
 101         BusMessage      *msg_req;
 102         BusMessage      *msg_rsp;
 103
 104         msg_req = new BusMessageInternal(req_seq_param);
 105         msg_rsp = NULL;
 106         _listener->request_received(msg_req, &msg_rsp);
 107         //rsp_seq_param = new DataItemSequence_out(msg_rsp._dataItemSeq);
 108         rsp_seq_param._data     = &(msg_rsp->_dataItemSeq);
 109
 110         return 0;
 111 }
 112
 113 void OrbServerImpl::add_event_listener(OrbClient_ptr client_param,
 114                         const char *msg,
 115                         CORBA::UShort period_secs)
 116 {
 117         ServerEventThread       *srvr_thread;
 118
 119         if (CORBA::is_nil(client_param) == false) {
 120                 log_debug("add_event_listener()\n");
 121                 server_thread_mutex.lock();
 122                 _server_thread_count++;
 123                 server_thread_mutex.unlock();
 124                 srvr_thread     = new ServerEventThread(client_param, msg, period_secs, this);
 125                 srvr_thread->start();
 126         }
 127         else {
 128                 cerr << "Failed to add event listener, listener NULL.\n";
 129         }
 130 }
 131
 132 void OrbServerImpl::shutdown()
 133 {
 134         omni_mutex_lock sync(server_thread_mutex);
 135         if (is_shutdown_pending() == 0) {
 136                 cout << "shutdown request received!" << endl;
 137                 // Tell the servers to exit, and wait for them to do so.
 138                 _shutdown_pending       = 1;
 139                 while(_server_thread_count > 0) {
 140                         server_thread_shutdown_signal.wait();
 141                 }
 142                 // Shutdown the ORB (but do not wait for completion).  This also
 143                 // causes the main thread to unblock from CORBA::ORB::run().
 144                 _orb->shutdown(0);
 145         }
 146 }
 147
 148 int OrbServerImpl::init() {
 149         int     argc;
 150         char    **argv;
 151         int     retVal;
 152
 153         argc    = -1;
 154         argv    = NULL;
 155         retVal  = 0;
 156         _orb    = CORBA::ORB_init(argc, argv);
 157         if (_orb != NULL) {
 158                 _poa    = create_poa(_orb);
 159                 retVal  = 0;
 160         }
 161         else {
 162                 cout << "init() failed"  << endl;
 163         }
 164         return retVal;
 165 }
 166
 167 int OrbServerImpl::launch(const char *server_name)
 168 {
 169         CORBA::Object_var               server_ref;
 170         PortableServer::ObjectId_var    server_id;
 171         bool                            ok_flg;
 172         int                             ret_val;
 173         CosNaming::NamingContext_var    naming_context;
 174
 175         ret_val         = -1;
 176         POA_plpbus_orb::OrbServer_tie<OrbServerImpl> server_impl(this);
 177         server_id       = _poa->activate_object(&server_impl);
 178         server_ref      = server_impl._this();
 179         //this->_remove_ref();
 180         naming_context  = get_service_naming_context(CONST_CONTEXT_NAME__PLPBUS, CONST_CONTEXT_KIND__PLPBUS);
 181         if (naming_context != NULL) {
 182                 ok_flg  = bind_naming_context_and_service(naming_context, server_ref, server_name, CONST_CONTEXT_KIND__PLPBUS);
 183                 if (ok_flg == true) {
 184                         cout << "Registered to naming service: " << server_name << endl;
 185                         _orb->run();
 186                         ret_val = 0;
 187                 }
 188                 else {
 189                         cout << "Failed to register to naming service: " << server_name << endl;
 190                 }
 191         }
 192         return ret_val;
 193 }
 194
 195 PortableServer::POA_var OrbServerImpl::create_poa(CORBA::ORB_var orb) {
 196         PortableServer::POA_var         ret_val;
 197         CORBA::Object_var               poa_obj;
 198         CORBA::PolicyList               policy_list;
 199         CORBA::Any                      policyVal;
 200         PortableServer::POAManager_var  poa_man;
 201         PortableServer::POA_var         rootpoa;
 202
 203         ret_val = NULL;
 204         poa_obj = orb->resolve_initial_references(CONST_ROOT_POA_NAME);
 205         if (poa_obj != NULL) {
 206                 rootpoa = PortableServer::POA::_narrow(poa_obj);
 207                 if (rootpoa != NULL) {
 208                         poa_man = rootpoa->the_POAManager();
 209                         if (poa_man != NULL) {
 210                                 poa_man->activate();
 211                                 // bidirectional policy
 212                                 policy_list.length(1);
 213                                 policyVal       <<= BiDirPolicy::BOTH;
 214                                 policy_list[0] = orb->create_policy(BiDirPolicy::BIDIRECTIONAL_POLICY_TYPE, policyVal);
 215                                 ret_val = rootpoa->create_POA(CONST_ROOT_POA_BIDIR_POLICY_NAME,
 216                                                         poa_man,
 217                                                         policy_list);
 218                         }
 219                 }
 220         }
 221         else {
 222                 cerr << "Failed to create RootPOA." << endl;
 223         }
 224         return ret_val;
 225 }
 226
 227 CosNaming::NamingContext_var OrbServerImpl::get_service_naming_context(const char *service_name_param,
 228                                                                 const char *service_kind_param)
 229 {
 230         CosNaming::NamingContext_var    ret_val;
 231         CosNaming::NamingContext_var    ns_context;
 232         CORBA::Object_var               ns_obj;
 233         CORBA::Object_var               service_obj;
 234         CosNaming::Name                 context_data;
 235
 236         ret_val = NULL;
 237         try {
 238                 // get nameservice reference
 239                 ns_obj          = _orb->resolve_initial_references(CONST_NAME_SERVICE_NAME);
 240                 // get nameservice context
 241                 ns_context      = CosNaming::NamingContext::_narrow(ns_obj);
 242                 if (CORBA::is_nil(ns_context) == false) {
 243                         context_data.length(1);
 244                         context_data[0].id   = service_name_param;
 245                         context_data[0].kind = service_kind_param;
 246                         try {
 247                                 service_obj     = ns_context->resolve(context_data);
 248                                 if (CORBA::is_nil(service_obj)) {
 249                                         // not found, try to bind the new context to name service
 250                                         ret_val = ns_context->bind_new_context(context_data);
 251                                         if (CORBA::is_nil(ret_val) ) {
 252                                                 cerr << "Failed to create new context to name service for " << service_name_param << "." << endl;
 253                                         }
 254                                 }
 255                                 else {
 256                                         ret_val = CosNaming::NamingContext::_narrow(service_obj);
 257                                         if (CORBA::is_nil(ret_val) ) {
 258                                                 cerr << "Failed to get existing context from name service for " << service_name_param << ", narrowing failed." << endl;
 259                                         }
 260                                 }
 261                         }
 262                         catch(CosNaming::NamingContext::AlreadyBound& ex) {
 263                                 cerr << "Could not get context from nameservice for " << service_name_param << ". Context with same name already existed."<< endl;
 264                         }
 265                 }
 266         }
 267         catch (CORBA::ORB::InvalidName&) {
 268                 // This should not happen!
 269                 cerr << "Could not get context from name service for " << service_name_param << ", name service does not exist." << endl;
 270         }
 271         catch(CORBA::TRANSIENT& ex) {
 272                 cerr << "Could not get context from name service for " << service_name_param << ",  verify that name service is running. " << service_name_param << endl;
 273         }
 274         catch (CORBA::NO_RESOURCES&) {
 275                 cerr << "Could not get context from name service for " << service_name_param << ". Name service is not running or has configuration problem." << endl;
 276         }
 277         catch(CORBA::SystemException& ex) {
 278                 cerr << "Could not get context from name service for " << service_name_param << ", could not determine reason." << endl;
 279         }
 280         return ret_val;
 281 }
 282
 283 bool OrbServerImpl::bind_naming_context_and_service(CosNaming::NamingContext_var service_context_param,
 284                                         CORBA::Object_ptr service_ref_param,
 285                                         const char *service_name_param,
 286                                         const char *service_kind_param)
 287 {
 288         bool            retVal;
 289         CosNaming::Name context_data;
 290
 291         retVal = false;
 292         try {
 293                 context_data.length(1);
 294                 context_data[0].id   = service_name_param;
 295                 context_data[0].kind = service_kind_param;
 296                 try {
 297                         service_context_param->bind(context_data, service_ref_param);
 298                         retVal  = true;
 299                 }
 300                 catch(CosNaming::NamingContext::AlreadyBound& ex) {
 301                         /**
 302                          * service existed already for the naming context with similar description.
 303                          * Replace the existing one with a new one.
 304                          */
 305                         cout << "service " << service_name_param << " existed, replacing it." << endl;
 306                         service_context_param->rebind(context_data, service_ref_param);
 307                         retVal  = true;
 308                 }
 309         }
 310         catch (CosNaming::NamingContext::InvalidName&) {
 311                 cerr << "Could not register service to name server, invalid service name." << endl;
 312         }
 313         catch (CosNaming::NamingContext::NotFound&) {
 314                 cerr << "Could not register service to name server, service object reference is invalid." << endl;
 315         }
 316         catch (CosNaming::NamingContext::CannotProceed&) {
 317                 // This should not happen!
 318                 cerr << "Could not register service to name server, unknown error." << endl;
 319         }
 320         catch(CORBA::SystemException& ex) {
 321                 cerr << "Could not register service to name server, unknown error." << endl;
 322         }
 323         return retVal;
 324 }
 325
 326 int OrbServerImpl::add_server_listener(IServerListener *listener_param)
 327 {
 328         int     retVal;
 329
 330         retVal          = 0;
 331         cout << "register_request_received_callback() started"  << endl;
 332         _listener       = listener_param;
 333         cout << "register_callback() done"  << endl;
 334         return retVal;
 335 }
 336
 337 int OrbServerImpl::is_shutdown_pending() {
 338         return _shutdown_pending;
 339 }
 340
 341 void OrbServerImpl::server_thread_closed() {
 342         bool send_signal = false;
 343
 344         server_thread_mutex.lock();
 345         _server_thread_count--;
 346         if (_server_thread_count == 0) {
 347                 send_signal     = true;
 348         }
 349         server_thread_mutex.unlock();
 350         if (send_signal == true) {
 351                 server_thread_shutdown_signal.signal();
 352         }
 353 }