src/plpbus/BusClient.cc

   1 /*
   2  * BusClient.cc
   3  *
   4  *  Created on: Jun 7, 2010
   5  *      Author: lamikr
   6  */
   7
   8 #include <malloc.h>
   9 #include <plp/log.h>
  10 #include <plp/retval.h>
  11
  12 #include "BusClient.hh"
  13 #include "ClientServerCommon.hh"
  14 #include "BusMessageInternal.hh"
  15 #include "RequestResponseBus.hh"
  16
  17 using namespace std;
  18 using namespace plpbus;
  19 using namespace plpbus_orb;
  20
  21 class OrbClientImpl : public virtual POA_plpbus_orb::OrbClient {
  22         private:
  23                 IClientListener *listener;
  24         public:
  25                 OrbClientImpl(IClientListener *listener_param)  {
  26                         listener        = listener_param;
  27                 }
  28                 virtual ~OrbClientImpl() {}
  29
  30                 virtual void receive_event_message(const char *event_param) {
  31                         if (listener != NULL) {
  32                                 listener->event_received(event_param);
  33                         }
  34                 }
  35
  36                 virtual void receive_event_dataitem_sequence(const ::plpbus_orb::DataItemSequence& event_param) {
  37                         BusMessageInternal      *bus_msg;
  38
  39                         if (listener != NULL) {
  40                                 bus_msg = new BusMessageInternal(event_param);
  41                                 listener->event_received(bus_msg);
  42                         }
  43                 }
  44
  45                 virtual void receive_response_message(const char *msg_rsp_param) {
  46                         if (listener != NULL) {
  47                                 listener->response_received(msg_rsp_param);
  48                         }
  49                 }
  50
  51                 virtual void receive_response_dataitem_sequence(const ::plpbus_orb::DataItemSequence& msg_rsp_param) {
  52                         const BusMessageInternal        *bus_msg;
  53
  54                         if (listener != NULL) {
  55                                 bus_msg = new BusMessageInternal(msg_rsp_param);
  56                                 listener->response_received(bus_msg);
  57                                 delete(bus_msg);
  58                         }
  59                 }
  60 };
  61
  62 BusClient::BusClient()
  63 {
  64         log_debug("created\n");;
  65         _orb            = NULL;
  66         _poa            = NULL;
  67 }
  68
  69 BusClient::~BusClient()
  70 {
  71         if (_poa != NULL) {
  72                 //_poa->destroy(true, true);
  73         }
  74         if (_orb != NULL) {
  75                 _orb->destroy();
  76                 _orb    = NULL;
  77         }
  78         log_debug("destroyed\n");
  79 }
  80
  81 int BusClient::init(const char *server_name) {
  82         int                     argc;
  83         char                    **argv;
  84         CORBA::Object_var       server_obj;
  85         int                     ret_val;
  86
  87         argc    = 0;
  88         argv    = NULL;
  89         ret_val = PLP_ERR;
  90         log_info("CORBA client init() started\n");
  91         _orb    = CORBA::ORB_init(argc, argv);
  92         if (CORBA::is_nil(_orb) == false) {
  93                 log_debug("ORB_init() done, finding server: %s\n", server_name);
  94                 server_obj      = find_server_object_by_name(_orb,
  95                                                         CONST_CONTEXT_NAME__PLPBUS,
  96                                                         CONST_CONTEXT_KIND__PLPBUS,
  97                                                         server_name,
  98                                                         CONST_CONTEXT_NAME__PLPBUS);
  99                 if (CORBA::is_nil(server_obj) == false) {
 100                         _server         = plpbus_orb::OrbServer::_narrow(server_obj);
 101                         if (CORBA::is_nil(_server) == false) {
 102                                 log_info("CORBA service found: %s", server_name);
 103                                 ret_val = PLP_OK;
 104                         }
 105                 }
 106                 else {
 107                         log_error("Could not find CORBA service: %s\n", server_name);
 108                 }
 109         }
 110         else {
 111                 log_error("Could not init CORBA client\n");
 112         }
 113         return ret_val;
 114 }
 115
 116 int BusClient::add_client_listener(IClientListener *listener_param) {
 117         log_debug("add_client_listener() started\n");
 118         if (_poa == NULL) {
 119                 OrbClientImpl                   *client_cb;
 120                 PortableServer::ObjectId_var    oid;
 121
 122                 _poa            = create_poa(_orb);
 123                 client_cb       = new OrbClientImpl(listener_param);
 124                 oid             = _poa->activate_object(client_cb);
 125                 _client         = client_cb->_this();
 126                 client_cb->_remove_ref();
 127                 _server->add_event_listener(_client, "event_message", CONST_DEFAULT_EVENT_INTERVAL);
 128         }
 129         log_debug("add_client_listener() done\n");
 130         return 0;
 131 }
 132
 133 int BusClient::send_message_and_request_response(const char *msg_req_param) {
 134         _server->send_message_and_request_response(_client, msg_req_param);
 135         return 0;
 136 }
 137
 138 int BusClient::send_message_and_wait_response(const char *msg_req_param, char **msg_rsp_param) {
 139         CORBA::Long     ret_val;
 140         char            *rsp;
 141
 142         rsp             = _server->send_message_and_wait_response(msg_req_param, ret_val);
 143         *msg_rsp_param  = strdup(rsp);
 144         CORBA::string_free(rsp);
 145         return ret_val;
 146 }
 147
 148 int BusClient::send_message_and_request_response(const BusMessage *msg_req_param) {
 149         DataItemSequence        *seq;
 150
 151         seq     = (DataItemSequence *)msg_req_param->_dataItemSeq;
 152         _server->send_dataitem_message_and_request_response(_client, *seq);
 153         return 0;
 154 }
 155
 156 int BusClient::send_message_and_wait_response(const BusMessage *msg_req_param, BusMessage **msg_rsp_param) {
 157         DataItemSequence        *seq_rsp;
 158         DataItemSequence        *seq;
 159
 160         seq_rsp = NULL;
 161         seq     = (DataItemSequence *)msg_req_param->_dataItemSeq;
 162         _server->send_dataitem_message_and_wait_response(*seq, seq_rsp);
 163         *msg_rsp_param  = new BusMessageInternal(*seq_rsp);
 164         delete(seq_rsp);
 165         return 0;
 166 }
 167
 168 void BusClient::request_shutdown() {
 169         if (CORBA::is_nil(_server) == false) {
 170                 _server->shutdown();
 171         }
 172 }
 173
 174 PortableServer::POA_var BusClient::create_poa(CORBA::ORB_var orb) {
 175         PortableServer::POA_var         ret_val;
 176         CORBA::Object_var               poa_obj;
 177         CORBA::PolicyList               policy_list;
 178         CORBA::Any                      policyVal;
 179         PortableServer::POAManager_var  poa_man;
 180         PortableServer::POA_var         rootpoa;
 181
 182         ret_val = NULL;
 183         poa_obj = orb->resolve_initial_references(CONST_ROOT_POA_NAME);
 184         if (poa_obj != NULL) {
 185                 rootpoa = PortableServer::POA::_narrow(poa_obj);
 186                 if (rootpoa != NULL) {
 187                         poa_man = rootpoa->the_POAManager();
 188                         if (poa_man != NULL) {
 189                                 poa_man->activate();
 190                                 // bidirectional policy
 191                                 policy_list.length(1);
 192                                 policyVal       <<= BiDirPolicy::BOTH;
 193                                 policy_list[0] = orb->create_policy(BiDirPolicy::BIDIRECTIONAL_POLICY_TYPE, policyVal);
 194                                 ret_val = rootpoa->create_POA(CONST_ROOT_POA_BIDIR_POLICY_NAME,
 195                                                         poa_man,
 196                                                         policy_list);
 197                         }
 198                 }
 199         }
 200         else {
 201                 cerr << "Failed to create RootPOA." << endl;
 202         }
 203         return ret_val;
 204 }
 205
 206 CosNaming::NamingContext_var BusClient::get_name_service_context(CORBA::ORB_var orb_param,
 207                                                                 const char *nameservice_name_param) {
 208         CORBA::Object_var               ns_obj;
 209         CosNaming::NamingContext_var    ret_val;
 210
 211         ret_val = NULL;
 212         try {
 213                 ns_obj  = orb_param->resolve_initial_references(nameservice_name_param);
 214                 if (CORBA::is_nil(ns_obj) == false) {
 215                         // narrow the object reference
 216                         ret_val = CosNaming::NamingContext::_narrow(ns_obj);
 217                 }
 218         }
 219         catch (CORBA::ORB::InvalidName&) {
 220                 // This should not happen!
 221                 cerr << "Could not find name service." << endl;
 222         }
 223         catch(CORBA::TRANSIENT& ex) {
 224                 cerr << "Could not find name service, verify that name service is running. " << endl;
 225         }
 226         catch (CORBA::NO_RESOURCES&) {
 227                 cerr << "Could not find name service, verify that name service is running. " << endl;
 228         }
 229         catch(CORBA::SystemException& ex) {
 230                 cerr << "Could not find name service, unknown reason. " << endl;
 231         }
 232         return ret_val;
 233 }
 234
 235 CORBA::Object_var BusClient::find_server_object_by_name(CORBA::ORB_var orb_param,
 236                                                         const char *leaf_name_param,
 237                                                         const char *leaf_kind_param,
 238                                                         const char *service_name_param,
 239                                                         const char *service_kind_param)
 240 {
 241         CORBA::Object_var               ret_val;
 242         CosNaming::NamingContext_var    ns_root_context;
 243         CosNaming::Name                 service_data;
 244         char *n;
 245
 246         ret_val = CORBA::Object::_nil();
 247         n       = strdup(CONST_NAME_SERVICE_NAME);
 248         ns_root_context = get_name_service_context(orb_param, n);
 249         free(n);
 250         if (CORBA::is_nil(ns_root_context) == false) {
 251                 // define search criteria
 252                 service_data.length(2);
 253                 service_data[0].id   = leaf_name_param;
 254                 service_data[0].kind = leaf_kind_param;
 255                 service_data[1].id   = service_name_param;
 256                 service_data[1].kind = service_kind_param;
 257                 try {
 258                         ret_val = ns_root_context->resolve(service_data);
 259                 }
 260                 catch(CosNaming::NamingContext::NotFound& ex) {
 261                         log_error("Could not find service from the name server.\n");
 262                 }
 263                 catch(CORBA::TRANSIENT& ex) {
 264                         log_error("Could not find service from the name server.\n");
 265                 }
 266                 catch(CORBA::SystemException& ex) {
 267                         log_error("Could not find service from the name server, unknown error.\n");
 268                 }
 269         }
 270         else {
 271                 log_error("Could not find naming service.\n");
 272         }
 273         return ret_val;
 274 }