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    std::string sub_id = "1";
 130
 131    auto gnblist = get_rnib_gnblist();
 132    int sz = gnblist.size();
 133
 134    if(sz <= 0)
 135            mdclog_write(MDCLOG_INFO,"Subscriptions cannot be sent as GNBList in RNIB is NULL");
 136
 137    for(int i = 0; i<sz; i++){
 138
 139          //give the message to subscription handler, along with the transmitter.
 140          strcpy((char*)meid,gnblist[i].c_str());
 141
 142         // char *strMsg = "Subscription Request from HelloWorld XApp\0";
 143         // strncpy((char *)data,strMsg,strlen(strMsg));
 144         // data_size = strlen(strMsg);
 145
 146          unsigned char buf[1024];
 147          size_t buf_size = 1024;
 148
 149
 150          HWEventTriggerDefinition eventObj;
 151          eventObj.set_triggerNature(0);
 152
 153          //creating Action Definition
 154          HWActionDefinition e2sm_actdefn1;
 155          e2sm_actdefn1.add(HWActionDefinition::RANParamIEs().set_param_id(1).set_param_name("ENodeBID").set_param_test(1).set_param_value("SR123"));
 156
 157          //first Action Object
 158          E2APAction<HWActionDefinition> actionObj;
 159          actionObj.add(E2APAction<HWActionDefinition>::ActionIEs().set_ricActionID(1).set_ricActionType(1).set_ricActionDefinition(e2sm_actdefn1));
 160
 161          E2APSubscriptionRequest<HWEventTriggerDefinition, HWActionDefinition> requestObj(E2APSubscriptionRequest<HWEventTriggerDefinition, HWActionDefinition>::SubscriptionRequestIEs().set_ranFunctionID(1).set_ricInstanceID(1).set_ricRequestorID(3).set_ricAction_ToBeSetup_List(actionObj).set_ricEventTriggerDefinition(eventObj));
 162
 163          bool res = requestObj.encode(&buf[0], &buf_size);
 164          if(!res)
 165                   mdclog_write(MDCLOG_ERR,"SubscriptionRequest ENCODING Error: %s",requestObj.get_error().c_str());
 166          else
 167                   mdclog_write(MDCLOG_INFO,"SubscriptionRequest ENCODING SUCESS");
 168
 169
 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
 175          strcpy((char*)rmr_header.sid,sub_id.c_str());
 176          strcpy((char*)rmr_header.meid,gnblist[i].c_str());
 177
 178          mdclog_write(MDCLOG_INFO,"Sending subscription in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 179      auto transmitter = std::bind(&XappRmr::xapp_rmr_send,rmr_ref, &rmr_header, (void*)buf );//(void*)data);
 180
 181      int result = subhandler_ref->manage_subscription_request(gnblist[i], transmitter);
 182      if(result==SUBSCR_SUCCESS){
 183         mdclog_write(MDCLOG_INFO,"Subscription SUCCESSFUL in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 184      }
 185      else {
 186                 mdclog_write(MDCLOG_ERR,"Subscription FAILED in file= %s, line=%d for MEID %s",__FILE__,__LINE__, meid);
 187      }
 188    }
 189
 190 }
 191
 192 void Xapp::startup_get_policies(void){
 193
 194   int policy_id = HELLOWORLD_POLICY_ID;
 195
 196   std::string policy_query = "{\"policy_type_id\":" + std::to_string(policy_id) + "}";
 197   unsigned char * message = (unsigned char *)calloc(policy_query.length(), sizeof(unsigned char));
 198   memcpy(message, policy_query.c_str(),  policy_query.length());
 199   xapp_rmr_header header;
 200   header.payload_length = policy_query.length();
 201   header.message_type = A1_POLICY_QUERY;
 202   mdclog_write(MDCLOG_INFO, "Sending request for policy id %d\n", policy_id);
 203   rmr_ref->xapp_rmr_send(&header, (void *)message);
 204   free(message);
 205
 206 }
 207
 208 void Xapp::set_rnib_gnblist(void) {
 209
 210            openSdl();
 211
 212            void *result = getListGnbIds();
 213            if(strlen((char*)result) < 1){
 214                     mdclog_write(MDCLOG_ERR, "ERROR: no data from getListGnbIds\n");
 215                 return;
 216             }
 217
 218             mdclog_write(MDCLOG_INFO, "GNB List in R-NIB %s\n", (char*)result);
 219
 220
 221             Document doc;
 222             ParseResult parseJson = doc.Parse<kParseStopWhenDoneFlag>((char*)result);
 223             if (!parseJson) {
 224                 std::cerr << "JSON parse error: %s (%u)", GetParseErrorFunc(parseJson.Code());
 225                 return;
 226             }
 227
 228             if(!doc.HasMember("gnb_list")){
 229                 mdclog_write(MDCLOG_INFO, "JSON Has No GNB List Object");
 230                 return;
 231             }
 232             assert(doc.HasMember("gnb_list"));
 233
 234             const Value& gnblist = doc["gnb_list"];
 235             if (gnblist.IsNull())
 236               return;
 237
 238             if(!gnblist.IsArray()){
 239                 mdclog_write(MDCLOG_INFO, "GNB List is not an array");
 240                 return;
 241             }
 242
 243
 244                 assert(gnblist.IsArray());
 245             for (SizeType i = 0; i < gnblist.Size(); i++) // Uses SizeType instead of size_t
 246             {
 247                 assert(gnblist[i].IsObject());
 248                 const Value& gnbobj = gnblist[i];
 249                 assert(gnbobj.HasMember("inventory_name"));
 250                 assert(gnbobj["inventory_name"].IsString());
 251                 std::string name = gnbobj["inventory_name"].GetString();
 252                 rnib_gnblist.push_back(name);
 253
 254             }
 255             closeSdl();
 256             return;
 257
 258 }
 259
 260
 261
 262
 263
 264