src/E2AP-c/subscription/subscription_handler.cc

   1 /*
   2 ==================================================================================
   3         Copyright (c) 2018-2019 AT&T Intellectual Property.
   4
   5    Licensed under the Apache License, Version 2.0 (the "License");
   6    you may not use this file except in compliance with the License.
   7    You may obtain a copy of the License at
   8
   9        http://www.apache.org/licenses/LICENSE-2.0
  10
  11    Unless required by applicable law or agreed to in writing, software
  12    distributed under the License is distributed on an "AS IS" BASIS,
  13    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14    See the License for the specific language governing permissions and
  15    limitations under the License.
  16 ==================================================================================
  17 */
  18
  19 /* Author : Ashwin Sridharan
  20    Date    : Feb 2019
  21 */
  22
  23
  24 #include <subscription_handler.hpp>
  25 #include <errno.h>
  26
  27 subscription_handler::subscription_handler(void){
  28
  29   init();
  30   _time_out =  std::chrono::seconds(5);
  31   _num_retries = 2;
  32
  33   // bool res;
  34   // unsigned char buffer[128];
  35   // size_t buf_len = 128;
  36
  37   // E2N_E2AP_PDU_t e2ap_pdu;
  38   // subscription_request e2ap_sub_req;
  39
  40   // int request_id = 2;
  41   // int req_seq = 1;
  42   // int function_id = 0;
  43   // int action_id = 0;
  44   // int action_type = 0;
  45   // int message_type = 1;
  46
  47   // subscription_helper sgnb_add_subscr_req;
  48
  49   // //sgnb_add_subscr_req.clear();
  50   // sgnb_add_subscr_req.set_request(request_id, req_seq);
  51   // sgnb_add_subscr_req.set_function_id(function_id);
  52   // sgnb_add_subscr_req.add_action(action_id, action_type);
  53   // std::string test = "This is a test";
  54   // sgnb_add_subscr_req.set_event_def(test.c_str(), test.length());
  55   // std::cout <<"Constructor ........" << std::endl;
  56   // // generate the request pdu
  57   // res = e2ap_sub_req.encode_e2ap_subscription(&buffer[0], &buf_len, &e2ap_pdu, sgnb_add_subscr_req);
  58   // if(! res){
  59   //   mdclog_write(MDCLOG_ERR, "%s, %d: Error encoding subscription pdu. Reason = ", __FILE__, __LINE__);
  60
  61   // }
  62   // std::cout <<"Encoded subscription request pdu " << std::endl;
  63
  64
  65 }
  66
  67 subscription_handler::subscription_handler(unsigned int timeout_seconds, unsigned int num_tries):_time_out(std::chrono::seconds(timeout_seconds)), _num_retries(num_tries){
  68   init();
  69 };
  70
  71 void subscription_handler::init(void){
  72
  73   _data_lock = std::make_unique<std::mutex>();
  74   _cv = std::make_unique<std::condition_variable>();
  75
  76 }
  77
  78 void subscription_handler::clear(void){
  79   {
  80     std::lock_guard<std::mutex> lock(*(_data_lock).get());
  81     requests_table.clear();
  82     subscription_responses.clear();
  83   }
  84
  85 };
  86
  87 size_t subscription_handler::num_pending(void) const {
  88   return requests_table.size();
  89 }
  90
  91 size_t subscription_handler::num_complete(void) const {
  92   return subscription_responses.size();
  93 }
  94
  95
  96 void subscription_handler::set_timeout(unsigned int timeout_seconds){
  97   _time_out = std::chrono::seconds(timeout_seconds);
  98 }
  99
 100 void subscription_handler::set_num_retries(unsigned int num_tries){
 101   _num_retries = num_tries;
 102 };
 103
 104
 105 unsigned int subscription_handler::get_next_id(void){
 106   std::lock_guard<std::mutex> lock(*(_data_lock).get());
 107   unique_request_id ++;
 108   return unique_request_id;
 109 }
 110
 111 bool subscription_handler::add_request_entry(int id, int status){
 112
 113   // add entry in hash table if it does not exist
 114   auto search = requests_table.find(id);
 115   if(search != requests_table.end()){
 116     return false;
 117   }
 118
 119   requests_table[id] = status;
 120   return true;
 121
 122 };
 123
 124 bool subscription_handler::set_request_status(int id, int status){
 125
 126   // change status of a request only if it exists.
 127
 128   auto search = requests_table.find(id);
 129   if(search != requests_table.end()){
 130     requests_table[id] = status;
 131     return true;
 132   }
 133
 134   return false;
 135
 136 };
 137
 138
 139 bool subscription_handler::delete_request_entry(int id){
 140
 141   auto search = requests_table.find(id);
 142   if (search != requests_table.end()){
 143     requests_table.erase(search);
 144     return true;
 145   }
 146
 147   return false;
 148 };
 149
 150 bool subscription_handler::add_subscription_entry(int id, subscription_response_helper &he){
 151
 152   auto search = subscription_responses.find(id);
 153   if (search == subscription_responses.end()){
 154     subscription_responses[id] = he;
 155     return true;
 156   }
 157
 158   return false;
 159 }
 160
 161
 162 bool subscription_handler::delete_subscription_entry(int id){
 163
 164   auto search = subscription_responses.find(id);
 165   if(search == subscription_responses.end()){
 166     return false;
 167   }
 168   else{
 169     subscription_responses.erase(search);
 170     return true;
 171   }
 172
 173 }
 174
 175 subscription_response_helper *  const subscription_handler::get_subscription(int id){
 176   auto search = subscription_responses.find(id);
 177   if(search == subscription_responses.end()){
 178     return NULL;
 179   }
 180   else{
 181     return &(subscription_responses[id]);
 182   }
 183 };
 184
 185
 186 // Handles responses from RMR
 187 void subscription_handler::Response(int message_type, unsigned char *payload, int payload_length){
 188
 189   bool res;
 190   int id;
 191   int type;
 192   int procedureCode;
 193   bool valid_response  =false;
 194
 195   E2N_E2AP_PDU_t * e2ap_recv;
 196   asn_dec_rval_t retval;
 197
 198   subscription_response sub_resp;
 199   subscription_delete_response sub_del_resp;
 200
 201   subscription_response_helper he_response;
 202
 203
 204   e2ap_recv = 0;
 205   retval = asn_decode(0, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2N_E2AP_PDU, (void**)&(e2ap_recv), payload, payload_length);
 206
 207   if(retval.code != RC_OK){
 208     mdclog_write(MDCLOG_ERR, "%s, %d: Error decoding E2AP PDU of RMR type %d. Bytes decoded = %lu out of %d\n", __FILE__, __LINE__, message_type, retval.consumed, payload_length);
 209     ASN_STRUCT_FREE(asn_DEF_E2N_E2AP_PDU, e2ap_recv);
 210     return ;
 211   }
 212
 213   type = e2ap_recv->present;
 214   mdclog_write(MDCLOG_INFO, "Received RMR message of type = %d", type);
 215
 216   if(type == E2N_E2AP_PDU_PR_successfulOutcome){
 217
 218     procedureCode =  e2ap_recv->choice.successfulOutcome->procedureCode;
 219     mdclog_write(MDCLOG_INFO, "Received E2N_E2AP PDU  successful outcome message with procedureCode = %d", procedureCode);
 220
 221     if( procedureCode == E2N_ProcedureCode_id_ricSubscription){
 222       // subscription response
 223       // decode the message
 224       sub_resp.get_fields(e2ap_recv->choice.successfulOutcome, he_response);
 225
 226       {
 227         std::lock_guard<std::mutex> lock(*(_data_lock.get()));
 228         // get the id
 229         id = he_response.get_request_id();
 230         // get status of id
 231         int req_status = get_request_status(id);
 232         if (req_status == request_pending ){
 233           res = add_subscription_entry(id, he_response);
 234           if(res)
 235             set_request_status(id, request_success);
 236
 237           else{
 238             set_request_status(id, request_duplicate);
 239             mdclog_write(MDCLOG_ERR, "Error:: %s, %d: Request %d seems to be a duplicate\n", __FILE__, __LINE__, id);
 240           }
 241
 242           valid_response = true;
 243         }
 244         else if (req_status > 0){
 245           // we don't change status of response since it was not in pending
 246           // we simply fail
 247           mdclog_write(MDCLOG_ERR, "Error:: %s, %d: Request %d is not in request_pending state, is in State = %d\n", __FILE__, __LINE__, id, req_status);
 248
 249         }
 250         else{
 251           mdclog_write(MDCLOG_ERR,  "%s, %d: Could not find id %d in request queue for subscription", __FILE__, __LINE__,  id);
 252         }
 253
 254       }
 255
 256     }
 257
 258     else if( procedureCode == E2N_ProcedureCode_id_ricSubscriptionDelete){
 259
 260       res = sub_del_resp.get_fields(e2ap_recv->choice.successfulOutcome, he_response);
 261       {
 262         std::lock_guard<std::mutex> lock(*(_data_lock.get()));
 263         // get the id
 264         id = he_response.get_request_id();
 265         int req_status = get_request_status(id);
 266         if (req_status == delete_request_pending ){
 267           // Remove the subscription from the table
 268           res = delete_subscription_entry(id);
 269           if(res){
 270             set_request_status(id, delete_request_success);
 271             valid_response = true;
 272           }
 273           else{
 274             set_request_status(id, delete_request_failed);
 275             std::string error_string = "Error deleting subscription entry for id = ";
 276             mdclog_write(MDCLOG_ERR,  "%s, %d: %s, %d", __FILE__, __LINE__, error_string.c_str(), id);
 277             valid_response = true;
 278           }
 279         }
 280         else if (req_status > 0){
 281           // we don't change status since it was not in pending
 282           // we simply fail
 283           mdclog_write(MDCLOG_ERR, "Error:: %s, %d: Request %d for deletion  is not in delete_pending state, is in State = %d\n", __FILE__, __LINE__, id, req_status);
 284         }
 285         else{
 286           mdclog_write(MDCLOG_ERR,  "%s, %d: Could not find id  %d in request queue for deletion ", __FILE__, __LINE__,  id);
 287         }
 288
 289       }
 290     }
 291
 292     else{
 293       mdclog_write(MDCLOG_ERR,  "%s, %d: Subscription Handler Response received E2AP PDU success  response with an non-subscription response related type  %d", __FILE__, __LINE__, procedureCode);
 294     }
 295
 296   }
 297
 298   else if(type == E2N_E2AP_PDU_PR_unsuccessfulOutcome){
 299
 300     procedureCode = e2ap_recv->choice.unsuccessfulOutcome->procedureCode;
 301     mdclog_write(MDCLOG_INFO, "Received E2AP PDU  unsuccessful outcome message with procedureCode = %d", procedureCode);
 302
 303     if(procedureCode == E2N_ProcedureCode_id_ricSubscription){
 304
 305       sub_resp.get_fields(e2ap_recv->choice.unsuccessfulOutcome, he_response);
 306       {
 307         std::lock_guard<std::mutex> lock(*(_data_lock.get()));
 308         id = he_response.get_request_id();
 309         int req_status = get_request_status(id);
 310         if(req_status == request_pending){
 311           set_request_status(id, request_failed);
 312           valid_response = true;
 313           mdclog_write(MDCLOG_ERR, "Subscription request %d failed", id);
 314         }
 315         else if (req_status > 0){
 316           // we don't changet status since it was not in pending
 317           // we simply fail
 318           mdclog_write(MDCLOG_ERR, "Error:: %s, %d: Request %d is not in request_pending state, is in State = %d\n", __FILE__, __LINE__, id, req_status);
 319         }
 320         else{
 321           mdclog_write(MDCLOG_ERR,  "%s, %d: Could not find id   %d in request queue for subscription ", __FILE__, __LINE__, id);
 322         }
 323       }
 324     }
 325
 326     else if(procedureCode == E2N_ProcedureCode_id_ricSubscriptionDelete){
 327
 328       res = sub_del_resp.get_fields(e2ap_recv->choice.unsuccessfulOutcome, he_response);
 329       {
 330         std::lock_guard<std::mutex> lock(*(_data_lock.get()));
 331         // get the id
 332         id = he_response.get_request_id();
 333         int req_status = get_request_status(id);
 334         if(req_status == delete_request_pending){
 335           set_request_status(id, delete_request_failed);
 336           mdclog_write(MDCLOG_INFO, "Subscription delete request %d failed", id);
 337           valid_response = true;
 338         }
 339         else if (req_status > 0){
 340           mdclog_write(MDCLOG_ERR, "Error:: %s, %d: Request %d for deletion  is not in delete_pending state, is in State = %d\n", __FILE__, __LINE__, id, req_status);
 341         }
 342         else{
 343           mdclog_write(MDCLOG_ERR,  "%s, %d: Could not find id  %d  in request queue for deletion ", __FILE__, __LINE__, id);
 344         }
 345
 346       }
 347     }
 348     else{
 349       mdclog_write(MDCLOG_ERR,  "%s, %d: Susbcription Handler Response received E2AP PDU failure response with a non-subscription response related type  %d", __FILE__, __LINE__,  procedureCode);
 350
 351     }
 352   }
 353   else{
 354     mdclog_write(MDCLOG_ERR,  "%s, %d: Susbcription Handler Response received E2AP PDU with non response type  %d", __FILE__, __LINE__, type);
 355   }
 356
 357
 358   ASN_STRUCT_FREE(asn_DEF_E2N_E2AP_PDU, e2ap_recv);
 359
 360   // wake up all waiting users ...
 361   if(valid_response){
 362     _cv.get()->notify_all();
 363   }
 364
 365 }
 366
 367
 368 int const subscription_handler::get_request_status(int id){
 369   auto search = requests_table.find(id);
 370   if (search == requests_table.end()){
 371     return -1;
 372   }
 373
 374   return search->second;
 375 }
 376
 377  bool subscription_handler::is_subscription_entry(int id){
 378   auto search = subscription_responses.find(id);
 379   if (search != subscription_responses.end())
 380     return true;
 381   else
 382     return false;
 383 }
 384
 385 bool subscription_handler::is_request_entry(int id){
 386   auto search = requests_table.find(id);
 387   if (search != requests_table.end())
 388     return true;
 389   else
 390     return false;
 391 }