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