src/xapp.cc

   1 /*
   2 ==================================================================================
   3
   4         Copyright (c) 2019-2020 AT&T Intellectual Property.
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18  */
  19 /*
  20  * xapp.cc
  21  *
  22  *  Mar, 2020 (Shraboni Jana)
  23  */
  24
  25 #include "xapp.hpp"
  26 #define BUFFER_SIZE 1024
  27
  28  Xapp::Xapp(XappSettings &config, XappRmr &rmr){
  29
  30           rmr_ref = &rmr;
  31           config_ref = &config;
  32           xapp_mutex = NULL;
  33           subhandler_ref = NULL;
  34           return;
  35   }
  36
  37 Xapp::~Xapp(void){
  38
  39         //Joining the threads
  40         int threadcnt = xapp_rcv_thread.size();
  41                 for(int i=0; i<threadcnt; i++){
  42                         if(xapp_rcv_thread[i].joinable())
  43                                 xapp_rcv_thread[i].join();
  44         }
  45         xapp_rcv_thread.clear();
  46
  47         if(xapp_mutex!=NULL){
  48                 xapp_mutex->~mutex();
  49                 delete xapp_mutex;
  50         }
  51 };
  52
  53 //Stop the xapp. Note- To be run only from unit test scripts.
  54 void Xapp::stop(void){
  55   // Get the mutex lock
  56         std::lock_guard<std::mutex> guard(*xapp_mutex);
  57         rmr_ref->set_listen(false);
  58         rmr_ref->~XappRmr();
  59
  60         //Detaching the threads....not sure if this is the right way to stop the receiver threads.
  61         //Hence function should be called only in Unit Tests
  62         int threadcnt = xapp_rcv_thread.size();
  63         for(int i=0; i<threadcnt; i++){
  64                 xapp_rcv_thread[i].detach();
  65         }
  66         sleep(10);
  67 }
  68
  69 void Xapp::startup(SubscriptionHandler &sub_ref) {
  70
  71         subhandler_ref = &sub_ref;
  72         set_rnib_gnblist();
  73
  74         //send subscriptions.
  75         startup_subscribe_requests();
  76
  77         //read A1 policies
  78         startup_get_policies();
  79         return;
  80 }
  81 void Xapp::Run(){
  82         rmr_ref->set_listen(true);
  83         if(xapp_mutex == NULL){
  84                 xapp_mutex = new std::mutex();
  85         }
  86         std::lock_guard<std::mutex> guard(*xapp_mutex);
  87
  88         for(int j=0; j < _callbacks.size(); j++){
  89                 std::thread th_recv([&](){ rmr_ref->xapp_rmr_receive(std::move(_callbacks[j]), rmr_ref);});
  90                 xapp_rcv_thread.push_back(std::move(th_recv));
  91         }
  92
  93         return;
  94 }
  95 //Starting a seperate single receiver
  96 void Xapp::start_xapp_receiver(XappMsgHandler& mp_handler){
  97         //start a receiver thread. Can be multiple receiver threads for more than 1 listening port.
  98         rmr_ref->set_listen(true);
  99         if(xapp_mutex == NULL){
 100                 xapp_mutex = new std::mutex();
 101         }
 102
 103         mdclog_write(MDCLOG_INFO,"Receiver Thread file= %s, line=%d",__FILE__,__LINE__);
 104         std::lock_guard<std::mutex> guard(*xapp_mutex);
 105         std::thread th_recv([&](){ rmr_ref->xapp_rmr_receive(std::move(mp_handler), rmr_ref);});
 106         xapp_rcv_thread.push_back(std::move(th_recv));
 107         return;
 108
 109
 110
 111 }
 112 void Xapp::shutdown(){
 113
 114         return;
 115
 116 }
 117
 118
 119 void Xapp::startup_subscribe_requests(void ){
 120
 121    bool res;
 122    size_t data_size = ASN_BUFF_MAX_SIZE;
 123    unsigned char        data[data_size];
 124    unsigned char meid[RMR_MAX_MEID];
 125    std::string xapp_id = config_ref->operator [](XappSettings::SettingName::XAPP_ID);
 126
 127    mdclog_write(MDCLOG_INFO,"Preparing to send subscription in file= %s, line=%d",__FILE__,__LINE__);
 128
 129    auto gnblist = get_rnib_gnblist();
 130
 131    int sz = gnblist.size();
 132
 133    if(sz <= 0)
 134            mdclog_write(MDCLOG_INFO,"Subscriptions cannot be sent as GNBList in RNIB is NULL");
 135
 136    for(int i = 0; i<sz; i++){
 137
 138          //give the message to subscription handler, along with the transmitter.
 139          strcpy((char*)meid,gnblist[i].c_str());
 140
 141         // char *strMsg = "Subscription Request from HelloWorld XApp\0";
 142         // strncpy((char *)data,strMsg,strlen(strMsg));
 143         // data_size = strlen(strMsg);
 144
 145          subscription_helper  din;
 146          subscription_helper  dout;
 147
 148          subscription_request sub_req;
 149          subscription_request sub_recv;
 150
 151          unsigned char buf[BUFFER_SIZE];
 152          size_t buf_size = BUFFER_SIZE;
 153          bool res;
 154
 155
 156          //Random Data  for request
 157          int request_id = 1;
 158          int function_id = 0;
 159          std::string event_def = "HelloWorld Event Definition";
 160
 161          din.set_request(request_id);
 162          din.set_function_id(function_id);
 163          din.set_event_def(event_def.c_str(), event_def.length());
 164
 165          std::string act_def = "HelloWorld Action Definition";
 166
 167          din.add_action(1,1,(void*)act_def.c_str(), act_def.length(), 0);
 168
 169          res = sub_req.encode_e2ap_subscription(&buf[0], &buf_size, din);
 170
 171          xapp_rmr_header rmr_header;
 172          rmr_header.message_type = RIC_SUB_REQ;
 173          rmr_header.payload_length = buf_size; //data_size
 174          strcpy((char*)rmr_header.meid,gnblist[i].c_str());
 175
 176          mdclog_write(MDCLOG_INFO,"Sending subscription in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 177      auto transmitter = std::bind(&XappRmr::xapp_rmr_send,rmr_ref, &rmr_header, (void*)buf );//(void*)data);
 178
 179      int result = subhandler_ref->manage_subscription_request(gnblist[i], transmitter);
 180      if(result==SUBSCR_SUCCESS){
 181         mdclog_write(MDCLOG_INFO,"Subscription SUCCESSFUL in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 182      }
 183      else {
 184                 mdclog_write(MDCLOG_ERR,"Subscription FAILED in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 185      }
 186    }
 187
 188 }
 189
 190 void Xapp::startup_get_policies(void){
 191
 192   int policy_id = HELLOWORLD_POLICY_ID;
 193
 194   std::string policy_query = "{\"policy_type_id\":" + std::to_string(policy_id) + "}";
 195   unsigned char * message = (unsigned char *)calloc(policy_query.length(), sizeof(unsigned char));
 196   memcpy(message, policy_query.c_str(),  policy_query.length());
 197   xapp_rmr_header header;
 198   header.payload_length = policy_query.length();
 199   header.message_type = A1_POLICY_QUERY;
 200   mdclog_write(MDCLOG_INFO, "Sending request for policy id %d\n", policy_id);
 201   rmr_ref->xapp_rmr_send(&header, (void *)message);
 202   free(message);
 203
 204 }
 205
 206 void Xapp::set_rnib_gnblist(void) {
 207
 208            openSdl();
 209
 210            void *result = getListGnbIds();
 211            if(strlen((char*)result) < 1){
 212                     mdclog_write(MDCLOG_ERR, "ERROR: no data from getListGnbIds\n");
 213                 return;
 214             }
 215
 216             mdclog_write(MDCLOG_INFO, "GNB List in R-NIB %s\n", (char*)result);
 217
 218
 219             Document doc;
 220             ParseResult parseJson = doc.Parse<kParseStopWhenDoneFlag>((char*)result);
 221             if (!parseJson) {
 222                 std::cerr << "JSON parse error: %s (%u)", GetParseErrorFunc(parseJson.Code());
 223                 return;
 224             }
 225
 226             if(!doc.HasMember("gnb_list")){
 227                 mdclog_write(MDCLOG_INFO, "JSON Has No GNB List Object");
 228                 return;
 229             }
 230             assert(doc.HasMember("gnb_list"));
 231
 232             const Value& gnblist = doc["gnb_list"];
 233             if (gnblist.IsNull())
 234               return;
 235
 236             if(!gnblist.IsArray()){
 237                 mdclog_write(MDCLOG_INFO, "GNB List is not an array");
 238                 return;
 239             }
 240
 241
 242                 assert(gnblist.IsArray());
 243             for (SizeType i = 0; i < gnblist.Size(); i++) // Uses SizeType instead of size_t
 244             {
 245                 assert(gnblist[i].IsObject());
 246                 const Value& gnbobj = gnblist[i];
 247                 assert(gnbobj.HasMember("inventory_name"));
 248                 assert(gnbobj["inventory_name"].IsString());
 249                 std::string name = gnbobj["inventory_name"].GetString();
 250                 rnib_gnblist.push_back(name);
 251
 252             }
 253             closeSdl();
 254             return;
 255
 256 }
 257
 258
 259
 260
 261
 262