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