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
  36 using namespace std;
  37 //using namespace std::placeholders;
  38 using namespace boost::filesystem;
  39 using namespace prometheus;
  40
  41
  42 //#ifdef __cplusplus
  43 //extern "C"
  44 //{
  45 //#endif
  46
  47 // need to expose without the include of gcov
  48 extern "C" void __gcov_flush(void);
  49 #define LOG_FILE_CONFIG_MAP "CONFIG_MAP_NAME"
  50
  51 static void catch_function(int signal) {
  52     __gcov_flush();
  53     exit(signal);
  54 }
  55
  56
  57 BOOST_LOG_INLINE_GLOBAL_LOGGER_DEFAULT(my_logger, src::logger_mt)
  58
  59 boost::shared_ptr<sinks::synchronous_sink<sinks::text_file_backend>> boostLogger;
  60 double cpuClock = 0.0;
  61 bool jsonTrace = false;
  62
  63 char* getinterfaceip()
  64 {
  65    char hostname[256];
  66    char *IP;
  67    struct hostent *host_entry;
  68    int retVal;
  69    retVal = gethostname(hostname, sizeof(hostname));
  70    if ( retVal == -1 )
  71        return NULL;
  72    host_entry = gethostbyname(hostname);
  73    if ( host_entry == NULL )
  74        return NULL;
  75    IP = inet_ntoa(*((struct in_addr*) host_entry->h_addr_list[0]));
  76    return IP;
  77 }
  78
  79
  80 static int enable_log_change_notify(const char* fileName)
  81 {
  82     int ret = -1;
  83     struct stat fileInfo;
  84     if ( lstat(fileName,&fileInfo) == 0 )
  85     {
  86         ret = register_log_change_notify(fileName);
  87     }
  88     return ret;
  89 }
  90
  91
  92 static int register_log_change_notify(const char *fileName)
  93 {
  94     pthread_attr_t cb_attr;
  95     pthread_t tid;
  96     pthread_attr_init(&cb_attr);
  97     pthread_attr_setdetachstate(&cb_attr,PTHREAD_CREATE_DETACHED);
  98     return pthread_create(&tid, &cb_attr,&monitor_loglevel_change_handler,(void *)fileName);
  99 }
 100
 101
 102 static void * monitor_loglevel_change_handler(void* arg)
 103 {
 104     char *fileName = (char*) arg;
 105     int ifd;                   // the inotify file des
 106     int wfd;                   // the watched file des
 107     ssize_t n = 0;
 108     char rbuf[4096];           // large read buffer as the event is var len
 109     fd_set fds;
 110     int res = 0;
 111     struct timeval timeout;
 112     char* dname=NULL;          // directory name
 113     char* bname = NULL;        // basename
 114     char* tok=NULL;
 115     char* log_level=NULL;
 116
 117     dname = strdup( fileName); // defrock the file name into dir and basename
 118     if( (tok = strrchr( dname, '/' )) != NULL ) {
 119         *tok = '\0';
 120         bname = strdup( tok+1 );
 121     }
 122
 123
 124     ifd = inotify_init1( 0 ); // initialise watcher setting blocking read (no option)
 125     if( ifd < 0 ) {
 126         fprintf( stderr, "### ERR ### unable to initialise file watch %s\n", strerror( errno ) );
 127     } else {
 128         wfd = inotify_add_watch( ifd, dname, IN_MOVED_TO | IN_CLOSE_WRITE ); // we only care about close write changes
 129
 130         if( wfd < 0 ) {
 131             fprintf( stderr, "### ERR ### unable to add watch on config file %s: %s\n", fileName, strerror( errno ) );
 132         } else {
 133
 134
 135             memset( &timeout, 0, sizeof(timeout) );
 136             while( 1 ) {
 137                 FD_ZERO (&fds);
 138                 FD_SET (ifd, &fds);
 139                 timeout.tv_sec=1;
 140                 res = select (ifd + 1, &fds, NULL, NULL, &timeout);
 141                 if(res)
 142                 {
 143                     n = read( ifd, rbuf, sizeof( rbuf ) ); // read the event
 144                     if( n < 0  ) {
 145 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 146                         if( errno == EAGAIN ) {
 147                         } else {
 148                             printf(  "### CRIT ### config listener read err: %s\n", strerror( errno ) );
 149                         }
 150                         continue;
 151 #endif
 152                     }
 153
 154                     //Retrieving Log Level from configmap by parsing configmap file
 155                     log_level = parse_file(fileName);
 156                     update_mdc_log_level_severity(log_level); //setting log level
 157                     free(log_level);
 158                 }
 159             }
 160             inotify_rm_watch(ifd,wfd);
 161         }
 162         close(ifd);
 163     }
 164     free(bname);
 165     free(dname);
 166
 167     pthread_exit(NULL);
 168 }
 169
 170 void  update_mdc_log_level_severity(char* log_level)
 171 {
 172     mdclog_severity_t level = MDCLOG_ERR;
 173
 174     if(log_level == NULL)
 175     {
 176         printf("### ERR ### Invalid Log-Level Configuration in ConfigMap, Default Log-Level Applied:   %d\n",level);
 177     }
 178     else if(strcasecmp(log_level,"1")==0)
 179     {
 180         level = MDCLOG_ERR;
 181     }
 182     else if(strcasecmp(log_level,"2")==0)
 183     {
 184         level = MDCLOG_WARN;
 185     }
 186     else if(strcasecmp(log_level,"3")==0)
 187     {
 188         level = MDCLOG_INFO;
 189     }
 190     else if(strcasecmp(log_level,"4")==0)
 191     {
 192         level = MDCLOG_DEBUG;
 193     }
 194
 195     mdclog_level_set(level);
 196 }
 197 static char* parse_file(char* filename)
 198 {
 199     char *token=NULL;
 200     char *search = ": ";
 201     char *string_match = "log-level";
 202     bool found = false;
 203     FILE *file = fopen ( filename, "r" );
 204     if ( file != NULL )
 205     {
 206         char line [ 128 ];
 207         while ( fgets ( line, sizeof line, file ) != NULL )
 208         {
 209             token = strtok(line, search);
 210             if(strcmp(token,string_match)==0)
 211             {
 212                 found = true;
 213                 token = strtok(NULL, search);
 214                 token = strtok(token, "\n");//removing newline if any
 215                 break;
 216             }
 217         }
 218         fclose ( file );
 219      }
 220      if(found)
 221          return(strdup(token));
 222      else
 223          return(NULL);
 224 }
 225
 226 char *read_env_param(const char*envkey)
 227 {
 228     if(envkey)
 229     {
 230         char *value = getenv(envkey);
 231         if(value)
 232             return strdup(value);
 233     }
 234     return NULL;
 235 }
 236
 237 void dynamic_log_level_change()
 238 {
 239     char *logFile_Name = read_env_param(LOG_FILE_CONFIG_MAP);
 240     char* log_level_init=NULL;
 241     if(logFile_Name)
 242     {
 243         log_level_init = parse_file(logFile_Name);
 244         update_mdc_log_level_severity(log_level_init); //setting log level
 245         free(log_level_init);
 246
 247     }
 248     enable_log_change_notify(logFile_Name);
 249     free(logFile_Name);
 250
 251 }
 252
 253 void init_log() {
 254     int log_change_monitor = 0;
 255     mdclog_attr_t *attr;
 256     mdclog_attr_init(&attr);
 257     mdclog_attr_set_ident(attr, "E2Terminator");
 258     mdclog_init(attr);
 259     if(mdclog_format_initialize(log_change_monitor)!=0)
 260         mdclog_write(MDCLOG_ERR, "Failed to intialize MDC log format !!!");
 261     dynamic_log_level_change();
 262     mdclog_attr_destroy(attr);
 263 }
 264 auto start_time = std::chrono::high_resolution_clock::now();
 265 typedef std::chrono::duration<double, std::ratio<1,1>> seconds_t;
 266
 267 double age() {
 268     return seconds_t(std::chrono::high_resolution_clock::now() - start_time).count();
 269 }
 270
 271 double approx_CPU_MHz(unsigned sleepTime) {
 272     using namespace std::chrono_literals;
 273     uint32_t aux = 0;
 274     uint64_t cycles_start = rdtscp(aux);
 275     double time_start = age();
 276     std::this_thread::sleep_for(sleepTime * 1ms);
 277     uint64_t elapsed_cycles = rdtscp(aux) - cycles_start;
 278     double elapsed_time = age() - time_start;
 279     return elapsed_cycles / elapsed_time;
 280 }
 281
 282 //std::atomic<int64_t> rmrCounter{0};
 283 std::atomic<int64_t> num_of_messages{0};
 284 std::atomic<int64_t> num_of_XAPP_messages{0};
 285 static long transactionCounter = 0;
 286
 287 int buildListeningPort(sctp_params_t &sctpParams) {
 288     sctpParams.listenFD = socket(AF_INET6, SOCK_STREAM, IPPROTO_SCTP);
 289     if (sctpParams.listenFD <= 0) {
 290 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 291         mdclog_write(MDCLOG_ERR, "Error Opening socket, %s", strerror(errno));
 292         return -1;
 293 #endif
 294     }
 295
 296     struct sockaddr_in6 serverAddress {};
 297     serverAddress.sin6_family = AF_INET6;
 298     serverAddress.sin6_addr   = in6addr_any;
 299     serverAddress.sin6_port = htons(sctpParams.sctpPort);
 300     if (bind(sctpParams.listenFD, (SA *)&serverAddress, sizeof(serverAddress)) < 0 ) {
 301 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 302         mdclog_write(MDCLOG_ERR, "Error binding port %d. %s", sctpParams.sctpPort, strerror(errno));
 303         return -1;
 304 #endif
 305     }
 306     if (setSocketNoBlocking(sctpParams.listenFD) == -1) {
 307         //mdclog_write(MDCLOG_ERR, "Error binding. %s", strerror(errno));
 308         return -1;
 309     }
 310     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 311         struct sockaddr_in6 clientAddress {};
 312         socklen_t len = sizeof(clientAddress);
 313         getsockname(sctpParams.listenFD, (SA *)&clientAddress, &len);
 314         char buff[1024] {};
 315         inet_ntop(AF_INET6, &clientAddress.sin6_addr, buff, sizeof(buff));
 316         mdclog_write(MDCLOG_DEBUG, "My address: %s, port %d\n", buff, htons(clientAddress.sin6_port));
 317     }
 318
 319     if (listen(sctpParams.listenFD, SOMAXCONN) < 0) {
 320 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 321         mdclog_write(MDCLOG_ERR, "Error listening. %s\n", strerror(errno));
 322         return -1;
 323 #endif
 324     }
 325     struct epoll_event event {};
 326     event.events = EPOLLIN | EPOLLET;
 327     event.data.fd = sctpParams.listenFD;
 328
 329     // add listening port to epoll
 330     if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.listenFD, &event)) {
 331 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 332         printf("Failed to add descriptor to epoll\n");
 333         mdclog_write(MDCLOG_ERR, "Failed to add descriptor to epoll. %s\n", strerror(errno));
 334         return -1;
 335 #endif
 336     }
 337
 338     return 0;
 339 }
 340
 341 int buildConfiguration(sctp_params_t &sctpParams) {
 342     path p = (sctpParams.configFilePath + "/" + sctpParams.configFileName).c_str();
 343     if (exists(p)) {
 344         const int size = 2048;
 345         auto fileSize = file_size(p);
 346         if (fileSize > size) {
 347 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 348             mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
 349             return -1;
 350 #endif
 351         }
 352     } else {
 353 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 354         mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
 355         return -1;
 356 #endif
 357     }
 358
 359     ReadConfigFile conf;
 360     if (conf.openConfigFile(p.string()) == -1) {
 361 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 362         mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
 363                      p.string().c_str(), strerror(errno));
 364         return -1;
 365 #endif
 366     }
 367     int rmrPort = conf.getIntValue("nano");
 368     if (rmrPort == -1) {
 369 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 370         mdclog_write(MDCLOG_ERR, "illegal RMR port ");
 371         return -1;
 372 #endif
 373     }
 374     sctpParams.rmrPort = (uint16_t)rmrPort;
 375     snprintf(sctpParams.rmrAddress, sizeof(sctpParams.rmrAddress), "%d", (int) (sctpParams.rmrPort));
 376     auto tmpStr = conf.getStringValue("volume");
 377     if (tmpStr.length() == 0) {
 378 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 379         mdclog_write(MDCLOG_ERR, "illegal volume.");
 380         return -1;
 381 #endif
 382     }
 383
 384     char tmpLogFilespec[VOLUME_URL_SIZE];
 385     tmpLogFilespec[0] = 0;
 386     sctpParams.volume[0] = 0;
 387     snprintf(sctpParams.volume, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
 388     // copy the name to temp file as well
 389     snprintf(tmpLogFilespec, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
 390
 391
 392     // define the file name in the tmp directory under the volume
 393     strcat(tmpLogFilespec,"/tmp/E2Term_%Y-%m-%d_%H-%M-%S.%N.tmpStr");
 394
 395     sctpParams.myIP = conf.getStringValue("local-ip");
 396     if (sctpParams.myIP.length() == 0) {
 397 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 398         mdclog_write(MDCLOG_ERR, "illegal local-ip.");
 399         return -1;
 400 #endif
 401     }
 402
 403     int sctpPort = conf.getIntValue("sctp-port");
 404     if (sctpPort == -1) {
 405 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 406         mdclog_write(MDCLOG_ERR, "illegal SCTP port ");
 407         return -1;
 408 #endif
 409     }
 410     sctpParams.sctpPort = (uint16_t)sctpPort;
 411
 412     sctpParams.fqdn = conf.getStringValue("external-fqdn");
 413     if (sctpParams.fqdn.length() == 0) {
 414 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 415         mdclog_write(MDCLOG_ERR, "illegal external-fqdn");
 416         return -1;
 417 #endif
 418     }
 419
 420     std::string pod = conf.getStringValue("pod_name");
 421 #ifndef UNIT_TEST
 422     if (pod.length() == 0) {
 423         mdclog_write(MDCLOG_ERR, "illegal pod_name in config file");
 424         return -1;
 425     }
 426     auto *podName = getenv(pod.c_str());
 427     if (podName == nullptr) {
 428         mdclog_write(MDCLOG_ERR, "illegal pod_name or environment variable not exists : %s", pod.c_str());
 429         return -1;
 430
 431     } else {
 432         sctpParams.podName.assign(podName);
 433         if (sctpParams.podName.length() == 0) {
 434             mdclog_write(MDCLOG_ERR, "illegal pod_name");
 435             return -1;
 436         }
 437     }
 438 #endif
 439     tmpStr = conf.getStringValue("trace");
 440     transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
 441     if ((tmpStr.compare("start")) == 0) {
 442 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 443         mdclog_write(MDCLOG_INFO, "Trace set to: start");
 444         sctpParams.trace = true;
 445 #endif
 446     } else if ((tmpStr.compare("stop")) == 0) {
 447         mdclog_write(MDCLOG_INFO, "Trace set to: stop");
 448         sctpParams.trace = false;
 449     } else {
 450 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 451         mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
 452         sctpParams.trace = false;
 453 #endif
 454     }
 455     jsonTrace = sctpParams.trace;
 456
 457     sctpParams.epollTimeOut = -1;
 458
 459     tmpStr = conf.getStringValue("prometheusPort");
 460     if (tmpStr.length() != 0) {
 461         sctpParams.prometheusPort = tmpStr;
 462     }
 463
 464     sctpParams.ka_message_length = snprintf(sctpParams.ka_message, KA_MESSAGE_SIZE, "{\"address\": \"%s:%d\","
 465                                                                                     "\"fqdn\": \"%s\","
 466                                                                                     "\"pod_name\": \"%s\"}",
 467                                             (const char *)sctpParams.myIP.c_str(),
 468                                             sctpParams.rmrPort,
 469                                             sctpParams.fqdn.c_str(),
 470                                             sctpParams.podName.c_str());
 471
 472     if (mdclog_level_get() >= MDCLOG_INFO) {
 473         mdclog_write(MDCLOG_DEBUG,"RMR Port: %s", to_string(sctpParams.rmrPort).c_str());
 474         mdclog_write(MDCLOG_DEBUG,"LogLevel: %s", to_string(sctpParams.logLevel).c_str());
 475         mdclog_write(MDCLOG_DEBUG,"volume: %s", sctpParams.volume);
 476         mdclog_write(MDCLOG_DEBUG,"tmpLogFilespec: %s", tmpLogFilespec);
 477         mdclog_write(MDCLOG_DEBUG,"my ip: %s", sctpParams.myIP.c_str());
 478         mdclog_write(MDCLOG_DEBUG,"pod name: %s", sctpParams.podName.c_str());
 479
 480         mdclog_write(MDCLOG_INFO, "running parameters for instance : %s", sctpParams.ka_message);
 481     }
 482
 483     // Files written to the current working directory
 484     boostLogger = logging::add_file_log(
 485             keywords::file_name = tmpLogFilespec, // to temp directory
 486             keywords::rotation_size = 10 * 1024 * 1024,
 487             keywords::time_based_rotation = sinks::file::rotation_at_time_interval(posix_time::hours(1)),
 488             keywords::format = "%Message%"
 489             //keywords::format = "[%TimeStamp%]: %Message%" // use each tmpStr with time stamp
 490     );
 491
 492     // Setup a destination folder for collecting rotated (closed) files --since the same volume can use rename()
 493     boostLogger->locked_backend()->set_file_collector(sinks::file::make_collector(
 494             keywords::target = sctpParams.volume
 495     ));
 496
 497     // Upon restart, scan the directory for files matching the file_name pattern
 498     boostLogger->locked_backend()->scan_for_files();
 499
 500     // Enable auto-flushing after each tmpStr record written
 501     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 502         boostLogger->locked_backend()->auto_flush(true);
 503     }
 504
 505     return 0;
 506 }
 507
 508 void startPrometheus(sctp_params_t &sctpParams) {
 509     auto podName = std::getenv("POD_NAME");
 510     string metric = "E2TBeta";
 511     if (strstr(podName, "alpha") != NULL) {
 512         metric = "E2TAlpha";
 513     }
 514     //Get eth0 interface IP
 515     char* host = getinterfaceip();
 516     string hostip = host;
 517
 518     sctpParams.prometheusFamily = &BuildCounter()
 519             .Name(metric.c_str())
 520             .Help("E2T instance metrics")
 521             .Labels({{"POD_NAME", sctpParams.podName}})
 522             .Register(*sctpParams.prometheusRegistry);
 523
 524     // Build E2T instance level metrics
 525     buildE2TPrometheusCounters(sctpParams);
 526
 527     string prometheusPath;
 528     if (hostip.empty())
 529         prometheusPath = sctpParams.prometheusPort + "," + "[::]:" + sctpParams.prometheusPort;
 530     else
 531         prometheusPath = hostip + ":" + sctpParams.prometheusPort;
 532
 533     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 534         mdclog_write(MDCLOG_DEBUG, "Start Prometheus Pull mode on %s", prometheusPath.c_str());
 535     }
 536     sctpParams.prometheusExposer = new Exposer(prometheusPath, 1);
 537     sctpParams.prometheusExposer->RegisterCollectable(sctpParams.prometheusRegistry);
 538 }
 539 #ifndef UNIT_TEST
 540
 541 int main(const int argc, char **argv) {
 542     sctp_params_t sctpParams;
 543     {
 544         std::random_device device{};
 545         std::mt19937 generator(device());
 546         std::uniform_int_distribution<long> distribution(1, (long) 1e12);
 547         transactionCounter = distribution(generator);
 548     }
 549
 550 //    uint64_t st = 0;
 551 //    uint32_t aux1 = 0;
 552 //   st = rdtscp(aux1);
 553
 554     unsigned num_cpus = std::thread::hardware_concurrency();
 555     init_log();
 556     if (std::signal(SIGINT, catch_function) == SIG_ERR) {
 557         mdclog_write(MDCLOG_ERR, "Error initializing SIGINT");
 558         exit(1);
 559     }
 560     if (std::signal(SIGABRT, catch_function)== SIG_ERR) {
 561         mdclog_write(MDCLOG_ERR, "Error initializing SIGABRT");
 562         exit(1);
 563     }
 564     if (std::signal(SIGTERM, catch_function)== SIG_ERR) {
 565         mdclog_write(MDCLOG_ERR, "Error initializing SIGTERM");
 566         exit(1);
 567     }
 568
 569     cpuClock = approx_CPU_MHz(100);
 570
 571     mdclog_write(MDCLOG_DEBUG, "CPU speed %11.11f", cpuClock);
 572
 573     auto result = parse(argc, argv, sctpParams);
 574
 575     if (buildConfiguration(sctpParams) != 0) {
 576         exit(-1);
 577     }
 578
 579     //auto registry = std::make_shared<Registry>();
 580     sctpParams.prometheusRegistry = std::make_shared<Registry>();
 581
 582     //sctpParams.prometheusFamily = new Family<Counter>("E2T", "E2T message counter", {{"E", sctpParams.podName}});
 583
 584     startPrometheus(sctpParams);
 585
 586     // start epoll
 587     sctpParams.epoll_fd = epoll_create1(0);
 588     if (sctpParams.epoll_fd == -1) {
 589         mdclog_write(MDCLOG_ERR, "failed to open epoll descriptor");
 590         exit(-1);
 591     }
 592     getRmrContext(sctpParams);
 593     if (sctpParams.rmrCtx == nullptr) {
 594         close(sctpParams.epoll_fd);
 595         exit(-1);
 596     }
 597
 598     if (buildInotify(sctpParams) == -1) {
 599         close(sctpParams.rmrListenFd);
 600         rmr_close(sctpParams.rmrCtx);
 601         close(sctpParams.epoll_fd);
 602         exit(-1);
 603     }
 604
 605     if (buildListeningPort(sctpParams) != 0) {
 606         close(sctpParams.rmrListenFd);
 607         rmr_close(sctpParams.rmrCtx);
 608         close(sctpParams.epoll_fd);
 609         exit(-1);
 610     }
 611
 612     sctpParams.sctpMap = new mapWrapper();
 613
 614     std::vector<std::thread> threads(num_cpus);
 615 //    std::vector<std::thread> threads;
 616
 617     num_cpus = 3;
 618     for (unsigned int i = 0; i < num_cpus; i++) {
 619         threads[i] = std::thread(listener, &sctpParams);
 620
 621         cpu_set_t cpuset;
 622         CPU_ZERO(&cpuset);
 623         CPU_SET(i, &cpuset);
 624         int rc = pthread_setaffinity_np(threads[i].native_handle(), sizeof(cpu_set_t), &cpuset);
 625         if (rc != 0) {
 626             mdclog_write(MDCLOG_ERR, "Error calling pthread_setaffinity_np: %d", rc);
 627         }
 628     }
 629
 630
 631     //loop over term_init until first message from xApp
 632     handleTermInit(sctpParams);
 633
 634     for (auto &t : threads) {
 635         t.join();
 636     }
 637
 638     return 0;
 639 }
 640 #endif
 641 void handleTermInit(sctp_params_t &sctpParams) {
 642     sendTermInit(sctpParams);
 643     //send to e2 manager init of e2 term
 644     //E2_TERM_INIT
 645
 646     int count = 0;
 647     while (true) {
 648         auto xappMessages = num_of_XAPP_messages.load(std::memory_order_acquire);
 649         if (xappMessages > 0) {
 650             if (mdclog_level_get() >=  MDCLOG_INFO) {
 651                 mdclog_write(MDCLOG_INFO, "Got a message from some application, stop sending E2_TERM_INIT");
 652             }
 653             return;
 654         }
 655         usleep(100000);
 656         count++;
 657         if (count % 1000 == 0) {
 658             mdclog_write(MDCLOG_ERR, "GOT No messages from any xApp");
 659             sendTermInit(sctpParams);
 660         }
 661     }
 662 }
 663
 664 void sendTermInit(sctp_params_t &sctpParams) {
 665     rmr_mbuf_t *msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length);
 666     auto count = 0;
 667     while (true) {
 668         msg->mtype = E2_TERM_INIT;
 669         msg->state = 0;
 670         rmr_bytes2payload(msg, (unsigned char *)sctpParams.ka_message, sctpParams.ka_message_length);
 671         static unsigned char tx[32];
 672         auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
 673         rmr_bytes2xact(msg, tx, txLen);
 674         msg = rmr_send_msg(sctpParams.rmrCtx, msg);
 675         if (msg == nullptr) {
 676             msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length);
 677         } else if (msg->state == 0) {
 678             rmr_free_msg(msg);
 679             if (mdclog_level_get() >=  MDCLOG_INFO) {
 680                 mdclog_write(MDCLOG_INFO, "E2_TERM_INIT successfully sent ");
 681             }
 682             return;
 683         } else {
 684             if (count % 100 == 0) {
 685                 mdclog_write(MDCLOG_ERR, "Error sending E2_TERM_INIT cause : %s ", translateRmrErrorMessages(msg->state).c_str());
 686             }
 687             sleep(1);
 688         }
 689         count++;
 690     }
 691 }
 692
 693 /**
 694  *
 695  * @param argc
 696  * @param argv
 697  * @param sctpParams
 698  * @return
 699  */
 700 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &sctpParams) {
 701     cxxopts::Options options(argv[0], "e2 term help");
 702     options.positional_help("[optional args]").show_positional_help();
 703     options.allow_unrecognised_options().add_options()
 704             ("p,path", "config file path", cxxopts::value<std::string>(sctpParams.configFilePath)->default_value("config"))
 705             ("f,file", "config file name", cxxopts::value<std::string>(sctpParams.configFileName)->default_value("config.conf"))
 706             ("h,help", "Print help");
 707
 708     auto result = options.parse(argc, (const char **&)argv);
 709
 710     if (result.count("help")) {
 711         std::cout << options.help({""}) << std::endl;
 712         exit(0);
 713     }
 714     return result;
 715 }
 716
 717 /**
 718  *
 719  * @param sctpParams
 720  * @return -1 failed 0 success
 721  */
 722 int buildInotify(sctp_params_t &sctpParams) {
 723     sctpParams.inotifyFD = inotify_init1(IN_NONBLOCK);
 724     if (sctpParams.inotifyFD == -1) {
 725         mdclog_write(MDCLOG_ERR, "Failed to init inotify (inotify_init1) %s", strerror(errno));
 726         return -1;
 727     }
 728
 729     sctpParams.inotifyWD = inotify_add_watch(sctpParams.inotifyFD,
 730                                              (const char *)sctpParams.configFilePath.c_str(),
 731                                              (unsigned)IN_OPEN | (unsigned)IN_CLOSE_WRITE | (unsigned)IN_CLOSE_NOWRITE); //IN_CLOSE = (IN_CLOSE_WRITE | IN_CLOSE_NOWRITE)
 732     if (sctpParams.inotifyWD == -1) {
 733         mdclog_write(MDCLOG_ERR, "Failed to add directory : %s to  inotify (inotify_add_watch) %s",
 734                      sctpParams.configFilePath.c_str(),
 735                      strerror(errno));
 736         close(sctpParams.inotifyFD);
 737         return -1;
 738     }
 739
 740     struct epoll_event event{};
 741     event.events = (EPOLLIN);
 742     event.data.fd = sctpParams.inotifyFD;
 743     // add listening RMR FD to epoll
 744     if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.inotifyFD, &event)) {
 745         mdclog_write(MDCLOG_ERR, "Failed to add inotify FD to epoll");
 746         close(sctpParams.inotifyFD);
 747         return -1;
 748     }
 749     return 0;
 750 }
 751
 752 /**
 753  *
 754  * @param args
 755  * @return
 756  */
 757 void listener(sctp_params_t *params) {
 758     int num_of_SCTP_messages = 0;
 759     auto totalTime = 0.0;
 760     std::thread::id this_id = std::this_thread::get_id();
 761     //save cout
 762     auto pod_name = std::getenv("POD_NAME");
 763     auto container_name = std::getenv("CONTAINER_NAME");
 764     auto service_name = std::getenv("SERVICE_NAME");
 765     auto host_name = std::getenv("HOST_NAME");
 766     auto system_name = std::getenv("SYSTEM_NAME");
 767     auto pid = std::to_string(getpid()).c_str();
 768     streambuf *oldCout = cout.rdbuf();
 769     ostringstream memCout;
 770     // create new cout
 771     cout.rdbuf(memCout.rdbuf());
 772     cout << this_id;
 773     //return to the normal cout
 774     cout.rdbuf(oldCout);
 775
 776     char tid[32];
 777     memcpy(tid, memCout.str().c_str(), memCout.str().length() < 32 ? memCout.str().length() : 31);
 778     tid[memCout.str().length()] = 0;
 779     mdclog_mdc_add("SYSTEM_NAME", system_name);
 780     mdclog_mdc_add("HOST_NAME", host_name);
 781     mdclog_mdc_add("SERVICE_NAME", service_name);
 782     mdclog_mdc_add("CONTAINER_NAME", container_name);
 783     mdclog_mdc_add("POD_NAME", pod_name);
 784     mdclog_mdc_add("PID", pid);
 785
 786     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 787         mdclog_write(MDCLOG_DEBUG, "started thread number %s", tid);
 788     }
 789
 790     RmrMessagesBuffer_t rmrMessageBuffer{};
 791     //create and init RMR
 792     rmrMessageBuffer.rmrCtx = params->rmrCtx;
 793
 794     auto *events = (struct epoll_event *) calloc(MAXEVENTS, sizeof(struct epoll_event));
 795     struct timespec end{0, 0};
 796     struct timespec start{0, 0};
 797
 798     rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 799     rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 800
 801     memcpy(rmrMessageBuffer.ka_message, params->ka_message, params->ka_message_length);
 802     rmrMessageBuffer.ka_message_len = params->ka_message_length;
 803     rmrMessageBuffer.ka_message[rmrMessageBuffer.ka_message_len] = 0;
 804
 805     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 806         mdclog_write(MDCLOG_DEBUG, "keep alive message is : %s", rmrMessageBuffer.ka_message);
 807     }
 808
 809     ReportingMessages_t message {};
 810
 811 //    for (int i = 0; i < MAX_RMR_BUFF_ARRAY; i++) {
 812 //        rmrMessageBuffer.rcvBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 813 //        rmrMessageBuffer.sendBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
 814 //    }
 815
 816     while (true) {
 817         if (mdclog_level_get() >= MDCLOG_DEBUG) {
 818             mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait. Timeout = %d", params->epollTimeOut);
 819         }
 820 #ifndef UNIT_TEST
 821         auto numOfEvents = epoll_wait(params->epoll_fd, events, MAXEVENTS, params->epollTimeOut);
 822 #else
 823         auto numOfEvents = 1;
 824 #endif
 825         if (numOfEvents == 0) { // time out
 826 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 827             if (mdclog_level_get() >= MDCLOG_DEBUG) {
 828                 mdclog_write(MDCLOG_DEBUG, "got epoll timeout");
 829             }
 830             continue;
 831         } else if (numOfEvents < 0) {
 832             if (errno == EINTR) {
 833                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
 834                     mdclog_write(MDCLOG_DEBUG, "got EINTR : %s", strerror(errno));
 835                 }
 836                 continue;
 837             }
 838             mdclog_write(MDCLOG_ERR, "Epoll wait failed, errno = %s", strerror(errno));
 839             if(events)
 840             {
 841                 free(events);
 842             }
 843             return;
 844 #endif
 845         }
 846         for (auto i = 0; i < numOfEvents; i++) {
 847             if (mdclog_level_get() >= MDCLOG_DEBUG) {
 848                 mdclog_write(MDCLOG_DEBUG, "handling epoll event %d out of %d", i + 1, numOfEvents);
 849             }
 850             clock_gettime(CLOCK_MONOTONIC, &message.message.time);
 851             start.tv_sec = message.message.time.tv_sec;
 852             start.tv_nsec = message.message.time.tv_nsec;
 853
 854
 855             if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP)) {
 856                 handlepoll_error(events[i], message, rmrMessageBuffer, params);
 857             } else if (events[i].events & EPOLLOUT) {
 858                 handleEinprogressMessages(events[i], message, rmrMessageBuffer, params);
 859             } else if (params->listenFD == events[i].data.fd) {
 860                 if (mdclog_level_get() >= MDCLOG_INFO) {
 861                     mdclog_write(MDCLOG_INFO, "New connection request from sctp network\n");
 862                 }
 863                 // new connection is requested from RAN  start build connection
 864                 while (true) {
 865                     struct sockaddr in_addr {};
 866                     socklen_t in_len;
 867                     char hostBuff[NI_MAXHOST];
 868                     char portBuff[NI_MAXSERV];
 869
 870                     in_len = sizeof(in_addr);
 871                     auto *peerInfo = (ConnectedCU_t *)calloc(1, sizeof(ConnectedCU_t));
 872                     if(peerInfo == nullptr){
 873                         mdclog_write(MDCLOG_ERR, "calloc failed");
 874                         break;
 875                     }
 876                     peerInfo->sctpParams = params;
 877                     peerInfo->fileDescriptor = accept(params->listenFD, &in_addr, &in_len);
 878                     if (peerInfo->fileDescriptor == -1) {
 879 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 880                         if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) {
 881                             /* We have processed all incoming connections. */
 882                             if(peerInfo)
 883                                 free(peerInfo);
 884                             break;
 885                         } else {
 886                             if(peerInfo)
 887                                 free(peerInfo);
 888                             mdclog_write(MDCLOG_ERR, "Accept error, errno = %s", strerror(errno));
 889                             break;
 890                         }
 891                     }
 892                     if (setSocketNoBlocking(peerInfo->fileDescriptor) == -1) {
 893                         mdclog_write(MDCLOG_ERR, "setSocketNoBlocking failed to set new connection %s on port %s\n", hostBuff, portBuff);
 894                         close(peerInfo->fileDescriptor);
 895                         if(peerInfo)
 896                             free(peerInfo);
 897                         break;
 898 #endif
 899                     }
 900                     auto  ans = getnameinfo(&in_addr, in_len,
 901                                             peerInfo->hostName, NI_MAXHOST,
 902                                             peerInfo->portNumber, NI_MAXSERV, (unsigned )((unsigned int)NI_NUMERICHOST | (unsigned int)NI_NUMERICSERV));
 903                     if (ans < 0) {
 904                         mdclog_write(MDCLOG_ERR, "Failed to get info on connection request. %s\n", strerror(errno));
 905                         close(peerInfo->fileDescriptor);
 906                         if(peerInfo)
 907                             free(peerInfo);
 908                         break;
 909                     }
 910                     if (mdclog_level_get() >= MDCLOG_DEBUG) {
 911                         mdclog_write(MDCLOG_DEBUG, "Accepted connection on descriptor %d (host=%s, port=%s)\n", peerInfo->fileDescriptor, peerInfo->hostName, peerInfo->portNumber);
 912                     }
 913                     peerInfo->isConnected = false;
 914                     peerInfo->gotSetup = false;
 915                     if (addToEpoll(params->epoll_fd,
 916                                    peerInfo,
 917                                    (EPOLLIN | EPOLLET),
 918                                    params->sctpMap, nullptr,
 919                                    0) != 0) {
 920                         if(peerInfo)
 921                             free(peerInfo);
 922                         break;
 923                     }
 924                     break;
 925                 }
 926             } else if (params->rmrListenFd == events[i].data.fd) {
 927                 // got message from XAPP
 928                 //num_of_XAPP_messages.fetch_add(1, std::memory_order_release);
 929                 num_of_messages.fetch_add(1, std::memory_order_release);
 930                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
 931                     mdclog_write(MDCLOG_DEBUG, "new RMR message");
 932                 }
 933                 if (receiveXappMessages(params->sctpMap,
 934                                         rmrMessageBuffer,
 935                                         message.message.time) != 0) {
 936                     mdclog_write(MDCLOG_ERR, "Error handling Xapp message");
 937                 }
 938             } else if (params->inotifyFD == events[i].data.fd) {
 939                 mdclog_write(MDCLOG_INFO, "Got event from inotify (configuration update)");
 940                 handleConfigChange(params);
 941             } else {
 942                 /* We RMR_ERR_RETRY have data on the fd waiting to be read. Read and display it.
 943                  * We must read whatever data is available completely, as we are running
 944                  *  in edge-triggered mode and won't get a notification again for the same data. */
 945                 num_of_messages.fetch_add(1, std::memory_order_release);
 946                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
 947                     mdclog_write(MDCLOG_DEBUG, "new message from SCTP, epoll flags are : %0x", events[i].events);
 948                 }
 949                 receiveDataFromSctp(&events[i],
 950                                     params->sctpMap,
 951                                     num_of_SCTP_messages,
 952                                     rmrMessageBuffer,
 953                                     message.message.time);
 954             }
 955
 956             clock_gettime(CLOCK_MONOTONIC, &end);
 957             if (mdclog_level_get() >= MDCLOG_INFO) {
 958                 totalTime += ((end.tv_sec + 1.0e-9 * end.tv_nsec) -
 959                               ((double) start.tv_sec + 1.0e-9 * start.tv_nsec));
 960             }
 961             if (mdclog_level_get() >= MDCLOG_DEBUG) {
 962                 mdclog_write(MDCLOG_DEBUG, "message handling is %ld seconds %ld nanoseconds",
 963                              end.tv_sec - start.tv_sec,
 964                              end.tv_nsec - start.tv_nsec);
 965             }
 966         }
 967 #ifdef UNIT_TEST
 968     break;
 969 #endif
 970     }
 971 }
 972
 973 /**
 974  *
 975  * @param sctpParams
 976  */
 977 void handleConfigChange(sctp_params_t *sctpParams) {
 978     char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));
 979     const struct inotify_event *event;
 980     char *ptr;
 981 #ifdef UNIT_TEST
 982     struct inotify_event tmpEvent;
 983 #endif
 984     path p = (sctpParams->configFilePath + "/" + sctpParams->configFileName).c_str();
 985     auto endlessLoop = true;
 986     while (endlessLoop) {
 987 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 988         auto len = read(sctpParams->inotifyFD, buf, sizeof buf);
 989 #else
 990     auto len=10;
 991 #endif
 992         if (len == -1) {
 993 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
 994             if (errno != EAGAIN) {
 995                 mdclog_write(MDCLOG_ERR, "read %s ", strerror(errno));
 996                 endlessLoop = false;
 997                 continue;
 998             }
 999             else {
1000                 endlessLoop = false;
1001                 continue;
1002             }
1003 #endif
1004         }
1005
1006         for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
1007 #ifndef UNIT_TEST
1008     event = (const struct inotify_event *)ptr;
1009 #else
1010     tmpEvent.mask = (uint32_t)IN_CLOSE_WRITE;
1011     event = &tmpEvent;
1012 #endif
1013             if (event->mask & (uint32_t)IN_ISDIR) {
1014                 continue;
1015             }
1016
1017             // the directory name
1018             if (sctpParams->inotifyWD == event->wd) {
1019                 // not the directory
1020             }
1021             if (event->len) {
1022 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1023                 auto  retVal = strcmp(sctpParams->configFileName.c_str(), event->name);
1024                 if (retVal != 0) {
1025                     continue;
1026                 }
1027 #endif
1028             }
1029             // only the file we want
1030             if (event->mask & (uint32_t)IN_CLOSE_WRITE) {
1031                 if (mdclog_level_get() >= MDCLOG_INFO) {
1032                     mdclog_write(MDCLOG_INFO, "Configuration file changed");
1033                 }
1034                 if (exists(p)) {
1035                     const int size = 2048;
1036                     auto fileSize = file_size(p);
1037                     if (fileSize > size) {
1038                         mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
1039                         return;
1040                     }
1041                 } else {
1042                     mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
1043                     return;
1044                 }
1045
1046                 ReadConfigFile conf;
1047                 if (conf.openConfigFile(p.string()) == -1) {
1048                     mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
1049                                  p.string().c_str(), strerror(errno));
1050                     return;
1051                 }
1052                 auto tmpStr = conf.getStringValue("loglevel");
1053                 if (tmpStr.length() == 0) {
1054                     mdclog_write(MDCLOG_ERR, "illegal loglevel. Set loglevel to MDCLOG_INFO");
1055                     tmpStr = "info";
1056                 }
1057                 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
1058
1059                 if ((tmpStr.compare("debug")) == 0) {
1060                     mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_DEBUG");
1061                     sctpParams->logLevel = MDCLOG_DEBUG;
1062                 }
1063 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1064                 else if ((tmpStr.compare("info")) == 0) {
1065                     mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_INFO");
1066                     sctpParams->logLevel = MDCLOG_INFO;
1067                 } else if ((tmpStr.compare("warning")) == 0) {
1068                     mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_WARN");
1069                     sctpParams->logLevel = MDCLOG_WARN;
1070                 } else if ((tmpStr.compare("error")) == 0) {
1071                     mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_ERR");
1072                     sctpParams->logLevel = MDCLOG_ERR;
1073                 } else {
1074                     mdclog_write(MDCLOG_ERR, "illegal loglevel = %s. Set loglevel to MDCLOG_INFO", tmpStr.c_str());
1075                     sctpParams->logLevel = MDCLOG_INFO;
1076                 }
1077 #endif
1078                 mdclog_level_set(sctpParams->logLevel);
1079                 tmpStr = conf.getStringValue("trace");
1080                 if (tmpStr.length() == 0) {
1081                     mdclog_write(MDCLOG_ERR, "illegal trace. Set trace to stop");
1082                     tmpStr = "stop";
1083                 }
1084
1085                 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
1086                 if ((tmpStr.compare("start")) == 0) {
1087                     mdclog_write(MDCLOG_INFO, "Trace set to: start");
1088                     sctpParams->trace = true;
1089                 } else if ((tmpStr.compare("stop")) == 0) {
1090                     mdclog_write(MDCLOG_INFO, "Trace set to: stop");
1091                     sctpParams->trace = false;
1092                 } else {
1093                     mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
1094                     sctpParams->trace = false;
1095                 }
1096                 jsonTrace = sctpParams->trace;
1097
1098
1099                 endlessLoop = false;
1100             }
1101 #ifdef UNIT_TEST
1102             break;
1103 #endif
1104         }
1105     }
1106 }
1107
1108 /**
1109  *
1110  * @param event
1111  * @param message
1112  * @param rmrMessageBuffer
1113  * @param params
1114  */
1115 void handleEinprogressMessages(struct epoll_event &event,
1116                                ReportingMessages_t &message,
1117                                RmrMessagesBuffer_t &rmrMessageBuffer,
1118                                sctp_params_t *params) {
1119     auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
1120     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1121
1122     mdclog_write(MDCLOG_INFO, "file descriptor %d got EPOLLOUT", peerInfo->fileDescriptor);
1123     auto retVal = 0;
1124     socklen_t retValLen = 0;
1125     auto rc = getsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_ERROR, &retVal, &retValLen);
1126     if (rc != 0 || retVal != 0) {
1127 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1128         if (rc != 0) {
1129             rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1130                                                          "%s|Failed SCTP Connection, after EINPROGRESS the getsockopt%s",
1131                                                          peerInfo->enodbName, strerror(errno));
1132         } else if (retVal != 0) {
1133             rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1134                                                          "%s|Failed SCTP Connection after EINPROGRESS, SO_ERROR",
1135                                                          peerInfo->enodbName);
1136         }
1137
1138         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1139         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1140         mdclog_write(MDCLOG_ERR, "%s", rmrMessageBuffer.sendMessage->payload);
1141         message.message.direction = 'N';
1142         if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != 0) {
1143             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1144         }
1145 #endif
1146         memset(peerInfo->asnData, 0, peerInfo->asnLength);
1147         peerInfo->asnLength = 0;
1148         peerInfo->mtype = 0;
1149         return;
1150     }
1151 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1152     peerInfo->isConnected = true;
1153
1154     if (modifyToEpoll(params->epoll_fd, peerInfo, (EPOLLIN | EPOLLET), params->sctpMap, peerInfo->enodbName,
1155                       peerInfo->mtype) != 0) {
1156         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_MOD");
1157         return;
1158     }
1159
1160     message.message.asndata = (unsigned char *)peerInfo->asnData;
1161     message.message.asnLength = peerInfo->asnLength;
1162     message.message.messageType = peerInfo->mtype;
1163     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1164     num_of_messages.fetch_add(1, std::memory_order_release);
1165     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1166         mdclog_write(MDCLOG_DEBUG, "send the delayed SETUP/ENDC SETUP to sctp for %s",
1167                      message.message.enodbName);
1168     }
1169     if (sendSctpMsg(peerInfo, message, params->sctpMap) != 0) {
1170         if (mdclog_level_get() >= MDCLOG_DEBUG) {
1171             mdclog_write(MDCLOG_DEBUG, "Error write to SCTP  %s %d", __func__, __LINE__);
1172         }
1173         return;
1174     }
1175
1176     memset(peerInfo->asnData, 0, peerInfo->asnLength);
1177     peerInfo->asnLength = 0;
1178     peerInfo->mtype = 0;
1179 #endif
1180 }
1181
1182
1183 void handlepoll_error(struct epoll_event &event,
1184                       ReportingMessages_t &message,
1185                       RmrMessagesBuffer_t &rmrMessageBuffer,
1186                       sctp_params_t *params) {
1187     if (event.data.fd != params->rmrListenFd) {
1188         auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
1189         mdclog_write(MDCLOG_ERR, "epoll error, events %0x on fd %d, RAN NAME : %s",
1190                      event.events, peerInfo->fileDescriptor, peerInfo->enodbName);
1191 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1192         rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1193                                                      "%s|Failed SCTP Connection",
1194                                                      peerInfo->enodbName);
1195         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1196         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1197
1198         memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1199         message.message.direction = 'N';
1200         if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != 0) {
1201             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1202         }
1203 #endif
1204         close(peerInfo->fileDescriptor);
1205         params->sctpMap->erase(peerInfo->enodbName);
1206         cleanHashEntry((ConnectedCU_t *) event.data.ptr, params->sctpMap);
1207     } else {
1208         mdclog_write(MDCLOG_ERR, "epoll error, events %0x on RMR FD", event.events);
1209     }
1210 }
1211 /**
1212  *
1213  * @param socket
1214  * @return
1215  */
1216 int setSocketNoBlocking(int socket) {
1217     auto flags = fcntl(socket, F_GETFL, 0);
1218
1219     if (flags == -1) {
1220         mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
1221         return -1;
1222     }
1223
1224     flags = (unsigned) flags | (unsigned) O_NONBLOCK;
1225     if (fcntl(socket, F_SETFL, flags) == -1) {
1226         mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
1227         return -1;
1228     }
1229
1230     return 0;
1231 }
1232
1233 /**
1234  *
1235  * @param val
1236  * @param m
1237  */
1238 void cleanHashEntry(ConnectedCU_t *val, Sctp_Map_t *m) {
1239     char *dummy;
1240     auto port = (uint16_t) strtol(val->portNumber, &dummy, 10);
1241     char searchBuff[2048]{};
1242
1243     snprintf(searchBuff, sizeof searchBuff, "host:%s:%d", val->hostName, port);
1244     m->erase(searchBuff);
1245
1246     m->erase(val->enodbName);
1247 #ifndef UNIT_TEST
1248     free(val);
1249 #endif
1250 }
1251
1252 /**
1253  *
1254  * @param fd file descriptor
1255  * @param data the asn data to send
1256  * @param len  length of the data
1257  * @param enodbName the enodbName as in the map for printing purpose
1258  * @param m map host information
1259  * @param mtype message number
1260  * @return 0 success, a negative number on fail
1261  */
1262 int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_t *m) {
1263     auto loglevel = mdclog_level_get();
1264 #ifndef UNIT_TEST
1265     int fd = peerInfo->fileDescriptor;
1266 #else
1267     int fd = FILE_DESCRIPTOR;
1268 #endif
1269     if (loglevel >= MDCLOG_DEBUG) {
1270         mdclog_write(MDCLOG_DEBUG, "Send SCTP message for CU %s, %s",
1271                      message.message.enodbName, __FUNCTION__);
1272     }
1273
1274     while (true) {
1275         if (send(fd,message.message.asndata, message.message.asnLength,MSG_NOSIGNAL) < 0) {
1276             if (errno == EINTR) {
1277                 continue;
1278             }
1279             mdclog_write(MDCLOG_ERR, "error writing to CU a message, %s ", strerror(errno));
1280 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1281             if (!peerInfo->isConnected) {
1282                 mdclog_write(MDCLOG_ERR, "connection to CU %s is still in progress.", message.message.enodbName);
1283                 return -1;
1284             }
1285 #endif
1286 #ifndef UNIT_TEST
1287             cleanHashEntry(peerInfo, m);
1288             close(fd);
1289 #endif
1290             char key[MAX_ENODB_NAME_SIZE * 2];
1291             snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName,
1292                      message.message.messageType);
1293             if (loglevel >= MDCLOG_DEBUG) {
1294                 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
1295             }
1296             auto tmp = m->find(key);
1297             if (tmp) {
1298                 free(tmp);
1299             }
1300             m->erase(key);
1301 #ifndef UNIT_TEST
1302             return -1;
1303 #endif
1304         }
1305         message.message.direction = 'D';
1306         // send report.buffer of size
1307         buildJsonMessage(message);
1308
1309         if (loglevel >= MDCLOG_DEBUG) {
1310             mdclog_write(MDCLOG_DEBUG,
1311                          "SCTP message for CU %s sent from %s",
1312                          message.message.enodbName,
1313                          __FUNCTION__);
1314         }
1315         return 0;
1316     }
1317 }
1318
1319 /**
1320  *
1321  * @param message
1322  * @param rmrMessageBuffer
1323  */
1324 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
1325     message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
1326     message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
1327
1328     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1329         mdclog_write(MDCLOG_DEBUG, "Message from Xapp RAN name = %s message length = %ld",
1330                      message.message.enodbName, (unsigned long) message.message.asnLength);
1331     }
1332 }
1333
1334
1335
1336 /**
1337  *
1338  * @param events
1339  * @param sctpMap
1340  * @param numOfMessages
1341  * @param rmrMessageBuffer
1342  * @param ts
1343  * @return
1344  */
1345 int receiveDataFromSctp(struct epoll_event *events,
1346                         Sctp_Map_t *sctpMap,
1347                         int &numOfMessages,
1348                         RmrMessagesBuffer_t &rmrMessageBuffer,
1349                         struct timespec &ts) {
1350     /* We have data on the fd waiting to be read. Read and display it.
1351  * We must read whatever data is available completely, as we are running
1352  *  in edge-triggered mode and won't get a notification again for the same data. */
1353     ReportingMessages_t message {};
1354     auto done = 0;
1355     auto loglevel = mdclog_level_get();
1356
1357     // get the identity of the interface
1358     message.peerInfo = (ConnectedCU_t *)events->data.ptr;
1359
1360     struct timespec start{0, 0};
1361     struct timespec decodeStart{0, 0};
1362     struct timespec end{0, 0};
1363
1364     E2AP_PDU_t *pdu = nullptr;
1365
1366     while (true) {
1367         if (loglevel >= MDCLOG_DEBUG) {
1368             mdclog_write(MDCLOG_DEBUG, "Start Read from SCTP %d fd", message.peerInfo->fileDescriptor);
1369             clock_gettime(CLOCK_MONOTONIC, &start);
1370         }
1371         // read the buffer directly to rmr payload
1372         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1373 #ifndef UNIT_TEST
1374         message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1375                 read(message.peerInfo->fileDescriptor, rmrMessageBuffer.sendMessage->payload, RECEIVE_SCTP_BUFFER_SIZE);
1376 #else
1377         message.message.asnLength = rmrMessageBuffer.sendMessage->len;
1378 #endif
1379
1380         if (loglevel >= MDCLOG_DEBUG) {
1381             mdclog_write(MDCLOG_DEBUG, "Finish Read from SCTP %d fd message length = %ld",
1382                          message.peerInfo->fileDescriptor, message.message.asnLength);
1383         }
1384
1385         memcpy(message.message.enodbName, message.peerInfo->enodbName, sizeof(message.peerInfo->enodbName));
1386         message.message.direction = 'U';
1387         message.message.time.tv_nsec = ts.tv_nsec;
1388         message.message.time.tv_sec = ts.tv_sec;
1389
1390         if (message.message.asnLength < 0) {
1391             if (errno == EINTR) {
1392                 continue;
1393             }
1394             /* If errno == EAGAIN, that means we have read all
1395                data. So goReportingMessages_t back to the main loop. */
1396             if (errno != EAGAIN) {
1397                 mdclog_write(MDCLOG_ERR, "Read error, %s ", strerror(errno));
1398                 done = 1;
1399             } else if (loglevel >= MDCLOG_DEBUG) {
1400                 mdclog_write(MDCLOG_DEBUG, "EAGAIN - descriptor = %d", message.peerInfo->fileDescriptor);
1401             }
1402             break;
1403         } else if (message.message.asnLength == 0) {
1404             /* End of file. The remote has closed the connection. */
1405             if (loglevel >= MDCLOG_INFO) {
1406                 mdclog_write(MDCLOG_INFO, "END of File Closed connection - descriptor = %d",
1407                              message.peerInfo->fileDescriptor);
1408             }
1409             done = 1;
1410             break;
1411         }
1412
1413         if (loglevel >= MDCLOG_DEBUG) {
1414             char printBuffer[RECEIVE_SCTP_BUFFER_SIZE]{};
1415             char *tmp = printBuffer;
1416             for (size_t i = 0; i < (size_t)message.message.asnLength; ++i) {
1417                 snprintf(tmp, 3, "%02x", message.message.asndata[i]);
1418                 tmp += 2;
1419             }
1420             printBuffer[message.message.asnLength] = 0;
1421             clock_gettime(CLOCK_MONOTONIC, &end);
1422             mdclog_write(MDCLOG_DEBUG, "Before Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1423                          message.peerInfo->enodbName, end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1424             mdclog_write(MDCLOG_DEBUG, "PDU buffer length = %ld, data =  : %s", message.message.asnLength,
1425                          printBuffer);
1426             clock_gettime(CLOCK_MONOTONIC, &decodeStart);
1427         }
1428 #ifndef UNIT_TEST
1429         auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu,
1430                         message.message.asndata, message.message.asnLength);
1431 #else
1432         asn_dec_rval_t rval = {RC_OK, 0};
1433         pdu = (E2AP_PDU_t*)rmrMessageBuffer.sendMessage->tp_buf;
1434 #endif
1435         if (rval.code != RC_OK) {
1436             mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from RAN : %s", rval.code,
1437                          message.peerInfo->enodbName);
1438             if (pdu != nullptr) {
1439                 ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1440                 pdu = nullptr;
1441             }
1442             break;
1443         }
1444
1445         if (loglevel >= MDCLOG_DEBUG) {
1446             clock_gettime(CLOCK_MONOTONIC, &end);
1447             mdclog_write(MDCLOG_DEBUG, "After Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1448                          message.peerInfo->enodbName, end.tv_sec - decodeStart.tv_sec, end.tv_nsec - decodeStart.tv_nsec);
1449             char *printBuffer;
1450             size_t size;
1451             FILE *stream = open_memstream(&printBuffer, &size);
1452             asn_fprint(stream, &asn_DEF_E2AP_PDU, pdu);
1453             mdclog_write(MDCLOG_DEBUG, "Encoding E2AP PDU past : %s", printBuffer);
1454             clock_gettime(CLOCK_MONOTONIC, &decodeStart);
1455
1456             fclose(stream);
1457             free(printBuffer);
1458         }
1459
1460         switch (pdu->present) {
1461             case E2AP_PDU_PR_initiatingMessage: {//initiating message
1462                 asnInitiatingRequest(pdu, sctpMap,message, rmrMessageBuffer);
1463                 break;
1464             }
1465             case E2AP_PDU_PR_successfulOutcome: { //successful outcome
1466                 asnSuccessfulMsg(pdu, sctpMap, message, rmrMessageBuffer);
1467                 break;
1468             }
1469             case E2AP_PDU_PR_unsuccessfulOutcome: { //Unsuccessful Outcome
1470                 asnUnSuccsesfulMsg(pdu, sctpMap, message, rmrMessageBuffer);
1471                 break;
1472             }
1473             default:
1474                 mdclog_write(MDCLOG_ERR, "Unknown index %d in E2AP PDU", pdu->present);
1475                 break;
1476         }
1477         if (loglevel >= MDCLOG_DEBUG) {
1478             clock_gettime(CLOCK_MONOTONIC, &end);
1479             mdclog_write(MDCLOG_DEBUG,
1480                          "After processing message and sent to rmr for : %s, Read time is : %ld seconds, %ld nanoseconds",
1481                          message.peerInfo->enodbName, end.tv_sec - decodeStart.tv_sec, end.tv_nsec - decodeStart.tv_nsec);
1482         }
1483         numOfMessages++;
1484 #ifndef UNIT_TEST
1485         if (pdu != nullptr) {
1486             // ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); /* With reset we were not freeing the memory and was causing the leak here. */
1487             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1488             pdu = nullptr;
1489         }
1490 #else
1491     done = 1;
1492     break;
1493 #endif
1494     }
1495
1496     if (done) {
1497         if (loglevel >= MDCLOG_INFO) {
1498             mdclog_write(MDCLOG_INFO, "Closed connection - descriptor = %d", message.peerInfo->fileDescriptor);
1499         }
1500         message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1501                 snprintf((char *)rmrMessageBuffer.sendMessage->payload,
1502                          256,
1503                          "%s|CU disconnected unexpectedly",
1504                          message.peerInfo->enodbName);
1505         message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1506 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1507         if (sendRequestToXapp(message,
1508                               RIC_SCTP_CONNECTION_FAILURE,
1509                               rmrMessageBuffer) != 0) {
1510             mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1511         }
1512 #endif
1513
1514         /* Closing descriptor make epoll remove it from the set of descriptors which are monitored. */
1515         close(message.peerInfo->fileDescriptor);
1516         cleanHashEntry((ConnectedCU_t *) events->data.ptr, sctpMap);
1517     }
1518     if (loglevel >= MDCLOG_DEBUG) {
1519         clock_gettime(CLOCK_MONOTONIC, &end);
1520         mdclog_write(MDCLOG_DEBUG, "from receive SCTP to send RMR time is %ld seconds and %ld nanoseconds",
1521                      end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1522
1523     }
1524     return 0;
1525 }
1526
1527 static void buildAndSendSetupRequest(ReportingMessages_t &message,
1528                                      RmrMessagesBuffer_t &rmrMessageBuffer,
1529                                      E2AP_PDU_t *pdu/*,
1530                                      string const &messageName,
1531                                      string const &ieName,
1532                                      vector<string> &functionsToAdd_v,
1533                                      vector<string> &functionsToModified_v*/) {
1534     auto logLevel = mdclog_level_get();
1535     // now we can send the data to e2Mgr
1536
1537     asn_enc_rval_t er;
1538     auto buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2;
1539     unsigned char *buffer = nullptr;
1540     buffer = (unsigned char *) calloc(buffer_size, sizeof(unsigned char));
1541     if(!buffer)
1542     {
1543 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1544         mdclog_write(MDCLOG_ERR, "Allocating buffer for %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1545         return;
1546 #endif
1547     }
1548     while (true) {
1549         er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu, buffer, buffer_size);
1550         if (er.encoded == -1) {
1551 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1552             mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1553             return;
1554 #endif
1555         } else if (er.encoded > (ssize_t) buffer_size) {
1556             buffer_size = er.encoded + 128;
1557 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
1558             mdclog_write(MDCLOG_WARN, "Buffer of size %d is to small for %s. Reallocate buffer of size %d",
1559                          (int) buffer_size,
1560                          asn_DEF_E2AP_PDU.name, buffer_size);
1561             buffer_size = er.encoded + 128;
1562
1563             unsigned char *newBuffer = nullptr;
1564             newBuffer = (unsigned char *) realloc(buffer, buffer_size);
1565             if (!newBuffer)
1566             {
1567                 // out of memory
1568                 mdclog_write(MDCLOG_ERR, "Reallocating buffer for %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1569                 free(buffer);
1570                 return;
1571             }
1572             buffer = newBuffer;
1573             continue;
1574 #endif
1575         }
1576         buffer[er.encoded] = '\0';
1577         break;
1578     }
1579     // encode to xml
1580
1581     string res((char *)buffer);
1582     res.erase(std::remove(res.begin(), res.end(), '\n'), res.end());
1583     res.erase(std::remove(res.begin(), res.end(), '\t'), res.end());
1584     res.erase(std::remove(res.begin(), res.end(), ' '), res.end());
1585
1586 //    string res {};
1587 //    if (!functionsToAdd_v.empty() || !functionsToModified_v.empty()) {
1588 //        res = buildXmlData(messageName, ieName, functionsToAdd_v, functionsToModified_v, buffer, (size_t) er.encoded);
1589 //    }
1590     rmr_mbuf_t *rmrMsg;
1591 //    if (res.length() == 0) {
1592 //        rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, buffer_size + 256);
1593 //        rmrMsg->len = snprintf((char *) rmrMsg->payload, RECEIVE_SCTP_BUFFER_SIZE * 2, "%s:%d|%s",
1594 //                               message.peerInfo->sctpParams->myIP.c_str(),
1595 //                               message.peerInfo->sctpParams->rmrPort,
1596 //                               buffer);
1597 //    } else {
1598         rmrMsg = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, (int)res.length() + 256);
1599         rmrMsg->len = snprintf((char *) rmrMsg->payload, res.length() + 256, "%s:%d|%s",
1600                                message.peerInfo->sctpParams->myIP.c_str(),
1601                                message.peerInfo->sctpParams->rmrPort,
1602                                res.c_str());
1603 //    }
1604
1605     if (logLevel >= MDCLOG_DEBUG) {
1606         mdclog_write(MDCLOG_DEBUG, "Setup request of size %d :\n %s\n", rmrMsg->len, rmrMsg->payload);
1607     }
1608     // send to RMR
1609     rmrMsg->mtype = message.message.messageType;
1610     rmrMsg->state = 0;
1611     rmr_bytes2meid(rmrMsg, (unsigned char *) message.message.enodbName, strlen(message.message.enodbName));
1612
1613     static unsigned char tx[32];
1614     snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1615     rmr_bytes2xact(rmrMsg, tx, strlen((const char *) tx));
1616 #ifndef UNIT_TEST
1617     rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg);
1618 #endif
1619     if (rmrMsg == nullptr) {
1620         mdclog_write(MDCLOG_ERR, "RMR failed to send returned nullptr");
1621     } else if (rmrMsg->state != 0) {
1622         char meid[RMR_MAX_MEID]{};
1623         if (rmrMsg->state == RMR_ERR_RETRY) {
1624             usleep(5);
1625             rmrMsg->state = 0;
1626             mdclog_write(MDCLOG_INFO, "RETRY sending Message %d to Xapp from %s",
1627                          rmrMsg->mtype, rmr_get_meid(rmrMsg, (unsigned char *) meid));
1628 #ifndef UNIT_TEST
1629             rmrMsg = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMsg);
1630 #endif
1631             if (rmrMsg == nullptr) {
1632                 mdclog_write(MDCLOG_ERR, "RMR failed send returned nullptr");
1633             } else if (rmrMsg->state != 0) {
1634                 mdclog_write(MDCLOG_ERR,
1635                              "RMR Retry failed %s sending request %d to Xapp from %s",
1636                              translateRmrErrorMessages(rmrMsg->state).c_str(),
1637                              rmrMsg->mtype,
1638                              rmr_get_meid(rmrMsg, (unsigned char *) meid));
1639             }
1640         } else {
1641             mdclog_write(MDCLOG_ERR, "RMR failed: %s. sending request %d to Xapp from %s",
1642                          translateRmrErrorMessages(rmrMsg->state).c_str(),
1643                          rmrMsg->mtype,
1644                          rmr_get_meid(rmrMsg, (unsigned char *) meid));
1645         }
1646     }
1647     message.peerInfo->gotSetup = true;
1648     buildJsonMessage(message);
1649
1650     if (rmrMsg != nullptr) {
1651         rmr_free_msg(rmrMsg);
1652     }
1653     free(buffer);
1654
1655     return;
1656 }
1657
1658 #if 0
1659 int RAN_Function_list_To_Vector(RANfunctions_List_t& list, vector <string> &runFunXML_v) {
1660     auto index = 0;
1661     runFunXML_v.clear();
1662     for (auto j = 0; j < list.list.count; j++) {
1663         auto *raNfunctionItemIEs = (RANfunction_ItemIEs_t *)list.list.array[j];
1664         if (raNfunctionItemIEs->id == ProtocolIE_ID_id_RANfunction_Item &&
1665             (raNfunctionItemIEs->value.present == RANfunction_ItemIEs__value_PR_RANfunction_Item)) {
1666             // encode to xml
1667             E2SM_gNB_NRT_RANfunction_Definition_t *ranFunDef = nullptr;
1668             auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER,
1669                                    &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition,
1670                                    (void **)&ranFunDef,
1671                                    raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.buf,
1672                                    raNfunctionItemIEs->value.choice.RANfunction_Item.ranFunctionDefinition.size);
1673             if (rval.code != RC_OK) {
1674                 mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2SM message from : %s",
1675                              rval.code,
1676                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name);
1677                 return -1;
1678             }
1679
1680             auto xml_buffer_size = RECEIVE_SCTP_BUFFER_SIZE * 2;
1681             unsigned char xml_buffer[RECEIVE_SCTP_BUFFER_SIZE * 2];
1682             memset(xml_buffer, 0, RECEIVE_SCTP_BUFFER_SIZE * 2);
1683             // encode to xml
1684             auto er = asn_encode_to_buffer(nullptr,
1685                                            ATS_BASIC_XER,
1686                                            &asn_DEF_E2SM_gNB_NRT_RANfunction_Definition,
1687                                            ranFunDef,
1688                                            xml_buffer,
1689                                            xml_buffer_size);
1690             if (er.encoded == -1) {
1691                 mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s",
1692                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name,
1693                              strerror(errno));
1694             } else if (er.encoded > (ssize_t)xml_buffer_size) {
1695                 mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
1696                              (int) xml_buffer_size,
1697                              asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name, __func__, __LINE__);
1698             } else {
1699                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1700                     mdclog_write(MDCLOG_DEBUG, "Encoding E2SM %s PDU number %d : %s",
1701                                  asn_DEF_E2SM_gNB_NRT_RANfunction_Definition.name,
1702                                  index++,
1703                                  xml_buffer);
1704                 }
1705
1706                 string runFuncs = (char *)(xml_buffer);
1707                 runFunXML_v.emplace_back(runFuncs);
1708             }
1709         }
1710     }
1711     return 0;
1712 }
1713
1714 int collectServiceUpdate_RequestData(E2AP_PDU_t *pdu,
1715                                      Sctp_Map_t *sctpMap,
1716                                      ReportingMessages_t &message,
1717                                      vector <string> &RANfunctionsAdded_v,
1718                                      vector <string> &RANfunctionsModified_v) {
1719     memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE);
1720     for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.count; i++) {
1721         auto *ie = pdu->choice.initiatingMessage->value.choice.RICserviceUpdate.protocolIEs.list.array[i];
1722         if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) {
1723             if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctionsID_List) {
1724                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1725                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
1726                                  ie->value.choice.RANfunctions_List.list.count);
1727                 }
1728                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) {
1729                     return -1;
1730                 }
1731             }
1732         } else if (ie->id == ProtocolIE_ID_id_RANfunctionsModified) {
1733             if (ie->value.present == RICserviceUpdate_IEs__value_PR_RANfunctions_List) {
1734                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1735                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
1736                                  ie->value.choice.RANfunctions_List.list.count);
1737                 }
1738                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsModified_v) != 0 ) {
1739                     return -1;
1740                 }
1741             }
1742         }
1743     }
1744     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1745         mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries",
1746                      RANfunctionsAdded_v.size());
1747     }
1748     return 0;
1749 }
1750
1751 #endif
1752
1753
1754 void buildE2TPrometheusCounters(sctp_params_t &sctpParams) {
1755     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestMsgs"}});
1756     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestBytes"}});
1757
1758     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupResponseMsgs"}});
1759     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &sctpParams.prometheusFamily->Add({{"counter", "SetupResponseBytes"}});
1760
1761     sctpParams.e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestFailureMsgs"}});
1762     sctpParams.e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup] = &sctpParams.prometheusFamily->Add({{"counter", "SetupRequestFailureBytes"}});
1763
1764     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &sctpParams.prometheusFamily->Add({{"counter", "ErrorIndicationMsgs"}});
1765     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &sctpParams.prometheusFamily->Add({{"counter", "ErrorIndicationBytes"}});
1766
1767     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetRequestMsgs"}});
1768     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetRequestBytes"}});
1769
1770     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetAckMsgs"}});
1771     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset] = &sctpParams.prometheusFamily->Add({{"counter", "ResetAckBytes"}});
1772
1773     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateMsgs"}});
1774     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateBytes"}});
1775
1776     sctpParams.e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateRespMsgs"}});
1777     sctpParams.e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateRespBytes"}});
1778
1779     sctpParams.e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateFailureMsgs"}});
1780     sctpParams.e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceUpdateFailureBytes"}});
1781
1782     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlMsgs"}});
1783     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlBytes"}});
1784
1785     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlAckMsgs"}});
1786     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlAckBytes"}});
1787
1788     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlFailureMsgs"}});
1789     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol] = &sctpParams.prometheusFamily->Add({{"counter", "RICControlFailureBytes"}});
1790
1791     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionMsgs"}});
1792     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionBytes"}});
1793
1794     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionAckMsgs"}});
1795     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionAckBytes"}});
1796
1797     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionFailureMsgs"}});
1798     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionFailureBytes"}});
1799
1800     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteMsgs"}});
1801     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteBytes"}});
1802
1803     sctpParams.e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteAckMsgs"}});
1804     sctpParams.e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteAckBytes"}});
1805
1806     sctpParams.e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteFailMsgs"}});
1807     sctpParams.e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete] = &sctpParams.prometheusFamily->Add({{"counter", "RICSubscriptionDeleteFailBytes"}});
1808
1809     sctpParams.e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication] = &sctpParams.prometheusFamily->Add({{"counter", "RICIndicationMsgs"}});
1810     sctpParams.e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication] = &sctpParams.prometheusFamily->Add({{"counter", "RICIndicationBytes"}});
1811
1812     sctpParams.e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceQueryMsgs"}});
1813     sctpParams.e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery] = &sctpParams.prometheusFamily->Add({{"counter", "RICServiceQueryBytes"}});
1814 }
1815
1816 void buildPrometheusList(ConnectedCU_t *peerInfo, Family<Counter> *prometheusFamily) {
1817     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Messages"}});
1818     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"SetupRequest", "Bytes"}});
1819
1820     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Messages"}});
1821     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ErrorIndication", "Bytes"}});
1822
1823     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Messages"}});
1824     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICindication)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICindication", "Bytes"}});
1825
1826     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Messages"}});
1827     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetRequest", "Bytes"}});
1828
1829     peerInfo->counters[IN_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Messages"}});
1830     peerInfo->counters[IN_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICserviceUpdate", "Bytes"}});
1831     // ---------------------------------------------
1832     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Messages"}});
1833     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"ResetACK", "Bytes"}});
1834
1835     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Messages"}});
1836     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolACK", "Bytes"}});
1837
1838     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Messages"}});
1839     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionACK", "Bytes"}});
1840
1841     peerInfo->counters[IN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Messages"}});
1842     peerInfo->counters[IN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteACK", "Bytes"}});
1843     //-------------------------------------------------------------
1844
1845     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Messages"}});
1846     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICcontrolFailure", "Bytes"}});
1847
1848     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Messages"}});
1849     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionFailure", "Bytes"}});
1850
1851     peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Messages"}});
1852     peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "IN"}, {"RICsubscriptionDeleteFailure", "Bytes"}});
1853
1854     //====================================================================================
1855     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Messages"}});
1856     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_ErrorIndication)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ErrorIndication", "Bytes"}});
1857
1858     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Messages"}});
1859     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetRequest", "Bytes"}});
1860
1861     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Messages"}});
1862     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICcontrol)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICcontrol", "Bytes"}});
1863
1864     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Messages"}});
1865     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICserviceQuery)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceQuery", "Bytes"}});
1866
1867     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Messages"}});
1868     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscription)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscription", "Bytes"}});
1869
1870     peerInfo->counters[OUT_INITI][MSG_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Messages"}});
1871     peerInfo->counters[OUT_INITI][BYTES_COUNTER][(ProcedureCode_id_RICsubscriptionDelete)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICsubscriptionDelete", "Bytes"}});
1872     //---------------------------------------------------------------------------------------------------------
1873     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Messages"}});
1874     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupResponse", "Bytes"}});
1875
1876     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Messages"}});
1877     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_Reset)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"ResetACK", "Bytes"}});
1878
1879     peerInfo->counters[OUT_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Messages"}});
1880     peerInfo->counters[OUT_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateResponse", "Bytes"}});
1881     //----------------------------------------------------------------------------------------------------------------
1882     peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Messages"}});
1883     peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_E2setup)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"SetupRequestFailure", "Bytes"}});
1884
1885     peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Messages"}});
1886     peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][(ProcedureCode_id_RICserviceUpdate)] = &prometheusFamily->Add({{peerInfo->enodbName, "OUT"}, {"RICserviceUpdateFailure", "Bytes"}});
1887 }
1888
1889 /**
1890  *
1891  * @param pdu
1892  * @param sctpMap
1893  * @param message
1894  * @param RANfunctionsAdded_v
1895  * @return
1896  */
1897 int collectSetupRequestData(E2AP_PDU_t *pdu,
1898                                      Sctp_Map_t *sctpMap,
1899                                      ReportingMessages_t &message /*, vector <string> &RANfunctionsAdded_v*/) {
1900     memset(message.peerInfo->enodbName, 0 , MAX_ENODB_NAME_SIZE);
1901     for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.count; i++) {
1902         auto *ie = pdu->choice.initiatingMessage->value.choice.E2setupRequest.protocolIEs.list.array[i];
1903         if (ie->id == ProtocolIE_ID_id_GlobalE2node_ID) {
1904             // get the ran name for meid
1905             if (ie->value.present == E2setupRequestIEs__value_PR_GlobalE2node_ID) {
1906                 if (buildRanName(message.peerInfo->enodbName, ie) < 0) {
1907                     mdclog_write(MDCLOG_ERR, "Bad param in E2setupRequestIEs GlobalE2node_ID.\n");
1908                     // no message will be sent
1909                     return -1;
1910                 }
1911
1912                 memcpy(message.message.enodbName, message.peerInfo->enodbName, strlen(message.peerInfo->enodbName));
1913                 sctpMap->setkey(message.message.enodbName, message.peerInfo);
1914             }
1915         } /*else if (ie->id == ProtocolIE_ID_id_RANfunctionsAdded) {
1916             if (ie->value.present == E2setupRequestIEs__value_PR_RANfunctions_List) {
1917                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1918                     mdclog_write(MDCLOG_DEBUG, "Run function list have %d entries",
1919                                  ie->value.choice.RANfunctions_List.list.count);
1920                 }
1921                 if (RAN_Function_list_To_Vector(ie->value.choice.RANfunctions_List, RANfunctionsAdded_v) != 0 ) {
1922                     return -1;
1923                 }
1924             }
1925         } */
1926     }
1927 //    if (mdclog_level_get() >= MDCLOG_DEBUG) {
1928 //        mdclog_write(MDCLOG_DEBUG, "Run function vector have %ld entries",
1929 //                     RANfunctionsAdded_v.size());
1930 //    }
1931     return 0;
1932 }
1933
1934 int XML_From_PER(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
1935     E2AP_PDU_t *pdu = nullptr;
1936
1937     if (mdclog_level_get() >= MDCLOG_DEBUG) {
1938         mdclog_write(MDCLOG_DEBUG, "got PER message of size %d is:%s",
1939                      rmrMessageBuffer.sendMessage->len, rmrMessageBuffer.sendMessage->payload);
1940     }
1941     auto rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu,
1942                            rmrMessageBuffer.sendMessage->payload, rmrMessageBuffer.sendMessage->len);
1943     if (rval.code != RC_OK) {
1944         mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response  from E2MGR : %s",
1945                      rval.code,
1946                      message.message.enodbName);
1947         if (pdu != nullptr) {
1948             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1949             pdu = nullptr;
1950         }
1951         return -1;
1952     }
1953
1954     int buff_size = RECEIVE_XAPP_BUFFER_SIZE;
1955     auto er = asn_encode_to_buffer(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, pdu,
1956                                    rmrMessageBuffer.sendMessage->payload, buff_size);
1957     if (er.encoded == -1) {
1958         mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
1959         if (pdu != nullptr) {
1960             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1961             pdu = nullptr;
1962         }
1963         return -1;
1964     } else if (er.encoded > (ssize_t)buff_size) {
1965         mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
1966                      (int)rmrMessageBuffer.sendMessage->len,
1967                      asn_DEF_E2AP_PDU.name,
1968                      __func__,
1969                      __LINE__);
1970         if (pdu != nullptr) {
1971             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1972             pdu = nullptr;
1973         }
1974         return -1;
1975     }
1976     rmrMessageBuffer.sendMessage->len = er.encoded;
1977     if (pdu != nullptr) {
1978         ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1979         pdu = nullptr;
1980     }
1981     return 0;
1982
1983 }
1984
1985 /**
1986  *
1987  * @param pdu
1988  * @param message
1989  * @param rmrMessageBuffer
1990  */
1991 void asnInitiatingRequest(E2AP_PDU_t *pdu,
1992                           Sctp_Map_t *sctpMap,
1993                           ReportingMessages_t &message,
1994                           RmrMessagesBuffer_t &rmrMessageBuffer) {
1995     auto logLevel = mdclog_level_get();
1996     auto procedureCode = ((InitiatingMessage_t *) pdu->choice.initiatingMessage)->procedureCode;
1997     if (logLevel >= MDCLOG_DEBUG) {
1998         mdclog_write(MDCLOG_DEBUG, "Initiating message %ld\n", procedureCode);
1999     }
2000     switch (procedureCode) {
2001         case ProcedureCode_id_E2setup: {
2002             if (logLevel >= MDCLOG_DEBUG) {
2003                 mdclog_write(MDCLOG_DEBUG, "Got E2setup");
2004             }
2005
2006 //            vector <string> RANfunctionsAdded_v;
2007 //            vector <string> RANfunctionsModified_v;
2008 //            RANfunctionsAdded_v.clear();
2009 //            RANfunctionsModified_v.clear();
2010             if (collectSetupRequestData(pdu, sctpMap, message) != 0) {
2011                 break;
2012             }
2013
2014             buildPrometheusList(message.peerInfo, message.peerInfo->sctpParams->prometheusFamily);
2015
2016             string messageName("E2setupRequest");
2017             string ieName("E2setupRequestIEs");
2018             message.message.messageType = RIC_E2_SETUP_REQ;
2019             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2020             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment((double)message.message.asnLength);
2021
2022             // Update E2T instance level metrics
2023             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2024             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment((double)message.message.asnLength);
2025
2026             buildAndSendSetupRequest(message, rmrMessageBuffer, pdu);
2027             break;
2028         }
2029         case ProcedureCode_id_RICserviceUpdate: {
2030             if (logLevel >= MDCLOG_DEBUG) {
2031                 mdclog_write(MDCLOG_DEBUG, "Got RICserviceUpdate %s", message.message.enodbName);
2032             }
2033 //            vector <string> RANfunctionsAdded_v;
2034 //            vector <string> RANfunctionsModified_v;
2035 //            RANfunctionsAdded_v.clear();
2036 //            RANfunctionsModified_v.clear();
2037 //            if (collectServiceUpdate_RequestData(pdu, sctpMap, message,
2038 //                                                 RANfunctionsAdded_v, RANfunctionsModified_v) != 0) {
2039 //                break;
2040 //            }
2041
2042             string messageName("RICserviceUpdate");
2043             string ieName("RICserviceUpdateIEs");
2044             message.message.messageType = RIC_SERVICE_UPDATE;
2045 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2046             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2047             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment((double)message.message.asnLength);
2048
2049             // Update E2T instance level metrics
2050             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2051             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment((double)message.message.asnLength);
2052 #endif
2053             buildAndSendSetupRequest(message, rmrMessageBuffer, pdu);
2054             break;
2055         }
2056         case ProcedureCode_id_ErrorIndication: {
2057             if (logLevel >= MDCLOG_DEBUG) {
2058                 mdclog_write(MDCLOG_DEBUG, "Got ErrorIndication %s", message.message.enodbName);
2059             }
2060 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2061             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2062             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment((double)message.message.asnLength);
2063
2064             // Update E2T instance level metrics
2065             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2066             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment((double)message.message.asnLength);
2067 #endif
2068             if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer) != 0) {
2069                 mdclog_write(MDCLOG_ERR, "RIC_ERROR_INDICATION failed to send to xAPP");
2070             }
2071             break;
2072         }
2073         case ProcedureCode_id_Reset: {
2074             if (logLevel >= MDCLOG_DEBUG) {
2075                 mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName);
2076             }
2077 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2078             message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2079             message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2080
2081             // Update E2T instance level metrics
2082             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2083             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2084 #endif
2085             if (XML_From_PER(message, rmrMessageBuffer) < 0) {
2086                 break;
2087             }
2088
2089             if (sendRequestToXapp(message, RIC_E2_RESET_REQ, rmrMessageBuffer) != 0) {
2090                 mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_REQ message failed to send to xAPP");
2091             }
2092             break;
2093         }
2094         case ProcedureCode_id_RICindication: {
2095             if (logLevel >= MDCLOG_DEBUG) {
2096                 mdclog_write(MDCLOG_DEBUG, "Got RICindication %s", message.message.enodbName);
2097             }
2098             for (auto i = 0; i < pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.count; i++) {
2099                 auto messageSent = false;
2100                 RICindication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.array[i];
2101                 if (logLevel >= MDCLOG_DEBUG) {
2102                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2103                 }
2104                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2105                     if (logLevel >= MDCLOG_DEBUG) {
2106                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2107                     }
2108                     if (ie->value.present == RICindication_IEs__value_PR_RICrequestID) {
2109                         static unsigned char tx[32];
2110                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_INDICATION;
2111                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2112                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2113                         rmr_bytes2meid(rmrMessageBuffer.sendMessage,
2114                                        (unsigned char *)message.message.enodbName,
2115                                        strlen(message.message.enodbName));
2116                         rmrMessageBuffer.sendMessage->state = 0;
2117                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2118
2119                         //ie->value.choice.RICrequestID.ricInstanceID;
2120                         if (mdclog_level_get() >= MDCLOG_DEBUG) {
2121                             mdclog_write(MDCLOG_DEBUG, "sub id = %d, mtype = %d, ric instance id %ld, requestor id = %ld",
2122                                          rmrMessageBuffer.sendMessage->sub_id,
2123                                          rmrMessageBuffer.sendMessage->mtype,
2124                                          ie->value.choice.RICrequestID.ricInstanceID,
2125                                          ie->value.choice.RICrequestID.ricRequestorID);
2126                         }
2127 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2128                         message.peerInfo->counters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication]->Increment();
2129                         message.peerInfo->counters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication]->Increment((double)message.message.asnLength);
2130
2131                         // Update E2T instance level metrics
2132                         message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_RICindication]->Increment();
2133                         message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_RICindication]->Increment((double)message.message.asnLength);
2134 #endif
2135                         sendRmrMessage(rmrMessageBuffer, message);
2136                         messageSent = true;
2137                     } else {
2138                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2139                     }
2140                 }
2141                 if (messageSent) {
2142                     break;
2143                 }
2144             }
2145             break;
2146         }
2147         default: {
2148             mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
2149             message.message.messageType = 0; // no RMR message type yet
2150
2151             buildJsonMessage(message);
2152
2153             break;
2154         }
2155     }
2156 }
2157
2158 /**
2159  *
2160  * @param pdu
2161  * @param message
2162  * @param rmrMessageBuffer
2163  */
2164 void asnSuccessfulMsg(E2AP_PDU_t *pdu,
2165                       Sctp_Map_t *sctpMap,
2166                       ReportingMessages_t &message,
2167                       RmrMessagesBuffer_t &rmrMessageBuffer) {
2168     auto procedureCode = pdu->choice.successfulOutcome->procedureCode;
2169     auto logLevel = mdclog_level_get();
2170     if (logLevel >= MDCLOG_INFO) {
2171         mdclog_write(MDCLOG_INFO, "Successful Outcome %ld", procedureCode);
2172     }
2173     switch (procedureCode) {
2174         case ProcedureCode_id_Reset: {
2175             if (logLevel >= MDCLOG_DEBUG) {
2176                 mdclog_write(MDCLOG_DEBUG, "Got Reset %s", message.message.enodbName);
2177             }
2178 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2179             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2180             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2181
2182             // Update E2T instance level metrics
2183             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2184             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment((double)message.message.asnLength);
2185 #endif
2186             if (XML_From_PER(message, rmrMessageBuffer) < 0) {
2187                 break;
2188             }
2189             if (sendRequestToXapp(message, RIC_E2_RESET_RESP, rmrMessageBuffer) != 0) {
2190                 mdclog_write(MDCLOG_ERR, "RIC_E2_RESET_RESP message failed to send to xAPP");
2191             }
2192             break;
2193         }
2194         case ProcedureCode_id_RICcontrol: {
2195             if (logLevel >= MDCLOG_DEBUG) {
2196                 mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName);
2197             }
2198             for (auto i = 0;
2199                  i < pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.count; i++) {
2200                 auto messageSent = false;
2201                 RICcontrolAcknowledge_IEs_t *ie = pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.array[i];
2202                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2203                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2204                 }
2205                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2206                     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2207                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2208                     }
2209                     if (ie->value.present == RICcontrolAcknowledge_IEs__value_PR_RICrequestID) {
2210                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_ACK;
2211                         rmrMessageBuffer.sendMessage->state = 0;
2212 //                        rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID;
2213                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2214
2215                         static unsigned char tx[32];
2216                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2217                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2218                         rmr_bytes2meid(rmrMessageBuffer.sendMessage,
2219                                        (unsigned char *)message.message.enodbName,
2220                                        strlen(message.message.enodbName));
2221 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2222                         message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2223                         message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2224
2225                         // Update E2T instance level metrics
2226                         message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2227                         message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2228 #endif
2229                         sendRmrMessage(rmrMessageBuffer, message);
2230                         messageSent = true;
2231                     } else {
2232                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2233                     }
2234                 }
2235                 if (messageSent) {
2236                     break;
2237                 }
2238             }
2239
2240             break;
2241         }
2242         case ProcedureCode_id_RICsubscription: {
2243             if (logLevel >= MDCLOG_DEBUG) {
2244                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName);
2245             }
2246 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2247             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2248             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2249
2250             // Update E2T instance level metrics
2251             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2252             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2253 #endif
2254             if (sendRequestToXapp(message, RIC_SUB_RESP, rmrMessageBuffer) != 0) {
2255                 mdclog_write(MDCLOG_ERR, "Subscription successful message failed to send to xAPP");
2256             }
2257             break;
2258         }
2259         case ProcedureCode_id_RICsubscriptionDelete: {
2260             if (logLevel >= MDCLOG_DEBUG) {
2261                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName);
2262             }
2263 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2264             message.peerInfo->counters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2265             message.peerInfo->counters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2266
2267             // Update E2T instance level metrics
2268             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2269             message.peerInfo->sctpParams->e2tCounters[IN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2270 #endif
2271             if (sendRequestToXapp(message, RIC_SUB_DEL_RESP, rmrMessageBuffer) != 0) {
2272                 mdclog_write(MDCLOG_ERR, "Subscription delete successful message failed to send to xAPP");
2273             }
2274             break;
2275         }
2276         default: {
2277             mdclog_write(MDCLOG_WARN, "Undefined or not supported message = %ld", procedureCode);
2278             message.message.messageType = 0; // no RMR message type yet
2279             buildJsonMessage(message);
2280
2281             break;
2282         }
2283     }
2284 }
2285
2286 /**
2287  *
2288  * @param pdu
2289  * @param message
2290  * @param rmrMessageBuffer
2291  */
2292 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
2293                         Sctp_Map_t *sctpMap,
2294                         ReportingMessages_t &message,
2295                         RmrMessagesBuffer_t &rmrMessageBuffer) {
2296     auto procedureCode = pdu->choice.unsuccessfulOutcome->procedureCode;
2297     auto logLevel = mdclog_level_get();
2298     if (logLevel >= MDCLOG_INFO) {
2299         mdclog_write(MDCLOG_INFO, "Unsuccessful Outcome %ld", procedureCode);
2300     }
2301     switch (procedureCode) {
2302         case ProcedureCode_id_RICcontrol: {
2303             if (logLevel >= MDCLOG_DEBUG) {
2304                 mdclog_write(MDCLOG_DEBUG, "Got RICcontrol %s", message.message.enodbName);
2305             }
2306             for (int i = 0;
2307                  i < pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.count; i++) {
2308                 auto messageSent = false;
2309                 RICcontrolFailure_IEs_t *ie = pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.array[i];
2310                 if (logLevel >= MDCLOG_DEBUG) {
2311                     mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
2312                 }
2313                 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
2314                     if (logLevel >= MDCLOG_DEBUG) {
2315                         mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
2316                     }
2317                     if (ie->value.present == RICcontrolFailure_IEs__value_PR_RICrequestID) {
2318                         message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_FAILURE;
2319                         rmrMessageBuffer.sendMessage->state = 0;
2320 //                        rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricRequestorID;
2321                         rmrMessageBuffer.sendMessage->sub_id = (int)ie->value.choice.RICrequestID.ricInstanceID;
2322                         static unsigned char tx[32];
2323                         snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2324                         rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
2325                         rmr_bytes2meid(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.enodbName,
2326                                        strlen(message.message.enodbName));
2327 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2328                         message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2329                         message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2330
2331                         // Update E2T instance level metrics
2332                         message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2333                         message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment((double)message.message.asnLength);
2334 #endif
2335                         sendRmrMessage(rmrMessageBuffer, message);
2336                         messageSent = true;
2337                     } else {
2338                         mdclog_write(MDCLOG_ERR, "RIC request id missing illegal request");
2339                     }
2340                 }
2341                 if (messageSent) {
2342                     break;
2343                 }
2344             }
2345             break;
2346         }
2347         case ProcedureCode_id_RICsubscription: {
2348             if (logLevel >= MDCLOG_DEBUG) {
2349                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscription %s", message.message.enodbName);
2350             }
2351 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2352             message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2353             message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2354
2355             // Update E2T instance level metrics
2356             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2357             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment((double)message.message.asnLength);
2358 #endif
2359             if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) {
2360                 mdclog_write(MDCLOG_ERR, "Subscription unsuccessful message failed to send to xAPP");
2361             }
2362             break;
2363         }
2364         case ProcedureCode_id_RICsubscriptionDelete: {
2365             if (logLevel >= MDCLOG_DEBUG) {
2366                 mdclog_write(MDCLOG_DEBUG, "Got RICsubscriptionDelete %s", message.message.enodbName);
2367             }
2368 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2369             message.peerInfo->counters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2370             message.peerInfo->counters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2371
2372             // Update E2T instance level metrics
2373             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2374             message.peerInfo->sctpParams->e2tCounters[IN_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment((double)message.message.asnLength);
2375 #endif
2376             if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer) != 0) {
2377                 mdclog_write(MDCLOG_ERR, "Subscription Delete unsuccessful message failed to send to xAPP");
2378             }
2379             break;
2380         }
2381         default: {
2382             mdclog_write(MDCLOG_WARN, "Undefined or not supported message = %ld", procedureCode);
2383             message.message.messageType = 0; // no RMR message type yet
2384 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2385             buildJsonMessage(message);
2386 #endif
2387             break;
2388         }
2389     }
2390 }
2391
2392 /**
2393  *
2394  * @param message
2395  * @param requestId
2396  * @param rmrMmessageBuffer
2397  * @return
2398  */
2399 int sendRequestToXapp(ReportingMessages_t &message,
2400                       int requestId,
2401                       RmrMessagesBuffer_t &rmrMmessageBuffer) {
2402     rmr_bytes2meid(rmrMmessageBuffer.sendMessage,
2403                    (unsigned char *)message.message.enodbName,
2404                    strlen(message.message.enodbName));
2405     message.message.messageType = rmrMmessageBuffer.sendMessage->mtype = requestId;
2406     rmrMmessageBuffer.sendMessage->state = 0;
2407     static unsigned char tx[32];
2408     snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2409     rmr_bytes2xact(rmrMmessageBuffer.sendMessage, tx, strlen((const char *) tx));
2410
2411     auto rc = sendRmrMessage(rmrMmessageBuffer, message);
2412     return rc;
2413 }
2414
2415 /**
2416  *
2417  * @param pSctpParams
2418  */
2419 void getRmrContext(sctp_params_t &pSctpParams) {
2420     pSctpParams.rmrCtx = nullptr;
2421     pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RECEIVE_XAPP_BUFFER_SIZE, RMRFL_NONE);
2422     if (pSctpParams.rmrCtx == nullptr) {
2423         mdclog_write(MDCLOG_ERR, "Failed to initialize RMR");
2424         return;
2425     }
2426
2427     rmr_set_stimeout(pSctpParams.rmrCtx, 0);    // disable retries for any send operation
2428     // we need to find that routing table exist and we can run
2429     if (mdclog_level_get() >= MDCLOG_INFO) {
2430         mdclog_write(MDCLOG_INFO, "We are after RMR INIT wait for RMR_Ready");
2431     }
2432     int rmrReady = 0;
2433     int count = 0;
2434     while (!rmrReady) {
2435         if ((rmrReady = rmr_ready(pSctpParams.rmrCtx)) == 0) {
2436             sleep(1);
2437         }
2438         count++;
2439         if (count % 60 == 0) {
2440             mdclog_write(MDCLOG_INFO, "waiting to RMR ready state for %d seconds", count);
2441         }
2442     }
2443     if (mdclog_level_get() >= MDCLOG_INFO) {
2444         mdclog_write(MDCLOG_INFO, "RMR running");
2445     }
2446     rmr_init_trace(pSctpParams.rmrCtx, 200);
2447     // get the RMR fd for the epoll
2448     pSctpParams.rmrListenFd = rmr_get_rcvfd(pSctpParams.rmrCtx);
2449     struct epoll_event event{};
2450     // add RMR fd to epoll
2451     event.events = (EPOLLIN);
2452     event.data.fd = pSctpParams.rmrListenFd;
2453     // add listening RMR FD to epoll
2454     if (epoll_ctl(pSctpParams.epoll_fd, EPOLL_CTL_ADD, pSctpParams.rmrListenFd, &event)) {
2455         mdclog_write(MDCLOG_ERR, "Failed to add RMR descriptor to epoll");
2456         close(pSctpParams.rmrListenFd);
2457         rmr_close(pSctpParams.rmrCtx);
2458         pSctpParams.rmrCtx = nullptr;
2459     }
2460 }
2461
2462 /**
2463  *
2464  * @param message
2465  * @param rmrMessageBuffer
2466  * @return
2467  */
2468 int PER_FromXML(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer) {
2469     E2AP_PDU_t *pdu = nullptr;
2470
2471     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2472         mdclog_write(MDCLOG_DEBUG, "got xml Format  data from xApp of size %d is:%s",
2473                 rmrMessageBuffer.rcvMessage->len, rmrMessageBuffer.rcvMessage->payload);
2474     }
2475     auto rval = asn_decode(nullptr, ATS_BASIC_XER, &asn_DEF_E2AP_PDU, (void **) &pdu,
2476                            rmrMessageBuffer.rcvMessage->payload, rmrMessageBuffer.rcvMessage->len);
2477     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2478         mdclog_write(MDCLOG_DEBUG, "%s After  decoding the XML to PDU", __func__ );
2479     }
2480     if (rval.code != RC_OK) {
2481 #ifdef UNIT_TEST
2482     return 0;
2483 #endif
2484         mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) setup response  from E2MGR : %s",
2485                      rval.code,
2486                      message.message.enodbName);
2487         if (pdu != nullptr) {
2488             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2489             pdu = nullptr;
2490         }
2491         return -1;
2492     }
2493
2494     int buff_size = RECEIVE_XAPP_BUFFER_SIZE;
2495     auto er = asn_encode_to_buffer(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, pdu,
2496                                    rmrMessageBuffer.rcvMessage->payload, buff_size);
2497     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2498         mdclog_write(MDCLOG_DEBUG, "%s After encoding PDU to PER", __func__ );
2499     }
2500     if (er.encoded == -1) {
2501         mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2AP_PDU.name, strerror(errno));
2502         if (pdu != nullptr) {
2503             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2504             pdu = nullptr;
2505         }
2506         return -1;
2507     } else if (er.encoded > (ssize_t)buff_size) {
2508         mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s, at %s line %d",
2509                      (int)rmrMessageBuffer.rcvMessage->len,
2510                      asn_DEF_E2AP_PDU.name,
2511                      __func__,
2512                      __LINE__);
2513         if (pdu != nullptr) {
2514             ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2515             pdu = nullptr;
2516         }
2517         return -1;
2518     }
2519     rmrMessageBuffer.rcvMessage->len = er.encoded;
2520     if (pdu != nullptr) {
2521         ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
2522         pdu = nullptr;
2523     }
2524     return 0;
2525 }
2526
2527 /**
2528  *
2529  * @param sctpMap
2530  * @param rmrMessageBuffer
2531  * @param ts
2532  * @return
2533  */
2534 int receiveXappMessages(Sctp_Map_t *sctpMap,
2535                         RmrMessagesBuffer_t &rmrMessageBuffer,
2536                         struct timespec &ts) {
2537     int loglevel = mdclog_level_get();
2538     if (rmrMessageBuffer.rcvMessage == nullptr) {
2539         //we have error
2540         mdclog_write(MDCLOG_ERR, "RMR Allocation message, %s", strerror(errno));
2541         return -1;
2542     }
2543
2544 //    if (loglevel >= MDCLOG_DEBUG) {
2545 //        mdclog_write(MDCLOG_DEBUG, "Call to rmr_rcv_msg");
2546 //    }
2547     rmrMessageBuffer.rcvMessage = rmr_rcv_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage);
2548     if (rmrMessageBuffer.rcvMessage == nullptr) {
2549         mdclog_write(MDCLOG_ERR, "RMR Receiving message with null pointer, Reallocated rmr message buffer");
2550         rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2551         return -2;
2552     }
2553     ReportingMessages_t message;
2554     message.message.direction = 'D';
2555     message.message.time.tv_nsec = ts.tv_nsec;
2556     message.message.time.tv_sec = ts.tv_sec;
2557
2558     // get message payload
2559     //auto msgData = msg->payload;
2560 #ifdef UNIT_TEST
2561     rmrMessageBuffer.rcvMessage->state = 0;
2562 #endif
2563     if (rmrMessageBuffer.rcvMessage->state != 0) {
2564         mdclog_write(MDCLOG_ERR, "RMR Receiving message with stat = %d", rmrMessageBuffer.rcvMessage->state);
2565         return -1;
2566     }
2567     rmr_get_meid(rmrMessageBuffer.rcvMessage, (unsigned char *)message.message.enodbName);
2568     message.peerInfo = (ConnectedCU_t *) sctpMap->find(message.message.enodbName);
2569     if (message.peerInfo == nullptr) {
2570         auto type = rmrMessageBuffer.rcvMessage->mtype;
2571         switch (type) {
2572             case RIC_SCTP_CLEAR_ALL:
2573             case E2_TERM_KEEP_ALIVE_REQ:
2574             case RIC_HEALTH_CHECK_REQ:
2575                 break;
2576             default:
2577 #ifdef UNIT_TEST
2578     break;
2579 #endif
2580                 mdclog_write(MDCLOG_ERR, "Failed to send message no CU entry %s", message.message.enodbName);
2581                 return -1;
2582         }
2583     }
2584
2585     if (rmrMessageBuffer.rcvMessage->mtype != RIC_HEALTH_CHECK_REQ) {
2586         num_of_XAPP_messages.fetch_add(1, std::memory_order_release);
2587
2588     }
2589     switch (rmrMessageBuffer.rcvMessage->mtype) {
2590         case RIC_E2_SETUP_RESP : {
2591             if (loglevel >= MDCLOG_DEBUG) {
2592                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_SETUP_RESP");
2593             }
2594             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2595                 break;
2596             }
2597 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2598             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2599             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
2600
2601             // Update E2T instance level metrics
2602             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2603             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
2604 #endif
2605             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2606                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_RESP");
2607                 return -6;
2608             }
2609             break;
2610         }
2611         case RIC_E2_SETUP_FAILURE : {
2612             if (loglevel >= MDCLOG_DEBUG) {
2613                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_SETUP_FAILURE");
2614             }
2615             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2616                 break;
2617             }
2618 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2619             message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2620             message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
2621
2622             // Update E2T instance level metrics
2623             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_E2setup]->Increment();
2624             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_E2setup]->Increment(rmrMessageBuffer.rcvMessage->len);
2625 #endif
2626             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2627                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_SETUP_FAILURE");
2628                 return -6;
2629             }
2630             break;
2631         }
2632         case RIC_ERROR_INDICATION: {
2633             if (loglevel >= MDCLOG_DEBUG) {
2634                 mdclog_write(MDCLOG_DEBUG, "RIC_ERROR_INDICATION");
2635             }
2636 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2637             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2638             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(rmrMessageBuffer.rcvMessage->len);
2639
2640             // Update E2T instance level metrics
2641             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_ErrorIndication]->Increment();
2642             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_ErrorIndication]->Increment(rmrMessageBuffer.rcvMessage->len);
2643 #endif
2644             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2645                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ERROR_INDICATION");
2646                 return -6;
2647             }
2648             break;
2649         }
2650         case RIC_SUB_REQ: {
2651             if (loglevel >= MDCLOG_DEBUG) {
2652                 mdclog_write(MDCLOG_DEBUG, "RIC_SUB_REQ");
2653             }
2654 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2655             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2656             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment(rmrMessageBuffer.rcvMessage->len);
2657
2658             // Update E2T instance level metrics
2659             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscription]->Increment();
2660             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscription]->Increment(rmrMessageBuffer.rcvMessage->len);
2661 #endif
2662             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2663                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_REQ");
2664                 return -6;
2665             }
2666             break;
2667         }
2668         case RIC_SUB_DEL_REQ: {
2669             if (loglevel >= MDCLOG_DEBUG) {
2670                 mdclog_write(MDCLOG_DEBUG, "RIC_SUB_DEL_REQ");
2671             }
2672 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2673             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2674             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(rmrMessageBuffer.rcvMessage->len);
2675
2676             // Update E2T instance level metrics
2677             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment();
2678             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICsubscriptionDelete]->Increment(rmrMessageBuffer.rcvMessage->len);
2679 #endif
2680             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2681                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_DEL_REQ");
2682                 return -6;
2683             }
2684             break;
2685         }
2686         case RIC_CONTROL_REQ: {
2687             if (loglevel >= MDCLOG_DEBUG) {
2688                 mdclog_write(MDCLOG_DEBUG, "RIC_CONTROL_REQ");
2689             }
2690 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2691             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2692             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment(rmrMessageBuffer.rcvMessage->len);
2693
2694             // Update E2T instance level metrics
2695             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICcontrol]->Increment();
2696             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICcontrol]->Increment(rmrMessageBuffer.rcvMessage->len);
2697 #endif
2698             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2699                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_CONTROL_REQ");
2700                 return -6;
2701             }
2702             break;
2703         }
2704         case RIC_SERVICE_QUERY: {
2705             if (loglevel >= MDCLOG_DEBUG) {
2706                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_QUERY");
2707             }
2708             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2709                 break;
2710             }
2711 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2712             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment();
2713             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len);
2714
2715             // Update E2T instance level metrics
2716             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment();
2717             message.peerInfo->sctpParams->e2tCounters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_RICserviceQuery]->Increment(rmrMessageBuffer.rcvMessage->len);
2718 #endif
2719             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2720                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_QUERY");
2721                 return -6;
2722             }
2723             break;
2724         }
2725         case RIC_SERVICE_UPDATE_ACK: {
2726             if (loglevel >= MDCLOG_DEBUG) {
2727                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_UPDATE_ACK");
2728             }
2729             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2730                 mdclog_write(MDCLOG_ERR, "error in PER_FromXML");
2731                 break;
2732             }
2733 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2734             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2735             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
2736
2737             // Update E2T instance level metrics
2738             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2739             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
2740 #endif
2741             if (loglevel >= MDCLOG_DEBUG) {
2742                 mdclog_write(MDCLOG_DEBUG, "Before sending to CU");
2743             }
2744             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2745                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_ACK");
2746                 return -6;
2747             }
2748             break;
2749         }
2750         case RIC_SERVICE_UPDATE_FAILURE: {
2751             if (loglevel >= MDCLOG_DEBUG) {
2752                 mdclog_write(MDCLOG_DEBUG, "RIC_SERVICE_UPDATE_FAILURE");
2753             }
2754             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2755                 break;
2756             }
2757 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2758             message.peerInfo->counters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2759             message.peerInfo->counters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
2760
2761             // Update E2T instance level metrics
2762             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][MSG_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment();
2763             message.peerInfo->sctpParams->e2tCounters[OUT_UN_SUCC][BYTES_COUNTER][ProcedureCode_id_RICserviceUpdate]->Increment(rmrMessageBuffer.rcvMessage->len);
2764 #endif
2765             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2766                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_FAILURE");
2767                 return -6;
2768             }
2769             break;
2770         }
2771         case RIC_E2_RESET_REQ: {
2772             if (loglevel >= MDCLOG_DEBUG) {
2773                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_RESET_REQ");
2774             }
2775             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2776                 break;
2777             }
2778 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2779             message.peerInfo->counters[OUT_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2780             message.peerInfo->counters[OUT_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
2781
2782             // Update E2T instance level metrics
2783             message.peerInfo->sctpParams->e2tCounters[IN_INITI][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2784             message.peerInfo->sctpParams->e2tCounters[IN_INITI][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
2785 #endif
2786             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2787                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET");
2788                 return -6;
2789             }
2790             break;
2791         }
2792         case RIC_E2_RESET_RESP: {
2793             if (loglevel >= MDCLOG_DEBUG) {
2794                 mdclog_write(MDCLOG_DEBUG, "RIC_E2_RESET_RESP");
2795             }
2796             if (PER_FromXML(message, rmrMessageBuffer) != 0) {
2797                 break;
2798             }
2799 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2800             message.peerInfo->counters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2801             message.peerInfo->counters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
2802
2803             // Update E2T instance level metrics
2804             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][MSG_COUNTER][ProcedureCode_id_Reset]->Increment();
2805             message.peerInfo->sctpParams->e2tCounters[OUT_SUCC][BYTES_COUNTER][ProcedureCode_id_Reset]->Increment(rmrMessageBuffer.rcvMessage->len);
2806 #endif
2807             if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap) != 0) {
2808                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_E2_RESET_RESP");
2809                 return -6;
2810             }
2811             break;
2812         }
2813         case RIC_SCTP_CLEAR_ALL: {
2814             mdclog_write(MDCLOG_INFO, "RIC_SCTP_CLEAR_ALL");
2815             // loop on all keys and close socket and then erase all map.
2816             vector<char *> v;
2817             sctpMap->getKeys(v);
2818             for (auto const &iter : v) { //}; iter != sctpMap.end(); iter++) {
2819                 if (!boost::starts_with((string) (iter), "host:") && !boost::starts_with((string) (iter), "msg:")) {
2820                     auto *peerInfo = (ConnectedCU_t *) sctpMap->find(iter);
2821                     if (peerInfo == nullptr) {
2822                         continue;
2823                     }
2824                     close(peerInfo->fileDescriptor);
2825                     memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
2826                     message.message.direction = 'D';
2827                     message.message.time.tv_nsec = ts.tv_nsec;
2828                     message.message.time.tv_sec = ts.tv_sec;
2829
2830                     message.message.asnLength = rmrMessageBuffer.sendMessage->len =
2831                             snprintf((char *)rmrMessageBuffer.sendMessage->payload,
2832                                      256,
2833                                      "%s|RIC_SCTP_CLEAR_ALL",
2834                                      peerInfo->enodbName);
2835                     message.message.asndata = rmrMessageBuffer.sendMessage->payload;
2836                     mdclog_write(MDCLOG_INFO, "%s", message.message.asndata);
2837                     if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer) != 0) {
2838                         mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
2839                     }
2840                     free(peerInfo);
2841                 }
2842             }
2843
2844             sleep(1);
2845             sctpMap->clear();
2846             break;
2847         }
2848         case E2_TERM_KEEP_ALIVE_REQ: {
2849             // send message back
2850             rmr_bytes2payload(rmrMessageBuffer.sendMessage,
2851                               (unsigned char *)rmrMessageBuffer.ka_message,
2852                               rmrMessageBuffer.ka_message_len);
2853             rmrMessageBuffer.sendMessage->mtype = E2_TERM_KEEP_ALIVE_RESP;
2854             rmrMessageBuffer.sendMessage->state = 0;
2855             static unsigned char tx[32];
2856             auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2857             rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, txLen);
2858 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2859             rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
2860 #endif
2861             if (rmrMessageBuffer.sendMessage == nullptr) {
2862                 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2863                 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP RMR message returned NULL");
2864             } else if (rmrMessageBuffer.sendMessage->state != 0)  {
2865                 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP, on RMR state = %d ( %s)",
2866                              rmrMessageBuffer.sendMessage->state, translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str());
2867             } else if (loglevel >= MDCLOG_DEBUG) {
2868                 mdclog_write(MDCLOG_DEBUG, "Got Keep Alive Request send : %s", rmrMessageBuffer.ka_message);
2869             }
2870
2871             break;
2872         }
2873         case RIC_HEALTH_CHECK_REQ: {
2874             static int counter = 0;
2875             // send message back
2876             rmr_bytes2payload(rmrMessageBuffer.rcvMessage,
2877                               (unsigned char *)"OK",
2878                               2);
2879             rmrMessageBuffer.rcvMessage->mtype = RIC_HEALTH_CHECK_RESP;
2880             rmrMessageBuffer.rcvMessage->state = 0;
2881             static unsigned char tx[32];
2882             auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2883             rmr_bytes2xact(rmrMessageBuffer.rcvMessage, tx, txLen);
2884             rmrMessageBuffer.rcvMessage = rmr_rts_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage);
2885             //rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
2886             if (rmrMessageBuffer.rcvMessage == nullptr) {
2887                 rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2888                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP RMR message returned NULL");
2889             } else if (rmrMessageBuffer.rcvMessage->state != 0)  {
2890                 mdclog_write(MDCLOG_ERR, "Failed to send RIC_HEALTH_CHECK_RESP, on RMR state = %d ( %s)",
2891                              rmrMessageBuffer.rcvMessage->state, translateRmrErrorMessages(rmrMessageBuffer.rcvMessage->state).c_str());
2892             } else if (loglevel >= MDCLOG_DEBUG && (++counter % 100 == 0)) {
2893                 mdclog_write(MDCLOG_DEBUG, "Got %d RIC_HEALTH_CHECK_REQ Request send : OK", counter);
2894             }
2895
2896             break;
2897         }
2898
2899         default:
2900             mdclog_write(MDCLOG_WARN, "Message Type : %d is not supported", rmrMessageBuffer.rcvMessage->mtype);
2901             message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
2902             message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
2903             message.message.time.tv_nsec = ts.tv_nsec;
2904             message.message.time.tv_sec = ts.tv_sec;
2905             message.message.messageType = rmrMessageBuffer.rcvMessage->mtype;
2906
2907             buildJsonMessage(message);
2908
2909
2910             return -7;
2911     }
2912     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2913         mdclog_write(MDCLOG_DEBUG, "EXIT OK from %s", __FUNCTION__);
2914     }
2915     return 0;
2916 }
2917
2918 /**
2919  * Send message to the CU that is not expecting for successful or unsuccessful results
2920  * @param messageBuffer
2921  * @param message
2922  * @param failedMsgId
2923  * @param sctpMap
2924  * @return
2925  */
2926 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
2927                            ReportingMessages_t &message,
2928                            int failedMsgId,
2929                            Sctp_Map_t *sctpMap) {
2930     if (mdclog_level_get() >= MDCLOG_DEBUG) {
2931         mdclog_write(MDCLOG_DEBUG, "send message: %d to %s address", message.message.messageType, message.message.enodbName);
2932     }
2933
2934     getRequestMetaData(message, messageBuffer);
2935     if (mdclog_level_get() >= MDCLOG_INFO) {
2936         mdclog_write(MDCLOG_INFO, "send message to %s address", message.message.enodbName);
2937     }
2938
2939     auto rc = sendMessagetoCu(sctpMap, messageBuffer, message, failedMsgId);
2940     return rc;
2941 }
2942
2943 /**
2944  *
2945  * @param sctpMap
2946  * @param messageBuffer
2947  * @param message
2948  * @param failedMesgId
2949  * @return
2950  */
2951 int sendMessagetoCu(Sctp_Map_t *sctpMap,
2952                     RmrMessagesBuffer_t &messageBuffer,
2953                     ReportingMessages_t &message,
2954                     int failedMesgId) {
2955     // get the FD
2956     message.message.messageType = messageBuffer.rcvMessage->mtype;
2957     auto rc = sendSctpMsg(message.peerInfo, message, sctpMap);
2958     return rc;
2959 }
2960
2961
2962 /**
2963  *
2964  * @param epoll_fd
2965  * @param peerInfo
2966  * @param events
2967  * @param sctpMap
2968  * @param enodbName
2969  * @param msgType
2970  * @return
2971  */
2972 int addToEpoll(int epoll_fd,
2973                ConnectedCU_t *peerInfo,
2974                uint32_t events,
2975                Sctp_Map_t *sctpMap,
2976                char *enodbName,
2977                int msgType) {
2978     // Add to Epol
2979     struct epoll_event event{};
2980     event.data.ptr = peerInfo;
2981     event.events = events;
2982     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, peerInfo->fileDescriptor, &event) < 0) {
2983 #if !(defined(UNIT_TEST) || defined(MODULE_TEST))
2984         if (mdclog_level_get() >= MDCLOG_DEBUG) {
2985             mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here), %s, %s %d",
2986                          strerror(errno), __func__, __LINE__);
2987         }
2988         close(peerInfo->fileDescriptor);
2989         if (enodbName != nullptr) {
2990             cleanHashEntry(peerInfo, sctpMap);
2991             char key[MAX_ENODB_NAME_SIZE * 2];
2992             snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", enodbName, msgType);
2993             if (mdclog_level_get() >= MDCLOG_DEBUG) {
2994                 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
2995             }
2996             auto tmp = sctpMap->find(key);
2997             if (tmp) {
2998                 free(tmp);
2999                 sctpMap->erase(key);
3000             }
3001         } else {
3002             peerInfo->enodbName[0] = 0;
3003         }
3004         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here)");
3005         return -1;
3006 #endif
3007     }
3008     return 0;
3009 }
3010
3011 /**
3012  *
3013  * @param epoll_fd
3014  * @param peerInfo
3015  * @param events
3016  * @param sctpMap
3017  * @param enodbName
3018  * @param msgType
3019  * @return
3020  */
3021 int modifyToEpoll(int epoll_fd,
3022                   ConnectedCU_t *peerInfo,
3023                   uint32_t events,
3024                   Sctp_Map_t *sctpMap,
3025                   char *enodbName,
3026                   int msgType) {
3027     // Add to Epol
3028     struct epoll_event event{};
3029     event.data.ptr = peerInfo;
3030     event.events = events;
3031     if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, peerInfo->fileDescriptor, &event) < 0) {
3032         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3033             mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_MOD (may check not to quit here), %s, %s %d",
3034                          strerror(errno), __func__, __LINE__);
3035         }
3036         close(peerInfo->fileDescriptor);
3037         cleanHashEntry(peerInfo, sctpMap);
3038         char key[MAX_ENODB_NAME_SIZE * 2];
3039         snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", enodbName, msgType);
3040         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3041             mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
3042         }
3043         auto tmp = sctpMap->find(key);
3044         if (tmp) {
3045             free(tmp);
3046         }
3047         sctpMap->erase(key);
3048         mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may check not to quit here)");
3049         return -1;
3050     }
3051     return 0;
3052 }
3053
3054
3055 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message) {
3056     buildJsonMessage(message);
3057 #ifndef UNIT_TEST
3058     rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3059 #else
3060     rmrMessageBuffer.sendMessage->state = RMR_ERR_RETRY;
3061 #endif
3062     if (rmrMessageBuffer.sendMessage == nullptr) {
3063         rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3064         mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
3065         return -1;
3066     }
3067
3068     if (rmrMessageBuffer.sendMessage->state != 0) {
3069         char meid[RMR_MAX_MEID]{};
3070         if (rmrMessageBuffer.sendMessage->state == RMR_ERR_RETRY) {
3071             usleep(5);
3072             rmrMessageBuffer.sendMessage->state = 0;
3073             mdclog_write(MDCLOG_INFO, "RETRY sending Message type %d to Xapp from %s",
3074                          rmrMessageBuffer.sendMessage->mtype,
3075                          rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3076 #ifndef UNIT_TEST
3077             rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
3078 #endif
3079             if (rmrMessageBuffer.sendMessage == nullptr) {
3080                 mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
3081                 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
3082                 return -1;
3083             } else if (rmrMessageBuffer.sendMessage->state != 0) {
3084                 mdclog_write(MDCLOG_ERR,
3085                              "Message state %s while sending request %d to Xapp from %s after retry of 10 microseconds",
3086                              translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
3087                              rmrMessageBuffer.sendMessage->mtype,
3088                              rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3089                 auto rc = rmrMessageBuffer.sendMessage->state;
3090                 return rc;
3091             }
3092         } else {
3093             mdclog_write(MDCLOG_ERR, "Message state %s while sending request %d to Xapp from %s",
3094                          translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
3095                          rmrMessageBuffer.sendMessage->mtype,
3096                          rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
3097             return rmrMessageBuffer.sendMessage->state;
3098         }
3099     }
3100     return 0;
3101 }
3102
3103 void buildJsonMessage(ReportingMessages_t &message) {
3104 #ifdef UNIT_TEST
3105     jsonTrace = true;
3106 #endif
3107     if (jsonTrace) {
3108         message.outLen = sizeof(message.base64Data);
3109         base64::encode((const unsigned char *) message.message.asndata,
3110                        (const int) message.message.asnLength,
3111                        message.base64Data,
3112                        message.outLen);
3113         if (mdclog_level_get() >= MDCLOG_DEBUG) {
3114             mdclog_write(MDCLOG_DEBUG, "Tracing: ASN length = %d, base64 message length = %d ",
3115                          (int) message.message.asnLength,
3116                          (int) message.outLen);
3117         }
3118
3119         snprintf(message.buffer, sizeof(message.buffer),
3120                  "{\"header\": {\"ts\": \"%ld.%09ld\","
3121                  "\"ranName\": \"%s\","
3122                  "\"messageType\": %d,"
3123                  "\"direction\": \"%c\"},"
3124                  "\"base64Length\": %d,"
3125                  "\"asnBase64\": \"%s\"}",
3126                  message.message.time.tv_sec,
3127                  message.message.time.tv_nsec,
3128                  message.message.enodbName,
3129                  message.message.messageType,
3130                  message.message.direction,
3131                  (int) message.outLen,
3132                  message.base64Data);
3133         static src::logger_mt &lg = my_logger::get();
3134
3135         BOOST_LOG(lg) << message.buffer;
3136     }
3137 }
3138
3139
3140 /**
3141  * take RMR error code to string
3142  * @param state
3143  * @return
3144  */
3145 string translateRmrErrorMessages(int state) {
3146     string str = {};
3147     switch (state) {
3148         case RMR_OK:
3149             str = "RMR_OK - state is good";
3150             break;
3151         case RMR_ERR_BADARG:
3152             str = "RMR_ERR_BADARG - argument passed to function was unusable";
3153             break;
3154         case RMR_ERR_NOENDPT:
3155             str = "RMR_ERR_NOENDPT - send//call could not find an endpoint based on msg type";
3156             break;
3157         case RMR_ERR_EMPTY:
3158             str = "RMR_ERR_EMPTY - msg received had no payload; attempt to send an empty message";
3159             break;
3160         case RMR_ERR_NOHDR:
3161             str = "RMR_ERR_NOHDR - message didn't contain a valid header";
3162             break;
3163         case RMR_ERR_SENDFAILED:
3164             str = "RMR_ERR_SENDFAILED - send failed; errno has nano reason";
3165             break;
3166         case RMR_ERR_CALLFAILED:
3167             str = "RMR_ERR_CALLFAILED - unable to send call() message";
3168             break;
3169         case RMR_ERR_NOWHOPEN:
3170             str = "RMR_ERR_NOWHOPEN - no wormholes are open";
3171             break;
3172         case RMR_ERR_WHID:
3173             str = "RMR_ERR_WHID - wormhole id was invalid";
3174             break;
3175         case RMR_ERR_OVERFLOW:
3176             str = "RMR_ERR_OVERFLOW - operation would have busted through a buffer/field size";
3177             break;
3178         case RMR_ERR_RETRY:
3179             str = "RMR_ERR_RETRY - request (send/call/rts) failed, but caller should retry (EAGAIN for wrappers)";
3180             break;
3181         case RMR_ERR_RCVFAILED:
3182             str = "RMR_ERR_RCVFAILED - receive failed (hard error)";
3183             break;
3184         case RMR_ERR_TIMEOUT:
3185             str = "RMR_ERR_TIMEOUT - message processing call timed out";
3186             break;
3187         case RMR_ERR_UNSET:
3188             str = "RMR_ERR_UNSET - the message hasn't been populated with a transport buffer";
3189             break;
3190         case RMR_ERR_TRUNC:
3191             str = "RMR_ERR_TRUNC - received message likely truncated";
3192             break;
3193         case RMR_ERR_INITFAILED:
3194             str = "RMR_ERR_INITFAILED - initialisation of something (probably message) failed";
3195             break;
3196         case RMR_ERR_NOTSUPP:
3197             str = "RMR_ERR_NOTSUPP - the request is not supported, or RMr was not initialised for the request";
3198             break;
3199         default:
3200             char buf[128]{};
3201             snprintf(buf, sizeof buf, "UNDOCUMENTED RMR_ERR : %d", state);
3202             str = buf;
3203             break;
3204     }
3205     return str;
3206 }