X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=RIC-E2-TERMINATION%2FsctpThread.cpp;h=9fc72b460d7cf48c5250370d360e46f3577695bf;hb=16c799b750e66b24ff1ac71f5038c18757fd7028;hp=9b391a90d320d91300ba78f78d3dd2dbbda8e4b8;hpb=11b17d45d58d844fe6a61db1cf24f51a91a58f33;p=ric-plt%2Fe2.git

diff --git a/RIC-E2-TERMINATION/sctpThread.cpp b/RIC-E2-TERMINATION/sctpThread.cpp
index 9b391a9..9fc72b4 100644
--- a/RIC-E2-TERMINATION/sctpThread.cpp
+++ b/RIC-E2-TERMINATION/sctpThread.cpp
@@ -277,6 +277,8 @@ int buildConfiguration(sctp_params_t &sctpParams) {
     return 0;
 }
 
+
+
 int main(const int argc, char **argv) {
     sctp_params_t sctpParams;
 
@@ -363,6 +365,9 @@ int main(const int argc, char **argv) {
         }
     }
 
+    auto statFlag = false;
+    auto statThread = std::thread(statColectorThread, (void *)&statFlag);
+
     //loop over term_init until first message from xApp
     handleTermInit(sctpParams);
 
@@ -370,6 +375,9 @@ int main(const int argc, char **argv) {
         t.join();
     }
 
+    statFlag = true;
+    statThread.join();
+
     return 0;
 }
 
@@ -544,6 +552,8 @@ void listener(sctp_params_t *params) {
         rmrMessageBuffer.sendBufferedMessages[i] = rmr_alloc_msg(rmrMessageBuffer.rmrCtx, RECEIVE_XAPP_BUFFER_SIZE);
     }
 
+    message.statCollector = StatCollector::GetInstance();
+
     while (true) {
         if (mdclog_level_get() >= MDCLOG_DEBUG) {
             mdclog_write(MDCLOG_DEBUG, "Start EPOLL Wait");
@@ -573,6 +583,9 @@ void listener(sctp_params_t *params) {
             } else if (events[i].events & EPOLLOUT) {
                 handleEinprogressMessages(events[i], message, rmrMessageBuffer, params);
             } else if (params->listenFD == events[i].data.fd) {
+                if (mdclog_level_get() >= MDCLOG_INFO) {
+                    mdclog_write(MDCLOG_INFO, "New connection request from sctp network\n");
+                }
                 // new connection is requested from RAN  start build connection
                 while (true) {
                     struct sockaddr in_addr {};
@@ -600,7 +613,7 @@ void listener(sctp_params_t *params) {
                     }
                     auto  ans = getnameinfo(&in_addr, in_len,
                             peerInfo->hostName, NI_MAXHOST,
-                            peerInfo->portNumber, NI_MAXSERV, (signed )((unsigned)NI_NUMERICHOST | NI_NUMERICSERV));
+                            peerInfo->portNumber, NI_MAXSERV, (unsigned )((unsigned int)NI_NUMERICHOST | (unsigned int)NI_NUMERICSERV));
                     if (ans < 0) {
                         mdclog_write(MDCLOG_ERR, "Failed to get info on connection request. %s\n", strerror(errno));
                         close(peerInfo->fileDescriptor);
@@ -618,6 +631,7 @@ void listener(sctp_params_t *params) {
                                    0) != 0) {
                         break;
                     }
+                    break;
                 }
             } else if (params->rmrListenFd == events[i].data.fd) {
                 // got message from XAPP
@@ -626,8 +640,7 @@ void listener(sctp_params_t *params) {
                 if (mdclog_level_get() >= MDCLOG_DEBUG) {
                     mdclog_write(MDCLOG_DEBUG, "new message from RMR");
                 }
-                if (receiveXappMessages(params->epoll_fd,
-                                        params->sctpMap,
+                if (receiveXappMessages(params->sctpMap,
                                         rmrMessageBuffer,
                                         message.message.time) != 0) {
                     mdclog_write(MDCLOG_ERR, "Error handling Xapp message");
@@ -960,7 +973,7 @@ int sendSctpMsg(ConnectedCU_t *peerInfo, ReportingMessages_t &message, Sctp_Map_
             m->erase(key);
             return -1;
         }
-        peerInfo->sentMesgs++;
+        message.statCollector->incSentMessage(string(message.message.enodbName));
         message.message.direction = 'D';
         // send report.buffer of size
         buildJsonMessage(message);
@@ -1018,7 +1031,7 @@ int receiveDataFromSctp(struct epoll_event *events,
     // get the identity of the interface
     message.peerInfo = (ConnectedCU_t *)events->data.ptr;
 
-
+    message.statCollector = StatCollector::GetInstance();
     struct timespec start{0, 0};
     struct timespec decodestart{0, 0};
     struct timespec end{0, 0};
@@ -1040,8 +1053,9 @@ int receiveDataFromSctp(struct epoll_event *events,
             mdclog_write(MDCLOG_DEBUG, "Finish Read from SCTP %d fd message length = %ld",
                     message.peerInfo->fileDescriptor, message.message.asnLength);
         }
-        message.peerInfo->rcvMsgs++;
+
         memcpy(message.message.enodbName, message.peerInfo->enodbName, sizeof(message.peerInfo->enodbName));
+        message.statCollector->incRecvMessage(string(message.message.enodbName));
         message.message.direction = 'U';
         message.message.time.tv_nsec = ts.tv_nsec;
         message.message.time.tv_sec = ts.tv_sec;
@@ -1775,14 +1789,12 @@ void getRmrContext(sctp_params_t &pSctpParams) {
 
 /**
  *
- * @param epoll_fd
  * @param sctpMap
  * @param rmrMessageBuffer
  * @param ts
  * @return
  */
-int receiveXappMessages(int epoll_fd,
-                        Sctp_Map_t *sctpMap,
+int receiveXappMessages(Sctp_Map_t *sctpMap,
                         RmrMessagesBuffer_t &rmrMessageBuffer,
                         struct timespec &ts) {
     if (rmrMessageBuffer.rcvMessage == nullptr) {