1 // Copyright 2019 AT&T Intellectual Property
2 // Copyright 2019 Nokia
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
8 // http://www.apache.org/licenses/LICENSE-2.0
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.
16 // This source code is part of the near-RT RIC (RAN Intelligent Controller)
17 // platform project (RICP).
19 // TODO: High-level file comment.
22 #include "sctpThread.h"
25 using namespace std::placeholders;
26 using namespace boost::filesystem;
29 using namespace opentracing;
36 // need to expose without the include of gcov
37 extern "C" void __gcov_flush(void);
39 static void catch_function(int signal) {
45 BOOST_LOG_INLINE_GLOBAL_LOGGER_DEFAULT(my_logger, src::logger_mt)
47 boost::shared_ptr<sinks::synchronous_sink<sinks::text_file_backend>> boostLogger;
48 double cpuClock = 0.0;
49 bool jsonTrace = true;
53 mdclog_attr_init(&attr);
54 mdclog_attr_set_ident(attr, "E2Terminator");
56 mdclog_attr_destroy(attr);
58 auto start_time = std::chrono::high_resolution_clock::now();
59 typedef std::chrono::duration<double, std::ratio<1,1>> seconds_t;
62 return seconds_t(std::chrono::high_resolution_clock::now() - start_time).count();
65 double approx_CPU_MHz(unsigned sleeptime) {
66 using namespace std::chrono_literals;
68 uint64_t cycles_start = rdtscp(aux);
69 double time_start = age();
70 std::this_thread::sleep_for(sleeptime * 1ms);
71 uint64_t elapsed_cycles = rdtscp(aux) - cycles_start;
72 double elapsed_time = age() - time_start;
73 return elapsed_cycles / elapsed_time;
76 //std::atomic<int64_t> rmrCounter{0};
77 std::atomic<int64_t> num_of_messages{0};
78 std::atomic<int64_t> num_of_XAPP_messages{0};
79 static long transactionCounter = 0;
82 int main(const int argc, char **argv) {
83 sctp_params_t sctpParams;
88 opentracing::Tracer::InitGlobal(tracelibcpp::createTracer("E2 Terminator"));
89 auto span = opentracing::Tracer::Global()->StartSpan(__FUNCTION__);
95 std::random_device device{};
96 std::mt19937 generator(device());
97 std::uniform_int_distribution<long> distribution(1, (long) 1e12);
98 transactionCounter = distribution(generator);
101 uint64_t st = 0,en = 0;
106 unsigned num_cpus = std::thread::hardware_concurrency();
108 mdclog_level_set(MDCLOG_INFO);
110 if (std::signal(SIGINT, catch_function) == SIG_ERR) {
111 mdclog_write(MDCLOG_ERR, "Errir initializing SIGINT");
114 if (std::signal(SIGABRT, catch_function)== SIG_ERR) {
115 mdclog_write(MDCLOG_ERR, "Errir initializing SIGABRT");
118 if (std::signal(SIGTERM, catch_function)== SIG_ERR) {
119 mdclog_write(MDCLOG_ERR, "Errir initializing SIGTERM");
124 cpuClock = approx_CPU_MHz(100);
126 mdclog_write(MDCLOG_DEBUG, "CPU speed %11.11f", cpuClock);
127 auto result = parse(argc, argv, sctpParams);
129 path p = (sctpParams.configFilePath + "/" + sctpParams.configFileName).c_str();
131 const int size = 2048;
132 auto fileSize = file_size(p);
133 if (fileSize > size) {
134 mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
138 mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
144 if (conf.openConfigFile(p.string()) == -1) {
145 mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
146 p.string().c_str(), strerror(errno));
149 int rmrPort = conf.getIntValue("nano");
151 mdclog_write(MDCLOG_ERR, "illigal RMR port ");
154 sctpParams.rmrPort = (uint16_t)rmrPort;
155 snprintf(sctpParams.rmrAddress, sizeof(sctpParams.rmrAddress), "%d", (int) (sctpParams.rmrPort));
157 auto tmpStr = conf.getStringValue("loglevel");
158 if (tmpStr.length() == 0) {
159 mdclog_write(MDCLOG_ERR, "illigal loglevel. Set loglevel to MDCLOG_INFO");
162 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
164 if ((tmpStr.compare("debug")) == 0) {
165 sctpParams.logLevel = MDCLOG_DEBUG;
166 } else if ((tmpStr.compare("info")) == 0) {
167 sctpParams.logLevel = MDCLOG_INFO;
168 } else if ((tmpStr.compare("warning")) == 0) {
169 sctpParams.logLevel = MDCLOG_WARN;
170 } else if ((tmpStr.compare("error")) == 0) {
171 sctpParams.logLevel = MDCLOG_ERR;
173 mdclog_write(MDCLOG_ERR, "illigal loglevel = %s. Set loglevel to MDCLOG_INFO", tmpStr.c_str());
174 sctpParams.logLevel = MDCLOG_INFO;
176 mdclog_level_set(sctpParams.logLevel);
178 tmpStr = conf.getStringValue("volume");
179 if (tmpStr.length() == 0) {
180 mdclog_write(MDCLOG_ERR, "illigal volume.");
184 char tmpLogFilespec[VOLUME_URL_SIZE];
185 tmpLogFilespec[0] = 0;
186 sctpParams.volume[0] = 0;
187 snprintf(sctpParams.volume, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
188 // copy the name to temp file as well
189 snprintf(tmpLogFilespec, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
192 // define the file name in the tmp directory under the volume
193 strcat(tmpLogFilespec,"/tmp/E2Term_%Y-%m-%d_%H-%M-%S.%N.tmpStr");
195 // std::string localIP = conf.getStringValue("local-ip");
196 // if (localIP.length() == 0) {
197 // mdclog_write(MDCLOG_ERR, "illigal local-ip. environment variable");
201 //sctpParams.myIP.assign(getenv(localIP.c_str()));
202 sctpParams.myIP = conf.getStringValue("local-ip");
203 if (sctpParams.myIP.length() == 0) {
204 mdclog_write(MDCLOG_ERR, "illigal local-ip.");
208 sctpParams.myIP = conf.getStringValue("external-fqdn");
209 if (sctpParams.myIP.length() == 0) {
210 mdclog_write(MDCLOG_ERR, "illigal external-fqdn.");
214 std::string pod = conf.getStringValue("pod_name");
215 if (pod.length() == 0) {
216 mdclog_write(MDCLOG_ERR, "illigal pod_name");
219 sctpParams.podName.assign(getenv(sctpParams.podName.c_str()));
220 if (sctpParams.podName.length() == 0) {
221 mdclog_write(MDCLOG_ERR, "illigal pod_name");
225 tmpStr = conf.getStringValue("trace");
226 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
227 if ((tmpStr.compare("start")) == 0) {
228 mdclog_write(MDCLOG_INFO, "Trace set to: start");
229 sctpParams.trace = true;
230 } else if ((tmpStr.compare("stop")) == 0) {
231 mdclog_write(MDCLOG_INFO, "Trace set to: stop");
232 sctpParams.trace = false;
234 jsonTrace = sctpParams.trace;
238 mdclog_write(MDCLOG_INFO, "start = %lx end = %lx diff = %lx\n", st, en, en - st);
239 mdclog_write(MDCLOG_INFO, "start high = %lx start lo = %lx end high = %lx end lo = %lx\n",
240 st >> 32, st & 0xFFFFFFFF, (int64_t)en >> 32, en & 0xFFFFFFFF);
241 mdclog_write(MDCLOG_INFO, "ellapsed time = %5.9f\n", (double)(en - st)/cpuClock);
243 if (mdclog_level_get() >= MDCLOG_INFO) {
244 mdclog_mdc_add("RMR Port", to_string(sctpParams.rmrPort).c_str());
245 mdclog_mdc_add("LogLevel", to_string(sctpParams.logLevel).c_str());
246 mdclog_mdc_add("volume", sctpParams.volume);
247 mdclog_mdc_add("tmpLogFilespec", tmpLogFilespec);
248 mdclog_mdc_add("my ip", sctpParams.myIP.c_str());
250 mdclog_write(MDCLOG_INFO, "running parameters");
253 sctpParams.ka_message_length = snprintf(sctpParams.ka_message, 4096, "{\"address\": \"%s:%d\","
255 "\"pod_name\": \"%s\"}",
256 (const char *)sctpParams.myIP.c_str(),
258 sctpParams.fqdn.c_str(),
259 sctpParams.podName.c_str());
262 // Files written to the current working directory
263 boostLogger = logging::add_file_log(
264 keywords::file_name = tmpLogFilespec, // to temp directory
265 keywords::rotation_size = 10 * 1024 * 1024,
266 keywords::time_based_rotation = sinks::file::rotation_at_time_interval(posix_time::hours(1)),
267 keywords::format = "%Message%"
268 //keywords::format = "[%TimeStamp%]: %Message%" // use each tmpStr with time stamp
271 // Setup a destination folder for collecting rotated (closed) files --since the same volumn can use rename()
272 boostLogger->locked_backend()->set_file_collector(sinks::file::make_collector(
273 keywords::target = sctpParams.volume
276 // Upon restart, scan the directory for files matching the file_name pattern
277 boostLogger->locked_backend()->scan_for_files();
279 // Enable auto-flushing after each tmpStr record written
280 if (mdclog_level_get() >= MDCLOG_DEBUG) {
281 boostLogger->locked_backend()->auto_flush(true);
285 sctpParams.epoll_fd = epoll_create1(0);
286 if (sctpParams.epoll_fd == -1) {
287 mdclog_write(MDCLOG_ERR, "failed to open epoll descriptor");
291 getRmrContext(sctpParams, &span);
292 if (sctpParams.rmrCtx == nullptr) {
293 close(sctpParams.epoll_fd);
297 if (buildInotify(sctpParams) == -1) {
298 close(sctpParams.rmrListenFd);
299 rmr_close(sctpParams.rmrCtx);
300 close(sctpParams.epoll_fd);
304 sctpParams.sctpMap = new mapWrapper();
306 std::vector<std::thread> threads(num_cpus);
307 // std::vector<std::thread> threads;
310 for (unsigned int i = 0; i < num_cpus; i++) {
311 threads[i] = std::thread(listener, &sctpParams);
316 int rc = pthread_setaffinity_np(threads[i].native_handle(), sizeof(cpu_set_t), &cpuset);
318 mdclog_write(MDCLOG_ERR, "Error calling pthread_setaffinity_np: %d", rc);
322 //loop over term_init until first message from xApp
323 handleTermInit(sctpParams);
325 for (auto &t : threads) {
330 opentracing::Tracer::Global()->Close();
335 void handleTermInit(sctp_params_t &sctpParams) {
336 sendTermInit(sctpParams);
337 //send to e2 manager init of e2 term
342 auto xappMessages = num_of_XAPP_messages.load(std::memory_order_acquire);
343 if (xappMessages > 0) {
344 if (mdclog_level_get() >= MDCLOG_INFO) {
345 mdclog_write(MDCLOG_INFO, "Got a message from some appliction, stop sending E@_TERM_INIT");
351 if (count % 1000 == 0) {
352 mdclog_write(MDCLOG_ERR, "GOT No messages from any xApp");
353 sendTermInit(sctpParams);
358 void sendTermInit(sctp_params_t &sctpParams) {
359 rmr_mbuf_t *msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.ka_message_length);
362 msg->mtype = E2_TERM_INIT;
364 rmr_bytes2payload(msg, (unsigned char *)sctpParams.ka_message, sctpParams.ka_message_length);
365 static unsigned char tx[32];
366 auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
367 rmr_bytes2xact(msg, tx, txLen);
368 msg = rmr_send_msg(sctpParams.rmrCtx, msg);
369 if (msg == nullptr) {
370 msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.myIP.length());
371 } else if (msg->state == 0) {
373 if (mdclog_level_get() >= MDCLOG_INFO) {
374 mdclog_write(MDCLOG_INFO, "E2_TERM_INIT succsesfuly sent ");
378 if (count % 100 == 0) {
379 mdclog_write(MDCLOG_ERR, "Error sending E2_TERM_INIT cause : %d ", msg->state);
395 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &sctpParams) {
396 cxxopts::Options options(argv[0], "e2 term help");
397 options.positional_help("[optional args]").show_positional_help();
398 options.allow_unrecognised_options().add_options()
399 ("p,path", "config file path", cxxopts::value<std::string>(sctpParams.configFilePath)->default_value("config"))
400 ("f,file", "config file name", cxxopts::value<std::string>(sctpParams.configFileName)->default_value("config.conf"))
401 ("h,help", "Print help");
403 auto result = options.parse(argc, argv);
405 if (result.count("help")) {
406 std::cout << options.help({""}) << std::endl;
415 * @return -1 failed 0 success
417 int buildInotify(sctp_params_t &sctpParams) {
418 sctpParams.inotifyFD = inotify_init1(IN_NONBLOCK);
419 if (sctpParams.inotifyFD == -1) {
420 mdclog_write(MDCLOG_ERR, "Failed to init inotify (inotify_init1) %s", strerror(errno));
421 close(sctpParams.rmrListenFd);
422 rmr_close(sctpParams.rmrCtx);
423 close(sctpParams.epoll_fd);
427 sctpParams.inotifyWD = inotify_add_watch(sctpParams.inotifyFD,
428 (const char *)sctpParams.configFilePath.c_str(),
430 if (sctpParams.inotifyWD == -1) {
431 mdclog_write(MDCLOG_ERR, "Failed to add directory : %s to inotify (inotify_add_watch) %s",
432 sctpParams.configFilePath.c_str(),
434 close(sctpParams.inotifyFD);
438 struct epoll_event event{};
439 event.events = (EPOLLIN);
440 event.data.fd = sctpParams.inotifyFD;
441 // add listening RMR FD to epoll
442 if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.inotifyFD, &event)) {
443 mdclog_write(MDCLOG_ERR, "Failed to add inotify FD to epoll");
444 close(sctpParams.inotifyFD);
455 void listener(sctp_params_t *params) {
457 auto span = opentracing::Tracer::Global()->StartSpan(__FUNCTION__);
461 int num_of_SCTP_messages = 0;
462 auto totalTime = 0.0;
464 mdclog_level_set(params->logLevel);
466 std::thread::id this_id = std::this_thread::get_id();
468 streambuf *oldCout = cout.rdbuf();
469 ostringstream memCout;
471 cout.rdbuf(memCout.rdbuf());
473 //return to the normal cout
477 memcpy(tid, memCout.str().c_str(), memCout.str().length() < 32 ? memCout.str().length() : 31);
478 tid[memCout.str().length()] = 0;
479 mdclog_mdc_add("thread id", tid);
481 if (mdclog_level_get() >= MDCLOG_DEBUG) {
482 mdclog_write(MDCLOG_DEBUG, "started thread number %s", tid);
485 RmrMessagesBuffer_t rmrMessageBuffer{};
486 //create and init RMR
487 rmrMessageBuffer.rmrCtx = params->rmrCtx;
489 auto *events = (struct epoll_event *) calloc(MAXEVENTS, sizeof(struct epoll_event));
490 struct timespec end{0, 0};
491 struct timespec start{0, 0};
493 rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
494 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
496 memcpy(rmrMessageBuffer.ka_message, params->ka_message, params->ka_message_length);
497 rmrMessageBuffer.ka_message_len = params->ka_message_length;
498 rmrMessageBuffer.ka_message[rmrMessageBuffer.ka_message_len] = 0;
500 if (mdclog_level_get() >= MDCLOG_DEBUG) {
501 mdclog_write(MDCLOG_DEBUG, "keep alive message is : %s", rmrMessageBuffer.ka_message);
504 ReportingMessages_t message {};
506 for (int i = 0; i < MAX_RMR_BUFF_ARRY; i++) {
507 rmrMessageBuffer.rcvBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
508 rmrMessageBuffer.sendBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
512 if (mdclog_level_get() >= MDCLOG_DEBUG) {
513 mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait");
515 auto numOfEvents = epoll_wait(params->epoll_fd, events, MAXEVENTS, -1);
516 if (numOfEvents < 0 && errno == EINTR) {
517 if (mdclog_level_get() >= MDCLOG_DEBUG) {
518 mdclog_write(MDCLOG_DEBUG, "got EINTR : %s", strerror(errno));
522 if (numOfEvents < 0) {
523 mdclog_write(MDCLOG_ERR, "Epoll wait failed, errno = %s", strerror(errno));
526 for (auto i = 0; i < numOfEvents; i++) {
527 if (mdclog_level_get() >= MDCLOG_DEBUG) {
528 mdclog_write(MDCLOG_DEBUG, "handling epoll event %d out of %d", i + 1, numOfEvents);
530 clock_gettime(CLOCK_MONOTONIC, &message.message.time);
531 start.tv_sec = message.message.time.tv_sec;
532 start.tv_nsec = message.message.time.tv_nsec;
535 if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP)) {
536 handlepoll_error(events[i], message, rmrMessageBuffer, params, &span);
537 } else if (events[i].events & EPOLLOUT) {
538 handleEinprogressMessages(events[i], message, rmrMessageBuffer, params, &span);
539 } else if (params->rmrListenFd == events[i].data.fd) {
540 // got message from XAPP
541 num_of_XAPP_messages.fetch_add(1, std::memory_order_release);
542 num_of_messages.fetch_add(1, std::memory_order_release);
543 if (mdclog_level_get() >= MDCLOG_DEBUG) {
544 mdclog_write(MDCLOG_DEBUG, "new message from RMR");
546 if (receiveXappMessages(params->epoll_fd,
549 message.message.time,
551 mdclog_write(MDCLOG_ERR, "Error handling Xapp message");
553 } else if (params->inotifyFD == events[i].data.fd) {
554 mdclog_write(MDCLOG_INFO, "Got event from inotify (configuration update)");
555 handleConfigChange(params);
557 /* We RMR_ERR_RETRY have data on the fd waiting to be read. Read and display it.
558 * We must read whatever data is available completely, as we are running
559 * in edge-triggered mode and won't get a notification again for the same data. */
560 num_of_messages.fetch_add(1, std::memory_order_release);
561 if (mdclog_level_get() >= MDCLOG_DEBUG) {
562 mdclog_write(MDCLOG_DEBUG, "new message from SCTP, epoll flags are : %0x", events[i].events);
564 receiveDataFromSctp(&events[i],
566 num_of_SCTP_messages,
568 message.message.time,
572 clock_gettime(CLOCK_MONOTONIC, &end);
573 if (mdclog_level_get() >= MDCLOG_INFO) {
574 totalTime += ((end.tv_sec + 1.0e-9 * end.tv_nsec) -
575 ((double) start.tv_sec + 1.0e-9 * start.tv_nsec));
577 if (mdclog_level_get() >= MDCLOG_DEBUG) {
578 mdclog_write(MDCLOG_DEBUG, "message handling is %ld seconds %ld nanoseconds",
579 end.tv_sec - start.tv_sec,
580 end.tv_nsec - start.tv_nsec);
595 void handleConfigChange(sctp_params_t *sctpParams) {
596 char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));
597 const struct inotify_event *event;
600 path p = (sctpParams->configFilePath + "/" + sctpParams->configFileName).c_str();
601 auto endlessLoop = true;
602 while (endlessLoop) {
603 auto len = read(sctpParams->inotifyFD, buf, sizeof buf);
605 if (errno != EAGAIN) {
606 mdclog_write(MDCLOG_ERR, "read %s ", strerror(errno));
616 for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
617 event = (const struct inotify_event *)ptr;
618 if (event->mask & (uint32_t)IN_ISDIR) {
622 // the directory name
623 if (sctpParams->inotifyWD == event->wd) {
627 if (!(sctpParams->configFileName.compare(event->name))) {
631 // only the file we want
632 if (event->mask & (uint32_t)IN_CLOSE_WRITE) {
634 const int size = 2048;
635 auto fileSize = file_size(p);
636 if (fileSize > size) {
637 mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
641 mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
646 if (conf.openConfigFile(p.string()) == -1) {
647 mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
648 p.string().c_str(), strerror(errno));
652 auto tmpStr = conf.getStringValue("loglevel");
653 if (tmpStr.length() == 0) {
654 mdclog_write(MDCLOG_ERR, "illigal loglevel. Set loglevel to MDCLOG_INFO");
657 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
659 if ((tmpStr.compare("debug")) == 0) {
660 mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_DEBUG");
661 sctpParams->logLevel = MDCLOG_DEBUG;
662 } else if ((tmpStr.compare("info")) == 0) {
663 mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_INFO");
664 sctpParams->logLevel = MDCLOG_INFO;
665 } else if ((tmpStr.compare("warning")) == 0) {
666 mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_WARN");
667 sctpParams->logLevel = MDCLOG_WARN;
668 } else if ((tmpStr.compare("error")) == 0) {
669 mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_ERR");
670 sctpParams->logLevel = MDCLOG_ERR;
672 mdclog_write(MDCLOG_ERR, "illigal loglevel = %s. Set loglevel to MDCLOG_INFO", tmpStr.c_str());
673 sctpParams->logLevel = MDCLOG_INFO;
675 mdclog_level_set(sctpParams->logLevel);
678 tmpStr = conf.getStringValue("trace");
679 if (tmpStr.length() == 0) {
680 mdclog_write(MDCLOG_ERR, "illigal trace. Set trace to stop");
684 transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
685 if ((tmpStr.compare("start")) == 0) {
686 mdclog_write(MDCLOG_INFO, "Trace set to: start");
687 sctpParams->trace = true;
688 } else if ((tmpStr.compare("stop")) == 0) {
689 mdclog_write(MDCLOG_INFO, "Trace set to: stop");
690 sctpParams->trace = false;
692 mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
693 sctpParams->trace = false;
695 jsonTrace = sctpParams->trace;
706 * @param rmrMessageBuffer
710 void handleEinprogressMessages(struct epoll_event &event,
711 ReportingMessages_t &message,
712 RmrMessagesBuffer_t &rmrMessageBuffer,
713 sctp_params_t *params,
716 auto lspan = opentracing::Tracer::Global()->StartSpan(
717 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
721 auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
722 memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
724 mdclog_write(MDCLOG_INFO, "file descriptor %d got EPOLLOUT", peerInfo->fileDescriptor);
726 socklen_t retValLen = 0;
727 auto rc = getsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_ERROR, &retVal, &retValLen);
728 if (rc != 0 || retVal != 0) {
730 rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
731 "%s|Failed SCTP Connection, after EINPROGRESS the getsockopt%s",
732 peerInfo->enodbName, strerror(errno));
733 } else if (retVal != 0) {
734 rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
735 "%s|Failed SCTP Connection after EINPROGRESS, SO_ERROR",
736 peerInfo->enodbName);
739 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
740 message.message.asnLength = rmrMessageBuffer.sendMessage->len;
741 mdclog_write(MDCLOG_ERR, "%s", rmrMessageBuffer.sendMessage->payload);
742 message.message.direction = 'N';
743 if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &lspan) != 0) {
744 mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
746 memset(peerInfo->asnData, 0, peerInfo->asnLength);
747 peerInfo->asnLength = 0;
755 peerInfo->isConnected = true;
757 if (modifyToEpoll(params->epoll_fd, peerInfo, (EPOLLIN | EPOLLET), params->sctpMap, peerInfo->enodbName,
758 peerInfo->mtype, &lspan) != 0) {
759 mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_MOD");
766 message.message.asndata = (unsigned char *)peerInfo->asnData;
767 message.message.asnLength = peerInfo->asnLength;
768 message.message.messageType = peerInfo->mtype;
769 memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
770 num_of_messages.fetch_add(1, std::memory_order_release);
771 if (mdclog_level_get() >= MDCLOG_DEBUG) {
772 mdclog_write(MDCLOG_DEBUG, "send the delayed SETUP/ENDC SETUP to sctp for %s",
773 message.message.enodbName);
775 if (sendSctpMsg(peerInfo, message, params->sctpMap, &lspan) != 0) {
776 if (mdclog_level_get() >= MDCLOG_DEBUG) {
777 mdclog_write(MDCLOG_DEBUG, "Error write to SCTP %s %d", __func__, __LINE__);
785 memset(peerInfo->asnData, 0, peerInfo->asnLength);
786 peerInfo->asnLength = 0;
794 void handlepoll_error(struct epoll_event &event,
795 ReportingMessages_t &message,
796 RmrMessagesBuffer_t &rmrMessageBuffer,
797 sctp_params_t *params,
800 auto lspan = opentracing::Tracer::Global()->StartSpan(
801 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
805 if (event.data.fd != params->rmrListenFd) {
806 auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
807 mdclog_write(MDCLOG_ERR, "epoll error, events %0x on fd %d, RAN NAME : %s",
808 event.events, peerInfo->fileDescriptor, peerInfo->enodbName);
810 rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
811 "%s|Failed SCTP Connection",
812 peerInfo->enodbName);
813 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
814 message.message.asnLength = rmrMessageBuffer.sendMessage->len;
816 memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
817 message.message.direction = 'N';
818 if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &lspan) != 0) {
819 mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
822 close(peerInfo->fileDescriptor);
823 cleanHashEntry((ConnectedCU_t *) event.data.ptr, params->sctpMap, &lspan);
825 mdclog_write(MDCLOG_ERR, "epoll error, events %0x on RMR FD", event.events);
837 int setSocketNoBlocking(int socket) {
838 auto flags = fcntl(socket, F_GETFL, 0);
841 mdclog_mdc_add("func", "fcntl");
842 mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
847 flags = (unsigned) flags | (unsigned) O_NONBLOCK;
848 if (fcntl(socket, F_SETFL, flags) == -1) {
849 mdclog_mdc_add("func", "fcntl");
850 mdclog_write(MDCLOG_ERR, "%s, %s", __FUNCTION__, strerror(errno));
864 void cleanHashEntry(ConnectedCU_t *val, Sctp_Map_t *m, otSpan *pSpan) {
866 auto lspan = opentracing::Tracer::Global()->StartSpan(
867 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
872 auto port = (uint16_t) strtol(val->portNumber, &dummy, 10);
873 char searchBuff[256]{};
875 snprintf(searchBuff, sizeof searchBuff, "host:%s:%d", val->hostName, port);
876 m->erase(searchBuff);
878 m->erase(val->enodbName);
887 * @param fd file discriptor
888 * @param data the asn data to send
889 * @param len length of the data
890 * @param enodbName the enodbName as in the map for printing purpose
891 * @param m map host information
892 * @param mtype message number
893 * @return 0 success, anegative number on fail
895 int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_t *m, otSpan *pSpan) {
897 auto lspan = opentracing::Tracer::Global()->StartSpan(
898 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
902 auto loglevel = mdclog_level_get();
903 int fd = peerInfo->fileDescriptor;
904 if (loglevel >= MDCLOG_DEBUG) {
905 mdclog_write(MDCLOG_DEBUG, "Send SCTP message for CU %s, %s",
906 message.message.enodbName, __FUNCTION__);
910 //TODO add send to VES client or KAFKA
911 //format ts|mtype|direction(D/U)|length of asn data|raw data
912 // auto length = sizeof message.message.time
913 // + sizeof message.message.enodbName
914 // + sizeof message.message.messageType
915 // + sizeof message.message.direction
916 // + sizeof message.message.asnLength
917 // + message.message.asnLength;
919 if (send(fd,message.message.asndata, message.message.asnLength,MSG_NOSIGNAL) < 0) {
920 if (errno == EINTR) {
923 mdclog_write(MDCLOG_ERR, "error writing to CU a message, %s ", strerror(errno));
924 // Prevent double free() of peerInfo in the event of connection failure.
925 // Returning failure will trigger, in x2/endc setup flow, RIC_SCTP_CONNECTION_FAILURE rmr message causing the E2M to retry.
926 if (!peerInfo->isConnected){
927 mdclog_write(MDCLOG_ERR, "connection to CU %s is still in progress.", message.message.enodbName);
933 cleanHashEntry(peerInfo, m, &lspan);
935 char key[MAX_ENODB_NAME_SIZE * 2];
936 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName,
937 message.message.messageType);
938 if (loglevel >= MDCLOG_DEBUG) {
939 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
941 auto tmp = m->find(key);
951 message.message.direction = 'D';
952 // send report.buffer of size
953 buildJsonMessage(message);
955 if (loglevel >= MDCLOG_DEBUG) {
956 mdclog_write(MDCLOG_DEBUG,
957 "SCTP message for CU %s sent from %s",
958 message.message.enodbName,
972 * @param rmrMessageBuffer
975 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, otSpan *pSpan) {
977 auto lspan = opentracing::Tracer::Global()->StartSpan(
978 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
982 rmr_get_meid(rmrMessageBuffer.rcvMessage, (unsigned char *)(message.message.enodbName));
984 message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
985 message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
987 if (mdclog_level_get() >= MDCLOG_DEBUG) {
988 mdclog_write(MDCLOG_DEBUG, "Message from Xapp RAN name = %s message length = %ld",
989 message.message.enodbName, (unsigned long) message.message.asnLength);
1000 * @param metaData all the data strip to structure
1001 * @param data the data recived from xAPP
1002 * @return 0 success all other values are fault
1004 int getSetupRequestMetaData(ReportingMessages_t &message, char *data, char *host, uint16_t &port, otSpan *pSpan) {
1006 auto lspan = opentracing::Tracer::Global()->StartSpan(
1007 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1009 // otSpan lspan = 0;
1011 auto loglevel = mdclog_level_get();
1013 char delimiter[4] {};
1014 memset(delimiter, 0, (size_t)4);
1018 char *val = strtok_r(data, delimiter, &tmp);
1019 if (val != nullptr) {
1020 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1021 mdclog_write(MDCLOG_DEBUG, "SCTP ADDRESS parameter from message = %s", val);
1023 memcpy(host, val, tmp - val );
1025 mdclog_write(MDCLOG_ERR, "wrong Host Name for setup request %s", data);
1032 val = strtok_r(nullptr, delimiter, &tmp);
1033 if (val != nullptr) {
1034 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1035 mdclog_write(MDCLOG_DEBUG, "PORT parameter from message = %s", val);
1038 port = (uint16_t)strtol(val, &dummy, 10);
1040 mdclog_write(MDCLOG_ERR, "wrong Port for setup request %s", data);
1047 val = strtok_r(nullptr, delimiter, &tmp);
1048 if (val != nullptr) {
1049 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1050 mdclog_write(MDCLOG_DEBUG, "RAN NAME parameter from message = %s", val);
1052 memcpy(message.message.enodbName, val, tmp - val);
1054 mdclog_write(MDCLOG_ERR, "wrong gNb/Enodeb name for setup request %s", data);
1061 val = strtok_r(nullptr, delimiter, &tmp);
1062 if (val != nullptr) {
1063 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1064 mdclog_write(MDCLOG_DEBUG, "ASN length parameter from message = %s", val);
1067 message.message.asnLength = (uint16_t) strtol(val, &dummy, 10);
1069 mdclog_write(MDCLOG_ERR, "wrong ASN length for setup request %s", data);
1076 message.message.asndata = (unsigned char *)tmp; // tmp is local but point to the location in data
1078 if (loglevel >= MDCLOG_INFO) {
1079 mdclog_write(MDCLOG_INFO, "Message from Xapp RAN name = %s host address = %s port = %d",
1080 message.message.enodbName, host, port);
1093 * @param numOfMessages
1094 * @param rmrMessageBuffer
1099 int receiveDataFromSctp(struct epoll_event *events,
1100 Sctp_Map_t *sctpMap,
1102 RmrMessagesBuffer_t &rmrMessageBuffer,
1103 struct timespec &ts,
1106 auto lspan = opentracing::Tracer::Global()->StartSpan(
1107 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1111 /* We have data on the fd waiting to be read. Read and display it.
1112 * We must read whatever data is available completely, as we are running
1113 * in edge-triggered mode and won't get a notification again for the same data. */
1115 auto loglevel = mdclog_level_get();
1116 // get the identity of the interface
1117 auto *peerInfo = (ConnectedCU_t *)events->data.ptr;
1118 struct timespec start{0, 0};
1119 struct timespec decodestart{0, 0};
1120 struct timespec end{0, 0};
1122 E2AP_PDU_t *pdu = nullptr;
1124 ReportingMessages_t message {};
1127 if (loglevel >= MDCLOG_DEBUG) {
1128 mdclog_write(MDCLOG_DEBUG, "Start Read from SCTP %d fd", peerInfo->fileDescriptor);
1129 clock_gettime(CLOCK_MONOTONIC, &start);
1131 // read the buffer directly to rmr payload
1132 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1133 message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1134 read(peerInfo->fileDescriptor, rmrMessageBuffer.sendMessage->payload, RECEIVE_SCTP_BUFFER_SIZE);
1135 if (loglevel >= MDCLOG_DEBUG) {
1136 mdclog_write(MDCLOG_DEBUG, "Finish Read from SCTP %d fd message length = %ld",
1137 peerInfo->fileDescriptor, message.message.asnLength);
1139 memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
1140 message.message.direction = 'U';
1141 message.message.time.tv_nsec = ts.tv_nsec;
1142 message.message.time.tv_sec = ts.tv_sec;
1144 if (message.message.asnLength < 0) {
1145 if (errno == EINTR) {
1148 /* If errno == EAGAIN, that means we have read all
1149 data. So go back to the main loop. */
1150 if (errno != EAGAIN) {
1151 mdclog_write(MDCLOG_ERR, "Read error, %s ", strerror(errno));
1153 } else if (loglevel >= MDCLOG_DEBUG) {
1154 mdclog_write(MDCLOG_DEBUG, "EAGAIN - descriptor = %d", peerInfo->fileDescriptor);
1157 } else if (message.message.asnLength == 0) {
1158 /* End of file. The remote has closed the connection. */
1159 if (loglevel >= MDCLOG_INFO) {
1160 mdclog_write(MDCLOG_INFO, "END of File Closed connection - descriptor = %d",
1161 peerInfo->fileDescriptor);
1167 asn_dec_rval_t rval;
1168 if (loglevel >= MDCLOG_DEBUG) {
1169 char printBuffer[4096]{};
1170 char *tmp = printBuffer;
1171 for (size_t i = 0; i < (size_t)message.message.asnLength; ++i) {
1172 snprintf(tmp, 2, "%02x", message.message.asndata[i]);
1175 printBuffer[message.message.asnLength] = 0;
1176 clock_gettime(CLOCK_MONOTONIC, &end);
1177 mdclog_write(MDCLOG_DEBUG, "Before Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1178 peerInfo->enodbName, end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1179 mdclog_write(MDCLOG_DEBUG, "PDU buffer length = %ld, data = : %s", message.message.asnLength,
1181 clock_gettime(CLOCK_MONOTONIC, &decodestart);
1184 rval = asn_decode(nullptr, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2AP_PDU, (void **) &pdu,
1185 message.message.asndata, message.message.asnLength);
1186 if (rval.code != RC_OK) {
1187 mdclog_write(MDCLOG_ERR, "Error %d Decoding (unpack) E2AP PDU from RAN : %s", rval.code,
1188 peerInfo->enodbName);
1192 if (loglevel >= MDCLOG_DEBUG) {
1193 clock_gettime(CLOCK_MONOTONIC, &end);
1194 mdclog_write(MDCLOG_DEBUG, "After Encoding E2AP PDU for : %s, Read time is : %ld seconds, %ld nanoseconds",
1195 peerInfo->enodbName, end.tv_sec - decodestart.tv_sec, end.tv_nsec - decodestart.tv_nsec);
1198 FILE *stream = open_memstream(&printBuffer, &size);
1199 asn_fprint(stream, &asn_DEF_E2AP_PDU, pdu);
1200 mdclog_write(MDCLOG_DEBUG, "Encoding E2AP PDU past : %s", printBuffer);
1201 clock_gettime(CLOCK_MONOTONIC, &decodestart);
1204 switch (pdu->present) {
1205 case E2AP_PDU_PR_initiatingMessage: {//initiating message
1206 asnInitiatingRequest(pdu, message, rmrMessageBuffer, &lspan);
1209 case E2AP_PDU_PR_successfulOutcome: { //successful outcome
1210 asnSuccsesfulMsg(pdu, message, sctpMap, rmrMessageBuffer, &lspan);
1213 case E2AP_PDU_PR_unsuccessfulOutcome: { //Unsuccessful Outcome
1214 asnUnSuccsesfulMsg(pdu, message, sctpMap, rmrMessageBuffer, &lspan);
1218 mdclog_write(MDCLOG_ERR, "Unknown index %d in E2AP PDU", pdu->present);
1221 if (loglevel >= MDCLOG_DEBUG) {
1222 clock_gettime(CLOCK_MONOTONIC, &end);
1223 mdclog_write(MDCLOG_DEBUG,
1224 "After processing message and sent to rmr for : %s, Read time is : %ld seconds, %ld nanoseconds",
1225 peerInfo->enodbName, end.tv_sec - decodestart.tv_sec, end.tv_nsec - decodestart.tv_nsec);
1229 // remove the break for EAGAIN
1231 if (pdu != nullptr) {
1232 //TODO need to test ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu); to get better performance
1233 //ASN_STRUCT_RESET(asn_DEF_E2AP_PDU, pdu);
1234 ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1237 //clock_gettime(CLOCK_MONOTONIC, &start);
1239 // in case of break to avoid memory leak
1240 if (pdu != nullptr) {
1241 ASN_STRUCT_FREE(asn_DEF_E2AP_PDU, pdu);
1246 if (loglevel >= MDCLOG_INFO) {
1247 mdclog_write(MDCLOG_INFO, "Closed connection - descriptor = %d", peerInfo->fileDescriptor);
1249 message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1250 snprintf((char *)rmrMessageBuffer.sendMessage->payload,
1252 "%s|CU disconnected unexpectedly",
1253 peerInfo->enodbName);
1254 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1256 if (sendRequestToXapp(message,
1257 RIC_SCTP_CONNECTION_FAILURE,
1260 mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
1263 /* Closing descriptor make epoll remove it from the set of descriptors which are monitored. */
1264 close(peerInfo->fileDescriptor);
1265 cleanHashEntry((ConnectedCU_t *) events->data.ptr, sctpMap, &lspan);
1267 if (loglevel >= MDCLOG_DEBUG) {
1268 clock_gettime(CLOCK_MONOTONIC, &end);
1269 mdclog_write(MDCLOG_DEBUG, "from receive SCTP to send RMR time is %ld seconds and %ld nanoseconds",
1270 end.tv_sec - start.tv_sec, end.tv_nsec - start.tv_nsec);
1284 * @param rmrMessageBuffer
1287 void asnInitiatingRequest(E2AP_PDU_t *pdu,
1288 ReportingMessages_t &message,
1289 RmrMessagesBuffer_t &rmrMessageBuffer,
1292 auto lspan = opentracing::Tracer::Global()->StartSpan(
1293 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1298 auto procedureCode = ((InitiatingMessage_t *) pdu->choice.initiatingMessage)->procedureCode;
1299 if (mdclog_level_get() >= MDCLOG_INFO) {
1300 mdclog_write(MDCLOG_INFO, "Initiating message %ld", procedureCode);
1302 switch (procedureCode) {
1303 case ProcedureCode_id_x2Setup: {
1304 if (mdclog_level_get() >= MDCLOG_INFO) {
1305 mdclog_write(MDCLOG_INFO, "Got Setup Initiating message from CU - %s",
1306 message.message.enodbName);
1310 case ProcedureCode_id_endcX2Setup: {
1311 if (mdclog_level_get() >= MDCLOG_INFO) {
1312 mdclog_write(MDCLOG_INFO, "Got X2 EN-DC Setup Request from CU - %s",
1313 message.message.enodbName);
1317 case ProcedureCode_id_ricSubscription: {
1318 if (mdclog_level_get() >= MDCLOG_INFO) {
1319 mdclog_write(MDCLOG_INFO, "Got RIC Subscription Request message from CU - %s",
1320 message.message.enodbName);
1324 case ProcedureCode_id_ricSubscriptionDelete: {
1325 if (mdclog_level_get() >= MDCLOG_INFO) {
1326 mdclog_write(MDCLOG_INFO, "Got RIC Subscription Delete Request message from CU - %s",
1327 message.message.enodbName);
1331 case ProcedureCode_id_endcConfigurationUpdate: {
1332 if (sendRequestToXapp(message, RIC_ENDC_CONF_UPDATE, rmrMessageBuffer, &lspan) != 0) {
1333 mdclog_write(MDCLOG_ERR, "E2 EN-DC CONFIGURATION UPDATE message failed to send to xAPP");
1337 case ProcedureCode_id_eNBConfigurationUpdate: {
1338 if (sendRequestToXapp(message, RIC_ENB_CONF_UPDATE, rmrMessageBuffer, &lspan) != 0) {
1339 mdclog_write(MDCLOG_ERR, "E2 EN-BC CONFIGURATION UPDATE message failed to send to xAPP");
1343 case ProcedureCode_id_x2Removal: {
1344 if (mdclog_level_get() >= MDCLOG_INFO) {
1345 mdclog_write(MDCLOG_INFO, "Got E2 Removal Initiating message from CU - %s",
1346 message.message.enodbName);
1350 case ProcedureCode_id_loadIndication: {
1351 if (sendRequestToXapp(message, RIC_ENB_LOAD_INFORMATION, rmrMessageBuffer, &lspan) != 0) {
1352 mdclog_write(MDCLOG_ERR, "Load indication message failed to send to xAPP");
1356 case ProcedureCode_id_resourceStatusReportingInitiation: {
1357 if (mdclog_level_get() >= MDCLOG_INFO) {
1358 mdclog_write(MDCLOG_INFO, "Got Status reporting initiation message from CU - %s",
1359 message.message.enodbName);
1363 case ProcedureCode_id_resourceStatusReporting: {
1364 if (sendRequestToXapp(message, RIC_RESOURCE_STATUS_UPDATE, rmrMessageBuffer, &lspan) != 0) {
1365 mdclog_write(MDCLOG_ERR, "Resource Status Reporting message failed to send to xAPP");
1369 case ProcedureCode_id_reset: {
1370 if (sendRequestToXapp(message, RIC_X2_RESET, rmrMessageBuffer, &lspan) != 0) {
1371 mdclog_write(MDCLOG_ERR, "RIC_X2_RESET message failed to send to xAPP");
1375 case ProcedureCode_id_ricIndication: {
1376 for (int i = 0; i < pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.count; i++) {
1377 auto messageSent = false;
1378 RICindication_IEs_t *ie = pdu->choice.initiatingMessage->value.choice.RICindication.protocolIEs.list.array[i];
1379 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1380 mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
1382 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
1383 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1384 mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
1386 if (ie->value.present == RICindication_IEs__value_PR_RICrequestID) {
1387 static unsigned char tx[32];
1388 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_INDICATION;
1389 snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1390 rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
1391 rmr_bytes2meid(rmrMessageBuffer.sendMessage,
1392 (unsigned char *)message.message.enodbName,
1393 strlen(message.message.enodbName));
1394 rmrMessageBuffer.sendMessage->state = 0;
1395 rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID;
1396 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1397 mdclog_write(MDCLOG_DEBUG, "RIC sub id = %d, message type = %d",
1398 rmrMessageBuffer.sendMessage->sub_id,
1399 rmrMessageBuffer.sendMessage->mtype);
1401 sendRmrMessage(rmrMessageBuffer, message, &lspan);
1404 mdclog_write(MDCLOG_ERR, "RIC request id missing illigal request");
1413 case ProcedureCode_id_errorIndication: {
1414 if (sendRequestToXapp(message, RIC_ERROR_INDICATION, rmrMessageBuffer, &lspan) != 0) {
1415 mdclog_write(MDCLOG_ERR, "Error Indication message failed to send to xAPP");
1419 case ProcedureCode_id_ricServiceUpdate : {
1420 if (sendRequestToXapp(message, RIC_SERVICE_UPDATE, rmrMessageBuffer, &lspan) != 0) {
1421 mdclog_write(MDCLOG_ERR, "Service Update message failed to send to xAPP");
1425 case ProcedureCode_id_gNBStatusIndication : {
1426 if (sendRequestToXapp(message, RIC_GNB_STATUS_INDICATION, rmrMessageBuffer, &lspan) != 0) {
1427 mdclog_write(MDCLOG_ERR, "RIC_GNB_STATUS_INDICATION failed to send to xAPP");
1432 mdclog_write(MDCLOG_ERR, "Undefined or not supported message = %ld", procedureCode);
1433 message.message.messageType = 0; // no RMR message type yet
1435 buildJsonMessage(message);
1451 * @param rmrMessageBuffer
1454 void asnSuccsesfulMsg(E2AP_PDU_t *pdu, ReportingMessages_t &message, Sctp_Map_t *sctpMap,
1455 RmrMessagesBuffer_t &rmrMessageBuffer, otSpan *pSpan) {
1457 auto lspan = opentracing::Tracer::Global()->StartSpan(
1458 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1462 auto procedureCode = pdu->choice.successfulOutcome->procedureCode;
1463 if (mdclog_level_get() >= MDCLOG_INFO) {
1464 mdclog_write(MDCLOG_INFO, "Successful Outcome %ld", procedureCode);
1466 switch (procedureCode) {
1467 case ProcedureCode_id_x2Setup: {
1468 if (mdclog_level_get() >= MDCLOG_INFO) {
1469 mdclog_write(MDCLOG_INFO, "Got Succesful Setup response from CU - %s",
1470 message.message.enodbName);
1472 if (sendResponseToXapp(message, RIC_X2_SETUP_RESP,
1473 RIC_X2_SETUP_REQ, rmrMessageBuffer, sctpMap, &lspan) != 0) {
1474 mdclog_write(MDCLOG_ERR, "Failed to send Succesful Setup response for CU - %s",
1475 message.message.enodbName);
1479 case ProcedureCode_id_endcX2Setup: { //X2_EN_DC_SETUP_REQUEST_FROM_CU
1480 if (mdclog_level_get() >= MDCLOG_INFO) {
1481 mdclog_write(MDCLOG_INFO, "Got Succesful E2 EN-DC Setup response from CU - %s",
1482 message.message.enodbName);
1484 if (sendResponseToXapp(message, RIC_ENDC_X2_SETUP_RESP,
1485 RIC_ENDC_X2_SETUP_REQ, rmrMessageBuffer, sctpMap, &lspan) != 0) {
1486 mdclog_write(MDCLOG_ERR, "Failed to send Succesful X2 EN DC Setup response for CU - %s",
1487 message.message.enodbName);
1491 case ProcedureCode_id_endcConfigurationUpdate: {
1492 if (mdclog_level_get() >= MDCLOG_INFO) {
1493 mdclog_write(MDCLOG_INFO, "Got Succesful E2 EN-DC CONFIGURATION UPDATE from CU - %s",
1494 message.message.enodbName);
1496 if (sendRequestToXapp(message, RIC_ENDC_CONF_UPDATE_ACK, rmrMessageBuffer, &lspan) != 0) {
1497 mdclog_write(MDCLOG_ERR, "Failed to send Succesful E2 EN DC CONFIGURATION response for CU - %s",
1498 message.message.enodbName);
1502 case ProcedureCode_id_eNBConfigurationUpdate: {
1503 if (mdclog_level_get() >= MDCLOG_INFO) {
1504 mdclog_write(MDCLOG_INFO, "Got Succesful E2 ENB CONFIGURATION UPDATE from CU - %s",
1505 message.message.enodbName);
1507 if (sendRequestToXapp(message, RIC_ENB_CONF_UPDATE_ACK, rmrMessageBuffer, &lspan) != 0) {
1508 mdclog_write(MDCLOG_ERR, "Failed to send Succesful E2 ENB CONFIGURATION response for CU - %s",
1509 message.message.enodbName);
1513 case ProcedureCode_id_reset: {
1514 if (sendRequestToXapp(message, RIC_X2_RESET_RESP, rmrMessageBuffer, &lspan) != 0) {
1515 mdclog_write(MDCLOG_ERR, "Failed to send Succesful E2_RESET response for CU - %s",
1516 message.message.enodbName);
1521 case ProcedureCode_id_resourceStatusReportingInitiation: {
1522 if (sendRequestToXapp(message, RIC_RES_STATUS_RESP, rmrMessageBuffer, &lspan) != 0) {
1523 mdclog_write(MDCLOG_ERR,
1524 "Failed to send Succesful 2_REQUEST_STATUS_REPORTING_INITIATION response for CU - %s",
1525 message.message.enodbName);
1529 case ProcedureCode_id_ricSubscription: {
1530 if (mdclog_level_get() >= MDCLOG_INFO) {
1531 mdclog_write(MDCLOG_INFO, "Got Succesful RIC Subscription response from CU - %s",
1532 message.message.enodbName);
1534 if (sendRequestToXapp(message, RIC_SUB_RESP, rmrMessageBuffer, &lspan) != 0) {
1535 mdclog_write(MDCLOG_ERR, "Subscription successful message failed to send to xAPP");
1540 case ProcedureCode_id_ricSubscriptionDelete: {
1541 if (mdclog_level_get() >= MDCLOG_INFO) {
1542 mdclog_write(MDCLOG_INFO,
1543 "Got Succesful RIC Subscription Delete response from CU - %s",
1544 message.message.enodbName);
1546 if (sendRequestToXapp(message, RIC_SUB_DEL_RESP, rmrMessageBuffer, &lspan) != 0) {
1547 mdclog_write(MDCLOG_ERR, "Subscription delete successful message failed to send to xAPP");
1551 case ProcedureCode_id_ricControl: {
1552 if (mdclog_level_get() >= MDCLOG_INFO) {
1553 mdclog_write(MDCLOG_INFO,
1554 "Got Succesful RIC control response from CU - %s",
1555 message.message.enodbName);
1558 i < pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.count; i++) {
1559 auto messageSent = false;
1560 RICcontrolAcknowledge_IEs_t *ie = pdu->choice.successfulOutcome->value.choice.RICcontrolAcknowledge.protocolIEs.list.array[i];
1561 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1562 mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
1564 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
1565 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1566 mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
1568 if (ie->value.present == RICcontrolAcknowledge_IEs__value_PR_RICrequestID) {
1569 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_ACK;
1570 rmrMessageBuffer.sendMessage->state = 0;
1571 rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID;
1572 static unsigned char tx[32];
1573 snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1574 rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
1575 rmr_bytes2meid(rmrMessageBuffer.sendMessage,
1576 (unsigned char *)message.message.enodbName,
1577 strlen(message.message.enodbName));
1579 sendRmrMessage(rmrMessageBuffer, message, &lspan);
1582 mdclog_write(MDCLOG_ERR, "RIC request id missing illigal request");
1592 mdclog_write(MDCLOG_WARN, "Undefined or not supported message = %ld", procedureCode);
1593 message.message.messageType = 0; // no RMR message type yet
1594 buildJsonMessage(message);
1610 * @param rmrMessageBuffer
1613 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
1614 ReportingMessages_t &message,
1615 Sctp_Map_t *sctpMap,
1616 RmrMessagesBuffer_t &rmrMessageBuffer,
1619 auto lspan = opentracing::Tracer::Global()->StartSpan(
1620 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1624 auto procedureCode = pdu->choice.unsuccessfulOutcome->procedureCode;
1625 if (mdclog_level_get() >= MDCLOG_INFO) {
1626 mdclog_write(MDCLOG_INFO, "Unsuccessful Outcome %ld", procedureCode);
1628 switch (procedureCode) {
1629 case ProcedureCode_id_x2Setup: {
1630 if (mdclog_level_get() >= MDCLOG_INFO) {
1631 mdclog_write(MDCLOG_INFO,
1632 "Got Unsuccessful Setup response from CU - %s",
1633 message.message.enodbName);
1635 if (sendResponseToXapp(message,
1636 RIC_X2_SETUP_FAILURE, RIC_X2_SETUP_REQ,
1640 mdclog_write(MDCLOG_ERR,
1641 "Failed to send Unsuccessful Setup response for CU - %s",
1642 message.message.enodbName);
1647 case ProcedureCode_id_endcX2Setup: {
1648 if (mdclog_level_get() >= MDCLOG_INFO) {
1649 mdclog_write(MDCLOG_INFO,
1650 "Got Unsuccessful E2 EN-DC Setup response from CU - %s",
1651 message.message.enodbName);
1653 if (sendResponseToXapp(message, RIC_ENDC_X2_SETUP_FAILURE,
1654 RIC_ENDC_X2_SETUP_REQ,
1658 mdclog_write(MDCLOG_ERR, "Failed to send Unsuccessful E2 EN DC Setup response for CU - %s",
1659 message.message.enodbName);
1663 case ProcedureCode_id_endcConfigurationUpdate: {
1664 if (sendRequestToXapp(message, RIC_ENDC_CONF_UPDATE_FAILURE, rmrMessageBuffer, &lspan) != 0) {
1665 mdclog_write(MDCLOG_ERR, "Failed to send Unsuccessful E2 EN DC CONFIGURATION response for CU - %s",
1666 message.message.enodbName);
1670 case ProcedureCode_id_eNBConfigurationUpdate: {
1671 if (sendRequestToXapp(message, RIC_ENB_CONF_UPDATE_FAILURE, rmrMessageBuffer, &lspan) != 0) {
1672 mdclog_write(MDCLOG_ERR, "Failed to send Unsuccessful E2 ENB CONFIGURATION response for CU - %s",
1673 message.message.enodbName);
1677 case ProcedureCode_id_resourceStatusReportingInitiation: {
1678 if (sendRequestToXapp(message, RIC_RES_STATUS_FAILURE, rmrMessageBuffer, &lspan) != 0) {
1679 mdclog_write(MDCLOG_ERR,
1680 "Failed to send Succesful E2_REQUEST_STATUS_REPORTING_INITIATION response for CU - %s",
1681 message.message.enodbName);
1685 case ProcedureCode_id_ricSubscription: {
1686 if (mdclog_level_get() >= MDCLOG_INFO) {
1687 mdclog_write(MDCLOG_INFO, "Got Unsuccessful RIC Subscription Response from CU - %s",
1688 message.message.enodbName);
1690 if (sendRequestToXapp(message, RIC_SUB_FAILURE, rmrMessageBuffer, &lspan) != 0) {
1691 mdclog_write(MDCLOG_ERR, "Subscription unsuccessful message failed to send to xAPP");
1695 case ProcedureCode_id_ricSubscriptionDelete: {
1696 if (mdclog_level_get() >= MDCLOG_INFO) {
1697 mdclog_write(MDCLOG_INFO, "Got Unsuccessful RIC Subscription Delete Response from CU - %s",
1698 message.message.enodbName);
1700 if (sendRequestToXapp(message, RIC_SUB_DEL_FAILURE, rmrMessageBuffer, &lspan) != 0) {
1701 mdclog_write(MDCLOG_ERR, "Subscription Delete unsuccessful message failed to send to xAPP");
1705 case ProcedureCode_id_ricControl: {
1706 if (mdclog_level_get() >= MDCLOG_INFO) {
1707 mdclog_write(MDCLOG_INFO, "Got UNSuccesful RIC control response from CU - %s",
1708 message.message.enodbName);
1711 i < pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.count; i++) {
1712 auto messageSent = false;
1713 RICcontrolFailure_IEs_t *ie = pdu->choice.unsuccessfulOutcome->value.choice.RICcontrolFailure.protocolIEs.list.array[i];
1714 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1715 mdclog_write(MDCLOG_DEBUG, "ie type (ProtocolIE_ID) = %ld", ie->id);
1717 if (ie->id == ProtocolIE_ID_id_RICrequestID) {
1718 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1719 mdclog_write(MDCLOG_DEBUG, "Got RIC requestId entry, ie type (ProtocolIE_ID) = %ld", ie->id);
1721 if (ie->value.present == RICcontrolFailure_IEs__value_PR_RICrequestID) {
1722 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_CONTROL_FAILURE;
1723 rmrMessageBuffer.sendMessage->state = 0;
1724 rmrMessageBuffer.sendMessage->sub_id = (int) ie->value.choice.RICrequestID.ricRequestorID;
1725 static unsigned char tx[32];
1726 snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1727 rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
1728 rmr_bytes2meid(rmrMessageBuffer.sendMessage, (unsigned char *)message.message.enodbName, strlen(message.message.enodbName));
1729 sendRmrMessage(rmrMessageBuffer, message, &lspan);
1732 mdclog_write(MDCLOG_ERR, "RIC request id missing illigal request");
1742 mdclog_write(MDCLOG_WARN, "Undefined or not supported message = %ld", procedureCode);
1743 message.message.messageType = 0; // no RMR message type yet
1745 buildJsonMessage(message);
1760 * @param rmrMmessageBuffer
1764 int sendRequestToXapp(ReportingMessages_t &message,
1766 RmrMessagesBuffer_t &rmrMmessageBuffer,
1769 auto lspan = opentracing::Tracer::Global()->StartSpan(
1770 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1774 rmr_bytes2meid(rmrMmessageBuffer.sendMessage,
1775 (unsigned char *)message.message.enodbName,
1776 strlen(message.message.enodbName));
1777 message.message.messageType = rmrMmessageBuffer.sendMessage->mtype = requestId;
1778 rmrMmessageBuffer.sendMessage->state = 0;
1779 static unsigned char tx[32];
1780 snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
1781 rmr_bytes2xact(rmrMmessageBuffer.sendMessage, tx, strlen((const char *) tx));
1783 auto rc = sendRmrMessage(rmrMmessageBuffer, message, &lspan);
1792 void getRmrContext(sctp_params_t &pSctpParams, otSpan *pSpan) {
1794 auto lspan = opentracing::Tracer::Global()->StartSpan(
1795 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1797 // otSpan lspan = 0;
1799 pSctpParams.rmrCtx = nullptr;
1800 pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RMR_MAX_RCV_BYTES, RMRFL_NONE);
1801 if (pSctpParams.rmrCtx == nullptr) {
1802 mdclog_write(MDCLOG_ERR, "Failed to initialize RMR");
1809 rmr_set_stimeout(pSctpParams.rmrCtx, 0); // disable retries for any send operation
1810 // we need to find that routing table exist and we can run
1811 if (mdclog_level_get() >= MDCLOG_INFO) {
1812 mdclog_write(MDCLOG_INFO, "We are after RMR INIT wait for RMR_Ready");
1817 if ((rmrReady = rmr_ready(pSctpParams.rmrCtx)) == 0) {
1821 if (count % 60 == 0) {
1822 mdclog_write(MDCLOG_INFO, "waiting to RMR ready state for %d seconds", count);
1825 if (mdclog_level_get() >= MDCLOG_INFO) {
1826 mdclog_write(MDCLOG_INFO, "RMR running");
1831 rmr_init_trace(pSctpParams.rmrCtx, 200);
1832 // get the RMR fd for the epoll
1833 pSctpParams.rmrListenFd = rmr_get_rcvfd(pSctpParams.rmrCtx);
1834 struct epoll_event event{};
1835 // add RMR fd to epoll
1836 event.events = (EPOLLIN);
1837 event.data.fd = pSctpParams.rmrListenFd;
1838 // add listening RMR FD to epoll
1839 if (epoll_ctl(pSctpParams.epoll_fd, EPOLL_CTL_ADD, pSctpParams.rmrListenFd, &event)) {
1840 mdclog_write(MDCLOG_ERR, "Failed to add RMR descriptor to epoll");
1841 close(pSctpParams.rmrListenFd);
1842 rmr_close(pSctpParams.rmrCtx);
1843 pSctpParams.rmrCtx = nullptr;
1851 * @param rmrMessageBuffer
1856 int receiveXappMessages(int epoll_fd,
1857 Sctp_Map_t *sctpMap,
1858 RmrMessagesBuffer_t &rmrMessageBuffer,
1859 struct timespec &ts,
1862 auto lspan = opentracing::Tracer::Global()->StartSpan(
1863 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
1867 if (rmrMessageBuffer.rcvMessage == nullptr) {
1869 mdclog_write(MDCLOG_ERR, "RMR Allocation message, %s", strerror(errno));
1877 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1878 mdclog_write(MDCLOG_DEBUG, "Call to rmr_rcv_msg");
1880 rmrMessageBuffer.rcvMessage = rmr_rcv_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.rcvMessage);
1881 if (rmrMessageBuffer.rcvMessage == nullptr) {
1882 mdclog_write(MDCLOG_ERR, "RMR Receving message with null pointer, Realloc rmr mesage buffer");
1883 rmrMessageBuffer.rcvMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
1890 ReportingMessages_t message;
1891 message.message.direction = 'D';
1892 message.message.time.tv_nsec = ts.tv_nsec;
1893 message.message.time.tv_sec = ts.tv_sec;
1895 // get message payload
1896 //auto msgData = msg->payload;
1897 if (rmrMessageBuffer.rcvMessage->state != 0) {
1898 mdclog_write(MDCLOG_ERR, "RMR Receving message with stat = %d", rmrMessageBuffer.rcvMessage->state);
1905 switch (rmrMessageBuffer.rcvMessage->mtype) {
1906 case RIC_X2_SETUP_REQ: {
1907 if (connectToCUandSetUp(rmrMessageBuffer, message, epoll_fd, sctpMap, &lspan) != 0) {
1908 mdclog_write(MDCLOG_ERR, "ERROR in connectToCUandSetUp on RIC_X2_SETUP_REQ");
1909 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_SCTP_CONNECTION_FAILURE;
1910 message.message.direction = 'N';
1911 message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1912 snprintf((char *)rmrMessageBuffer.sendMessage->payload,
1914 "ERROR in connectToCUandSetUp on RIC_X2_SETUP_REQ");
1915 rmrMessageBuffer.sendMessage->state = 0;
1916 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1918 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1919 mdclog_write(MDCLOG_DEBUG, "start writing to rmr buffer");
1921 rmr_bytes2xact(rmrMessageBuffer.sendMessage, rmrMessageBuffer.rcvMessage->xaction, RMR_MAX_XID);
1922 rmr_str2meid(rmrMessageBuffer.sendMessage, (unsigned char *)message.message.enodbName);
1924 sendRmrMessage(rmrMessageBuffer, message, &lspan);
1932 case RIC_ENDC_X2_SETUP_REQ: {
1933 if (connectToCUandSetUp(rmrMessageBuffer, message, epoll_fd, sctpMap, &lspan) != 0) {
1934 mdclog_write(MDCLOG_ERR, "ERROR in connectToCUandSetUp on RIC_ENDC_X2_SETUP_REQ");
1935 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = RIC_SCTP_CONNECTION_FAILURE;
1936 message.message.direction = 'N';
1937 message.message.asnLength = rmrMessageBuffer.sendMessage->len =
1938 snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
1939 "ERROR in connectToCUandSetUp on RIC_ENDC_X2_SETUP_REQ");
1940 rmrMessageBuffer.sendMessage->state = 0;
1941 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
1943 if (mdclog_level_get() >= MDCLOG_DEBUG) {
1944 mdclog_write(MDCLOG_DEBUG, "start writing to rmr buffer");
1947 rmr_bytes2xact(rmrMessageBuffer.sendMessage, rmrMessageBuffer.rcvMessage->xaction, RMR_MAX_XID);
1948 rmr_str2meid(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.enodbName);
1950 sendRmrMessage(rmrMessageBuffer, message, &lspan);
1958 case RIC_ENDC_CONF_UPDATE: {
1959 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
1960 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENDC_CONF_UPDATE");
1968 case RIC_ENDC_CONF_UPDATE_ACK: {
1969 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
1970 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENDC_CONF_UPDATE_ACK");
1978 case RIC_ENDC_CONF_UPDATE_FAILURE: {
1979 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
1980 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENDC_CONF_UPDATE_FAILURE");
1989 case RIC_ENB_CONF_UPDATE: {
1990 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
1991 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENDC_CONF_UPDATE");
1999 case RIC_ENB_CONF_UPDATE_ACK: {
2000 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2001 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENB_CONF_UPDATE_ACK");
2009 case RIC_ENB_CONF_UPDATE_FAILURE: {
2010 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2011 mdclog_write(MDCLOG_ERR, "Failed to send RIC_ENB_CONF_UPDATE_FAILURE");
2019 case RIC_RES_STATUS_REQ: {
2020 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2021 mdclog_write(MDCLOG_ERR, "Failed to send RIC_RES_STATUS_REQ");
2030 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2031 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_REQ");
2039 case RIC_SUB_DEL_REQ: {
2040 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2041 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SUB_DEL_REQ");
2049 case RIC_CONTROL_REQ: {
2050 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2051 mdclog_write(MDCLOG_ERR, "Failed to send RIC_CONTROL_REQ");
2059 case RIC_SERVICE_QUERY: {
2060 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2061 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_QUERY");
2069 case RIC_SERVICE_UPDATE_ACK: {
2070 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2071 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_ACK");
2079 case RIC_SERVICE_UPDATE_FAILURE: {
2080 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2081 mdclog_write(MDCLOG_ERR, "Failed to send RIC_SERVICE_UPDATE_FAILURE");
2089 case RIC_X2_RESET: {
2090 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2091 mdclog_write(MDCLOG_ERR, "Failed to send RIC_X2_RESET");
2099 case RIC_X2_RESET_RESP: {
2100 if (sendDirectionalSctpMsg(rmrMessageBuffer, message, 0, sctpMap, &lspan) != 0) {
2101 mdclog_write(MDCLOG_ERR, "Failed to send RIC_X2_RESET_RESP");
2109 case RIC_SCTP_CLEAR_ALL: {
2110 mdclog_write(MDCLOG_INFO, "RIC_SCTP_CLEAR_ALL");
2111 // loop on all keys and close socket and then erase all map.
2113 sctpMap->getKeys(v);
2114 for (auto const &iter : v) { //}; iter != sctpMap.end(); iter++) {
2115 if (!boost::starts_with((string) (iter), "host:") && !boost::starts_with((string) (iter), "msg:")) {
2116 auto *peerInfo = (ConnectedCU_t *) sctpMap->find(iter);
2117 if (peerInfo == nullptr) {
2120 close(peerInfo->fileDescriptor);
2121 memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
2122 message.message.direction = 'D';
2123 message.message.time.tv_nsec = ts.tv_nsec;
2124 message.message.time.tv_sec = ts.tv_sec;
2126 message.message.asnLength = rmrMessageBuffer.sendMessage->len =
2127 snprintf((char *)rmrMessageBuffer.sendMessage->payload,
2129 "%s|RIC_SCTP_CLEAR_ALL",
2130 peerInfo->enodbName);
2131 message.message.asndata = rmrMessageBuffer.sendMessage->payload;
2132 mdclog_write(MDCLOG_INFO, "%s", message.message.asndata);
2133 if (sendRequestToXapp(message,
2134 RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &lspan) != 0) {
2135 mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
2145 case E2_TERM_KEEP_ALIVE_REQ: {
2146 // send message back
2147 rmr_bytes2payload(rmrMessageBuffer.sendMessage,
2148 (unsigned char *)rmrMessageBuffer.ka_message,
2149 rmrMessageBuffer.ka_message_len);
2150 rmrMessageBuffer.sendMessage->mtype = E2_TERM_KEEP_ALIVE_RESP;
2151 rmrMessageBuffer.sendMessage->state = 0;
2152 static unsigned char tx[32];
2153 auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2154 rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, txLen);
2155 rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
2156 if (rmrMessageBuffer.sendMessage == nullptr) {
2157 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2158 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP");
2159 } else if (rmrMessageBuffer.sendMessage->state != 0) {
2160 mdclog_write(MDCLOG_ERR, "Failed to send E2_TERM_KEEP_ALIVE_RESP, on RMR state = %d", rmrMessageBuffer.sendMessage->state);
2161 } else if (mdclog_level_get() >= MDCLOG_INFO) {
2162 mdclog_write(MDCLOG_INFO, "Got Keep Alive Request send : %s", rmrMessageBuffer.ka_message);
2168 mdclog_write(MDCLOG_WARN, "Message Type : %d is not seported", rmrMessageBuffer.rcvMessage->mtype);
2169 message.message.asndata = rmrMessageBuffer.rcvMessage->payload;
2170 message.message.asnLength = rmrMessageBuffer.rcvMessage->len;
2171 message.message.time.tv_nsec = ts.tv_nsec;
2172 message.message.time.tv_sec = ts.tv_sec;
2173 message.message.messageType = rmrMessageBuffer.rcvMessage->mtype;
2175 buildJsonMessage(message);
2183 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2184 mdclog_write(MDCLOG_DEBUG, "EXIT OK from %s", __FUNCTION__);
2193 * Send message to the CU that is not expecting for successful or unsuccessful results
2194 * @param messageBuffer
2196 * @param failedMsgId
2201 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
2202 ReportingMessages_t &message,
2204 Sctp_Map_t *sctpMap,
2207 auto lspan = opentracing::Tracer::Global()->StartSpan(
2208 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2213 getRequestMetaData(message, messageBuffer, &lspan);
2214 if (mdclog_level_get() >= MDCLOG_INFO) {
2215 mdclog_write(MDCLOG_INFO, "send message to %s address", message.message.enodbName);
2218 auto rc = sendMessagetoCu(sctpMap, messageBuffer, message, failedMsgId, &lspan);
2229 * @param messageBuffer
2231 * @param failedMesgId
2235 int sendMessagetoCu(Sctp_Map_t *sctpMap,
2236 RmrMessagesBuffer_t &messageBuffer,
2237 ReportingMessages_t &message,
2241 auto lspan = opentracing::Tracer::Global()->StartSpan(
2242 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2246 auto *peerInfo = (ConnectedCU_t *) sctpMap->find(message.message.enodbName);
2247 if (peerInfo == nullptr) {
2248 if (failedMesgId != 0) {
2249 sendFailedSendingMessagetoXapp(messageBuffer, message, failedMesgId, &lspan);
2251 mdclog_write(MDCLOG_ERR, "Failed to send message no CU entry %s", message.message.enodbName);
2261 message.message.messageType = messageBuffer.rcvMessage->mtype;
2262 auto rc = sendSctpMsg(peerInfo, message, sctpMap, &lspan);
2272 * @param rmrCtx the rmr context to send and receive
2273 * @param msg the msg we got fromxApp
2274 * @param metaData data from xApp in ordered struct
2275 * @param failedMesgId the return message type error
2278 sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, int failedMesgId,
2281 auto lspan = opentracing::Tracer::Global()->StartSpan(
2282 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2286 rmr_mbuf_t *msg = rmrMessageBuffer.sendMessage;
2287 msg->len = snprintf((char *) msg->payload, 200, "the gNb/eNode name %s not found",
2288 message.message.enodbName);
2289 if (mdclog_level_get() >= MDCLOG_INFO) {
2290 mdclog_write(MDCLOG_INFO, "%s", msg->payload);
2292 msg->mtype = failedMesgId;
2295 static unsigned char tx[32];
2296 snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
2297 rmr_bytes2xact(msg, tx, strlen((const char *) tx));
2299 sendRmrMessage(rmrMessageBuffer, message, &lspan);
2301 lspan->Finish();pLogSink
2307 * Send Response back to xApp, message is used only when there was a request from the xApp
2309 * @param enodbName the name of the gNb/eNodeB
2310 * @param msgType the value of the message to the xApp
2311 * @param requestType The request that was sent by the xAPP
2312 * @param rmrCtx the rmr identifier
2313 * @param sctpMap hash map holds data on the requestrs
2314 * @param buf the buffer to send to xAPP
2315 * @param size size of the buffer to send
2318 int sendResponseToXapp(ReportingMessages_t &message,
2321 RmrMessagesBuffer_t &rmrMessageBuffer,
2322 Sctp_Map_t *sctpMap,
2325 auto lspan = opentracing::Tracer::Global()->StartSpan(
2326 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2330 char key[MAX_ENODB_NAME_SIZE * 2];
2331 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName, requestType);
2333 auto xact = sctpMap->find(key);
2334 if (xact == nullptr) {
2335 mdclog_write(MDCLOG_ERR, "NO Request %s found for this response from CU: %s", key,
2336 message.message.enodbName);
2343 sctpMap->erase(key);
2345 message.message.messageType = rmrMessageBuffer.sendMessage->mtype = msgType; //SETUP_RESPONSE_MESSAGE_TYPE;
2346 rmr_bytes2payload(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.asndata,
2347 message.message.asnLength);
2348 rmr_bytes2xact(rmrMessageBuffer.sendMessage, (const unsigned char *)xact, strlen((const char *)xact));
2349 rmr_str2meid(rmrMessageBuffer.sendMessage, (unsigned char *) message.message.enodbName);
2350 rmrMessageBuffer.sendMessage->state = 0;
2352 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2353 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
2357 auto rc = sendRmrMessage(rmrMessageBuffer, message, &lspan);
2365 * build the SCTP connection to eNodB or gNb
2366 * @param rmrMessageBuffer
2373 int connectToCUandSetUp(RmrMessagesBuffer_t &rmrMessageBuffer,
2374 ReportingMessages_t &message,
2376 Sctp_Map_t *sctpMap,
2379 auto lspan = opentracing::Tracer::Global()->StartSpan(
2380 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2384 struct sockaddr_in6 servaddr{};
2385 struct addrinfo hints{}, *result;
2386 auto msgData = rmrMessageBuffer.rcvMessage->payload;
2387 unsigned char meid[RMR_MAX_MEID]{};
2391 message.message.messageType = rmrMessageBuffer.rcvMessage->mtype;
2392 rmr_mbuf_t *msg = rmrMessageBuffer.rcvMessage;
2393 rmr_get_meid(msg, meid);
2395 if (mdclog_level_get() >= MDCLOG_INFO) {
2396 mdclog_write(MDCLOG_INFO, "message %d Received for MEID :%s. SETUP/EN-DC Setup Request from xApp, Message = %s",
2397 msg->mtype, meid, msgData);
2399 if (getSetupRequestMetaData(message, (char *)msgData, host, port, &lspan) < 0) {
2400 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2401 mdclog_write(MDCLOG_DEBUG, "Error in setup parameters %s, %d", __func__, __LINE__);
2409 //// message asndata points to the start of the asndata of the message and not to start of payload
2410 // search if the same host:port but not the same enodbname
2411 char searchBuff[256]{};
2412 snprintf(searchBuff, sizeof searchBuff, "host:%s:%d", host, port);
2413 auto e = (char *)sctpMap->find(searchBuff);
2415 // found one compare if not the same
2416 if (strcmp(message.message.enodbName, e) != 0) {
2417 mdclog_write(MDCLOG_ERR,
2418 "Try to connect CU %s to Host %s but %s already connected",
2419 message.message.enodbName, host, e);
2427 // check if not alread connected. if connected send the request and return
2428 auto *peerInfo = (ConnectedCU_t *)sctpMap->find(message.message.enodbName);
2429 if (peerInfo != nullptr) {
2432 // "Device %s already connected please remove and then setup again",
2433 // message.message.enodbName);
2434 if (mdclog_level_get() >= MDCLOG_INFO) {
2435 mdclog_write(MDCLOG_INFO,
2436 "Device already connected to %s",
2437 message.message.enodbName);
2439 message.message.messageType = msg->mtype;
2440 auto rc = sendSctpMsg(peerInfo, message, sctpMap, &lspan);
2442 mdclog_write(MDCLOG_ERR, "failed write to SCTP %s, %d", __func__, __LINE__);
2449 char key[MAX_ENODB_NAME_SIZE * 2];
2450 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName, msg->mtype);
2451 int xaction_len = strlen((const char *) msg->xaction);
2452 auto *xaction = (unsigned char *) calloc(1, xaction_len);
2453 memcpy(xaction, msg->xaction, xaction_len);
2454 sctpMap->setkey(key, xaction);
2455 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2456 mdclog_write(MDCLOG_DEBUG, "set key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
2464 peerInfo = (ConnectedCU_t *) calloc(1, sizeof(ConnectedCU_t));
2465 memcpy(peerInfo->enodbName, message.message.enodbName, sizeof(message.message.enodbName));
2468 if ((peerInfo->fileDescriptor = socket(AF_INET6, SOCK_STREAM, IPPROTO_SCTP)) < 0) {
2469 mdclog_write(MDCLOG_ERR, "Socket Error, %s %s, %d", strerror(errno), __func__, __LINE__);
2477 if (setsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof optval) != 0) {
2478 mdclog_write(MDCLOG_ERR, "setsockopt SO_REUSEPORT Error, %s %s, %d", strerror(errno), __func__, __LINE__);
2485 if (setsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof optval) != 0) {
2486 mdclog_write(MDCLOG_ERR, "setsockopt SO_REUSEADDR Error, %s %s, %d", strerror(errno), __func__, __LINE__);
2492 servaddr.sin6_family = AF_INET6;
2494 struct sockaddr_in6 localAddr {};
2495 localAddr.sin6_family = AF_INET6;
2496 localAddr.sin6_addr = in6addr_any;
2497 localAddr.sin6_port = htons(SRC_PORT);
2499 if (bind(peerInfo->fileDescriptor, (struct sockaddr*)&localAddr , sizeof(struct sockaddr_in6)) < 0) {
2500 mdclog_write(MDCLOG_ERR, "bind Socket Error, %s %s, %d", strerror(errno), __func__, __LINE__);
2505 }//Ends the binding.
2507 memset(&hints, 0, sizeof hints);
2508 hints.ai_flags = AI_NUMERICHOST;
2509 if (getaddrinfo(host, nullptr, &hints, &result) < 0) {
2510 close(peerInfo->fileDescriptor);
2511 mdclog_write(MDCLOG_ERR, "getaddrinfo error for %s, Error = %s", host, strerror(errno));
2517 memcpy(&servaddr, result->ai_addr, sizeof(struct sockaddr_in6));
2518 freeaddrinfo(result);
2520 servaddr.sin6_port = htons(port); /* daytime server */
2521 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2522 mdclog_write(MDCLOG_DEBUG, "Send Connect FD = %d host : %s port %d",
2523 peerInfo->fileDescriptor,
2529 if (addToEpoll(epoll_fd, peerInfo, (EPOLLOUT | EPOLLIN | EPOLLET), sctpMap, message.message.enodbName,
2530 msg->mtype, &lspan) != 0) {
2537 char hostBuff[NI_MAXHOST];
2538 char portBuff[NI_MAXHOST];
2540 if (getnameinfo((SA *) &servaddr, sizeof(servaddr),
2541 hostBuff, sizeof(hostBuff),
2542 portBuff, sizeof(portBuff),
2543 (uint) (NI_NUMERICHOST) | (uint) (NI_NUMERICSERV)) != 0) {
2544 mdclog_write(MDCLOG_ERR, "getnameinfo() Error, %s %s %d", strerror(errno), __func__, __LINE__);
2551 if (setSocketNoBlocking(peerInfo->fileDescriptor) != 0) {
2552 mdclog_write(MDCLOG_ERR, "setSocketNoBlocking failed to set new connection %s on sctpPort %s", hostBuff,
2554 close(peerInfo->fileDescriptor);
2561 memcpy(peerInfo->hostName, hostBuff, strlen(hostBuff));
2562 peerInfo->hostName[strlen(hostBuff)] = 0;
2563 memcpy(peerInfo->portNumber, portBuff, strlen(portBuff));
2564 peerInfo->portNumber[strlen(portBuff)] = 0;
2566 // map by enoodb/gnb name
2567 sctpMap->setkey(message.message.enodbName, peerInfo);
2568 //map host and port to enodeb
2569 sctpMap->setkey(searchBuff, message.message.enodbName);
2571 // save message for the return values
2572 char key[MAX_ENODB_NAME_SIZE * 2];
2573 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", message.message.enodbName, msg->mtype);
2574 int xaction_len = strlen((const char *) msg->xaction);
2575 auto *xaction = (unsigned char *) calloc(1, xaction_len);
2576 memcpy(xaction, msg->xaction, xaction_len);
2577 sctpMap->setkey(key, xaction);
2578 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2579 mdclog_write(MDCLOG_DEBUG, "End building peerinfo: %s for CU %s", key, message.message.enodbName);
2582 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2583 mdclog_write(MDCLOG_DEBUG, "Send connect to FD %d, %s, %d",
2584 peerInfo->fileDescriptor, __func__, __LINE__);
2586 if (connect(peerInfo->fileDescriptor, (SA *) &servaddr, sizeof(servaddr)) < 0) {
2587 if (errno != EINPROGRESS) {
2588 mdclog_write(MDCLOG_ERR, "connect FD %d to host : %s port %d, %s",
2589 peerInfo->fileDescriptor, host, port, strerror(errno));
2590 close(peerInfo->fileDescriptor);
2596 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2597 mdclog_write(MDCLOG_DEBUG,
2598 "Connect to FD %d returned with EINPROGRESS : %s",
2599 peerInfo->fileDescriptor, strerror(errno));
2601 // since message.message.asndata is pointing to the asndata in the rmr message payload we copy it like this
2602 memcpy(peerInfo->asnData, message.message.asndata, message.message.asnLength);
2603 peerInfo->asnLength = message.message.asnLength;
2604 peerInfo->mtype = msg->mtype;
2611 if (mdclog_level_get() >= MDCLOG_INFO) {
2612 mdclog_write(MDCLOG_INFO, "Connect to FD %d returned OK without EINPROGRESS", peerInfo->fileDescriptor);
2615 peerInfo->isConnected = true;
2617 if (modifyToEpoll(epoll_fd, peerInfo, (EPOLLIN | EPOLLET), sctpMap, message.message.enodbName, msg->mtype,
2625 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2626 mdclog_write(MDCLOG_DEBUG, "Connected to host : %s port %d", host, port);
2629 message.message.messageType = msg->mtype;
2630 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2631 mdclog_write(MDCLOG_DEBUG, "Send SCTP message to FD %d", peerInfo->fileDescriptor);
2633 if (sendSctpMsg(peerInfo, message, sctpMap, &lspan) != 0) {
2634 mdclog_write(MDCLOG_ERR, "Error write to SCTP %s %d", __func__, __LINE__);
2640 memset(peerInfo->asnData, 0, message.message.asnLength);
2641 peerInfo->asnLength = 0;
2642 peerInfo->mtype = 0;
2644 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2645 mdclog_write(MDCLOG_DEBUG, "Sent message to SCTP for %s", message.message.enodbName);
2664 int addToEpoll(int epoll_fd,
2665 ConnectedCU_t *peerInfo,
2667 Sctp_Map_t *sctpMap,
2672 auto lspan = opentracing::Tracer::Global()->StartSpan(
2673 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2678 struct epoll_event event{};
2679 event.data.ptr = peerInfo;
2680 event.events = events;
2681 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, peerInfo->fileDescriptor, &event) < 0) {
2682 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2683 mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_ADD (may chack not to quit here), %s, %s %d",
2684 strerror(errno), __func__, __LINE__);
2686 close(peerInfo->fileDescriptor);
2687 cleanHashEntry(peerInfo, sctpMap, &lspan);
2688 char key[MAX_ENODB_NAME_SIZE * 2];
2689 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", enodbName, msgType);
2690 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2691 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
2693 auto tmp = sctpMap->find(key);
2697 sctpMap->erase(key);
2698 mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may chack not to quit here)");
2721 int modifyToEpoll(int epoll_fd,
2722 ConnectedCU_t *peerInfo,
2724 Sctp_Map_t *sctpMap,
2729 auto lspan = opentracing::Tracer::Global()->StartSpan(
2730 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2735 struct epoll_event event{};
2736 event.data.ptr = peerInfo;
2737 event.events = events;
2738 if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, peerInfo->fileDescriptor, &event) < 0) {
2739 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2740 mdclog_write(MDCLOG_DEBUG, "epoll_ctl EPOLL_CTL_MOD (may chack not to quit here), %s, %s %d",
2741 strerror(errno), __func__, __LINE__);
2743 close(peerInfo->fileDescriptor);
2744 cleanHashEntry(peerInfo, sctpMap, &lspan);
2745 char key[MAX_ENODB_NAME_SIZE * 2];
2746 snprintf(key, MAX_ENODB_NAME_SIZE * 2, "msg:%s|%d", enodbName, msgType);
2747 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2748 mdclog_write(MDCLOG_DEBUG, "remove key = %s from %s at line %d", key, __FUNCTION__, __LINE__);
2750 auto tmp = sctpMap->find(key);
2754 sctpMap->erase(key);
2755 mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_ADD (may chack not to quit here)");
2768 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, otSpan *pSpan) {
2770 auto lspan = opentracing::Tracer::Global()->StartSpan(
2771 __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
2773 // otSpan lspan = 0;
2775 //serialize the span
2777 std::unordered_map<std::string, std::string> data;
2778 RICCarrierWriter carrier(data);
2779 opentracing::Tracer::Global()->Inject((lspan.get())->context(), carrier);
2780 nlohmann::json j = data;
2781 std::string str = j.dump();
2782 static auto maxTraceLength = 0;
2784 maxTraceLength = str.length() > maxTraceLength ? str.length() : maxTraceLength;
2785 // serialized context can be put to RMR message using function:
2786 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2787 mdclog_write(MDCLOG_DEBUG, "max trace length is %d trace data length = %ld data = %s", maxTraceLength,
2788 str.length(), str.c_str());
2790 rmr_set_trace(rmrMessageBuffer.sendMessage, (const unsigned char *) str.c_str(), str.length());
2792 buildJsonMessage(message);
2794 rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
2796 if (rmrMessageBuffer.sendMessage == nullptr) {
2797 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2798 mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
2805 if (rmrMessageBuffer.sendMessage->state != 0) {
2806 char meid[RMR_MAX_MEID]{};
2807 if (rmrMessageBuffer.sendMessage->state == RMR_ERR_RETRY) {
2809 rmrMessageBuffer.sendMessage->state = 0;
2810 mdclog_write(MDCLOG_INFO, "RETRY sending Message type %d to Xapp from %s",
2811 rmrMessageBuffer.sendMessage->mtype,
2812 rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
2813 rmrMessageBuffer.sendMessage = rmr_send_msg(rmrMessageBuffer.rmrCtx, rmrMessageBuffer.sendMessage);
2814 if (rmrMessageBuffer.sendMessage == nullptr) {
2815 mdclog_write(MDCLOG_ERR, "RMR failed send message returned with NULL pointer");
2816 rmrMessageBuffer.sendMessage = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
2821 } else if (rmrMessageBuffer.sendMessage->state != 0) {
2822 mdclog_write(MDCLOG_ERR,
2823 "Message state %s while sending request %d to Xapp from %s after retry of 10 microseconds",
2824 translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
2825 rmrMessageBuffer.sendMessage->mtype,
2826 rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
2827 auto rc = rmrMessageBuffer.sendMessage->state;
2834 mdclog_write(MDCLOG_ERR, "Message state %s while sending request %d to Xapp from %s",
2835 translateRmrErrorMessages(rmrMessageBuffer.sendMessage->state).c_str(),
2836 rmrMessageBuffer.sendMessage->mtype,
2837 rmr_get_meid(rmrMessageBuffer.sendMessage, (unsigned char *)meid));
2841 return rmrMessageBuffer.sendMessage->state;
2847 void buildJsonMessage(ReportingMessages_t &message) {
2849 message.outLen = sizeof(message.base64Data);
2850 base64::encode((const unsigned char *) message.message.asndata,
2851 (const int) message.message.asnLength,
2854 if (mdclog_level_get() >= MDCLOG_DEBUG) {
2855 mdclog_write(MDCLOG_DEBUG, "asn data length = %d, base64 message length = %d ",
2856 (int) message.message.asnLength,
2857 (int) message.outLen);
2860 snprintf(message.buffer, sizeof(message.buffer),
2861 "{\"header\": {\"ts\": \"%ld.%09ld\","
2862 "\"ranName\": \"%s\","
2863 "\"messageType\": %d,"
2864 "\"direction\": \"%c\"},"
2865 "\"base64Length\": %d,"
2866 "\"asnBase64\": \"%s\"}",
2867 message.message.time.tv_sec,
2868 message.message.time.tv_nsec,
2869 message.message.enodbName,
2870 message.message.messageType,
2871 message.message.direction,
2872 (int) message.outLen,
2873 message.base64Data);
2874 static src::logger_mt &lg = my_logger::get();
2876 BOOST_LOG(lg) << message.buffer;
2882 * take RMR error code to string
2886 string translateRmrErrorMessages(int state) {
2890 str = "RMR_OK - state is good";
2892 case RMR_ERR_BADARG:
2893 str = "RMR_ERR_BADARG - argument passd to function was unusable";
2895 case RMR_ERR_NOENDPT:
2896 str = "RMR_ERR_NOENDPT - send//call could not find an endpoint based on msg type";
2899 str = "RMR_ERR_EMPTY - msg received had no payload; attempt to send an empty message";
2902 str = "RMR_ERR_NOHDR - message didn't contain a valid header";
2904 case RMR_ERR_SENDFAILED:
2905 str = "RMR_ERR_SENDFAILED - send failed; errno has nano reason";
2907 case RMR_ERR_CALLFAILED:
2908 str = "RMR_ERR_CALLFAILED - unable to send call() message";
2910 case RMR_ERR_NOWHOPEN:
2911 str = "RMR_ERR_NOWHOPEN - no wormholes are open";
2914 str = "RMR_ERR_WHID - wormhole id was invalid";
2916 case RMR_ERR_OVERFLOW:
2917 str = "RMR_ERR_OVERFLOW - operation would have busted through a buffer/field size";
2920 str = "RMR_ERR_RETRY - request (send/call/rts) failed, but caller should retry (EAGAIN for wrappers)";
2922 case RMR_ERR_RCVFAILED:
2923 str = "RMR_ERR_RCVFAILED - receive failed (hard error)";
2925 case RMR_ERR_TIMEOUT:
2926 str = "RMR_ERR_TIMEOUT - message processing call timed out";
2929 str = "RMR_ERR_UNSET - the message hasn't been populated with a transport buffer";
2932 str = "RMR_ERR_TRUNC - received message likely truncated";
2934 case RMR_ERR_INITFAILED:
2935 str = "RMR_ERR_INITFAILED - initialisation of something (probably message) failed";
2937 case RMR_ERR_NOTSUPP:
2938 str = "RMR_ERR_NOTSUPP - the request is not supported, or RMr was not initialised for the request";
2942 snprintf(buf, sizeof buf, "UNDOCUMENTED RMR_ERR : %d", state);