src/plpbus/BusClient.cc

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