RIC-E2-TERMINATION/sctpThread.cpp

   1 // Copyright 2019 AT&T Intellectual Property
   2 // Copyright 2019 Nokia
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //      http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15
  16 //  This source code is part of the near-RT RIC (RAN Intelligent Controller)
  17 //  platform project (RICP).
  18
  19 // TODO: High-level file comment.
  20
  21
  22
  23 #include <3rdparty/oranE2/RANfunctions-List.h>
  24 #include "sctpThread.h"
  25 #include "BuildRunName.h"
  26 #include <unistd.h>
  27 //#include "3rdparty/oranE2SM/E2SM-gNB-NRT-RANfunction-Definition.h"
  28 //#include "BuildXml.h"
  29 //#include "pugixml/src/pugixml.hpp"
  30 #include <pthread.h>
  31 #include <sys/time.h>
  32 #include <sys/inotify.h>
  33 #include <errno.h>
  34 #include <sys/stat.h>
  35 #include <arpa/inet.h>
  36
  37 using namespace std;
  38 //using namespace std::placeholders;
  39 using namespace boost::filesystem;
  40 using namespace prometheus;
  41
  42
  43 //#ifdef __cplusplus
  44 //extern "C"
  45 //{
  46 //#endif
  47
  48 // need to expose without the include of gcov
  49 extern "C" void __gcov_flush(void);
  50 #define LOG_FILE_CONFIG_MAP "CONFIG_MAP_NAME"
  51 #define E2AP_PPID 70 // as per E2GAP chapter 6.1
  52
  53 static void catch_function(int signal) {
  54     __gcov_flush();
  55     exit(signal);
  56 }
  57
  58
  59 BOOST_LOG_INLINE_GLOBAL_LOGGER_DEFAULT(my_logger, src::logger_mt)
  60
  61 boost::shared_ptr<sinks::synchronous_sink<sinks::text_file_backend>> boostLogger;
  62 // double cpuClock = 0.0;
  63 bool jsonTrace = false;
  64 std::map<std::string, E2NodeConnectionHandling> connectionHandlingPerE2NodeMap;
  65
  66 char* getinterfaceip()
  67 {
  68    char hostname[256];
  69    char *IP;
  70    struct hostent *host_entry;
  71    int retVal;
  72    retVal = gethostname(hostname, sizeof(hostname));
  73    if ( retVal == -1 )
  74        return NULL;
  75    host_entry = gethostbyname(hostname);
  76    if ( host_entry == NULL )
  77        return NULL;
  78    IP = inet_ntoa(*((struct in_addr*) host_entry->h_addr_list[0]));
  79    return IP;
  80 }
  81
  82
  83 static int enable_log_change_notify(const char* fileName)
  84 {
  85     int ret = -1;
  86     struct stat fileInfo;
  87     if ( lstat(fileName,&fileInfo) == 0 )
  88     {
  89         ret = register_log_change_notify(fileName);
  90     }
  91     return ret;
  92 }
  93
  94
  95 static int register_log_change_notify(const char *fileName)
  96 {
  97     pthread_attr_t cb_attr;
  98     pthread_t tid;
  99     pthread_attr_init(&cb_attr);
 100     pthread_attr_setdetachstate(&cb_attr,PTHREAD_CREATE_DETACHED);
 101     return pthread_create(&tid, &cb_attr,&monitor_loglevel_change_handler,(void *)fileName);
 102 }
 103
 104
 105 static void * monitor_loglevel_change_handler(void* arg)
 106 {
 107     char *fileName = (char*) arg;
 108     int ifd;                   // the inotify file des
 109     int wfd;                   // the watched file des
 110     ssize_t n = 0;
 111     char rbuf[4096];           // large read buffer as the event is var len
 112     fd_set fds;
 113     int res = 0;
 114     char* dname=NULL;          // directory name
 115     char* bname = NULL;        // basename
 116     char* tok=NULL;
 117     char* log_level=NULL;
 118
 119     dname = strdup( fileName ); // defrock the file name into dir and basename
 120     if( (tok = strrchr( dname, '/' )) != NULL ) {
 121         *tok = '\0';
 122         bname = strdup( tok+1 );
 123     }
 124
 125     ifd = inotify_init1( 0 ); // initialise watcher setting blocking read (no option)
 126     if( ifd < 0 ) {
 127         fprintf( stderr, "### ERR ### unable to initialise file watch %s\n", strerror( errno ) );
 128     } else {
 129         wfd = inotify_add_watch( ifd, dname, IN_MOVED_TO | IN_CLOSE_WRITE ); // we only care about close write changes
 130
 131         if( wfd < 0 ) {
 132             fprintf( stderr, "### ERR ### unable to add watch on config file %s: %s\n", fileName, strerror( errno ) );
 133         } else {
 134             FD_ZERO (&fds);
 135             FD_SET (ifd, &fds);
 136             while( 1 ) {
 137                 res = select (ifd + 1, &fds, NULL, NULL, NULL);
 138                 if(res)
 139                 {
 140                     n = read( ifd, rbuf, sizeof( rbuf ) ); // read the event
 141                     if( n < 0  ) {
 142 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 143                         if( errno == EAGAIN ) {
 144                         } else {
 145                             fprintf( stderr, "### CRIT ### config listener read err: %s\n", strerror( errno ) );
 146                         }
 147                         continue;
 148 #endif
 149                     }
 150
 151                     //Retrieving Log Level from configmap by parsing configmap file
 152                     log_level = parse_file(fileName);
 153                     update_mdc_log_level_severity(log_level); //setting log level
 154                     if(log_level != NULL) {
 155                         mdclog_write(MDCLOG_INFO, "MDC log level updated to %s", log_level);
 156                         free(log_level);
 157                     }
 158                 }
 159             }
 160             inotify_rm_watch(ifd,wfd);
 161         }
 162         close(ifd);
 163     }
 164     free(bname);
 165     free(dname);
 166     free(fileName);
 167
 168     pthread_exit(NULL);
 169 }
 170
 171 void  update_mdc_log_level_severity(char* log_level)
 172 {
 173     mdclog_severity_t level = MDCLOG_ERR;
 174
 175     if(log_level == NULL)
 176     {
 177         printf("### ERR ### Invalid Log-Level Configuration in ConfigMap, Default Log-Level Applied:   %d\n",level);
 178     }
 179     else if(strcasecmp(log_level,"1")==0)
 180     {
 181         level = MDCLOG_ERR;
 182     }
 183     else if(strcasecmp(log_level,"2")==0)
 184     {
 185         level = MDCLOG_WARN;
 186     }
 187     else if(strcasecmp(log_level,"3")==0)
 188     {
 189         level = MDCLOG_INFO;
 190     }
 191     else if(strcasecmp(log_level,"4")==0)
 192     {
 193         level = MDCLOG_DEBUG;
 194     }
 195
 196     mdclog_level_set(level);
 197 }
 198
 199 /**
 200  * @brief Remove leading and trailing spaces from s.
 201  *
 202  * If the string was allocated dynamically, the caller cannot
 203  * overwrite the returned pointer.
 204  *
 205  * @param s the string we want to remove spaces.
 206  * @return Returns a null-terminated substring of "s".
 207  */
 208 static inline char *trim(char *s)
 209 {
 210     char *end;
 211     /* skip leading spaces */
 212     while (isspace(*s)) s++;
 213
 214     /* all spaces */
 215     if (*s == '\0') return s;
 216
 217     /* skip trailing spaces */
 218     end = s + strlen(s) - 1;
 219     while (end > s && isspace(*end)) end--;
 220
 221     /* write null character */
 222     *(end+1) = '\0';
 223
 224     return s;
 225 }
 226
 227 static char* parse_file(char* filename)
 228 {
 229     char *token=NULL;
 230     bool found = false;
 231     FILE *file = fopen ( filename, "r" );
 232
 233     if ( file != NULL )
 234     {
 235         char line [ 128 ];
 236         while ( fgets ( line, sizeof(line), file ) != NULL )
 237         {
 238             token = strtok(line, ":");
 239             token = trim(token);
 240             if (strcmp(token,"log-level") == 0) {
 241                 found = true;
 242                 token = strtok(NULL, "\n");
 243                 token = trim(token);
 244                 break;
 245             }
 246         }
 247         fclose ( file );
 248      }
 249      if(found)
 250          return(strdup(token));
 251      else
 252          return(NULL);
 253 }
 254
 255 char *read_env_param(const char*envkey)
 256 {
 257     if(envkey)
 258     {
 259         char *value = getenv(envkey);
 260         if(value)
 261             return strdup(value);
 262     }
 263     return NULL;
 264 }
 265
 266 void dynamic_log_level_change()
 267 {
 268     char *logFile_Name = read_env_param(LOG_FILE_CONFIG_MAP);
 269     char* log_level_init=NULL;
 270     if(logFile_Name)
 271     {
 272         log_level_init = parse_file(logFile_Name);
 273         update_mdc_log_level_severity(log_level_init); //setting log level
 274         mdclog_write(MDCLOG_INFO, "MDC log level set to %s", log_level_init);
 275         free(log_level_init);
 276
 277     }
 278     enable_log_change_notify(logFile_Name);
 279 }
 280
 281 void init_log() {
 282     int log_change_monitor = 0;
 283     mdclog_attr_t *attr;
 284     mdclog_attr_init(&attr);
 285     mdclog_attr_set_ident(attr, "E2Terminator");
 286     mdclog_init(attr);
 287     if(mdclog_format_initialize(log_change_monitor)!=0)
 288         mdclog_write(MDCLOG_ERR, "Failed to intialize MDC log format !!!");
 289     dynamic_log_level_change();
 290     mdclog_attr_destroy(attr);
 291 }
 292 //auto start_time = std::chrono::high_resolution_clock::now();
 293 //typedef std::chrono::duration<double, std::ratio<1,1>> seconds_t;
 294
 295 //double age() {
 296 //    return seconds_t(std::chrono::high_resolution_clock::now() - start_time).count();
 297 //}
 298
 299 // If model name in "/proc/cpuinfo" is happens to be "Common KVM processor" then
 300 // approx_CPU_MHz() results in "SIGILL - Illegal Instruction" signal.
 301 //
 302 // double approx_CPU_MHz(unsigned sleepTime) {
 303 //    using namespace std::chrono_literals;
 304 //    uint32_t aux = 0;
 305 //    uint64_t cycles_start = rdtscp(aux);
 306 //    double time_start = age();
 307 //    std::this_thread::sleep_for(sleepTime * 1ms);
 308 //    uint64_t elapsed_cycles = rdtscp(aux) - cycles_start;
 309 //    double elapsed_time = age() - time_start;
 310 //    return elapsed_cycles / elapsed_time;
 311 //}
 312
 313 //std::atomic<int64_t> rmrCounter{0};
 314 std::atomic<int64_t> num_of_messages{0};
 315 std::atomic<int64_t> num_of_XAPP_messages{0};
 316 static long transactionCounter = 0;
 317 pthread_mutex_t thread_lock;
 318
 319 int buildListeningPort(sctp_params_t &sctpParams) {
 320     sctpParams.listenFD = socket(AF_INET6, SOCK_STREAM, IPPROTO_SCTP);
 321     if (sctpParams.listenFD <= 0) {
 322 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 323         mdclog_write(MDCLOG_ERR, "Error Opening socket, %s", strerror(errno));
 324         return -1;
 325 #endif
 326     }
 327     struct sctp_initmsg initmsg;
 328     memset (&initmsg, 0, sizeof (initmsg));
 329     initmsg.sinit_num_ostreams = 2;
 330     initmsg.sinit_max_instreams = 2;
 331     initmsg.sinit_max_attempts = 4;
 332     setsockopt (sctpParams.listenFD, IPPROTO_SCTP, SCTP_INITMSG, &initmsg, sizeof (initmsg));
 333
 334     struct sockaddr_in6 serverAddress {};
 335     serverAddress.sin6_family = AF_INET6;
 336     serverAddress.sin6_addr   = in6addr_any;
 337     serverAddress.sin6_port = htons(sctpParams.sctpPort);
 338     if (bind(sctpParams.listenFD, (SA *)&serverAddress, sizeof(serverAddress)) < 0 ) {
 339 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 340         mdclog_write(MDCLOG_ERR, "Error binding port %d. %s", sctpParams.sctpPort, strerror(errno));
 341         return -1;
 342 #endif
 343     }
 344     if (setSocketNoBlocking(sctpParams.listenFD) == -1) {
 345         //mdclog_write(MDCLOG_ERR, "Error binding. %s", strerror(errno));
 346         return -1;
 347     }
 348     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 349         struct sockaddr_in6 clientAddress {};
 350         socklen_t len = sizeof(clientAddress);
 351         getsockname(sctpParams.listenFD, (SA *)&clientAddress, &len);
 352         char buff[1024] {};
 353         inet_ntop(AF_INET6, &clientAddress.sin6_addr, buff, sizeof(buff));
 354         mdclog_write(MDCLOG_DEBUG, "My address: %s, port %d\n", buff, htons(clientAddress.sin6_port));
 355     }
 356
 357     // SCTP_HB_INTERVAL is in milisec
 358     char *sctp_hb_interval = std::getenv("SCTP_HB_INTERVAL");
 359     if (sctp_hb_interval) {
 360         // Setting the heartbeat interval timeout value
 361         struct sctp_paddrparams sckt_parms;
 362         memset(&sckt_parms, 0, sizeof(sckt_parms));
 363         unsigned int sckt_parms_size = sizeof(sckt_parms);
 364         sckt_parms.spp_address.ss_family = AF_INET6;
 365         sckt_parms.spp_flags |= SPP_HB_ENABLE;
 366         sctp_opt_info(sctpParams.listenFD, 0, SCTP_PEER_ADDR_PARAMS, &sckt_parms, &sckt_parms_size);
 367         if (sckt_parms_size != sizeof(sckt_parms)) {
 368             fprintf(stderr, "Invalid size of sctp_paddrparams socket option: {} / {}", sckt_parms_size, (socklen_t)sizeof(sckt_parms));
 369         } else {
 370             sckt_parms.spp_hbinterval = atoi(sctp_hb_interval);
 371             setsockopt(sctpParams.listenFD, IPPROTO_SCTP, SCTP_PEER_ADDR_PARAMS, &sckt_parms, sizeof(sckt_parms));
 372         }
 373     }
 374
 375     char *sctp_max_retries = std::getenv("SCTP_MAX_RETRIES");
 376     if (sctp_max_retries) {
 377         // Setting the max retries config for the socket if rechability loss
 378         struct sctp_assocparams sckt_assoc;
 379         memset(&sckt_assoc, 0, sizeof(sckt_assoc));
 380         unsigned int str_assoc_size = sizeof(sckt_assoc);
 381         sctp_opt_info(sctpParams.listenFD, 0, SCTP_ASSOCINFO, &sckt_assoc, &str_assoc_size);
 382         if (str_assoc_size != sizeof(sckt_assoc)) {
 383             fprintf(stderr, "Invalid size of sctp_assocparams socket option: {} / {}", str_assoc_size, (socklen_t)sizeof(sckt_assoc));
 384         } else {
 385             sckt_assoc.sasoc_asocmaxrxt = atoi(sctp_max_retries);
 386             setsockopt(sctpParams.listenFD, IPPROTO_SCTP, SCTP_ASSOCINFO, &sckt_assoc, sizeof(sckt_assoc));
 387         }
 388     }
 389
 390
 391     if (listen(sctpParams.listenFD, SOMAXCONN) < 0) {
 392 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 393         mdclog_write(MDCLOG_ERR, "Error listening. %s\n", strerror(errno));
 394         return -1;
 395 #endif
 396     }
 397     struct epoll_event event {};
 398     event.events = EPOLLIN | EPOLLET;
 399     event.data.fd = sctpParams.listenFD;
 400
 401     // add listening port to epoll
 402     if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.listenFD, &event)) {
 403 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 404         printf("Failed to add descriptor to epoll\n");
 405         mdclog_write(MDCLOG_ERR, "Failed to add descriptor to epoll. %s\n", strerror(errno));
 406         return -1;
 407 #endif
 408     }
 409
 410     return 0;
 411 }
 412
 413 int buildConfiguration(sctp_params_t &sctpParams) {
 414     path p = (sctpParams.configFilePath + "/" + sctpParams.configFileName).c_str();
 415     if (exists(p)) {
 416         const int size = 2048;
 417         auto fileSize = file_size(p);
 418         if (fileSize > size) {
 419 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 420             mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
 421             return -1;
 422 #endif
 423         }
 424     } else {
 425 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 426         mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
 427         return -1;
 428 #endif
 429     }
 430
 431     ReadConfigFile conf;
 432     if (conf.openConfigFile(p.string()) == -1) {
 433 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 434         mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
 435                      p.string().c_str(), strerror(errno));
 436         return -1;
 437 #endif
 438     }
 439     int rmrPort = conf.getIntValue("nano");
 440     if (rmrPort == -1) {
 441 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 442         mdclog_write(MDCLOG_ERR, "illegal RMR port ");
 443         return -1;
 444 #endif
 445     }
 446     sctpParams.rmrPort = (uint16_t)rmrPort;
 447     snprintf(sctpParams.rmrAddress, sizeof(sctpParams.rmrAddress), "%d", (int) (sctpParams.rmrPort));
 448     auto tmpStr = conf.getStringValue("volume");
 449     if (tmpStr.length() == 0) {
 450 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 451         mdclog_write(MDCLOG_ERR, "illegal volume.");
 452         return -1;
 453 #endif
 454     }
 455
 456     char tmpLogFilespec[VOLUME_URL_SIZE];
 457     tmpLogFilespec[0] = 0;
 458     sctpParams.volume[0] = 0;
 459     snprintf(sctpParams.volume, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
 460     // copy the name to temp file as well
 461     snprintf(tmpLogFilespec, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
 462
 463
 464     // define the file name in the tmp directory under the volume
 465     strcat(tmpLogFilespec,"/tmp/E2Term_%Y-%m-%d_%H-%M-%S.%N.tmpStr");
 466
 467     sctpParams.myIP = conf.getStringValue("local-ip");
 468     if (sctpParams.myIP.length() == 0) {
 469 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 470         mdclog_write(MDCLOG_ERR, "illegal local-ip.");
 471         return -1;
 472 #endif
 473     }
 474
 475     int sctpPort = conf.getIntValue("sctp-port");
 476     if (sctpPort == -1) {
 477 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 478         mdclog_write(MDCLOG_ERR, "illegal SCTP port ");
 479         return -1;
 480 #endif
 481     }
 482     sctpParams.sctpPort = (uint16_t)sctpPort;
 483
 484     sctpParams.fqdn = conf.getStringValue("external-fqdn");
 485     if (sctpParams.fqdn.length() == 0) {
 486 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 487         mdclog_write(MDCLOG_ERR, "illegal external-fqdn");
 488         return -1;
 489 #endif
 490     }
 491
 492     std::string pod = conf.getStringValue("pod_name");
 493 #ifndef UNIT_TEST
 494     if (pod.length() == 0) {
 495         mdclog_write(MDCLOG_ERR, "illegal pod_name in config file");
 496         return -1;
 497     }
 498     auto *podName = getenv(pod.c_str());
 499     if (podName == nullptr) {
 500         mdclog_write(MDCLOG_ERR, "illegal pod_name or environment variable not exists : %s", pod.c_str());
 501         return -1;
 502
 503     } else {
 504         sctpParams.podName.assign(podName);
 505         if (sctpParams.podName.length() == 0) {
 506             mdclog_write(MDCLOG_ERR, "illegal pod_name");
 507             return -1;
 508         }
 509     }
 510 #endif
 511     tmpStr = conf.getStringValue("trace");
 512     transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
 513     if ((tmpStr.compare("start")) == 0) {
 514 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 515         mdclog_write(MDCLOG_INFO, "Trace set to: start");
 516         sctpParams.trace = true;
 517 #endif
 518     } else if ((tmpStr.compare("stop")) == 0) {
 519         mdclog_write(MDCLOG_INFO, "Trace set to: stop");
 520         sctpParams.trace = false;
 521     } else {
 522 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 523         mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
 524         sctpParams.trace = false;
 525 #endif
 526     }
 527     jsonTrace = sctpParams.trace;
 528
 529     sctpParams.epollTimeOut = -1;
 530
 531     tmpStr = conf.getStringValue("prometheusPort");
 532     if (tmpStr.length() != 0) {
 533         sctpParams.prometheusPort = tmpStr;
 534     }
 535
 536     sctpParams.ka_message_length = snprintf(sctpParams.ka_message, KA_MESSAGE_SIZE, "{\"address\": \"%s:%d\","
 537                                                                                     "\"fqdn\": \"%s\","
 538                                                                                     "\"pod_name\": \"%s\"}",
 539                                             (const char *)sctpParams.myIP.c_str(),
 540                                             sctpParams.rmrPort,
 541                                             sctpParams.fqdn.c_str(),
 542                                             sctpParams.podName.c_str());
 543
 544     if (mdclog_level_get() >= MDCLOG_INFO) {
 545         mdclog_write(MDCLOG_DEBUG,"RMR Port: %s", to_string(sctpParams.rmrPort).c_str());
 546         mdclog_write(MDCLOG_DEBUG,"volume: %s", sctpParams.volume);
 547         mdclog_write(MDCLOG_DEBUG,"tmpLogFilespec: %s", tmpLogFilespec);
 548         mdclog_write(MDCLOG_DEBUG,"my ip: %s", sctpParams.myIP.c_str());
 549         mdclog_write(MDCLOG_DEBUG,"pod name: %s", sctpParams.podName.c_str());
 550
 551         mdclog_write(MDCLOG_INFO, "running parameters for instance : %s", sctpParams.ka_message);
 552     }
 553
 554     // Files written to the current working directory
 555     boostLogger = logging::add_file_log(
 556             keywords::file_name = tmpLogFilespec, // to temp directory
 557             keywords::rotation_size = 10 * 1024 * 1024,
 558             keywords::time_based_rotation = sinks::file::rotation_at_time_interval(posix_time::hours(1)),
 559             keywords::format = "%Message%"
 560             //keywords::format = "[%TimeStamp%]: %Message%" // use each tmpStr with time stamp
 561     );
 562
 563     // Setup a destination folder for collecting rotated (closed) files --since the same volume can use rename()
 564     boostLogger->locked_backend()->set_file_collector(sinks::file::make_collector(
 565             keywords::target = sctpParams.volume
 566     ));
 567
 568     // Upon restart, scan the directory for files matching the file_name pattern
 569     boostLogger->locked_backend()->scan_for_files();
 570
 571     // Enable auto-flushing after each tmpStr record written
 572     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 573         boostLogger->locked_backend()->auto_flush(true);
 574     }
 575
 576     return 0;
 577 }
 578
 579 void startPrometheus(sctp_params_t &sctpParams) {
 580     auto podName = std::getenv("POD_NAME");
 581     string metric = "E2TBeta";
 582     if (strstr(podName, "alpha") != NULL) {
 583         metric = "E2TAlpha";
 584     }
 585     //Get eth0 interface IP
 586     char* host = getinterfaceip();
 587     string hostip = host;
 588
 589     sctpParams.prometheusFamily = &BuildCounter()
 590             .Name(metric.c_str())
 591             .Help("E2T instance metrics")
 592             .Labels({{"POD_NAME", sctpParams.podName}})
 593             .Register(*sctpParams.prometheusRegistry);
 594
 595     // Build E2T instance level metrics
 596     buildE2TPrometheusCounters(sctpParams);
 597
 598     string prometheusPath;
 599     if (hostip.empty())
 600         prometheusPath = sctpParams.prometheusPort + "," + "[::]:" + sctpParams.prometheusPort;
 601     else
 602         prometheusPath = hostip + ":" + sctpParams.prometheusPort;
 603
 604     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 605         mdclog_write(MDCLOG_DEBUG, "Start Prometheus Pull mode on %s", prometheusPath.c_str());
 606     }
 607     sctpParams.prometheusExposer = new Exposer(prometheusPath, 1);
 608     sctpParams.prometheusExposer->RegisterCollectable(sctpParams.prometheusRegistry);
 609 }
 610 #ifndef UNIT_TEST
 611
 612 int main(const int argc, char **argv) {
 613     sctp_params_t sctpParams;
 614     {
 615         std::random_device device{};
 616         std::mt19937 generator(device());
 617         std::uniform_int_distribution<long> distribution(1, (long) 1e12);
 618         transactionCounter = distribution(generator);
 619     }
 620
 621 //    uint64_t st = 0;
 622 //    uint32_t aux1 = 0;
 623 //   st = rdtscp(aux1);
 624
 625     unsigned num_cpus = std::thread::hardware_concurrency();
 626     init_log();
 627     if (std::signal(SIGINT, catch_function) == SIG_ERR) {
 628         mdclog_write(MDCLOG_ERR, "Error initializing SIGINT");
 629         exit(1);
 630     }
 631     if (std::signal(SIGABRT, catch_function)== SIG_ERR) {
 632         mdclog_write(MDCLOG_ERR, "Error initializing SIGABRT");
 633         exit(1);
 634     }
 635     if (std::signal(SIGTERM, catch_function)== SIG_ERR) {
 636         mdclog_write(MDCLOG_ERR, "Error initializing SIGTERM");
 637         exit(1);
 638     }
 639
 640 //    cpuClock = approx_CPU_MHz(100);
 641
 642 //    mdclog_write(MDCLOG_DEBUG, "CPU speed %11.11f", cpuClock);
 643
 644     auto result = parse(argc, argv, sctpParams);
 645
 646     if (buildConfiguration(sctpParams) != numberZero) {
 647         exit(negativeOne);
 648     }
 649
 650     //auto registry = std::make_shared<Registry>();
 651     sctpParams.prometheusRegistry = std::make_shared<Registry>();
 652
 653     //sctpParams.prometheusFamily = new Family<Counter>("E2T", "E2T message counter", {{"E", sctpParams.podName}});
 654
 655     startPrometheus(sctpParams);
 656
 657     // start epoll
 658     sctpParams.epoll_fd = epoll_create1(0);
 659     if (sctpParams.epoll_fd == -1) {
 660         mdclog_write(MDCLOG_ERR, "failed to open epoll descriptor");
 661         exit(-1);
 662     }
 663     getRmrContext(sctpParams);
 664     if (sctpParams.rmrCtx == nullptr) {
 665         close(sctpParams.epoll_fd);
 666         exit(-1);
 667     }
 668
 669     if (buildInotify(sctpParams) == -1) {
 670         close(sctpParams.rmrListenFd);
 671         rmr_close(sctpParams.rmrCtx);
 672         close(sctpParams.epoll_fd);
 673         exit(-1);
 674     }
 675
 676     if (buildListeningPort(sctpParams) != 0) {
 677         close(sctpParams.rmrListenFd);
 678         rmr_close(sctpParams.rmrCtx);
 679         close(sctpParams.epoll_fd);
 680         exit(-1);
 681     }
 682
 683     sctpParams.sctpMap = new mapWrapper();
 684
 685     if (pthread_mutex_init(&thread_lock, NULL) != 0) {
 686         mdclog_write(MDCLOG_ERR, "failed to init thread lock");
 687         exit(-1);
 688     }
 689     std::vector<std::thread> threads(num_cpus);
 690 //    std::vector<std::thread> threads;
 691
 692     num_cpus = numberOne;
 693     for (unsigned int i = numberZero; i < num_cpus; i++) {
 694         threads[i] = std::thread(listener, &sctpParams);
 695
 696         cpu_set_t cpuset;
 697         CPU_ZERO(&cpuset);
 698         CPU_SET(i, &cpuset);
 699         int rc = pthread_setaffinity_np(threads[i].native_handle(), sizeof(cpu_set_t), &cpuset);
 700         if (rc != numberZero) {
 701             mdclog_write(MDCLOG_ERR, "Error calling pthread_setaffinity_np: %d", rc);
 702         }
 703     }
 704
 705
 706     //loop over term_init until first message from xApp
 707     handleTermInit(sctpParams);
 708
 709     for (auto &t : threads) {
 710         t.join();
 711     }
 712     pthread_mutex_destroy(&thread_lock);
 713     return 0;
 714 }
 715 #endif
 716 void handleTermInit(sctp_params_t &sctpParams) {
 717     sendTermInit(sctpParams);
 718     //send to e2 manager init of e2 term
 719     //E2_TERM_INIT
 720
 721     int count = 0;
 722     while (true) {
 723         auto xappMessages = num_of_XAPP_messages.load(std::memory_order_acquire);
 724         if (xappMessages > 0) {
 725             if (mdclog_level_get() >=  MDCLOG_INFO) {
 726                 mdclog_write(MDCLOG_INFO, "Got a message from some application, stop sending E2_TERM_INIT");
 727             }
 728             return;
 729         }
 730         usleep(100000);
 731         count++;
 732         if (count % 1000 == 0) {
 733             mdclog_write(MDCLOG_ERR, "GOT No messages from any xApp");
 734             sendTermInit(sctpParams);
 735         }
 736     }
 737 }
 738
 739 void sendTermInit(sctp_params_t &sctpParams) {
 740     rmr_mbuf_t *msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length);
 741     auto count = 0;
 742     while (true) {
 743         msg->mtype = E2_TERM_INIT;
 744         msg->state = 0;
 745         rmr_bytes2payload(msg, (unsigned char *)sctpParams.ka_message, sctpParams.ka_message_length);
 746         static unsigned char tx[32];
 747         auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
 748         rmr_bytes2xact(msg, tx, txLen);
 749         msg = rmr_send_msg(sctpParams.rmrCtx, msg);
 750         if (msg == nullptr) {
 751             msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length);
 752         } else if (msg->state == 0) {
 753             rmr_free_msg(msg);
 754             if (mdclog_level_get() >=  MDCLOG_INFO) {
 755                 mdclog_write(MDCLOG_INFO, "E2_TERM_INIT successfully sent ");
 756             }
 757             return;
 758         } else {
 759             if (count % 100 == 0) {
 760                 mdclog_write(MDCLOG_ERR, "Error sending E2_TERM_INIT cause : %s ", translateRmrErrorMessages(msg->state).c_str());
 761             }
 762             sleep(1);
 763         }
 764         count++;
 765     }
 766 }
 767
 768 /**
 769  *
 770  * @param argc
 771  * @param argv
 772  * @param sctpParams
 773  * @return
 774  */
 775 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &sctpParams) {
 776     cxxopts::Options options(argv[0], "e2 term help");
 777     options.positional_help("[optional args]").show_positional_help();
 778     options.allow_unrecognised_options().add_options()
 779             ("p,path", "config file path", cxxopts::value<std::string>(sctpParams.configFilePath)->default_value("config"))
 780             ("f,file", "config file name", cxxopts::value<std::string>(sctpParams.configFileName)->default_value("config.conf"))
 781             ("h,help", "Print help");
 782
 783     auto result = options.parse(argc, (const char **&)argv);
 784
 785     if (result.count("help")) {
 786         std::cout << options.help({""}) << std::endl;
 787         exit(0);
 788     }
 789     return result;
 790 }
 791
 792 /**
 793  *
 794  * @param sctpParams
 795  * @return -1 failed 0 success
 796  */
 797 int buildInotify(sctp_params_t &sctpParams) {
 798     sctpParams.inotifyFD = inotify_init1(IN_NONBLOCK);
 799     if (sctpParams.inotifyFD == -1) {
 800         mdclog_write(MDCLOG_ERR, "Failed to init inotify (inotify_init1) %s", strerror(errno));
 801         return -1;
 802     }
 803
 804     sctpParams.inotifyWD = inotify_add_watch(sctpParams.inotifyFD,
 805                                              (const char *)sctpParams.configFilePath.c_str(),
 806                                              (unsigned)IN_OPEN | (unsigned)IN_CLOSE_WRITE | (unsigned)IN_CLOSE_NOWRITE); //IN_CLOSE = (IN_CLOSE_WRITE | IN_CLOSE_NOWRITE)
 807     if (sctpParams.inotifyWD == -1) {
 808         mdclog_write(MDCLOG_ERR, "Failed to add directory : %s to  inotify (inotify_add_watch) %s",
 809                      sctpParams.configFilePath.c_str(),
 810                      strerror(errno));
 811         close(sctpParams.inotifyFD);
 812         return -1;
 813     }
 814
 815     struct epoll_event event{};
 816     event.events = (EPOLLIN);
 817     event.data.fd = sctpParams.inotifyFD;
 818     // add listening RMR FD to epoll
 819     if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.inotifyFD, &event)) {
 820         mdclog_write(MDCLOG_ERR, "Failed to add inotify FD to epoll");
 821         close(sctpParams.inotifyFD);
 822         return -1;
 823     }
 824     return 0;
 825 }
 826
 827 /**
 828  *
 829  * @param args
 830  * @return
 831  */
 832 void listener(sctp_params_t *params) {
 833     int num_of_SCTP_messages = 0;
 834     auto totalTime = 0.0;
 835     std::thread::id this_id = std::this_thread::get_id();
 836     //save cout
 837     auto pod_name = std::getenv("POD_NAME");
 838     auto container_name = std::getenv("CONTAINER_NAME");
 839     auto service_name = std::getenv("SERVICE_NAME");
 840     auto host_name = std::getenv("HOST_NAME");
 841     auto system_name = std::getenv("SYSTEM_NAME");
 842     auto pid = std::to_string(getpid()).c_str();
 843     streambuf *oldCout = cout.rdbuf();
 844     ostringstream memCout;
 845     // create new cout
 846     cout.rdbuf(memCout.rdbuf());
 847     cout << this_id;
 848     //return to the normal cout
 849     cout.rdbuf(oldCout);
 850
 851     char tid[32];
 852     memcpy(tid, memCout.str().c_str(), memCout.str().length() < 32 ? memCout.str().length() : 31);
 853     tid[memCout.str().length()] = 0;
 854     mdclog_mdc_add("SYSTEM_NAME", system_name);
 855     mdclog_mdc_add("HOST_NAME", host_name);
 856     mdclog_mdc_add("SERVICE_NAME", service_name);
 857     mdclog_mdc_add("CONTAINER_NAME", container_name);
 858     mdclog_mdc_add("POD_NAME", pod_name);
 859     mdclog_mdc_add("PID", pid);
 860
 861     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 862         mdclog_write(MDCLOG_DEBUG, "started thread number %s", tid);
 863     }
 864
 865     RmrMessagesBuffer_t rmrMessageBuffer{};
 866     //create and init RMR
 867     rmrMessageBuffer.rmrCtx = params->rmrCtx;
 868
 869     auto *events = (struct epoll_event *) calloc(MAXEVENTS, sizeof(struct epoll_event));
 870     struct timespec end{0, 0};
 871     struct timespec start{0, 0};
 872
 873     rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 874     rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 875
 876     memcpy(rmrMessageBuffer.ka_message, params->ka_message, params->ka_message_length);
 877     rmrMessageBuffer.ka_message_len = params->ka_message_length;
 878     rmrMessageBuffer.ka_message[rmrMessageBuffer.ka_message_len] = 0;
 879
 880     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 881         mdclog_write(MDCLOG_DEBUG, "keep alive message is : %s", rmrMessageBuffer.ka_message);
 882     }
 883
 884     ReportingMessages_t message {};
 885
 886 //    for (int i = 0; i < MAX_RMR_BUFF_ARRAY; i++) {
 887 //        rmrMessageBuffer.rcvBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 888 //        rmrMessageBuffer.sendBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 889 //    }
 890
 891     while (true) {
 892         if (mdclog_level_get() >= MDCLOG_DEBUG) {
 893             mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait. Timeout = %d", params->epollTimeOut);
 894         }
 895 #ifndef UNIT_TEST
 896         auto numOfEvents = epoll_wait(params->epoll_fd, events, MAXEVENTS, params->epollTimeOut);
 897 #else
 898         auto numOfEvents = 1;
 899 #endif
 900         if (numOfEvents == numberZero) { // time out
 901 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 902             if (mdclog_level_get() >= MDCLOG_DEBUG) {
 903                 mdclog_write(MDCLOG_DEBUG, "got epoll timeout");
 904             }
 905             continue;
 906         } else if (numOfEvents < numberZero) {
 907             if (errno == EINTR) {
 908                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
 909                     mdclog_write(MDCLOG_DEBUG, "got EINTR : %s", strerror(errno));
 910                 }
 911                 continue;
 912             }
 913             mdclog_write(MDCLOG_ERR, "Epoll wait failed, errno = %s", strerror(errno));
 914             if(events)
 915             {
 916                 free(events);
 917                 events = nullptr;
 918             }
 919             return;
 920 #endif
 921         }
 922         for (auto i = numberZero; i < numOfEvents; i++) {
 923             if (mdclog_level_get() >= MDCLOG_DEBUG) {
 924                 mdclog_write(MDCLOG_DEBUG, "handling epoll event %d out of %d", i + 1, numOfEvents);
 925             }
 926             clock_gettime(CLOCK_MONOTONIC, &message.message.time);
 927             start.tv_sec = message.message.time.tv_sec;
 928             start.tv_nsec = message.message.time.tv_nsec;
 929
 930
 931             if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP)) {
 932                 handlepoll_error(events[i], message, rmrMessageBuffer, params);
 933             } else if (events[i].events & EPOLLOUT) {
 934                 handleEinprogressMessages(events[i], message, rmrMessageBuffer, params);
 935             } else if (params->listenFD == events[i].data.fd) {
 936                 if (mdclog_level_get() >= MDCLOG_INFO) {
 937                     mdclog_write(MDCLOG_INFO, "New connection request from sctp network\n");
 938                 }
 939                 // new connection is requested from RAN  start build connection
 940                 while (true) {
 941                     struct sockaddr in_addr {};
 942                     socklen_t in_len;
 943                     char hostBuff[NI_MAXHOST];
 944                     char portBuff[NI_MAXSERV];
 945
 946                     in_len = sizeof(in_addr);
 947                     auto *peerInfo = (ConnectedCU_t *)calloc(1, sizeof(ConnectedCU_t));
 948                     if(peerInfo == nullptr){
 949                         mdclog_write(MDCLOG_ERR, "calloc failed");
 950                         break;
 951                     }
 952                     peerInfo->sctpParams = params;
 953                     peerInfo->fileDescriptor = accept(params->listenFD, &in_addr, &in_len);
 954                     if (peerInfo->fileDescriptor == negativeOne) {
 955 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 956                         if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
 957                             /* We have processed all incoming connections. */
 958                             if(peerInfo) {
 959                                 free(peerInfo);
 960                                 peerInfo = nullptr;
 961                             }
 962                             break;
 963                         } else {
 964                             if(peerInfo) {
 965                                 free(peerInfo);
 966                                 peerInfo = nullptr;
 967                             }
 968                             mdclog_write(MDCLOG_ERR, "Accept error, errno = %s", strerror(errno));
 969                             break;
 970                         }
 971                     }
 972                     if (setSocketNoBlocking(peerInfo->fileDescriptor) == -1) {
 973                         mdclog_write(MDCLOG_ERR, "setSocketNoBlocking failed to set new connection %s on port %s\n", hostBuff, portBuff);
 974                         close(peerInfo->fileDescriptor);
 975                         if(peerInfo) {
 976                             free(peerInfo);
 977                             peerInfo = nullptr;
 978                         }
 979                         break;
 980 #endif
 981                     }
 982                     struct sctp_event_subscribe sctpevents;
 983                     memset( (void *)&sctpevents, 0, sizeof(sctpevents) );
 984                     sctpevents.sctp_data_io_event = 1;
 985                     setsockopt(peerInfo->fileDescriptor, SOL_SCTP, SCTP_EVENTS,(const void *)&sctpevents, sizeof(sctpevents) );
 986
 987                     {
 988                         char *value = getenv("SCTP_ASSOC_MAX_RETRANS");
 989                         if (value)
 990                         {
 991                             int int_val = atoi(value);
 992                             mdclog_write(MDCLOG_INFO, "Changing sctp_association_max_retrans to %s, %d\n", value, int_val);
 993                             if (int_val > 0)
 994                             {
 995                                 struct sctp_assocparams sctpassocparams;
 996                                 memset((void *)&sctpassocparams, 0, sizeof(sctpassocparams));
 997                                 sctpassocparams.sasoc_asocmaxrxt = int_val;
 998                                 setsockopt(peerInfo->fileDescriptor, SOL_SCTP, SCTP_ASSOCINFO, (const void *)&sctpassocparams, sizeof(sctpassocparams));
 999                             }
1000                         }
1001                     }
1002
1003                     auto  ans = getnameinfo(&in_addr, in_len,
1004                                             peerInfo->hostName, NI_MAXHOST,
1005                                             peerInfo->portNumber, NI_MAXSERV, (unsigned )((unsigned int)NI_NUMERICHOST | (unsigned int)NI_NUMERICSERV));
1006                     if (ans < 0) {
1007                         mdclog_write(MDCLOG_ERR, "Failed to get info on connection request. %s\n", strerror(errno));
1008                         close(peerInfo->fileDescriptor);
1009                         if(peerInfo) {
1010                             free(peerInfo);
1011                             peerInfo = nullptr;
1012                         }
1013                         break;
1014                     }
1015                     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1016                         mdclog_write(MDCLOG_DEBUG, "Accepted connection on descriptor %d (host=%s, port=%s)\n", peerInfo->fileDescriptor, peerInfo->hostName, peerInfo->portNumber);
1017                     }
1018                     peerInfo->isConnected = false;
1019                     peerInfo->gotSetup = false;
1020                     if (addToEpoll(params->epoll_fd,
1021                                    peerInfo,
1022                                    (EPOLLIN | EPOLLET),
1023                                    params->sctpMap, nullptr,
1024                                    0) != 0) {
1025                         if(peerInfo) {
1026                             free(peerInfo);
1027                             peerInfo = nullptr;
1028                         }
1029                         break;
1030                     }
1031                     break;
1032                 }
1033             } else if (params->rmrListenFd == events[i].data.fd) {
1034                 // got message from XAPP
1035                 //num_of_XAPP_messages.fetch_add(1, std::memory_order_release);
1036                 num_of_messages.fetch_add(1, std::memory_order_release);
1037                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1038                     mdclog_write(MDCLOG_DEBUG, "new RMR message");
1039                 }
1040                 if (receiveXappMessages(params->sctpMap,
1041                                         rmrMessageBuffer,
1042                                         message.message.time) != 0) {
1043                     mdclog_write(MDCLOG_ERR, "Error handling Xapp message");
1044                 }
1045             } else if (params->inotifyFD == events[i].data.fd) {
1046                 mdclog_write(MDCLOG_INFO, "Got event from inotify (configuration update)");
1047                 handleConfigChange(params);
1048             } else {
1049                 /* We RMR_ERR_RETRY have data on the fd waiting to be read. Read and display it.
1050                  * We must read whatever data is available completely, as we are running
1051                  *  in edge-triggered mode and won't get a notification again for the same data. */
1052                 num_of_messages.fetch_add(1, std::memory_order_release);
1053                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1054                     mdclog_write(MDCLOG_DEBUG, "new message from SCTP, epoll flags are : %0x", events[i].events);
1055                 }
1056                 receiveDataFromSctp(&events[i],
1057                                     params->sctpMap,
1058                                     num_of_SCTP_messages,
1059                                     rmrMessageBuffer,
1060                                     message.message.time);
1061             }
1062
1063             clock_gettime(CLOCK_MONOTONIC, &end);
1064             if (mdclog_level_get() >= MDCLOG_INFO) {
1065                 totalTime += ((end.tv_sec + 1.0e-9 * end.tv_nsec) -
1066                               ((double) start.tv_sec + 1.0e-9 * start.tv_nsec));
1067             }
1068             if (mdclog_level_get() >= MDCLOG_DEBUG) {
1069                 mdclog_write(MDCLOG_DEBUG, "message handling is %ld seconds %ld nanoseconds",
1070                              end.tv_sec - start.tv_sec,
1071                              end.tv_nsec - start.tv_nsec);
1072             }
1073         }
1074 #ifdef UNIT_TEST
1075     break;
1076 #endif
1077     }
1078 }
1079
1080 /**
1081  *
1082  * @param sctpParams
1083  */
1084 void handleConfigChange(sctp_params_t *sctpParams) {
1085     char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));
1086     const struct inotify_event *event;
1087     char *ptr;
1088 #ifdef UNIT_TEST
1089     struct inotify_event tmpEvent;
1090 #endif
1091     path p = (sctpParams->configFilePath + "/" + sctpParams->configFileName).c_str();
1092     auto endlessLoop = true;
1093     while (endlessLoop) {
1094 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1095         auto len = read(sctpParams->inotifyFD, buf, sizeof buf);
1096 #else
1097     auto len=10;
1098 #endif
1099         if (len == -1) {
1100 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1101             if (errno != EAGAIN) {
1102                 mdclog_write(MDCLOG_ERR, "read %s ", strerror(errno));
1103                 endlessLoop = false;
1104                 continue;
1105             }
1106             else {
1107                 endlessLoop = false;
1108                 continue;
1109             }
1110 #endif
1111         }
1112
1113         for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
1114 #ifndef UNIT_TEST
1115     event = (const struct inotify_event *)ptr;
1116 #else
1117     tmpEvent.mask = (uint32_t)IN_CLOSE_WRITE;
1118     event = &tmpEvent;
1119 #endif
1120             if (event->mask & (uint32_t)IN_ISDIR) {
1121                 continue;
1122             }
1123
1124             // the directory name
1125             if (sctpParams->inotifyWD == event->wd) {
1126                 // not the directory
1127             }
1128             if (event->len) {
1129 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1130                 auto  retVal = strcmp(sctpParams->configFileName.c_str(), event->name);
1131                 if (retVal != 0) {
1132                     continue;
1133                 }
1134 #endif
1135             }
1136             // only the file we want
1137             if (event->mask & (uint32_t)IN_CLOSE_WRITE) {
1138                 if (mdclog_level_get() >= MDCLOG_INFO) {
1139                     mdclog_write(MDCLOG_INFO, "Configuration file changed");
1140                 }
1141                 if (exists(p)) {
1142                     const int size = 2048;
1143                     auto fileSize = file_size(p);
1144                     if (fileSize > size) {
1145                         mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
1146                         return;
1147                     }
1148                 } else {
1149                     mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
1150                     return;
1151                 }
1152
1153                 ReadConfigFile conf;
1154                 if (conf.openConfigFile(p.string()) == -1) {
1155                     mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
1156                                  p.string().c_str(), strerror(errno));
1157                     return;
1158                 }
1159                 auto tmpStr = conf.getStringValue("trace");
1160                 if (tmpStr.length() == 0) {
1161                     mdclog_write(MDCLOG_ERR, "illegal trace. Set trace to stop");
1162                     tmpStr = "stop";
1163                 }
1164
1165                 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
1166                 if ((tmpStr.compare("start")) == 0) {
1167                     mdclog_write(MDCLOG_INFO, "Trace set to: start");
1168                     sctpParams->trace = true;
1169                 } else if ((tmpStr.compare("stop")) == 0) {
1170                     mdclog_write(MDCLOG_INFO, "Trace set to: stop");
1171                     sctpParams->trace = false;
1172                 } else {
1173                     mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
1174                     sctpParams->trace = false;
1175                 }
1176                 jsonTrace = sctpParams->trace;
1177
1178
1179                 endlessLoop = false;
1180             }
1181 #ifdef UNIT_TEST
1182             break;
1183 #endif
1184         }
1185     }
1186 }
1187
1188 /**
1189  *
1190  * @param event
1191  * @param message
1192  * @param rmrMessageBuffer
1193  * @param params
1194  */
1195 void handleEinprogressMessages(struct epoll_event &event,
1196                                ReportingMessages_t &message,
1197                                RmrMessagesBuffer_t &rmrMessageBuffer,
1198                                sctp_params_t *params) {
1199     auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
1200     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1201
1202     mdclog_write(MDCLOG_INFO, "file descriptor %d got EPOLLOUT", peerInfo->fileDescriptor);
1203     auto retVal = 0;
1204     socklen_t retValLen = 0;
1205     auto rc = getsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_ERROR, &retVal, &retValLen);
1206     if (rc != 0 || retVal != 0) {
1207 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1208         if (rc != 0) {
1209             rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1210                                                          "%s|Failed SCTP Connection, after EINPROGRESS the getsockopt%s",
1211                                                          peerInfo->enodbName, strerror(errno));
1212         } else if (retVal != 0) {
1213             rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1214                                                          "%s|Failed SCTP Connection after EINPROGRESS, SO_ERROR",
1215                                                          peerInfo->enodbName);
1216         }
1217
1218         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1219         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1220         mdclog_write(MDCLOG_ERR, "%s", rmrMessageBuffer.sendMessage->payload);
1221         message.message.direction = 'N';
1222         mdclog_write(MDCLOG_DEBUG, "Checking the peerInfo->enodbName - %s", peerInfo->enodbName);
1223         if (peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE])
1224         {
1225             mdclog_write(MDCLOG_DEBUG, "Incrementing SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1226             peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE]->Increment();
1227             mdclog_write(MDCLOG_DEBUG, "Incremented SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1228         }
1229         if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != 0) {
1230             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1231         }
1232 #endif
1233         memset(peerInfo->asnData, 0, peerInfo->asnLength);
1234         peerInfo->asnLength = 0;
1235         peerInfo->mtype = 0;
1236         return;
1237     }
1238 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1239     peerInfo->isConnected = true;
1240
1241     if (modifyToEpoll(params->epoll_fd, peerInfo, (EPOLLIN | EPOLLET), params->sctpMap, peerInfo->enodbName,
1242                       peerInfo->mtype) != 0) {
1243         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_MOD");
1244         return;
1245     }
1246
1247     message.message.asndata = (unsigned char *)peerInfo->asnData;
1248     message.message.asnLength = peerInfo->asnLength;
1249     message.message.messageType = peerInfo->mtype;
1250     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1251     num_of_messages.fetch_add(1, std::memory_order_release);
1252     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1253         mdclog_write(MDCLOG_DEBUG, "send the delayed SETUP/ENDC SETUP to sctp for %s",
1254                      message.message.enodbName);
1255     }
1256     if (sendSctpMsg(peerInfo, message, params->sctpMap) != 0) {
1257         if (mdclog_level_get() >= MDCLOG_DEBUG) {
1258             mdclog_write(MDCLOG_ERR, "Error write to SCTP  %s %d", __func__, __LINE__);
1259         }
1260         return;
1261     }
1262
1263     memset(peerInfo->asnData, 0, peerInfo->asnLength);
1264     peerInfo->asnLength = 0;
1265     peerInfo->mtype = 0;
1266 #endif
1267 }
1268
1269
1270 void handlepoll_error(struct epoll_event &event,
1271                       ReportingMessages_t &message,
1272                       RmrMessagesBuffer_t &rmrMessageBuffer,
1273                       sctp_params_t *params) {
1274     if ((event.data.fd != params->rmrListenFd) && (event.data.ptr != nullptr)) {
1275         auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
1276         mdclog_write(MDCLOG_ERR, "epoll error, events %0x on fd %d, RAN NAME : %s",
1277                      event.events, peerInfo->fileDescriptor, peerInfo->enodbName);
1278 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1279         rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1280                                                      "%s|Failed SCTP Connection",
1281                                                      peerInfo->enodbName);
1282         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1283         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1284
1285         memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1286         message.message.direction = 'N';
1287         mdclog_write(MDCLOG_DEBUG, "Checking the peerInfo->enodbName - %s", peerInfo->enodbName);
1288         if (peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE])
1289         {
1290             mdclog_write(MDCLOG_DEBUG, "Incrementing SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1291             peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE]->Increment();
1292             mdclog_write(MDCLOG_DEBUG, "Incremented SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1293         }
1294         if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != 0) {
1295             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1296         }
1297 #endif
1298         close(peerInfo->fileDescriptor);
1299         mdclog_write(MDCLOG_DEBUG, "Before event.data.ptr %p and data is %d", (void *)event.data.ptr, *((ConnectedCU_t *)event.data.ptr));
1300         mdclog_write(MDCLOG_DEBUG, "Erasing Entry from Map for Key %s", peerInfo->enodbName);
1301         removeE2ConnectionEntryFromMap(peerInfo->enodbName);
1302         //params->sctpMap->erase(peerInfo->enodbName);
1303         cleanHashEntry((ConnectedCU_t *) event.data.ptr, params->sctpMap);
1304         if (event.data.ptr != nullptr)
1305         {
1306             mdclog_write(MDCLOG_DEBUG, "Not assigned event.data.ptr = NULL");
1307             event.data.ptr = nullptr;
1308         }
1309         mdclog_write(MDCLOG_DEBUG, "After event.data.ptr %p", (void *)event.data.ptr);
1310     } else {
1311         mdclog_write(MDCLOG_ERR, "epoll error, events %0x on RMR FD", event.events);
1312     }
1313 }
1314 /**
1315  *
1316  * @param socket
1317  * @return
1318  */
1319 int setSocketNoBlocking(int socket) {
1320     auto flags = fcntl(socket, F_GETFL, 0);
1321
1322     if (flags == -1) {
1323         mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
1324         return -1;
1325     }
1326
1327     flags = (unsigned) flags | (unsigned) O_NONBLOCK;
1328     if (fcntl(socket, F_SETFL, flags) == -1) {
1329         mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
1330         return -1;
1331     }
1332
1333     return 0;
1334 }
1335
1336 /**
1337  *
1338  * @param val
1339  * @param m
1340  */
1341 void cleanHashEntry(ConnectedCU_t *val, Sctp_Map_t *m) {
1342     if(val != nullptr)
1343     {
1344         mdclog_write(MDCLOG_DEBUG, "Inside cleanHashEntry");
1345         char *dummy;
1346         auto port = (uint16_t) strtol(val->portNumber, &dummy, 10);
1347         char searchBuff[2048]{};
1348
1349         snprintf(searchBuff, sizeof searchBuff, "host:%s:%d", val->hostName, port);
1350         if(m->find(searchBuff)) {
1351             m->erase(searchBuff);
1352         }
1353
1354         if(m->find(val->enodbName)) {
1355             mdclog_write(MDCLOG_DEBUG, "remove key enodbName = %s from %s at line %d", val->enodbName, __FUNCTION__, __LINE__);
1356             m->erase(val->enodbName);
1357         }
1358         free(val);
1359         val = nullptr;
1360         mdclog_write(MDCLOG_DEBUG, "After free");
1361     }
1362 }
1363
1364 /**
1365  *
1366  * @param fd file descriptor
1367  * @param data the asn data to send
1368  * @param len  length of the data
1369  * @param enodbName the enodbName as in the map for printing purpose
1370  * @param m map host information
1371  * @param mtype message number
1372  * @return 0 success, a negative number on fail
1373  */
1374 int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_t *m) {
1375     auto loglevel = mdclog_level_get();
1376 #ifndef UNIT_TEST
1377     int fd = peerInfo->fileDescriptor;
1378     int streamId = fetchStreamId(peerInfo,message);
1379 #else
1380     int fd = FILE_DESCRIPTOR;
1381     int streamId = 0;
1382 #endif
1383     if (loglevel >= MDCLOG_DEBUG) {
1384         mdclog_write(MDCLOG_DEBUG, "Send SCTP message for CU %s, %s",
1385                      message.message.enodbName, __FUNCTION__);
1386     }
1387
1388     while (true) {
1389         if (sctp_sendmsg(fd,message.message.asndata, message.message.asnLength,(struct sockaddr *) NULL, 0, htonl(E2AP_PPID), 0,streamId,0,0) < 0) {
1390             if (errno == EINTR) {
1391                 continue;
1392             }
1393             mdclog_write(MDCLOG_ERR, "error writing to CU a message, %s ", strerror(errno));
1394 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1395             if (!peerInfo->isConnected) {
1396                 mdclog_write(MDCLOG_ERR, "connection to CU %s is still in progress.", message.message.enodbName);
1397                 return -1;
1398             }
1399 #endif
1400 #ifndef UNIT_TEST
1401             mdclog_write(MDCLOG_DEBUG, "Erasing Entry from Map for Key %s", peerInfo->enodbName);
1402             removeE2ConnectionEntryFromMap(peerInfo->enodbName);
1403             cleanHashEntry(peerInfo, m);
1404             if (peerInfo != nullptr)
1405             {
1406                 mdclog_write(MDCLOG_DEBUG, "Not assigned peerInfo = NULL");
1407                 peerInfo = nullptr;
1408             }
1409             close(fd);
1410 #endif
1411             char key[MAX_ENODB_NAME_SIZE * 2];
1412             snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName,
1413                      message.message.messageType);
1414             if (loglevel >= MDCLOG_DEBUG) {
1415                 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
1416             }
1417             auto tmp = m->find(key);
1418             if (tmp) {
1419                 free(tmp);
1420                 tmp = nullptr;
1421             }
1422             m->erase(key);
1423 #ifndef UNIT_TEST
1424             return -1;
1425 #endif
1426         }
1427         message.message.direction = 'D';
1428         // send report.buffer of size
1429         buildJsonMessage(message);
1430
1431         if (loglevel >= MDCLOG_DEBUG) {
1432             mdclog_write(MDCLOG_DEBUG,
1433                          "SCTP message for CU %s sent from %s",
1434                          message.message.enodbName,
1435                          __FUNCTION__);
1436         }
1437         return 0;
1438     }
1439 }
1440
1441 /**
1442  *
1443  * @param message
1444  * @param rmrMessageBuffer
1445  */
1446 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
1447     message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
1448     message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
1449
1450     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1451         mdclog_write(MDCLOG_DEBUG, "Message from Xapp RAN name = %s message length = %ld",
1452                      message.message.enodbName, (unsigned long) message.message.asnLength);
1453     }
1454 }
1455
1456 void removeE2ConnectionEntryFromMap(char* enbName)
1457 {
1458     if(connectionHandlingPerE2NodeMap.empty())
1459     {
1460         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
1461         return;
1462     }
1463
1464     std::map<std::string, E2NodeConnectionHandling>::iterator itr = connectionHandlingPerE2NodeMap.begin();
1465     mdclog_write(MDCLOG_DEBUG, "Iterating Map e2NodeIdConnectionStatusMap");
1466     while (itr != connectionHandlingPerE2NodeMap.end())
1467     {
1468         int result = itr->first.compare(enbName);
1469         if (result == 0)
1470         {
1471             mdclog_write(MDCLOG_DEBUG, "Key %s Found in connectionHandlingPerE2NodeMap Map, Current Procedure is %d", enbName, itr->second.e2tProcedureOngoingStatus);
1472             mdclog_write(MDCLOG_DEBUG, "Removing the Entry from map for %s", enbName);
1473             connectionHandlingPerE2NodeMap.erase(itr);
1474             mdclog_write(MDCLOG_DEBUG, "Deleted the Entry Successfully from map for %s", enbName);
1475             break;
1476         }
1477         ++itr;
1478     }
1479 }
1480
1481 /**
1482  *
1483  * @param events
1484  * @param sctpMap
1485  * @param numOfMessages
1486  * @param rmrMessageBuffer
1487  * @param ts
1488  * @return
1489  */
1490 int receiveDataFromSctp(struct epoll_event *events,
1491                         Sctp_Map_t *sctpMap,
1492                         int &numOfMessages,
1493                         RmrMessagesBuffer_t &rmrMessageBuffer,
1494                         struct timespec &ts) {
1495     /* We have data on the fd waiting to be read. Read and display it.
1496  * We must read whatever data is available completely, as we are running
1497  *  in edge-triggered mode and won't get a notification again for the same data. */
1498     ReportingMessages_t message {};
1499     auto done = 0;
1500     auto loglevel = mdclog_level_get();
1501     struct sctp_sndrcvinfo sndrcvinfo;
1502     int flags;
1503     int streamId;
1504
1505     // get the identity of the interface
1506     if (events->data.ptr != nullptr){
1507         message.peerInfo = (ConnectedCU_t *)events->data.ptr;
1508     }
1509
1510     struct timespec start{0, 0};
1511     struct timespec decodeStart{0, 0};
1512     struct timespec end{0, 0};
1513
1514     E2AP_PDU_t *pdu = nullptr;
1515
1516     while (true) {
1517         if (loglevel >= MDCLOG_DEBUG) {
1518             mdclog_write(MDCLOG_DEBUG, "Start Read from SCTP %d fd", message.peerInfo->fileDescriptor);
1519             clock_gettime(CLOCK_MONOTONIC, &start);
1520         }
1521         // read the buffer directly to rmr payload
1522         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1523 #ifndef UNIT_TEST
1524         message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1525                 sctp_recvmsg(message.peerInfo->fileDescriptor, rmrMessageBuffer.sendMessage->payload, RECEIVE_SCTP_BUFFER_SIZE,(struct sockaddr *) NULL, 0, &sndrcvinfo, &flags);
1526         mdclog_write(MDCLOG_DEBUG, "Start Read from SCTP fd %d stream %d ", message.peerInfo->fileDescriptor, sndrcvinfo.sinfo_stream);
1527         streamId = sndrcvinfo.sinfo_stream;
1528 #else
1529         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1530         streamId = 0;
1531 #endif
1532
1533         if (loglevel >= MDCLOG_DEBUG) {
1534             mdclog_write(MDCLOG_DEBUG, "Finish Read from SCTP %d fd message length = %ld",
1535                          message.peerInfo->fileDescriptor, message.message.asnLength);
1536         }
1537
1538         memcpy(message.message.enodbName, message.peerInfo->enodbName, sizeof(message.peerInfo->enodbName));
1539         message.message.direction = 'U';
1540         message.message.time.tv_nsec = ts.tv_nsec;
1541         message.message.time.tv_sec = ts.tv_sec;
1542
1543         if (message.message.asnLength < 0) {
1544             if (errno == EINTR) {
1545                 continue;
1546             }
1547             /* If errno == EAGAIN, that means we have read all
1548                data. So goReportingMessages_t back to the main loop. */
1549             if (errno != EAGAIN) {
1550                 mdclog_write(MDCLOG_ERR, "Read error, %s ", strerror(errno));
1551                 done = 1;
1552             } else if (loglevel >= MDCLOG_DEBUG) {
1553                 mdclog_write(MDCLOG_DEBUG, "EAGAIN - descriptor = %d", message.peerInfo->fileDescriptor);
1554             }
1555             break;
1556         } else if (message.message.asnLength == 0) {
1557             /* End of file. The remote has closed the connection. */
1558             if (loglevel >= MDCLOG_INFO) {
1559                 mdclog_write(MDCLOG_INFO, "END of File Closed connection - descriptor = %d and %s",
1560                              message.peerInfo->fileDescriptor, message.peerInfo->enodbName);
1561             }
1562             done = 1;
1563             break;
1564         }
1565
1566         if (loglevel >= MDCLOG_DEBUG) {
1567             char printBuffer[RECEIVE_SCTP_BUFFER_SIZE]{};
1568             char *tmp = printBuffer;
1569             for (size_t i = 0; i < (size_t)message.message.asnLength; ++i) {
1570                 snprintf(tmp, 3, "%02x", message.message.asndata[i]);
1571                 tmp += 2;
1572             }
1573             printBuffer[message.message.asnLength] = 0;
1574             clock_gettime(CLOCK_MONOTONIC, &end);
1575             mdclog_write(MDCLOG_DEBUG, "Before Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1576                          message.peerInfo->enodbName, end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1577             mdclog_write(MDCLOG_DEBUG, "PDU buffer length = %ld, data =  : %s", message.message.asnLength,
1578                          printBuffer);
1579             clock_gettime(CLOCK_MONOTONIC, &decodeStart);
1580         }
1581 #ifndef UNIT_TEST
1582         auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu,
1583                         message.message.asndata, message.message.asnLength);
1584 #else
1585         asn_dec_rval_t rval = {RC_OK, 0};
1586         pdu = (E2AP_PDU_t*)rmrMessageBuffer.sendMessage->tp_buf;
1587 #endif
1588         if (rval.code != RC_OK) {
1589             mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from RAN : %s", rval.code,
1590                          message.peerInfo->enodbName);
1591             if (pdu != nullptr) {
1592                 ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1593                 pdu = nullptr;
1594             }
1595             break;
1596         }
1597
1598         if (loglevel >= MDCLOG_DEBUG) {
1599             clock_gettime(CLOCK_MONOTONIC, &end);
1600             mdclog_write(MDCLOG_DEBUG, "After Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1601                          message.peerInfo->enodbName, end.tv_sec - decodeStart.tv_sec, end.tv_nsec - decodeStart.tv_nsec);
1602             char *printBuffer;
1603             size_t size;
1604             FILE *stream = open_memstream(&printBuffer, &size);
1605             asn_fprint(stream, &asn_DEF_E2AP_PDU, pdu);
1606             mdclog_write(MDCLOG_DEBUG, "Encoding E2AP PDU past : %s", printBuffer);
1607             clock_gettime(CLOCK_MONOTONIC, &decodeStart);
1608
1609             fclose(stream);
1610             free(printBuffer);
1611         }
1612
1613         switch (pdu->present) {
1614             case E2AP_PDU_PR_initiatingMessage: {//initiating message
1615                 asnInitiatingRequest(pdu, sctpMap,message, rmrMessageBuffer, streamId);
1616                 break;
1617             }
1618             case E2AP_PDU_PR_successfulOutcome: { //successful outcome
1619                 asnSuccessfulMsg(pdu, sctpMap, message, rmrMessageBuffer);
1620                 break;
1621             }
1622             case E2AP_PDU_PR_unsuccessfulOutcome: { //Unsuccessful Outcome
1623                 asnUnSuccsesfulMsg(pdu, sctpMap, message, rmrMessageBuffer);
1624                 break;
1625             }
1626             default:
1627                 mdclog_write(MDCLOG_ERR, "Unknown index %d in E2AP PDU", pdu->present);
1628                 break;
1629         }
1630         if (loglevel >= MDCLOG_DEBUG) {
1631             clock_gettime(CLOCK_MONOTONIC, &end);
1632             mdclog_write(MDCLOG_DEBUG,
1633                          "After processing message and sent to rmr for : %s, Read time is : %ld seconds, %ld nanoseconds",
1634                          message.peerInfo->enodbName, end.tv_sec - decodeStart.tv_sec, end.tv_nsec - decodeStart.tv_nsec);
1635         }
1636         numOfMessages++;
1637 #ifndef UNIT_TEST
1638         if (pdu != nullptr) {
1639             // ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); /* With reset we were not freeing the memory and was causing the leak here. */
1640             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1641             pdu = nullptr;
1642         }
1643 #else
1644     done = 1;
1645     break;
1646 #endif
1647     }
1648
1649     if (done) {
1650         if (loglevel >= MDCLOG_INFO) {
1651             mdclog_write(MDCLOG_INFO, "Closed connection - descriptor = %d", message.peerInfo->fileDescriptor);
1652         }
1653         message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1654                 snprintf((char *)rmrMessageBuffer.sendMessage->payload,
1655                          256,
1656                          "%s|CU disconnected unexpectedly",
1657                          message.peerInfo->enodbName);
1658         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1659         #ifndef UNIT_TEST
1660         if (message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE])
1661         {
1662             mdclog_write(MDCLOG_DEBUG, "Incrementing SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1663             message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE]->Increment();
1664             mdclog_write(MDCLOG_DEBUG, "Incremented SCTP_ABORT_INITIATED_BY_E2NODE Counter");
1665         }
1666         #endif
1667 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1668         if (sendRequestToXapp(message,
1669                               RIC_SCTP_CONNECTION_FAILURE,
1670                               rmrMessageBuffer) != 0) {
1671             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1672         }
1673 #endif
1674
1675         mdclog_write(MDCLOG_DEBUG, "Erasing Entry from Map for Key %s", message.peerInfo->enodbName);
1676         removeE2ConnectionEntryFromMap(message.peerInfo->enodbName);
1677         /* Closing descriptor make epoll remove it from the set of descriptors which are monitored. */
1678 #ifndef UNIT_TEST
1679         pthread_mutex_lock(&thread_lock);
1680         if (fcntl(message.peerInfo->fileDescriptor, F_GETFD) != -1) {
1681             mdclog_write(MDCLOG_DEBUG, "Closing connection - descriptor = %d", message.peerInfo->fileDescriptor);
1682             close(message.peerInfo->fileDescriptor);
1683             cleanHashEntry((ConnectedCU_t *) events->data.ptr, sctpMap);
1684             if (events->data.ptr != nullptr)
1685             {
1686                 mdclog_write(MDCLOG_DEBUG, "Not assigned events->data.ptr = NULL");
1687                 events->data.ptr = nullptr;
1688             }
1689         }
1690         pthread_mutex_unlock(&thread_lock);
1691 #else
1692         close(message.peerInfo->fileDescriptor);
1693         cleanHashEntry((ConnectedCU_t *) events->data.ptr, sctpMap);
1694         if (events->data.ptr != nullptr)
1695         {
1696             mdclog_write(MDCLOG_DEBUG, "[SCTP_ABORT]: Not assigned events->data.ptr = NULL");
1697             events->data.ptr = nullptr;
1698         }
1699 #endif
1700     }
1701     if (loglevel >= MDCLOG_DEBUG) {
1702         clock_gettime(CLOCK_MONOTONIC, &end);
1703         mdclog_write(MDCLOG_DEBUG, "from receive SCTP to send RMR time is %ld seconds and %ld nanoseconds",
1704                      end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1705
1706     }
1707     return 0;
1708 }
1709 #ifndef UNIT_TEST
1710 static void buildAndSendSetupRequest(ReportingMessages_t &message,
1711                                      RmrMessagesBuffer_t &rmrMessageBuffer,
1712                                      E2AP_PDU_t *pdu/*,
1713                                      string const &messageName,
1714                                      string const &ieName,
1715                                      vector<string> &functionsToAdd_v,
1716                                      vector<string> &functionsToModified_v*/) {
1717     auto logLevel = mdclog_level_get();
1718     // now we can send the data to e2Mgr
1719
1720     asn_enc_rval_t er;
1721     auto buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2;
1722     unsigned char *buffer = nullptr;
1723     buffer = (unsigned char *) calloc(buffer_size, sizeof(unsigned char));
1724     if(!buffer)
1725     {
1726 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1727         mdclog_write(MDCLOG_ERR, "Allocating buffer for %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1728         return;
1729 #endif
1730     }
1731     while (true) {
1732         er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, buffer, buffer_size);
1733         if (er.encoded == -1) {
1734 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1735             mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1736             return;
1737 #endif
1738         } else if (er.encoded > (ssize_t) buffer_size) {
1739             buffer_size = er.encoded + 128;
1740 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1741             mdclog_write(MDCLOG_WARN, "Buffer of size %d is to small for %s. Reallocate buffer of size %d",
1742                          (int) buffer_size,
1743                          asn_DEF_E2AP_PDU.name, buffer_size);
1744             buffer_size = er.encoded + 128;
1745
1746             unsigned char *newBuffer = nullptr;
1747             newBuffer = (unsigned char *) realloc(buffer, buffer_size);
1748             if (!newBuffer)
1749             {
1750                 // out of memory
1751                 mdclog_write(MDCLOG_ERR, "Reallocating buffer for %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1752                 free(buffer);
1753                 buffer = nullptr;
1754                 return;
1755             }
1756             buffer = newBuffer;
1757             continue;
1758 #endif
1759         }
1760         buffer[er.encoded] = '\0';
1761         break;
1762     }
1763     // encode to xml
1764
1765     string res((char *)buffer);
1766     res.erase(std::remove(res.begin(), res.end(), '\n'), res.end());
1767     res.erase(std::remove(res.begin(), res.end(), '\t'), res.end());
1768     res.erase(std::remove(res.begin(), res.end(), ' '), res.end());
1769
1770 //    string res {};
1771 //    if (!functionsToAdd_v.empty() || !functionsToModified_v.empty()) {
1772 //        res = buildXmlData(messageName, ieName, functionsToAdd_v, functionsToModified_v, buffer, (size_t) er.encoded);
1773 //    }
1774     rmr_mbuf_t *rmrMsg;
1775 //    if (res.length() == 0) {
1776 //        rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, buffer_size + 256);
1777 //        rmrMsg->len = snprintf((char *) rmrMsg->payload, RECEIVE_SCTP_BUFFER_SIZE * 2, "%s:%d|%s",
1778 //                               message.peerInfo->sctpParams->myIP.c_str(),
1779 //                               message.peerInfo->sctpParams->rmrPort,
1780 //                               buffer);
1781 //    } else {
1782         rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, (int)res.length() + 256);
1783         rmrMsg->len = snprintf((char *) rmrMsg->payload, res.length() + 256, "%s:%d|%s",
1784                                message.peerInfo->sctpParams->myIP.c_str(),
1785                                message.peerInfo->sctpParams->rmrPort,
1786                                res.c_str());
1787 //    }
1788
1789     if (logLevel >= MDCLOG_DEBUG) {
1790         mdclog_write(MDCLOG_DEBUG, "Setup request of size %d :\n %s\n", rmrMsg->len, rmrMsg->payload);
1791     }
1792     // send to RMR
1793     rmrMsg->mtype = message.message.messageType;
1794     rmrMsg->state = 0;
1795     rmr_bytes2meid(rmrMsg, (unsigned char *) message.message.enodbName, strlen(message.message.enodbName));
1796
1797     static unsigned char tx[32];
1798     snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1799     rmr_bytes2xact(rmrMsg, tx, strlen((const char *) tx));
1800 #ifndef UNIT_TEST
1801     rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg);
1802 #endif
1803     if (rmrMsg == nullptr) {
1804         mdclog_write(MDCLOG_ERR, "RMR failed to send returned nullptr");
1805     } else if (rmrMsg->state != 0) {
1806         char meid[RMR_MAX_MEID]{};
1807         if (rmrMsg->state == RMR_ERR_RETRY) {
1808             usleep(5);
1809             rmrMsg->state = 0;
1810             mdclog_write(MDCLOG_INFO, "RETRY sending Message %d to Xapp from %s",
1811                          rmrMsg->mtype, rmr_get_meid(rmrMsg, (unsigned char *) meid));
1812 #ifndef UNIT_TEST
1813             rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg);
1814 #endif
1815             if (rmrMsg == nullptr) {
1816                 mdclog_write(MDCLOG_ERR, "RMR failed send returned nullptr");
1817             } else if (rmrMsg->state != 0) {
1818                 mdclog_write(MDCLOG_ERR,
1819                              "RMR Retry failed %s sending request %d to Xapp from %s",
1820                              translateRmrErrorMessages(rmrMsg->state).c_str(),
1821                              rmrMsg->mtype,
1822                              rmr_get_meid(rmrMsg, (unsigned char *) meid));
1823             }
1824         } else {
1825             mdclog_write(MDCLOG_ERR, "RMR failed: %s. sending request %d to Xapp from %s",
1826                          translateRmrErrorMessages(rmrMsg->state).c_str(),
1827                          rmrMsg->mtype,
1828                          rmr_get_meid(rmrMsg, (unsigned char *) meid));
1829         }
1830     }
1831     message.peerInfo->gotSetup = true;
1832     buildJsonMessage(message);
1833
1834     if (rmrMsg != nullptr) {
1835         rmr_free_msg(rmrMsg);
1836     }
1837     free(buffer);
1838     buffer = nullptr;
1839
1840     return;
1841 }
1842 #endif
1843
1844 #if 0
1845 int RAN_Function_list_To_Vector(RANfunctions_List_t& list, vector <string> &runFunXML_v) {
1846     auto index = 0;
1847     runFunXML_v.clear();
1848     for (auto j = 0; j < list.list.count; j++) {
1849         auto *raNfunctionItemIEs = (RANfunction_ItemIEs_t *)list.list.array[j];
1850         if (raNfunctionItemIEs->id == ProtocolIE_ID_id_RANfunction_Item &&
1851             (raNfunctionItemIEs->value.present == RANfunction_ItemIEs__value_PR_RANfunction_Item)) {
1852             // encode to xml
1853             E2SM_gNB_NRT_RANfunction_Definition_t *ranFunDef = nullptr;
1854             auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER,
1855                                    &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition,
1856                                    (void **)&ranFunDef,
1857                                    raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.buf,
1858                                    raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.size);
1859             if (rval.code != RC_OK) {
1860                 mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2SM message from : %s",
1861                              rval.code,
1862                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name);
1863                 return -1;
1864             }
1865
1866             auto xml_buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2;
1867             unsigned char xml_buffer[RECEIVE_SCTP_BUFFER_SIZE * 2];
1868             memset(xml_buffer, 0, RECEIVE_SCTP_BUFFER_SIZE * 2);
1869             // encode to xml
1870             auto er = asn_encode_to_buffer(nullptr,
1871                                            ATS_BASIC_XER,
1872                                            &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition,
1873                                            ranFunDef,
1874                                            xml_buffer,
1875                                            xml_buffer_size);
1876             if (er.encoded == -1) {
1877                 mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s",
1878                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name,
1879                              strerror(errno));
1880             } else if (er.encoded > (ssize_t)xml_buffer_size) {
1881                 mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
1882                              (int) xml_buffer_size,
1883                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, __func__, __LINE__);
1884             } else {
1885                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1886                     mdclog_write(MDCLOG_DEBUG, "Encoding E2SM %s PDU number %d : %s",
1887                                  asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name,
1888                                  index++,
1889                                  xml_buffer);
1890                 }
1891
1892                 string runFuncs = (char *)(xml_buffer);
1893                 runFunXML_v.emplace_back(runFuncs);
1894             }
1895         }
1896     }
1897     return 0;
1898 }
1899
1900 int collectServiceUpdate_RequestData(E2AP_PDU_t *pdu,
1901                                      Sctp_Map_t *sctpMap,
1902                                      ReportingMessages_t &message,
1903                                      vector <string> &RANfunctionsAdded_v,
1904                                      vector <string> &RANfunctionsModified_v) {
1905     memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE);
1906     for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.count; i++) {
1907         auto *ie = pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.array[i];
1908         if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) {
1909             if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctionsID_List) {
1910                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1911                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
1912                                  ie->value.choice.RANfunctions_List.list.count);
1913                 }
1914                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) {
1915                     return -1;
1916                 }
1917             }
1918         } else if (ie->id == ProtocolIE_ID_id_RANfunctionsModified) {
1919             if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctions_List) {
1920                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1921                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
1922                                  ie->value.choice.RANfunctions_List.list.count);
1923                 }
1924                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsModified_v) != 0 ) {
1925                     return -1;
1926                 }
1927             }
1928         }
1929     }
1930     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1931         mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries",
1932                      RANfunctionsAdded_v.size());
1933     }
1934     return 0;
1935 }
1936
1937 #endif
1938
1939
1940 void buildE2TPrometheusCounters(sctp_params_t &sctpParams) {
1941     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestMsgs"}});
1942     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestBytes"}});
1943
1944     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupResponseMsgs"}});
1945     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupResponseBytes"}});
1946
1947     sctpParams.e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestFailureMsgs"}});
1948     sctpParams.e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestFailureBytes"}});
1949
1950     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateMsgs"}});
1951     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateBytes"}});
1952
1953     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateResponseMsgs"}});
1954     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateResponseBytes"}});
1955
1956     sctpParams.e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateFailureMsgs"}});
1957     sctpParams.e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "E2NodeConfigUpdateFailureBytes"}});
1958
1959     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &sctpParams.prometheusFamily->Add({{"counter", "ErrorIndicationMsgs"}});
1960     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &sctpParams.prometheusFamily->Add({{"counter", "ErrorIndicationBytes"}});
1961
1962     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetRequestMsgs"}});
1963     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetRequestBytes"}});
1964
1965     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetAckMsgs"}});
1966     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetAckBytes"}});
1967
1968     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateMsgs"}});
1969     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateBytes"}});
1970
1971     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateRespMsgs"}});
1972     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateRespBytes"}});
1973
1974     sctpParams.e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateFailureMsgs"}});
1975     sctpParams.e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateFailureBytes"}});
1976
1977     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlMsgs"}});
1978     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlBytes"}});
1979
1980     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlAckMsgs"}});
1981     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlAckBytes"}});
1982
1983     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlFailureMsgs"}});
1984     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlFailureBytes"}});
1985
1986     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionMsgs"}});
1987     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionBytes"}});
1988
1989     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionAckMsgs"}});
1990     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionAckBytes"}});
1991
1992     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionFailureMsgs"}});
1993     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionFailureBytes"}});
1994
1995     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteMsgs"}});
1996     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteBytes"}});
1997
1998     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteAckMsgs"}});
1999     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteAckBytes"}});
2000
2001     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteFailMsgs"}});
2002     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteFailBytes"}});
2003
2004     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication] = &sctpParams.prometheusFamily->Add({{"counter", "RICIndicationMsgs"}});
2005     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication] = &sctpParams.prometheusFamily->Add({{"counter", "RICIndicationBytes"}});
2006
2007     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceQueryMsgs"}});
2008     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceQueryBytes"}});
2009
2010     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteRequiredMsgs"}});
2011     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteRequiredBytes"}});
2012
2013 }
2014
2015 void buildPrometheusList(ConnectedCU_t *peerInfo, Family<Counter> *prometheusFamily) {
2016     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Messages"}});
2017     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Bytes"}});
2018
2019     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"E2NodeConfigUpdate", "Messages"}});
2020     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"E2NodeConfigUpdate", "Bytes"}});
2021
2022     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Messages"}});
2023     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Bytes"}});
2024
2025     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Messages"}});
2026     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Bytes"}});
2027
2028     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Messages"}});
2029     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Bytes"}});
2030
2031     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Messages"}});
2032     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Bytes"}});
2033     // ---------------------------------------------
2034     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Messages"}});
2035     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Bytes"}});
2036
2037     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Messages"}});
2038     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Bytes"}});
2039
2040     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Messages"}});
2041     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Bytes"}});
2042
2043     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Messages"}});
2044     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Bytes"}});
2045     //-------------------------------------------------------------
2046
2047     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Messages"}});
2048     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Bytes"}});
2049
2050     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Messages"}});
2051     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Bytes"}});
2052
2053     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Messages"}});
2054     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Bytes"}});
2055
2056     //====================================================================================
2057     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Messages"}});
2058     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Bytes"}});
2059
2060     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Messages"}});
2061     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Bytes"}});
2062
2063     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Messages"}});
2064     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Bytes"}});
2065
2066     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Messages"}});
2067     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Bytes"}});
2068
2069     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Messages"}});
2070     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Bytes"}});
2071
2072     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Messages"}});
2073     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Bytes"}});
2074     //---------------------------------------------------------------------------------------------------------
2075     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Messages"}});
2076     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Bytes"}});
2077
2078     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"E2NodeConfigUpdateSuccess", "Messages"}});
2079     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"E2NodeConfigUpdateSuccess", "Bytes"}});
2080
2081     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Messages"}});
2082     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Bytes"}});
2083
2084     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Messages"}});
2085     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Bytes"}});
2086     //----------------------------------------------------------------------------------------------------------------
2087     peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Messages"}});
2088     peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Bytes"}});
2089
2090     peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"E2NodeConfigUpdateFailure", "Messages"}});
2091     peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2nodeConfigurationUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"E2NodeConfigUpdateFailure", "Bytes"}});
2092
2093     peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Messages"}});
2094     peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Bytes"}});
2095
2096
2097     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDeleteRequired)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteRequired", "Messages"}});
2098     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDeleteRequired)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteRequired", "Bytes"}});
2099
2100     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2connectionUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"UnsupportedE2ConnectionUpdateAck", "Messages"}});
2101     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2connectionUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"UnsupportedE2ConnectionUpdateFail", "Messages"}});
2102 }
2103
2104 #ifndef UNIT_TEST
2105 void buildInternalCounterPrometheusList(ConnectedCU_t *peerInfo, Family<Counter> *prometheusFamily)
2106 {
2107     mdclog_write(MDCLOG_DEBUG, "Inside buildInternalCounterPrometheusList");
2108     peerInfo->e2tInternalCounters[E2T_Internal_Counters::SCTP_ABORT_INITIATED_BY_E2NODE] = &prometheusFamily->Add({{"NODEB", peerInfo->enodbName}, {"counter", "SctpAbortInitiatedByE2Node"}});
2109     peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED] = &prometheusFamily->Add({{"NODEB", peerInfo->enodbName}, {"counter", "InvalidMessageReceived"}});
2110 }
2111 #endif
2112 /**
2113  *
2114  * @param pdu
2115  * @param sctpMap
2116  * @param message
2117  * @param RANfunctionsAdded_v
2118  * @return
2119  */
2120 long collectSetupRequestData(E2AP_PDU_t *pdu,
2121                             Sctp_Map_t *sctpMap,
2122                             ReportingMessages_t &message /*, vector <string> &RANfunctionsAdded_v*/) {
2123     long transactionId = negativeOne;
2124     memset(message.peerInfo->enodbName, numberZero , MAX_ENODB_NAME_SIZE);
2125     for (auto i = numberZero; i < pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.count; i++) {
2126         auto *ie = pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.array[i];
2127         if (ie->id == ProtocolIE_ID_id_GlobalE2node_ID) {
2128             // get the ran name for meid
2129             if (ie->value.present == E2setupRequestIEs__value_PR_GlobalE2node_ID) {
2130                 mdclog_write(MDCLOG_DEBUG, "GnbId is present");
2131                 #ifndef UNIT_TEST
2132                 if (buildRanName(message.peerInfo->enodbName, ie) < numberZero) {
2133                     mdclog_write(MDCLOG_ERR, "Bad param in E2setupRequestIEs GlobalE2node_ID.\n");
2134                     // no message will be sent
2135                     return negativeOne;
2136                 }
2137                 #endif
2138                 memcpy(message.message.enodbName, message.peerInfo->enodbName, strlen(message.peerInfo->enodbName));
2139                 sctpMap->setkey(message.message.enodbName, message.peerInfo);
2140             }
2141         } /*else if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) {
2142             if (ie->value.present == E2setupRequestIEs__value_PR_RANfunctions_List) {
2143                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2144                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
2145                                  ie->value.choice.RANfunctions_List.list.count);
2146                 }
2147                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) {
2148                     return -1;
2149                 }
2150             }
2151         } */
2152         if (ie->id == ProtocolIE_ID_id_TransactionID) {
2153             if (ie->value.present == E2setupRequestIEs__value_PR_TransactionID) {
2154                 mdclog_write(MDCLOG_DEBUG, "TransactionId is present in E2SetupReq");
2155
2156                 transactionId = ie->value.choice.TransactionID;
2157                 mdclog_write(MDCLOG_DEBUG, "TransactionId is %ld", transactionId);
2158             }
2159         }
2160     }
2161 //    if (mdclog_level_get() >= MDCLOG_DEBUG) {
2162 //        mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries",
2163 //                     RANfunctionsAdded_v.size());
2164 //    }
2165     return transactionId;
2166 }
2167
2168 int XML_From_PER(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
2169     E2AP_PDU_t *pdu = nullptr;
2170
2171     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2172         mdclog_write(MDCLOG_DEBUG, "got PER message of size %d is:%s",
2173                      rmrMessageBuffer.sendMessage->len, rmrMessageBuffer.sendMessage->payload);
2174     }
2175     auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu,
2176                            rmrMessageBuffer.sendMessage->payload, rmrMessageBuffer.sendMessage->len);
2177     if (rval.code != RC_OK) {
2178         mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response  from E2MGR : %s",
2179                      rval.code,
2180                      message.message.enodbName);
2181         if (pdu != nullptr) {
2182             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2183             pdu = nullptr;
2184         }
2185         return -1;
2186     }
2187
2188     int buff_size = RECEIVE_XAPP_BUFFER_SIZE;
2189     auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu,
2190                                    rmrMessageBuffer.sendMessage->payload, buff_size);
2191     if (er.encoded == -1) {
2192         mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
2193         if (pdu != nullptr) {
2194             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2195             pdu = nullptr;
2196         }
2197         return -1;
2198     } else if (er.encoded > (ssize_t)buff_size) {
2199         mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
2200                      (int)rmrMessageBuffer.sendMessage->len,
2201                      asn_DEF_E2AP_PDU.name,
2202                      __func__,
2203                      __LINE__);
2204         if (pdu != nullptr) {
2205             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2206             pdu = nullptr;
2207         }
2208         return -1;
2209     }
2210     rmrMessageBuffer.sendMessage->len = er.encoded;
2211     if (pdu != nullptr) {
2212         ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2213         pdu = nullptr;
2214     }
2215     return 0;
2216
2217 }
2218
2219 /**
2220  *
2221  * @param pdu
2222  * @param message
2223  * @param rmrMessageBuffer
2224  */
2225 void asnInitiatingRequest(E2AP_PDU_t *pdu,
2226                           Sctp_Map_t *sctpMap,
2227                           ReportingMessages_t &message,
2228                           RmrMessagesBuffer_t &rmrMessageBuffer, int streamId) {
2229     auto logLevel = mdclog_level_get();
2230     auto procedureCode = ((InitiatingMessage_t *) pdu->choice.initiatingMessage)->procedureCode;
2231     if (logLevel >= MDCLOG_DEBUG) {
2232         mdclog_write(MDCLOG_DEBUG, "Initiating message %ld\n", procedureCode);
2233     }
2234     switch (procedureCode) {
2235         case ProcedureCode_id_E2setup: {
2236
2237             if (logLevel >= MDCLOG_DEBUG) {
2238                 mdclog_write(MDCLOG_DEBUG, "Got E2setup");
2239             }
2240
2241 //            vector <string> RANfunctionsAdded_v;
2242 //            vector <string> RANfunctionsModified_v;
2243 //            RANfunctionsAdded_v.clear();
2244 //            RANfunctionsModified_v.clear();
2245             long transactionID = collectSetupRequestData(pdu, sctpMap, message);
2246             mdclog_write(MDCLOG_DEBUG, "transactionID returned is %ld", transactionID);
2247
2248             if (transactionID < numberZero)
2249             {
2250                 mdclog_write(MDCLOG_ERR, "Invalid TransactionID or GnbIb");
2251                 break;
2252             }
2253
2254             mdclog_write(MDCLOG_DEBUG, "Calling handleE2SetupReq");
2255             handleE2SetupReq(message, rmrMessageBuffer, pdu, transactionID, streamId, sctpMap);
2256             break;
2257         }
2258         case ProcedureCode_id_RICserviceUpdate: {
2259             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2260             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2261             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2262             if (!isE2SetupRequestReceived)
2263             {
2264                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICserviceUpdate");
2265                 return ;
2266             }
2267             if (logLevel >= MDCLOG_DEBUG) {
2268                 mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName);
2269             }
2270
2271             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
2272             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
2273 //            vector <string> RANfunctionsAdded_v;
2274 //            vector <string> RANfunctionsModified_v;
2275 //            RANfunctionsAdded_v.clear();
2276 //            RANfunctionsModified_v.clear();
2277 //            if (collectServiceUpdate_RequestData(pdu, sctpMap, message,
2278 //                                                 RANfunctionsAdded_v, RANfunctionsModified_v) != 0) {
2279 //                break;
2280 //            }
2281
2282             string messageName("RICserviceUpdate");
2283             string ieName("RICserviceUpdateIEs");
2284             message.message.messageType = RIC_SERVICE_UPDATE;
2285 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2286             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2287             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment((double)message.message.asnLength);
2288
2289             // Update E2T instance level metrics
2290             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2291             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment((double)message.message.asnLength);
2292 #endif
2293             #ifndef UNIT_TEST
2294             buildAndSendSetupRequest(message, rmrMessageBuffer, pdu);
2295             #endif
2296
2297             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
2298             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_SERVICE_UPDATE_PROCEDURE_ONGOING);
2299             break;
2300         }
2301
2302 case ProcedureCode_id_E2nodeConfigurationUpdate: {
2303             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2304             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2305             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2306             if (!isE2SetupRequestReceived)
2307             {
2308                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received E2nodeConfigurationUpdate");
2309                 return ;
2310             }
2311             if (logLevel >= MDCLOG_DEBUG) {
2312                 mdclog_write(MDCLOG_DEBUG, "Got E2nodeConfigurationUpdate %s", message.message.enodbName);
2313             }
2314
2315             string messageName("RICE2nodeConfigurationUpdate");
2316             string ieName("RICE2nodeConfigurationUpdateIEs");
2317             message.message.messageType = RIC_E2NODE_CONFIG_UPDATE;
2318 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2319             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
2320             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment((double)message.message.asnLength);
2321
2322             // Update E2T instance level metrics
2323             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
2324             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment((double)message.message.asnLength);
2325
2326             buildAndSendSetupRequest(message, rmrMessageBuffer, pdu);
2327 #endif
2328             break;
2329         }
2330
2331         case ProcedureCode_id_ErrorIndication: {
2332             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2333             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2334             if (!isE2SetupRequestReceived){
2335                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received ErrorIndication");
2336                 return ;
2337             }
2338
2339             asn_fprint(stdout, &asn_DEF_E2AP_PDU, pdu);
2340
2341             mdclog_write(MDCLOG_INFO, "Got ErrorIndication %s sub id = %d, mtype = %d", message.message.enodbName,
2342                         rmrMessageBuffer.sendMessage->sub_id, rmrMessageBuffer.sendMessage->mtype);
2343
2344             int procedureCode = numberZero;
2345             bitset<requiredIePresentMaxBitSetPosition> isRequiredIesPresent(0); // TransactionId==0, RicRequestId==1, Cause==2
2346
2347             for (auto i = numberZero; i < pdu->choice.initiatingMessage->value.choice.ErrorIndication.protocolIEs.list.count; i++) {
2348                 ErrorIndication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.ErrorIndication.protocolIEs.list.array[i];
2349                 if (ie->id == ProtocolIE_ID_id_TransactionID) {
2350                     if (ie->value.present == ErrorIndication_IEs__value_PR_TransactionID) {
2351                         mdclog_write(MDCLOG_DEBUG, "TransactionID in ErrorIndication Message is %ld ", ie->value.choice.TransactionID);
2352                         if (e2NodeConnectionHandling.e2SetupProcedureTransactionId == ie->value.choice.TransactionID)
2353                         {
2354                             isRequiredIesPresent.set(transactionIdIeBitSetPosition);
2355                         }
2356                     }
2357                 }
2358                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2359                     if (ie->value.present == ErrorIndication_IEs__value_PR_RICrequestID) {
2360                         isRequiredIesPresent.set(ricRequestIdIeBitSetPosition);
2361                         mdclog_write(MDCLOG_DEBUG, "RIC instance id %ld, RIC requestor id = %ld",
2362                                     ie->value.choice.RICrequestID.ricInstanceID,
2363                                     ie->value.choice.RICrequestID.ricRequestorID);
2364                     }
2365                 }
2366                 if (ie->id == ProtocolIE_ID_id_Cause) {
2367                     if (ie->value.present == ErrorIndication_IEs__value_PR_Cause) {
2368                         isRequiredIesPresent.set(causeIeBitSetPosition);
2369                     }
2370                 }
2371                 if (ie->id == ProtocolIE_ID_id_CriticalityDiagnostics) {
2372                     if (ie->value.present == ErrorIndication_IEs__value_PR_CriticalityDiagnostics) {
2373                         if (ie->value.choice.CriticalityDiagnostics.procedureCode)
2374                         {
2375                             procedureCode = *(ie->value.choice.CriticalityDiagnostics.procedureCode);
2376                             mdclog_write(MDCLOG_DEBUG, "ProcedureCode Value is %ld", procedureCode);
2377                         }
2378                     }
2379                 }
2380             }
2381
2382             // Msg being sent to E2M = 0, Msg being sent to SubMgr = 1
2383             bitset<sendMsgMaxBitPosition> sendMsgTo = getSendMsgBitSetValue(procedureCode, isRequiredIesPresent, message.message.enodbName);
2384
2385             if(sendMsgTo.test(sendMsgToE2MBitSetPosition))
2386             {
2387                 message.message.messageType = RIC_E2_RIC_ERROR_INDICATION;
2388                 mdclog_write(MDCLOG_INFO, "Sending ErrorIndication Message to E2Manager");
2389                 #ifndef UNIT_TEST
2390                 buildAndSendSetupRequest(message, rmrMessageBuffer, pdu);
2391                 #endif
2392             }
2393             else if (sendMsgTo.test(sendMsgToSubMgrBitSetPosition))
2394             {
2395                 if (sendRequestToXapp(message, RIC_E2_RAN_ERROR_INDICATION, rmrMessageBuffer) != 0) {
2396                     mdclog_write(MDCLOG_ERR, "Subscription successful message failed to send to SubMgr");
2397                 }
2398                 else {
2399                     mdclog_write(MDCLOG_INFO, "Successfully Sent ErrorIndication Message to SubMgr");
2400                 }
2401             }
2402             else
2403             {
2404                 mdclog_write(MDCLOG_INFO, "No action taken on ErrorIndication Message as TransactionId/Ric RequestID/Cause not present/invalid");
2405             }
2406
2407             #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2408             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2409             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment((double)message.message.asnLength);
2410
2411             // Update E2T instance level metrics
2412             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2413             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment((double)message.message.asnLength);
2414             #endif
2415
2416             /*Not to break below functionality which is used to send to Xapp*/
2417             if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer) != 0) {
2418                 mdclog_write(MDCLOG_ERR, "RIC_ERROR_INDICATION failed to send to xAPP");
2419             }
2420             break;
2421         }
2422         case ProcedureCode_id_Reset: {
2423             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2424             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2425             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2426             if (!isE2SetupRequestReceived){
2427                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received Reset");
2428                 return ;
2429             }
2430             if (logLevel >= MDCLOG_DEBUG) {
2431                 mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName);
2432             }
2433 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2434             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2435             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2436
2437             // Update E2T instance level metrics
2438             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2439             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2440 #endif
2441             if (XML_From_PER(message, rmrMessageBuffer) < numberZero) {
2442                 break;
2443             }
2444
2445             if (sendRequestToXapp(message, RIC_E2_RESET_REQ, rmrMessageBuffer) != numberZero) {
2446                 mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_REQ message failed to send to xAPP");
2447             }
2448             break;
2449         }
2450         case ProcedureCode_id_RICindication: {
2451             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2452             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2453             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2454             if (!isE2SetupRequestReceived){
2455                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICIndication");
2456                 return ;
2457             }
2458             if (logLevel >= MDCLOG_DEBUG) {
2459                 mdclog_write(MDCLOG_DEBUG, "Got RICindication %s", message.message.enodbName);
2460             }
2461
2462             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2463             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2464
2465             for (auto i = numberZero; i < pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.count; i++) {
2466                 auto messageSent = false;
2467                 RICindication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.array[i];
2468                 if (logLevel >= MDCLOG_DEBUG) {
2469                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2470                 }
2471                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2472                     if (logLevel >= MDCLOG_DEBUG) {
2473                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2474                     }
2475                     if (ie->value.present == RICindication_IEs__value_PR_RICrequestID) {
2476                         static unsigned char tx[32];
2477                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_INDICATION;
2478                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2479                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2480                         rmr_bytes2meid(rmrMessageBuffer.sendMessage,
2481                                        (unsigned char *)message.message.enodbName,
2482                                        strlen(message.message.enodbName));
2483                         rmrMessageBuffer.sendMessage->state = numberZero;
2484                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2485
2486                         //ie->value.choice.RICrequestID.ricInstanceID;
2487                         if (mdclog_level_get() >= MDCLOG_DEBUG) {
2488                             mdclog_write(MDCLOG_DEBUG, "sub id = %d, mtype = %d, ric instance id %ld, requestor id = %ld",
2489                                          rmrMessageBuffer.sendMessage->sub_id,
2490                                          rmrMessageBuffer.sendMessage->mtype,
2491                                          ie->value.choice.RICrequestID.ricInstanceID,
2492                                          ie->value.choice.RICrequestID.ricRequestorID);
2493                         }
2494 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2495                         message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication]->Increment();
2496                         message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication]->Increment((double)message.message.asnLength);
2497
2498                         // Update E2T instance level metrics
2499                         message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication]->Increment();
2500                         message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication]->Increment((double)message.message.asnLength);
2501 #endif
2502                         sendRmrMessage(rmrMessageBuffer, message);
2503                         messageSent = true;
2504                     } else {
2505                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2506                     }
2507                 }
2508                 if (messageSent) {
2509                     mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, RIC_INDICATION_PROCEDURE_COMPLETED);
2510                     setE2ProcedureOngoingStatus(message.message.enodbName, RIC_INDICATION_PROCEDURE_COMPLETED);
2511                     break;
2512                 }
2513             }
2514             break;
2515         }
2516         case ProcedureCode_id_RICsubscriptionDeleteRequired: {
2517             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2518             mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
2519             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2520             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2521             if (!isE2SetupRequestReceived){
2522                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring Unsupported Message with procedureCode = %ld", procedureCode);
2523                 return ;
2524             }
2525             if (logLevel >= MDCLOG_DEBUG) {
2526                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDeleteRequired %s", message.message.enodbName);
2527             }
2528         #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2529             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired]->Increment();
2530             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired]->Increment((double)message.message.asnLength);
2531
2532             // Update E2T instance level metrics
2533             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired]->Increment();
2534             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDeleteRequired]->Increment((double)message.message.asnLength);
2535         #endif
2536             if (sendRequestToXapp(message, RIC_SUB_DEL_REQUIRED, rmrMessageBuffer) != 0) {
2537                 mdclog_write(MDCLOG_ERR, "Subscription Delete Required message failed to send to xAPP");
2538             }
2539             break;
2540         }
2541
2542         default: {
2543             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2544             mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
2545             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2546             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2547             if (!isE2SetupRequestReceived){
2548                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring Unsupported Message with procedureCode = %ld", procedureCode);
2549                 return ;
2550             }
2551             message.message.messageType = numberZero; // no RMR message type yet
2552             mdclog_write(MDCLOG_ERR, "Invalid Message Received at E2T");
2553             mdclog_write(MDCLOG_ERR, "Counter Incrementing for Invalid Message");
2554             #ifndef UNIT_TEST
2555             if (message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED])
2556             {
2557                 mdclog_write(MDCLOG_DEBUG, "Incrementing INVALID_MESSAGE_RECEIVED Counter");
2558                 message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED]->Increment();
2559                 mdclog_write(MDCLOG_DEBUG, "Incremented INVALID_MESSAGE_RECEIVED Counter");
2560             }
2561             #endif
2562
2563             buildJsonMessage(message);
2564
2565             break;
2566         }
2567     }
2568 }
2569
2570 /**
2571  *
2572  * @param pdu
2573  * @param message
2574  * @param rmrMessageBuffer
2575  */
2576 void asnSuccessfulMsg(E2AP_PDU_t *pdu,
2577                       Sctp_Map_t *sctpMap,
2578                       ReportingMessages_t &message,
2579                       RmrMessagesBuffer_t &rmrMessageBuffer) {
2580     auto procedureCode = pdu->choice.successfulOutcome->procedureCode;
2581     auto logLevel = mdclog_level_get();
2582     if (logLevel >= MDCLOG_INFO) {
2583         mdclog_write(MDCLOG_INFO, "Successful Outcome %ld", procedureCode);
2584     }
2585     switch (procedureCode) {
2586         case ProcedureCode_id_Reset: {
2587             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2588             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2589             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2590             if (!isE2SetupRequestReceived)
2591             {
2592                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received Reset message");
2593                 return ;
2594             }
2595             if (logLevel >= MDCLOG_DEBUG) {
2596                 mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName);
2597             }
2598 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2599             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2600             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2601
2602             // Update E2T instance level metrics
2603             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2604             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2605 #endif
2606             if (XML_From_PER(message, rmrMessageBuffer) < numberZero) {
2607                 break;
2608             }
2609             if (sendRequestToXapp(message, RIC_E2_RESET_RESP, rmrMessageBuffer) != 0) {
2610                 mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_RESP message failed to send to xAPP");
2611             }
2612             break;
2613         }
2614         case ProcedureCode_id_RICcontrol: {
2615             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2616             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2617             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2618             if (!isE2SetupRequestReceived){
2619                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICcontrol");
2620                 return ;
2621             }
2622             if (logLevel >= MDCLOG_DEBUG) {
2623                 mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName);
2624             }
2625             for (auto i = numberZero;
2626                  i < pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.count; i++) {
2627                 auto messageSent = false;
2628                 RICcontrolAcknowledge_IEs_t *ie = pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.array[i];
2629                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2630                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2631                 }
2632                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2633                     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2634                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2635                     }
2636                     if (ie->value.present == RICcontrolAcknowledge_IEs__value_PR_RICrequestID) {
2637                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_ACK;
2638                         rmrMessageBuffer.sendMessage->state = numberZero;
2639 //                        rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID;
2640                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2641
2642                         static unsigned char tx[32];
2643                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2644                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2645                         rmr_bytes2meid(rmrMessageBuffer.sendMessage,
2646                                        (unsigned char *)message.message.enodbName,
2647                                        strlen(message.message.enodbName));
2648 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2649                         message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2650                         message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2651
2652                         // Update E2T instance level metrics
2653                         message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2654                         message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2655 #endif
2656                         sendRmrMessage(rmrMessageBuffer, message);
2657                         messageSent = true;
2658                     } else {
2659                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2660                     }
2661                 }
2662                 if (messageSent) {
2663                     mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, CONTROL_PROCEDURE_COMPLETED);
2664                     setE2ProcedureOngoingStatus(message.message.enodbName, CONTROL_PROCEDURE_COMPLETED);
2665                     break;
2666                 }
2667             }
2668
2669             break;
2670         }
2671         case ProcedureCode_id_RICsubscription: {
2672             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2673             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2674             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2675             if (!isE2SetupRequestReceived){
2676                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICsubscription");
2677                 return ;
2678             }
2679             if (logLevel >= MDCLOG_DEBUG) {
2680                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName);
2681             }
2682 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2683             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2684             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2685
2686             // Update E2T instance level metrics
2687             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2688             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2689 #endif
2690             if (sendRequestToXapp(message, RIC_SUB_RESP, rmrMessageBuffer) != numberZero) {
2691                 mdclog_write(MDCLOG_ERR, "Subscription successful message failed to send to xAPP");
2692             }
2693             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2694             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2695             break;
2696         }
2697         case ProcedureCode_id_RICsubscriptionDelete: {
2698             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2699             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2700             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2701             if (!isE2SetupRequestReceived){
2702                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICsubscriptionDelete");
2703                 return ;
2704             }
2705             if (logLevel >= MDCLOG_DEBUG) {
2706                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName);
2707             }
2708
2709             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, RIC_SUBS_DEL_PROCEDURE_COMPLETED);
2710             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_SUBS_DEL_PROCEDURE_COMPLETED);
2711 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2712             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2713             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2714
2715             // Update E2T instance level metrics
2716             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2717             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2718 #endif
2719             if (sendRequestToXapp(message, RIC_SUB_DEL_RESP, rmrMessageBuffer) != numberZero) {
2720                 mdclog_write(MDCLOG_ERR, "Subscription delete successful message failed to send to xAPP");
2721             }
2722             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2723             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_INDICATION_PROCEDURE_ONGOING);
2724             break;
2725         }
2726         default: {
2727             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2728             mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
2729             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2730             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2731             if (!isE2SetupRequestReceived){
2732                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring Unsupported Message with procedureCode = %ld", procedureCode);
2733                 return ;
2734             }
2735             message.message.messageType = numberZero; // no RMR message type yet
2736             if (procedureCode == ProcedureCode_id_E2connectionUpdate)
2737             {
2738                 mdclog_write(MDCLOG_ERR, "Unsupported message E2Connection Update Ack Received at E2T, procedureCode = %ld", procedureCode);
2739
2740                 mdclog_write(MDCLOG_ERR, "Counter Incrementing for Unsupported E2Connection Update Ack Message");
2741                 #ifndef UNIT_TEST
2742                 message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_E2connectionUpdate]->Increment();
2743                 #endif
2744             }
2745             else {
2746                 mdclog_write(MDCLOG_ERR, "Invalid Message Received at E2T");
2747                 mdclog_write(MDCLOG_ERR, "Counter Incrementing for Invalid Message");
2748                 #ifndef UNIT_TEST
2749                 if (message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED])
2750                 {
2751                     mdclog_write(MDCLOG_DEBUG, "Incrementing INVALID_MESSAGE_RECEIVED Counter");
2752                     message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED]->Increment();
2753                     mdclog_write(MDCLOG_DEBUG, "Incremented INVALID_MESSAGE_RECEIVED Counter");
2754                 }
2755                 #endif
2756             }
2757             buildJsonMessage(message);
2758
2759             break;
2760         }
2761     }
2762 }
2763
2764 /**
2765  *
2766  * @param pdu
2767  * @param message
2768  * @param rmrMessageBuffer
2769  */
2770 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
2771                         Sctp_Map_t *sctpMap,
2772                         ReportingMessages_t &message,
2773                         RmrMessagesBuffer_t &rmrMessageBuffer) {
2774     auto procedureCode = pdu->choice.unsuccessfulOutcome->procedureCode;
2775     auto logLevel = mdclog_level_get();
2776     if (logLevel >= MDCLOG_INFO) {
2777         mdclog_write(MDCLOG_INFO, "Unsuccessful Outcome %ld", procedureCode);
2778     }
2779     switch (procedureCode) {
2780         case ProcedureCode_id_RICcontrol: {
2781             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2782             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2783             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2784             if (!isE2SetupRequestReceived)
2785             {
2786                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICcontrol");
2787                 return ;
2788             }
2789             if (logLevel >= MDCLOG_DEBUG) {
2790                 mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName);
2791             }
2792             for (int i = 0;
2793                  i < pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.count; i++) {
2794                 auto messageSent = false;
2795                 RICcontrolFailure_IEs_t *ie = pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.array[i];
2796                 if (logLevel >= MDCLOG_DEBUG) {
2797                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2798                 }
2799                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2800                     if (logLevel >= MDCLOG_DEBUG) {
2801                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2802                     }
2803                     if (ie->value.present == RICcontrolFailure_IEs__value_PR_RICrequestID) {
2804                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_FAILURE;
2805                         rmrMessageBuffer.sendMessage->state = 0;
2806 //                        rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricRequestorID;
2807                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2808                         static unsigned char tx[32];
2809                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2810                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2811                         rmr_bytes2meid(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.enodbName,
2812                                        strlen(message.message.enodbName));
2813 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2814                         message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2815                         message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2816
2817                         // Update E2T instance level metrics
2818                         message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2819                         message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2820 #endif
2821                         sendRmrMessage(rmrMessageBuffer, message);
2822                         messageSent = true;
2823                     } else {
2824                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2825                     }
2826                 }
2827                 if (messageSent) {
2828                     break;
2829                 }
2830             }
2831             break;
2832         }
2833         case ProcedureCode_id_RICsubscription: {
2834             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2835             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2836             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2837             if (!isE2SetupRequestReceived){
2838                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICsubscription");
2839                 return ;
2840             }
2841             if (logLevel >= MDCLOG_DEBUG) {
2842                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName);
2843             }
2844 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2845             message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2846             message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2847
2848             // Update E2T instance level metrics
2849             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2850             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2851 #endif
2852             if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) {
2853                 mdclog_write(MDCLOG_ERR, "Subscription unsuccessful message failed to send to xAPP");
2854             }
2855             break;
2856         }
2857         case ProcedureCode_id_RICsubscriptionDelete: {
2858             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2859             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2860             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2861             if (!isE2SetupRequestReceived){
2862                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring received RICsubscriptionDelete");
2863                 return ;
2864             }
2865             if (logLevel >= MDCLOG_DEBUG) {
2866                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName);
2867             }
2868 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2869             message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2870             message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2871
2872             // Update E2T instance level metrics
2873             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2874             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2875 #endif
2876             if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) {
2877                 mdclog_write(MDCLOG_ERR, "Subscription Delete unsuccessful message failed to send to xAPP");
2878             }
2879             break;
2880         }
2881         default: {
2882             struct E2NodeConnectionHandling e2NodeConnectionHandling;
2883             mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
2884             bool isE2SetupRequestReceived = getE2tProcedureOngoingStatus(message.message.enodbName, e2NodeConnectionHandling);
2885             mdclog_write(MDCLOG_DEBUG, "Value of isE2SetupRequestReceived is %s", isE2SetupRequestReceived == true ? "true" : "false");
2886             if (!isE2SetupRequestReceived){
2887                 mdclog_write(MDCLOG_WARN, "E2Setup procedure is not initiated, ignoring Unsupported Message with procedureCode = %ld", procedureCode);
2888                 return ;
2889             }
2890
2891             message.message.messageType = numberZero; // no RMR message type yet
2892             if (procedureCode == ProcedureCode_id_E2connectionUpdate)
2893             {
2894                 mdclog_write(MDCLOG_ERR, "Unsupported message E2Connection Update Failure Received at E2T, procedureCode = %ld", procedureCode);
2895
2896                 mdclog_write(MDCLOG_ERR, "Counter Incrementing for Unsupported E2Connection Update Failure Message");
2897                 #ifndef UNIT_TEST
2898                 message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2connectionUpdate]->Increment();
2899                 #endif
2900             }
2901             else {
2902                 mdclog_write(MDCLOG_ERR, "Invalid Message Received at E2T");
2903                 mdclog_write(MDCLOG_ERR, "Counter Incrementing for Invalid Message");
2904                 #ifndef UNIT_TEST
2905                 if (message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED])
2906                 {
2907                     mdclog_write(MDCLOG_DEBUG, "Incrementing INVALID_MESSAGE_RECEIVED Counter");
2908                     message.peerInfo->e2tInternalCounters[E2T_Internal_Counters::INVALID_MESSAGE_RECEIVED]->Increment();
2909                     mdclog_write(MDCLOG_DEBUG, "Incremented INVALID_MESSAGE_RECEIVED Counter");
2910                 }
2911                 #endif
2912             }
2913             buildJsonMessage(message);
2914             break;
2915         }
2916     }
2917 }
2918
2919 bool getE2tProcedureOngoingStatus(char *enbName, E2NodeConnectionHandling &e2NodeConnectionHandling)
2920 {
2921     mdclog_write(MDCLOG_DEBUG, "Inside getE2tProcedureOngoingStatus, Checking for enb %s", enbName);
2922     if(strcmp(enbName, "") == 0)
2923     {
2924         mdclog_write(MDCLOG_DEBUG, "enbName is empty");
2925         return false;
2926     }
2927     else if(connectionHandlingPerE2NodeMap.empty())
2928     {
2929         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
2930         return false;
2931     }
2932     else
2933     {
2934         std::map<std::string, E2NodeConnectionHandling>::iterator itr = connectionHandlingPerE2NodeMap.begin();
2935         while (itr != connectionHandlingPerE2NodeMap.end())
2936         {
2937             int result = itr->first.compare(enbName);
2938             if (result == 0)
2939             {
2940                 mdclog_write(MDCLOG_DEBUG, "enb name in map :%s, status :%d", itr->first.c_str(), itr->second.e2tProcedureOngoingStatus);
2941                 e2NodeConnectionHandling = itr->second;
2942                 return true;
2943                 break;
2944             }
2945             ++itr;
2946         }
2947     }
2948     return false;
2949 }
2950
2951 void setE2ProcedureOngoingStatus(char *enbName, E2T_Procedure_States state)
2952 {
2953     printEntryPresentInMap();
2954     if(connectionHandlingPerE2NodeMap.empty())
2955     {
2956         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
2957         return;
2958     }
2959     else
2960     {
2961         mdclog_write(MDCLOG_DEBUG, "Entry Present in connectionHandlingPerE2NodeMap Map");
2962         std::map<std::string, E2NodeConnectionHandling>::iterator itr = connectionHandlingPerE2NodeMap.begin();
2963         while(itr != connectionHandlingPerE2NodeMap.end())
2964         {
2965             int result = itr->first.compare(enbName);
2966             if (result == 0)
2967             {
2968                 mdclog_write(MDCLOG_DEBUG, "Key %s Found in connectionHandlingPerE2NodeMap Map, Current Procedure is %d", enbName, itr->second.e2tProcedureOngoingStatus);
2969                 itr->second.e2tProcedureOngoingStatus = state;
2970                 mdclog_write(MDCLOG_DEBUG, "Current procedure updated to %d", itr->second.e2tProcedureOngoingStatus);
2971                 break;
2972             }
2973             ++itr;
2974         }
2975     }
2976 }
2977
2978 void insertE2SetupProcedureOngoing(char *enbName, long &transactionID)
2979 {
2980     struct E2NodeConnectionHandling e2NodeConnectionHandling;
2981     e2NodeConnectionHandling.e2tProcedureOngoingStatus = E2_SETUP_PROCEDURE_ONGOING;
2982     e2NodeConnectionHandling.e2SetupProcedureTransactionId = transactionID;
2983
2984     std::string key(enbName);
2985     printEntryPresentInMap();
2986     if(connectionHandlingPerE2NodeMap.empty())
2987     {
2988         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
2989         mdclog_write(MDCLOG_DEBUG, "Inserting %s to connectionHandlingPerE2NodeMap Map", enbName);
2990
2991         connectionHandlingPerE2NodeMap.insert(std::make_pair(key, e2NodeConnectionHandling));
2992
2993         mdclog_write(MDCLOG_DEBUG, "Default Value after Inserting Key for %s - Value is {e2tProcedureOngoingStatus is %d, e2SetupProcedureTransactionId: %ld}",
2994                 key.c_str(),
2995                 e2NodeConnectionHandling.e2tProcedureOngoingStatus,
2996                 e2NodeConnectionHandling.e2SetupProcedureTransactionId);
2997         return;
2998     }
2999     else
3000     {
3001         mdclog_write(MDCLOG_DEBUG, "Entry Present in connectionHandlingPerE2NodeMap Map");
3002         std::map<std::string, E2NodeConnectionHandling>::iterator itr = connectionHandlingPerE2NodeMap.begin();
3003         bool enodebNameFound = false;
3004         while(itr != connectionHandlingPerE2NodeMap.end())
3005         {
3006             int result = itr->first.compare(enbName);
3007             if (result == 0)
3008             {
3009                 enodebNameFound = true;
3010                 break;
3011             }
3012             ++itr;
3013         }
3014         if (enodebNameFound == false)
3015         {
3016             mdclog_write(MDCLOG_DEBUG, "Inserting %s to connectionHandlingPerE2NodeMap Map", enbName);
3017             connectionHandlingPerE2NodeMap.insert(std::make_pair(key, e2NodeConnectionHandling));
3018
3019             mdclog_write(MDCLOG_DEBUG, "Default Value after Inserting Key for %s - Value is {e2tProcedureOngoingStatus is %d, e2SetupProcedureTransactionId: %ld}",
3020                 key.c_str(),
3021                 e2NodeConnectionHandling.e2tProcedureOngoingStatus,
3022                 e2NodeConnectionHandling.e2SetupProcedureTransactionId);
3023         }
3024         else
3025         {
3026             mdclog_write(MDCLOG_DEBUG, "Processing E2Setup Req received for same gnb - %s", enbName);
3027             setE2ProcedureOngoingStatus(enbName, E2_SETUP_PROCEDURE_ONGOING);
3028         }
3029     }
3030 }
3031
3032 E2T_Procedure_States currentE2tProcedureOngoingStatus(char *enbName)
3033 {
3034     printEntryPresentInMap();
3035     if(connectionHandlingPerE2NodeMap.empty())
3036     {
3037         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
3038         return E2_SETUP_PROCEDURE_NOT_INITIATED;
3039     }
3040     else
3041     {
3042         mdclog_write(MDCLOG_DEBUG, "Entry Present in connectionHandlingPerE2NodeMap Map");
3043         std::map<std::string, E2NodeConnectionHandling>::iterator itr = connectionHandlingPerE2NodeMap.begin();
3044         while(itr != connectionHandlingPerE2NodeMap.end())
3045         {
3046             int result = itr->first.compare(enbName);
3047             if (result == 0)
3048             {
3049                 mdclog_write(MDCLOG_DEBUG, "Key %s Found in connectionHandlingPerE2NodeMap Map, Current Procedure is %d", enbName, itr->second.e2tProcedureOngoingStatus);
3050                 mdclog_write(MDCLOG_DEBUG, "%s Entry has found", enbName);
3051                 return itr->second.e2tProcedureOngoingStatus;
3052             }
3053             ++itr;
3054         }
3055         mdclog_write(MDCLOG_DEBUG, "No Key %s Found in connectionHandlingPerE2NodeMap Map", enbName);
3056         return E2_SETUP_PROCEDURE_NOT_INITIATED;
3057     }
3058 }
3059
3060 bitset<sendMsgMaxBitPosition> getSendMsgBitSetValue(int procedureCode, bitset<requiredIePresentMaxBitSetPosition> isRequiredIesPresent, char* enbName)
3061 {
3062     bitset<sendMsgMaxBitPosition> sendMsgTo(0); // Msg being sent to E2M = 0, Msg being sent to SubMgr = 1
3063     switch(procedureCode)
3064     {
3065         case numberZero:
3066         {
3067             E2T_Procedure_States currentProcedureOngoing = currentE2tProcedureOngoingStatus(enbName);
3068             if ((currentProcedureOngoing == E2_SETUP_PROCEDURE_COMPLETED ||
3069                  currentProcedureOngoing == RIC_SERVICE_UPDATE_PROCEDURE_ONGOING) &&
3070                  (isRequiredIesPresent.test(transactionIdIeBitSetPosition)))
3071             {
3072                 sendMsgTo.set(sendMsgToE2MBitSetPosition);
3073             }
3074             else if ((currentProcedureOngoing == RIC_SUBS_PROCEDURE_ONGOING) &&
3075                       isRequiredIesPresent.test(ricRequestIdIeBitSetPosition) && isRequiredIesPresent.test(causeIeBitSetPosition))
3076             {
3077                 sendMsgTo.set(sendMsgToSubMgrBitSetPosition);
3078             }
3079             break;
3080         }
3081         case ProcedureCode_id_E2setup:
3082         case ProcedureCode_id_RICserviceUpdate:
3083         {
3084             if(isRequiredIesPresent.test(transactionIdIeBitSetPosition))
3085             {
3086                 sendMsgTo.set(sendMsgToE2MBitSetPosition);
3087             }
3088             break;
3089
3090         }
3091         case ProcedureCode_id_RICsubscription:
3092         {
3093             if(isRequiredIesPresent.test(ricRequestIdIeBitSetPosition) && isRequiredIesPresent.test(causeIeBitSetPosition))
3094             {
3095                 sendMsgTo.set(sendMsgToSubMgrBitSetPosition);
3096             }
3097             break;
3098         }
3099         default:
3100         {
3101             mdclog_write(MDCLOG_ERR, "Invalid procedure code received");
3102         }
3103     }
3104     return sendMsgTo;
3105 }
3106
3107 void printEntryPresentInMap()
3108 {
3109     mdclog_write(MDCLOG_DEBUG, "Inside printEntryPresentInMap");
3110     if(connectionHandlingPerE2NodeMap.empty())
3111     {
3112         mdclog_write(MDCLOG_DEBUG, "connectionHandlingPerE2Node map is empty");
3113         return;
3114     }
3115     else
3116     {
3117         mdclog_write(MDCLOG_DEBUG, "Entry Present in Map");
3118         std::map<std::string, E2NodeConnectionHandling>::iterator it = connectionHandlingPerE2NodeMap.begin();
3119         while (it != connectionHandlingPerE2NodeMap.end())
3120         {
3121             mdclog_write(MDCLOG_DEBUG, "Key -> { enb name in map : %s }, Value -> { e2tProcedureOngoingStatus is %d, e2SetupProcedureTransactionId is %ld }", it->first.c_str(), it->second.e2tProcedureOngoingStatus, it->second.e2SetupProcedureTransactionId);
3122             ++it;
3123         }
3124         mdclog_write(MDCLOG_DEBUG, "All entries currently present in connectionHandlingPerE2NodeMap are printed successfully");
3125     }
3126 }
3127
3128 void handleE2SetupReq(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, E2AP_PDU_t *pdu, long &transactionID, int streamId, Sctp_Map_t *sctpMap)
3129 {
3130     auto logLevel = mdclog_level_get();
3131
3132     #ifndef UNIT_TEST
3133     buildPrometheusList(message.peerInfo, message.peerInfo->sctpParams->prometheusFamily);
3134     buildInternalCounterPrometheusList(message.peerInfo, message.peerInfo->sctpParams->prometheusFamily);
3135     #endif
3136
3137     struct sctp_status status;
3138     int stat_size = sizeof(status);
3139     getsockopt( message.peerInfo->fileDescriptor, SOL_SCTP, SCTP_STATUS,(void *)&status, (socklen_t *)&stat_size );
3140
3141     if (logLevel >= MDCLOG_DEBUG) {
3142         mdclog_write(MDCLOG_DEBUG, "Start from SCTP %d fd", message.peerInfo->fileDescriptor);
3143         mdclog_write(MDCLOG_DEBUG, "SCTP status assoc id %d instrms %d outstrms %d", status.sstat_assoc_id,
3144                 status.sstat_instrms, status.sstat_outstrms);
3145     }
3146     if(status.sstat_outstrms == numberOne || status.sstat_instrms == numberOne)
3147     {
3148         message.peerInfo->isSingleStream = true;
3149         message.peerInfo->singleStreamId = streamId;
3150         if (status.sstat_outstrms == numberOne && status.sstat_instrms == numberOne){
3151             if (logLevel >= MDCLOG_DEBUG) {
3152                 mdclog_write(MDCLOG_DEBUG, "Single SCTP stream is used for sending from now on, assoc id %d streamId %d #instrms %d #outstrms %d, %s",status.sstat_assoc_id, streamId, status.sstat_instrms, status.sstat_outstrms, __FUNCTION__);
3153             }
3154         }
3155         else {
3156             mdclog_write(MDCLOG_ERR, "Single SCTP stream used for sending messages even if there is a mismatch in number of in & out streams, assoc id %d instrms %d outstrms %d", status.sstat_assoc_id,
3157                 status.sstat_instrms, status.sstat_outstrms);
3158         }
3159     }
3160
3161     string messageName("E2setupRequest");
3162     string ieName("E2setupRequestIEs");
3163     message.message.messageType = RIC_E2_SETUP_REQ;
3164     #ifndef UNIT_TEST
3165     message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
3166     message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment((double)message.message.asnLength);
3167
3168     buildAndSendSetupRequest(message, rmrMessageBuffer, pdu); //UT - Segmentation Fault Happening.
3169     #endif
3170
3171     mdclog_write(MDCLOG_DEBUG, "Called insertE2ProcedureOngoing: EnbName is %s and transactionID is %ld", message.message.enodbName, transactionID);
3172     insertE2SetupProcedureOngoing(message.message.enodbName, transactionID);
3173 }
3174
3175 /**
3176  *
3177  * @param message
3178  * @param requestId
3179  * @param rmrMmessageBuffer
3180  * @return
3181  */
3182 int sendRequestToXapp(ReportingMessages_t &message,
3183                       int requestId,
3184                       RmrMessagesBuffer_t &rmrMmessageBuffer) {
3185     rmr_bytes2meid(rmrMmessageBuffer.sendMessage,
3186                    (unsigned char *)message.message.enodbName,
3187                    strlen(message.message.enodbName));
3188     message.message.messageType = rmrMmessageBuffer.sendMessage->mtype = requestId;
3189     rmrMmessageBuffer.sendMessage->state = numberZero;
3190     static unsigned char tx[32];
3191     snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
3192     rmr_bytes2xact(rmrMmessageBuffer.sendMessage, tx, strlen((const char *) tx));
3193
3194     auto rc = sendRmrMessage(rmrMmessageBuffer, message);
3195     return rc;
3196 }
3197
3198 /**
3199  *
3200  * @param pSctpParams
3201  */
3202 void getRmrContext(sctp_params_t &pSctpParams) {
3203     pSctpParams.rmrCtx = nullptr;
3204     pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RECEIVE_XAPP_BUFFER_SIZE, RMRFL_NONE);
3205     if (pSctpParams.rmrCtx == nullptr) {
3206         mdclog_write(MDCLOG_ERR, "Failed to initialize RMR");
3207         return;
3208     }
3209
3210     rmr_set_low_latency(pSctpParams.rmrCtx);
3211     rmr_set_stimeout(pSctpParams.rmrCtx, 0);    // disable retries for any send operation
3212     // we need to find that routing table exist and we can run
3213     if (mdclog_level_get() >= MDCLOG_INFO) {
3214         mdclog_write(MDCLOG_INFO, "We are after RMR INIT wait for RMR_Ready");
3215     }
3216     int rmrReady = 0;
3217     int count = 0;
3218     while (!rmrReady) {
3219         if ((rmrReady = rmr_ready(pSctpParams.rmrCtx)) == 0) {
3220             sleep(1);
3221         }
3222         count++;
3223         if (count % 60 == 0) {
3224             mdclog_write(MDCLOG_INFO, "waiting to RMR ready state for %d seconds", count);
3225         }
3226     }
3227     if (mdclog_level_get() >= MDCLOG_INFO) {
3228         mdclog_write(MDCLOG_INFO, "RMR running");
3229     }
3230     rmr_init_trace(pSctpParams.rmrCtx, 200);
3231     // get the RMR fd for the epoll
3232     pSctpParams.rmrListenFd = rmr_get_rcvfd(pSctpParams.rmrCtx);
3233     struct epoll_event event{};
3234     // add RMR fd to epoll
3235     event.events = (EPOLLIN);
3236     event.data.fd = pSctpParams.rmrListenFd;
3237     // add listening RMR FD to epoll
3238     if (epoll_ctl(pSctpParams.epoll_fd, EPOLL_CTL_ADD, pSctpParams.rmrListenFd, &event)) {
3239         mdclog_write(MDCLOG_ERR, "Failed to add RMR descriptor to epoll");
3240         close(pSctpParams.rmrListenFd);
3241         rmr_close(pSctpParams.rmrCtx);
3242         pSctpParams.rmrCtx = nullptr;
3243     }
3244 }
3245
3246 /**
3247  *
3248  * @param message
3249  * @param rmrMessageBuffer
3250  * @return
3251  */
3252 int PER_FromXML(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
3253     E2AP_PDU_t *pdu = nullptr;
3254
3255     if (mdclog_level_get() >= MDCLOG_DEBUG) {
3256         mdclog_write(MDCLOG_DEBUG, "got xml Format  data from xApp of size %d is:%s",
3257                 rmrMessageBuffer.rcvMessage->len, rmrMessageBuffer.rcvMessage->payload);
3258     }
3259     auto rval = asn_decode(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, (void **) &pdu,
3260                            rmrMessageBuffer.rcvMessage->payload, rmrMessageBuffer.rcvMessage->len);
3261     if (mdclog_level_get() >= MDCLOG_DEBUG) {
3262         mdclog_write(MDCLOG_DEBUG, "%s After  decoding the XML to PDU", __func__ );
3263     }
3264     if (rval.code != RC_OK) {
3265 #ifdef UNIT_TEST
3266     return 0;
3267 #endif
3268         mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response  from E2MGR : %s",
3269                      rval.code,
3270                      message.message.enodbName);
3271         if (pdu != nullptr) {
3272             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
3273             pdu = nullptr;
3274         }
3275         return -1;
3276     }
3277
3278     int buff_size = RECEIVE_XAPP_BUFFER_SIZE;
3279     auto er = asn_encode_to_buffer(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, pdu,
3280                                    rmrMessageBuffer.rcvMessage->payload, buff_size);
3281     if (mdclog_level_get() >= MDCLOG_DEBUG) {
3282         mdclog_write(MDCLOG_DEBUG, "%s After encoding PDU to PER", __func__ );
3283     }
3284     if (er.encoded == -1) {
3285         mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
3286         if (pdu != nullptr) {
3287             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
3288             pdu = nullptr;
3289         }
3290         return -1;
3291     } else if (er.encoded > (ssize_t)buff_size) {
3292         mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
3293                      (int)rmrMessageBuffer.rcvMessage->len,
3294                      asn_DEF_E2AP_PDU.name,
3295                      __func__,
3296                      __LINE__);
3297         if (pdu != nullptr) {
3298             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
3299             pdu = nullptr;
3300         }
3301         return -1;
3302     }
3303     rmrMessageBuffer.rcvMessage->len = er.encoded;
3304     if (pdu != nullptr) {
3305         ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
3306         pdu = nullptr;
3307     }
3308     return 0;
3309 }
3310
3311 /**
3312  *
3313  * @param sctpMap
3314  * @param rmrMessageBuffer
3315  * @param ts
3316  * @return
3317  */
3318 int receiveXappMessages(Sctp_Map_t *sctpMap,
3319                         RmrMessagesBuffer_t &rmrMessageBuffer,
3320                         struct timespec &ts) {
3321     int loglevel = mdclog_level_get();
3322     if (rmrMessageBuffer.rcvMessage == nullptr) {
3323         //we have error
3324         mdclog_write(MDCLOG_ERR, "RMR Allocation message, %s", strerror(errno));
3325         return negativeOne;
3326     }
3327
3328 //    if (loglevel >= MDCLOG_DEBUG) {
3329 //        mdclog_write(MDCLOG_DEBUG, "Call to rmr_rcv_msg");
3330 //    }
3331     rmrMessageBuffer.rcvMessage = rmr_rcv_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage);
3332     if (rmrMessageBuffer.rcvMessage == nullptr) {
3333         mdclog_write(MDCLOG_ERR, "RMR Receiving message with null pointer, Reallocated rmr message buffer");
3334         rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3335         return -2;
3336     }
3337     ReportingMessages_t message;
3338     message.message.direction = 'D';
3339     message.message.time.tv_nsec = ts.tv_nsec;
3340     message.message.time.tv_sec = ts.tv_sec;
3341
3342     // get message payload
3343     //auto msgData = msg->payload;
3344 #ifdef UNIT_TEST
3345     rmrMessageBuffer.rcvMessage->state = 0;
3346 #endif
3347     if (rmrMessageBuffer.rcvMessage->state != numberZero) {
3348         mdclog_write(MDCLOG_ERR, "RMR Receiving message with stat = %d", rmrMessageBuffer.rcvMessage->state);
3349         return negativeOne;
3350     }
3351     rmr_get_meid(rmrMessageBuffer.rcvMessage, (unsigned char *)message.message.enodbName);
3352     message.peerInfo = (ConnectedCU_t *) sctpMap->find(message.message.enodbName);
3353     if (message.peerInfo == nullptr) {
3354         auto type = rmrMessageBuffer.rcvMessage->mtype;
3355         switch (type) {
3356             case RIC_SCTP_CLEAR_ALL:
3357             case E2_TERM_KEEP_ALIVE_REQ:
3358             case RIC_HEALTH_CHECK_REQ:
3359                 break;
3360             default:
3361 #ifdef UNIT_TEST
3362     break;
3363 #endif
3364                 mdclog_write(MDCLOG_ERR, "Failed to send message no CU entry %s", message.message.enodbName);
3365                 return negativeOne;
3366         }
3367     }
3368
3369     if (rmrMessageBuffer.rcvMessage->mtype != RIC_HEALTH_CHECK_REQ) {
3370         num_of_XAPP_messages.fetch_add(numberOne, std::memory_order_release);
3371
3372     }
3373     switch (rmrMessageBuffer.rcvMessage->mtype) {
3374         case RIC_E2_SETUP_RESP : {
3375             if (loglevel >= MDCLOG_DEBUG) {
3376                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_SETUP_RESP");
3377             }
3378             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3379                 break;
3380             }
3381 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3382             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
3383             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
3384
3385             // Update E2T instance level metrics
3386             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
3387             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
3388 #endif
3389             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3390                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_RESP");
3391                 return negativeSix;
3392             }
3393             mdclog_write(MDCLOG_DEBUG, "Successfully Sent E2_SETUP_RESP to E2Node");
3394             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3395             break;
3396         }
3397         case RIC_E2_SETUP_FAILURE : {
3398             if (loglevel >= MDCLOG_DEBUG) {
3399                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_SETUP_FAILURE");
3400             }
3401             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3402                 break;
3403             }
3404 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3405             message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
3406             message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
3407
3408             // Update E2T instance level metrics
3409             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
3410             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
3411 #endif
3412             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3413                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_FAILURE");
3414                 return negativeSix;
3415             }
3416
3417             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3418             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3419             break;
3420         }
3421
3422         case RIC_E2NODE_CONFIG_UPDATE_ACK: {
3423             if (loglevel >= MDCLOG_DEBUG) {
3424                 mdclog_write(MDCLOG_DEBUG, "RIC_E2NODE_CONFIG_UPDATE_ACK");
3425             }
3426             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3427                 break;
3428             }
3429 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3430             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
3431             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3432
3433             // Update E2T instance level metrics
3434             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
3435             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3436 #endif
3437             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3438                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2NODE_CONFIG_UPDATE_ACK");
3439                 return -6;
3440             }
3441             break;
3442         }
3443
3444         case RIC_E2NODE_CONFIG_UPDATE_FAILURE: {
3445             if (loglevel >= MDCLOG_DEBUG) {
3446                 mdclog_write(MDCLOG_DEBUG, "RIC_E2NODE_CONFIG_UPDATE_FAILURE");
3447             }
3448             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3449                 break;
3450             }
3451 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3452             message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
3453             message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3454
3455             // Update E2T instance level metrics
3456             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment();
3457             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2nodeConfigurationUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3458 #endif
3459             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3460                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2NODE_CONFIG_UPDATE_FAILURE");
3461                 return -6;
3462             }
3463             break;
3464         }
3465
3466         case RIC_ERROR_INDICATION: {
3467             if (loglevel >= MDCLOG_DEBUG) {
3468                 mdclog_write(MDCLOG_DEBUG, "RIC_ERROR_INDICATION");
3469             }
3470 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3471             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
3472             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(rmrMessageBuffer.rcvMessage->len);
3473
3474             // Update E2T instance level metrics
3475             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
3476             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(rmrMessageBuffer.rcvMessage->len);
3477 #endif
3478             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3479                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ERROR_INDICATION");
3480                 return negativeSix;
3481             }
3482             break;
3483         }
3484         case RIC_SUB_REQ: {
3485             if (loglevel >= MDCLOG_DEBUG) {
3486                 mdclog_write(MDCLOG_DEBUG, "RIC_SUB_REQ");
3487             }
3488
3489             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3490             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3491 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3492             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
3493             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment(rmrMessageBuffer.rcvMessage->len);
3494
3495             // Update E2T instance level metrics
3496             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
3497             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment(rmrMessageBuffer.rcvMessage->len);
3498 #endif
3499             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3500                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_REQ");
3501                 return negativeSix;
3502             }
3503             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state", message.message.enodbName);
3504             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_SUBS_PROCEDURE_ONGOING);
3505             break;
3506         }
3507         case RIC_SUB_DEL_REQ: {
3508             if (loglevel >= MDCLOG_DEBUG) {
3509                 mdclog_write(MDCLOG_DEBUG, "RIC_SUB_DEL_REQ");
3510             }
3511
3512             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3513             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3514 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3515             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
3516             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(rmrMessageBuffer.rcvMessage->len);
3517
3518             // Update E2T instance level metrics
3519             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
3520             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(rmrMessageBuffer.rcvMessage->len);
3521 #endif
3522             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3523                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_DEL_REQ");
3524                 return negativeSix;
3525             }
3526             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure State is %d", message.message.enodbName, RIC_SUBS_DEL_PROCEDURE_ONGOING);
3527             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_SUBS_DEL_PROCEDURE_ONGOING);
3528             break;
3529         }
3530         case RIC_CONTROL_REQ: {
3531             if (loglevel >= MDCLOG_DEBUG) {
3532                 mdclog_write(MDCLOG_DEBUG, "RIC_CONTROL_REQ");
3533             }
3534             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3535             setE2ProcedureOngoingStatus(message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3536 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3537             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
3538             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment(rmrMessageBuffer.rcvMessage->len);
3539
3540             // Update E2T instance level metrics
3541             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
3542             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment(rmrMessageBuffer.rcvMessage->len);
3543 #endif
3544             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3545                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_CONTROL_REQ");
3546                 return negativeSix;
3547             }
3548             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Proedure State is %d", message.message.enodbName, CONTROL_PROCEDURE_ONGOING);
3549             setE2ProcedureOngoingStatus(message.message.enodbName, CONTROL_PROCEDURE_ONGOING);
3550             break;
3551         }
3552         case RIC_SERVICE_QUERY: {
3553             if (loglevel >= MDCLOG_DEBUG) {
3554                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_QUERY");
3555             }
3556             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3557                 break;
3558             }
3559 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3560             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment();
3561             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len);
3562
3563             // Update E2T instance level metrics
3564             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment();
3565             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len);
3566 #endif
3567             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3568                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_QUERY");
3569                 return negativeSix;
3570             }
3571             mdclog_write(MDCLOG_DEBUG, "EnbName is %s New Procedure state is %d", message.message.enodbName, E2_SETUP_PROCEDURE_COMPLETED);
3572             setE2ProcedureOngoingStatus(message.message.enodbName, RIC_SERVICE_UPDATE_PROCEDURE_ONGOING);
3573             break;
3574         }
3575         case RIC_SERVICE_UPDATE_ACK: {
3576             if (loglevel >= MDCLOG_DEBUG) {
3577                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_UPDATE_ACK");
3578             }
3579             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3580                 mdclog_write(MDCLOG_ERR, "error in PER_FromXML");
3581                 break;
3582             }
3583 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3584             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
3585             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3586
3587             // Update E2T instance level metrics
3588             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
3589             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3590 #endif
3591             if (loglevel >= MDCLOG_DEBUG) {
3592                 mdclog_write(MDCLOG_DEBUG, "Before sending to CU");
3593             }
3594             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3595                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_ACK");
3596                 return negativeSix;
3597             }
3598             break;
3599         }
3600         case RIC_SERVICE_UPDATE_FAILURE: {
3601             if (loglevel >= MDCLOG_DEBUG) {
3602                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_UPDATE_FAILURE");
3603             }
3604             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3605                 break;
3606             }
3607 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3608             message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
3609             message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3610
3611             // Update E2T instance level metrics
3612             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
3613             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
3614 #endif
3615             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3616                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_FAILURE");
3617                 return negativeSix;
3618             }
3619             break;
3620         }
3621         case RIC_E2_RESET_REQ: {
3622             if (loglevel >= MDCLOG_DEBUG) {
3623                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_RESET_REQ");
3624             }
3625             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3626                 break;
3627             }
3628 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3629             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
3630             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
3631
3632             // Update E2T instance level metrics
3633             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
3634             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
3635 #endif
3636             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
3637                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET");
3638                 return -6;
3639             }
3640             break;
3641         }
3642         case RIC_E2_RESET_RESP: {
3643             if (loglevel >= MDCLOG_DEBUG) {
3644                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_RESET_RESP");
3645             }
3646             if (PER_FromXML(message, rmrMessageBuffer) != numberZero) {
3647                 break;
3648             }
3649 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3650             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
3651             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
3652
3653             // Update E2T instance level metrics
3654             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
3655             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
3656 #endif
3657             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, numberZero, sctpMap) != numberZero) {
3658                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET_RESP");
3659                 return -6;
3660             }
3661             break;
3662         }
3663         case RIC_SCTP_CLEAR_ALL: {
3664             mdclog_write(MDCLOG_INFO, "RIC_SCTP_CLEAR_ALL");
3665             // loop on all keys and close socket and then erase all map.
3666             vector<char *> v;
3667             sctpMap->getKeys(v);
3668             for (auto const &iter : v) { //}; iter != sctpMap.end(); iter++) {
3669                 if (!boost::starts_with((string) (iter), "host:") && !boost::starts_with((string) (iter), "msg:")) {
3670                     auto *peerInfo = (ConnectedCU_t *) sctpMap->find(iter);
3671                     if (peerInfo == nullptr) {
3672                         continue;
3673                     }
3674                     close(peerInfo->fileDescriptor);
3675                     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
3676                     message.message.direction = 'D';
3677                     message.message.time.tv_nsec = ts.tv_nsec;
3678                     message.message.time.tv_sec = ts.tv_sec;
3679
3680                     message.message.asnLength = rmrMessageBuffer.sendMessage->len =
3681                             snprintf((char *)rmrMessageBuffer.sendMessage->payload,
3682                                      256,
3683                                      "%s|RIC_SCTP_CLEAR_ALL",
3684                                      peerInfo->enodbName);
3685                     message.message.asndata = rmrMessageBuffer.sendMessage->payload;
3686                     mdclog_write(MDCLOG_INFO, "%s", message.message.asndata);
3687                     if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != numberZero) {
3688                         mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
3689                     }
3690                     free(peerInfo);
3691                     peerInfo = nullptr;
3692                 }
3693             }
3694
3695             sleep(1);
3696             sctpMap->clear();
3697             break;
3698         }
3699         case E2_TERM_KEEP_ALIVE_REQ: {
3700             // send message back
3701             rmr_bytes2payload(rmrMessageBuffer.sendMessage,
3702                               (unsigned char *)rmrMessageBuffer.ka_message,
3703                               rmrMessageBuffer.ka_message_len);
3704             rmrMessageBuffer.sendMessage->mtype = E2_TERM_KEEP_ALIVE_RESP;
3705             rmrMessageBuffer.sendMessage->state = numberZero;
3706             static unsigned char tx[32];
3707             auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
3708             rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, txLen);
3709 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3710             rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3711 #endif
3712             if (rmrMessageBuffer.sendMessage == nullptr) {
3713                 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3714                 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP RMR message returned NULL");
3715             } else if (rmrMessageBuffer.sendMessage->state != numberZero)  {
3716                 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP, on RMR state = %d ( %s)",
3717                              rmrMessageBuffer.sendMessage->state, translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str());
3718             } else if (loglevel >= MDCLOG_DEBUG) {
3719                 mdclog_write(MDCLOG_DEBUG, "Got Keep Alive Request send : %s", rmrMessageBuffer.ka_message);
3720             }
3721
3722             break;
3723         }
3724         case RIC_HEALTH_CHECK_REQ: {
3725             static int counter = numberZero;
3726             // send message back
3727             rmr_bytes2payload(rmrMessageBuffer.rcvMessage,
3728                               (unsigned char *)"OK",
3729                               numberTwo);
3730             rmrMessageBuffer.rcvMessage->mtype = RIC_HEALTH_CHECK_RESP;
3731             rmrMessageBuffer.rcvMessage->state = numberZero;
3732             static unsigned char tx[32];
3733             auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
3734             rmr_bytes2xact(rmrMessageBuffer.rcvMessage, tx, txLen);
3735             rmrMessageBuffer.rcvMessage = rmr_rts_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage);
3736             //rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3737             if (rmrMessageBuffer.rcvMessage == nullptr) {
3738                 rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3739                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP RMR message returned NULL");
3740             } else if (rmrMessageBuffer.rcvMessage->state != numberZero)  {
3741                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP, on RMR state = %d ( %s)",
3742                              rmrMessageBuffer.rcvMessage->state, translateRmrErrorMessages(rmrMessageBuffer.rcvMessage->state).c_str());
3743             } else if (loglevel >= MDCLOG_DEBUG && (++counter % 100 == numberZero)) {
3744                 mdclog_write(MDCLOG_DEBUG, "Got %d RIC_HEALTH_CHECK_REQ Request send : OK", counter);
3745             }
3746
3747             break;
3748         }
3749
3750         default:
3751             mdclog_write(MDCLOG_ERR, "Message Type : %d is not supported", rmrMessageBuffer.rcvMessage->mtype);
3752             message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
3753             message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
3754             message.message.time.tv_nsec = ts.tv_nsec;
3755             message.message.time.tv_sec = ts.tv_sec;
3756             message.message.messageType = rmrMessageBuffer.rcvMessage->mtype;
3757
3758             buildJsonMessage(message);
3759
3760
3761             return negativeSeven;
3762     }
3763     if (mdclog_level_get() >= MDCLOG_DEBUG) {
3764         mdclog_write(MDCLOG_DEBUG, "EXIT OK from %s", __FUNCTION__);
3765     }
3766     return numberZero;
3767 }
3768
3769 /**
3770  * Send message to the CU that is not expecting for successful or unsuccessful results
3771  * @param messageBuffer
3772  * @param message
3773  * @param failedMsgId
3774  * @param sctpMap
3775  * @return
3776  */
3777 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
3778                            ReportingMessages_t &message,
3779                            int failedMsgId,
3780                            Sctp_Map_t *sctpMap) {
3781     if (mdclog_level_get() >= MDCLOG_DEBUG) {
3782         mdclog_write(MDCLOG_DEBUG, "send message: %d to %s address", message.message.messageType, message.message.enodbName);
3783     }
3784
3785     getRequestMetaData(message, messageBuffer);
3786     if (mdclog_level_get() >= MDCLOG_INFO) {
3787         mdclog_write(MDCLOG_INFO, "send message to %s address", message.message.enodbName);
3788     }
3789
3790     auto rc = sendMessagetoCu(sctpMap, messageBuffer, message, failedMsgId);
3791     return rc;
3792 }
3793
3794 /**
3795  *
3796  * @param sctpMap
3797  * @param messageBuffer
3798  * @param message
3799  * @param failedMesgId
3800  * @return
3801  */
3802 int sendMessagetoCu(Sctp_Map_t *sctpMap,
3803                     RmrMessagesBuffer_t &messageBuffer,
3804                     ReportingMessages_t &message,
3805                     int failedMesgId) {
3806     // get the FD
3807     message.message.messageType = messageBuffer.rcvMessage->mtype;
3808     mdclog_write(MDCLOG_DEBUG, "message.message.messageType is %d", message.message.messageType);
3809     auto rc = sendSctpMsg(message.peerInfo, message, sctpMap);
3810     return rc;
3811 }
3812
3813
3814 /**
3815  *
3816  * @param epoll_fd
3817  * @param peerInfo
3818  * @param events
3819  * @param sctpMap
3820  * @param enodbName
3821  * @param msgType
3822  * @return
3823  */
3824 int addToEpoll(int epoll_fd,
3825                ConnectedCU_t *peerInfo,
3826                uint32_t events,
3827                Sctp_Map_t *sctpMap,
3828                char *enodbName,
3829                int msgType) {
3830     // Add to Epol
3831     struct epoll_event event{};
3832     event.data.ptr = peerInfo;
3833     event.events = events;
3834     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, peerInfo->fileDescriptor, &event) < numberZero) {
3835 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
3836         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3837             mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here), %s, %s %d",
3838                          strerror(errno), __func__, __LINE__);
3839         }
3840         close(peerInfo->fileDescriptor);
3841         if (enodbName != nullptr) {
3842             mdclog_write(MDCLOG_DEBUG, "Erasing Entry from Map for Key %s", enodbName);
3843             removeE2ConnectionEntryFromMap(enodbName);
3844             cleanHashEntry(peerInfo, sctpMap);
3845             if (peerInfo != nullptr)
3846             {
3847                 mdclog_write(MDCLOG_DEBUG, "Not assigned peerInfo = NULL");
3848                 peerInfo = nullptr;
3849             }
3850             char key[MAX_ENODB_NAME_SIZE * numberTwo];
3851             snprintf(key, MAX_ENODB_NAME_SIZE * numberTwo, "msg:%s|%d", enodbName, msgType);
3852             if (mdclog_level_get() >= MDCLOG_DEBUG) {
3853                 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
3854             }
3855             auto tmp = sctpMap->find(key);
3856             if (tmp) {
3857                 free(tmp);
3858                 tmp = nullptr;
3859                 sctpMap->erase(key);
3860             }
3861         } else {
3862             peerInfo->enodbName[numberZero] = numberZero;
3863         }
3864         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here)");
3865         return negativeOne;
3866 #endif
3867     }
3868     return numberZero;
3869 }
3870
3871 /**
3872  *
3873  * @param epoll_fd
3874  * @param peerInfo
3875  * @param events
3876  * @param sctpMap
3877  * @param enodbName
3878  * @param msgType
3879  * @return
3880  */
3881 int modifyToEpoll(int epoll_fd,
3882                   ConnectedCU_t *peerInfo,
3883                   uint32_t events,
3884                   Sctp_Map_t *sctpMap,
3885                   char *enodbName,
3886                   int msgType) {
3887     // Add to Epol
3888     struct epoll_event event{};
3889     event.data.ptr = peerInfo;
3890     event.events = events;
3891     if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, peerInfo->fileDescriptor, &event) < numberZero) {
3892         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3893             mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_MOD (may check not to quit here), %s, %s %d",
3894                          strerror(errno), __func__, __LINE__);
3895         }
3896         close(peerInfo->fileDescriptor);
3897         mdclog_write(MDCLOG_DEBUG, "Erasing Entry from Map for Key %s", enodbName);
3898         removeE2ConnectionEntryFromMap(enodbName);
3899         cleanHashEntry(peerInfo, sctpMap);
3900         if (peerInfo != nullptr)
3901         {
3902             mdclog_write(MDCLOG_DEBUG, "Not assigned peerInfo = NULL");
3903             peerInfo = nullptr;
3904         }
3905         char key[MAX_ENODB_NAME_SIZE * numberTwo];
3906         snprintf(key, MAX_ENODB_NAME_SIZE * numberTwo, "msg:%s|%d", enodbName, msgType);
3907         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3908             mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
3909         }
3910         auto tmp = sctpMap->find(key);
3911         if (tmp) {
3912             free(tmp);
3913             tmp = nullptr;
3914             sctpMap->erase(key);
3915         }
3916         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here)");
3917         return negativeOne;
3918     }
3919     return numberZero;
3920 }
3921
3922
3923 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message) {
3924     buildJsonMessage(message);
3925 #ifndef UNIT_TEST
3926     rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3927 #else
3928     rmrMessageBuffer.sendMessage->state = RMR_ERR_RETRY;
3929 #endif
3930     if (rmrMessageBuffer.sendMessage == nullptr) {
3931         rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3932         mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
3933         return negativeOne;
3934     }
3935
3936     if (rmrMessageBuffer.sendMessage->state != numberZero) {
3937         char meid[RMR_MAX_MEID]{};
3938         if (rmrMessageBuffer.sendMessage->state == RMR_ERR_RETRY) {
3939             usleep(numberFive);
3940             rmrMessageBuffer.sendMessage->state = numberZero;
3941             mdclog_write(MDCLOG_INFO, "RETRY sending Message type %d to Xapp from %s",
3942                          rmrMessageBuffer.sendMessage->mtype,
3943                          rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3944 #ifndef UNIT_TEST
3945             rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3946 #endif
3947             if (rmrMessageBuffer.sendMessage == nullptr) {
3948                 mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
3949                 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3950                 return negativeOne;
3951             } else if (rmrMessageBuffer.sendMessage->state != numberZero) {
3952                 mdclog_write(MDCLOG_ERR,
3953                              "Message state %s while sending request %d to Xapp from %s after retry of 10 microseconds",
3954                              translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
3955                              rmrMessageBuffer.sendMessage->mtype,
3956                              rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3957                 auto rc = rmrMessageBuffer.sendMessage->state;
3958                 return rc;
3959             }
3960         } else {
3961             mdclog_write(MDCLOG_ERR, "Message state %s while sending request %d to Xapp from %s",
3962                          translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
3963                          rmrMessageBuffer.sendMessage->mtype,
3964                          rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3965             return rmrMessageBuffer.sendMessage->state;
3966         }
3967     }
3968     return 0;
3969 }
3970
3971 void buildJsonMessage(ReportingMessages_t &message) {
3972 #ifdef UNIT_TEST
3973     jsonTrace = true;
3974 #endif
3975     if (jsonTrace) {
3976         message.outLen = sizeof(message.base64Data);
3977         base64::encode((const unsigned char *) message.message.asndata,
3978                        (const int) message.message.asnLength,
3979                        message.base64Data,
3980                        message.outLen);
3981         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3982             mdclog_write(MDCLOG_DEBUG, "Tracing: ASN length = %d, base64 message length = %d ",
3983                          (int) message.message.asnLength,
3984                          (int) message.outLen);
3985         }
3986
3987         snprintf(message.buffer, sizeof(message.buffer),
3988                  "{\"header\": {\"ts\": \"%ld.%09ld\","
3989                  "\"ranName\": \"%s\","
3990                  "\"messageType\": %d,"
3991                  "\"direction\": \"%c\"},"
3992                  "\"base64Length\": %d,"
3993                  "\"asnBase64\": \"%s\"}",
3994                  message.message.time.tv_sec,
3995                  message.message.time.tv_nsec,
3996                  message.message.enodbName,
3997                  message.message.messageType,
3998                  message.message.direction,
3999                  (int) message.outLen,
4000                  message.base64Data);
4001         static src::logger_mt &lg = my_logger::get();
4002
4003         BOOST_LOG(lg) << message.buffer;
4004     }
4005 }
4006
4007
4008 /**
4009  * take RMR error code to string
4010  * @param state
4011  * @return
4012  */
4013 string translateRmrErrorMessages(int state) {
4014     string str = {};
4015     switch (state) {
4016         case RMR_OK:
4017             str = "RMR_OK - state is good";
4018             break;
4019         case RMR_ERR_BADARG:
4020             str = "RMR_ERR_BADARG - argument passed to function was unusable";
4021             break;
4022         case RMR_ERR_NOENDPT:
4023             str = "RMR_ERR_NOENDPT - send//call could not find an endpoint based on msg type";
4024             break;
4025         case RMR_ERR_EMPTY:
4026             str = "RMR_ERR_EMPTY - msg received had no payload; attempt to send an empty message";
4027             break;
4028         case RMR_ERR_NOHDR:
4029             str = "RMR_ERR_NOHDR - message didn't contain a valid header";
4030             break;
4031         case RMR_ERR_SENDFAILED:
4032             str = "RMR_ERR_SENDFAILED - send failed; errno has nano reason";
4033             break;
4034         case RMR_ERR_CALLFAILED:
4035             str = "RMR_ERR_CALLFAILED - unable to send call() message";
4036             break;
4037         case RMR_ERR_NOWHOPEN:
4038             str = "RMR_ERR_NOWHOPEN - no wormholes are open";
4039             break;
4040         case RMR_ERR_WHID:
4041             str = "RMR_ERR_WHID - wormhole id was invalid";
4042             break;
4043         case RMR_ERR_OVERFLOW:
4044             str = "RMR_ERR_OVERFLOW - operation would have busted through a buffer/field size";
4045             break;
4046         case RMR_ERR_RETRY:
4047             str = "RMR_ERR_RETRY - request (send/call/rts) failed, but caller should retry (EAGAIN for wrappers)";
4048             break;
4049         case RMR_ERR_RCVFAILED:
4050             str = "RMR_ERR_RCVFAILED - receive failed (hard error)";
4051             break;
4052         case RMR_ERR_TIMEOUT:
4053             str = "RMR_ERR_TIMEOUT - message processing call timed out";
4054             break;
4055         case RMR_ERR_UNSET:
4056             str = "RMR_ERR_UNSET - the message hasn't been populated with a transport buffer";
4057             break;
4058         case RMR_ERR_TRUNC:
4059             str = "RMR_ERR_TRUNC - received message likely truncated";
4060             break;
4061         case RMR_ERR_INITFAILED:
4062             str = "RMR_ERR_INITFAILED - initialisation of something (probably message) failed";
4063             break;
4064         case RMR_ERR_NOTSUPP:
4065             str = "RMR_ERR_NOTSUPP - the request is not supported, or RMr was not initialised for the request";
4066             break;
4067         default:
4068             char buf[128]{};
4069             snprintf(buf, sizeof buf, "UNDOCUMENTED RMR_ERR : %d", state);
4070             str = buf;
4071             break;
4072     }
4073     return str;
4074 }
4075 int fetchStreamId(ConnectedCU_t *peerInfo, ReportingMessages_t &message)
4076 {
4077     auto loglevel = mdclog_level_get();
4078     int streamId = INVALID_STREAM_ID;
4079     if(message.peerInfo->isSingleStream != false)
4080     {
4081         streamId = message.peerInfo->singleStreamId;
4082         if (loglevel >= MDCLOG_DEBUG) {
4083         mdclog_write(MDCLOG_DEBUG, "Send SCTP message for SINGLE_STREAM streamId %d , Messeage Type %d ,%s",
4084                      streamId,message.message.messageType,  __FUNCTION__);
4085         }
4086         return streamId;
4087     }
4088     int msgType = message.message.messageType;
4089     switch (msgType){
4090         case RIC_E2_RESET_REQ:
4091         case RIC_E2_RESET_RESP:
4092         case RIC_E2_SETUP_RESP:
4093         case RIC_E2_SETUP_FAILURE:
4094         case RIC_ERROR_INDICATION:
4095         case RIC_SERVICE_QUERY:
4096         case RIC_SERVICE_UPDATE_ACK:
4097         case RIC_SERVICE_UPDATE_FAILURE:
4098             streamId = numberZero;
4099             break;
4100         case RIC_SUB_REQ:
4101         case RIC_SUB_DEL_REQ:
4102         case RIC_CONTROL_REQ:
4103             streamId = numberOne;
4104             break;
4105         default:
4106             streamId = numberZero;
4107             break;
4108     }
4109     if (loglevel >= MDCLOG_DEBUG) {
4110         mdclog_write(MDCLOG_DEBUG, "Send SCTP message for streamId %d Messeage Type %d, %s",
4111                      streamId, message.message.messageType, __FUNCTION__);
4112     }
4113     return streamId;
4114 }