Bouncer/src/xapp-utils/xapp_rmr.hpp

   1 /*
   2 ==================================================================================
   3
   4         Copyright (c) 2019-2020 AT&T Intellectual Property.
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19
  20 #ifndef XAPP_RMR_XAPP_RMR_H_
  21 #define XAPP_RMR_XAPP_RMR_H_
  22
  23
  24 #ifdef __GNUC__
  25 #define likely(x)  __builtin_expect((x), 1)
  26 #define unlikely(x) __builtin_expect((x), 0)
  27 #else
  28 #define likely(x) (x)
  29 #define unlikely(x) (x)
  30 #endif
  31
  32 #include <iostream>
  33 #include <fstream>
  34 #include <stdlib.h>
  35 #include <unistd.h>
  36 #include <string.h>
  37 #include <error.h>
  38 #include <assert.h>
  39 #include <sstream>
  40 #include <thread>
  41 #include <functional>
  42 #include <map>
  43 #include <mutex>
  44 #include <sys/epoll.h>
  45 #include <rmr/rmr.h>
  46 #include <rmr/RIC_message_types.h>
  47 #include <mdclog/mdclog.h>
  48 #include <vector>
  49 #include <ctime>
  50 #include <chrono>
  51 #include <sys/time.h>
  52
  53 #include "a1_helper.hpp"
  54 #include "e2ap_control.hpp"
  55 #include "e2ap_control_response.hpp"
  56 #include "e2ap_indication.hpp"
  57 #include "subscription_delete_request.hpp"
  58 #include "subscription_delete_response.hpp"
  59 #include "subscription_helper.hpp"
  60 #include "subscription_request.hpp"
  61 #include "subscription_request.hpp"
  62 #include "subscription_response.hpp"
  63 #include "e2sm_subscription.hpp"
  64 #include "subs_mgmt.hpp"
  65
  66 typedef struct{
  67         struct timespec ts;
  68         int32_t message_type;
  69         int32_t state;
  70         int32_t payload_length;
  71
  72         unsigned char sid[RMR_MAX_SID]; //Subscription ID.
  73         unsigned char src[RMR_MAX_SRC]; //Xapp Name
  74         unsigned char meid[RMR_MAX_MEID]={};
  75
  76 }  xapp_rmr_header;
  77
  78
  79 class XappRmr{
  80 private:
  81         std::string _proto_port;
  82         int _nattempts;
  83         bool _rmr_is_ready;
  84     bool _listen;
  85         void* _xapp_rmr_ctx;
  86         rmr_mbuf_t*             _xapp_send_buff;                                        // send buffer
  87         rmr_mbuf_t*             _xapp_received_buff;                                    // received buffer
  88
  89
  90 public:
  91
  92         XappRmr(std::string, int rmrattempts=10);
  93         ~XappRmr(void);
  94         void xapp_rmr_init(bool);
  95
  96         template <class MessageProcessor>
  97         void xapp_rmr_receive(MessageProcessor&&, XappRmr *parent);
  98
  99         bool xapp_rmr_send(xapp_rmr_header*, void*);
 100
 101         bool rmr_header(xapp_rmr_header*);
 102         void set_listen(bool);
 103         bool get_listen(void);
 104         int get_is_ready(void);
 105         bool get_isRunning(void);
 106         void* get_rmr_context(void);
 107
 108 };
 109
 110
 111 // main workhorse thread which does the listen->process->respond loop
 112 template <class MsgHandler>
 113 void XappRmr::xapp_rmr_receive(MsgHandler&& msgproc, XappRmr *parent){
 114
 115         bool* resend = new bool(false);
 116         // Get the thread id
 117         std::thread::id my_id = std::this_thread::get_id();
 118         std::stringstream thread_id;
 119         std::stringstream ss;
 120         std::fstream io_file;
 121
 122         thread_id << my_id;
 123
 124         // Get the rmr context from parent (all threads and parent use same rmr context. rmr context is expected to be thread safe)
 125         if(!parent->get_is_ready()){
 126                         mdclog_write( MDCLOG_ERR, "RMR Shows Not Ready in RECEIVER, file= %s, line=%d ",__FILE__,__LINE__);
 127                         return;
 128         }
 129         void *rmr_context = parent->get_rmr_context();
 130         assert(rmr_context != NULL);
 131
 132         // Get buffer specific to this thread
 133         this->_xapp_received_buff = NULL;
 134         this->_xapp_received_buff = rmr_alloc_msg(rmr_context, RMR_DEF_SIZE);
 135         assert(this->_xapp_received_buff != NULL);
 136
 137         mdclog_write(MDCLOG_INFO, "Starting receiver thread %s",  thread_id.str().c_str());
 138         io_file.open("/tmp/timestamp.txt", std::ios::in|std::ios::out|std::ios::app);
 139         std::time_t sentMsg_time;
 140         std::time_t recvMsg_time;
 141         struct timeval ts_recv;
 142         struct timeval ts_sent;
 143         int num = 0;
 144
 145         while(parent->get_listen()) {
 146                 mdclog_write(MDCLOG_INFO, "Listening at Thread: %s",  thread_id.str().c_str());
 147
 148                 this->_xapp_received_buff = rmr_rcv_msg( rmr_context, this->_xapp_received_buff );
 149                 //this->_xapp_received_buff = rmr_call( rmr_context, this->_xapp_received_buff);
 150
 151                 if (io_file) {
 152                         gettimeofday(&ts_recv, NULL);
 153                         io_file << "Received Msg with msgType: " << this->_xapp_received_buff->mtype << " at time: " <<  (ts_recv.tv_sec * 1000) + (ts_recv.tv_usec/1000) << std::endl;
 154                 }
 155
 156                 if( this->_xapp_received_buff->mtype < 0 || this->_xapp_received_buff->state != RMR_OK ) {
 157                         mdclog_write(MDCLOG_ERR, "bad msg:  state=%d  errno=%d, file= %s, line=%d", this->_xapp_received_buff->state, errno, __FILE__,__LINE__ );
 158                         return;
 159                 }
 160                 else
 161                 {
 162                         mdclog_write(MDCLOG_INFO,"RMR Received Message of Type: %d",this->_xapp_received_buff->mtype);
 163                         mdclog_write(MDCLOG_INFO,"RMR Received Message: %s",(char*)this->_xapp_received_buff->payload);
 164
 165                     //in case message handler returns true, need to resend the message.
 166                         msgproc(this->_xapp_received_buff, resend);
 167
 168                         //start of code to check decoding indication payload
 169
 170                          num++;
 171                          mdclog_write(MDCLOG_DEBUG, "Total Indications received : %d", num);
 172
 173                         if(*resend){
 174                                 mdclog_write(MDCLOG_INFO,"RMR Return to Sender Message of Type: %d",this->_xapp_received_buff->mtype);
 175                                 mdclog_write(MDCLOG_INFO,"RMR Return to Sender Message: %s",(char*)this->_xapp_received_buff->payload);
 176
 177                                 if (io_file) {
 178                                         gettimeofday(&ts_sent, NULL);
 179
 180                                         io_file << "Send Msg with msgType: " << this->_xapp_received_buff->mtype << " at time: " << (ts_sent.tv_sec * 1000) + (ts_sent.tv_usec/1000) << std::endl;
 181
 182                                         io_file << "Time diff: " << ((ts_sent.tv_sec - ts_recv.tv_sec)*1000 + (ts_sent.tv_usec - ts_recv.tv_usec)/1000) << std::endl;
 183                                 }
 184
 185                                 rmr_rts_msg(rmr_context, this->_xapp_received_buff );
 186                                 //sleep(1);
 187
 188                                 *resend = false;
 189                         }
 190                         continue;
 191                 }
 192
 193         }
 194
 195         if (io_file) {
 196                 io_file.close();
 197         }
 198
 199         // Clean up
 200         try{
 201                 delete resend;
 202                 rmr_free_msg(this->_xapp_received_buff);
 203         }
 204         catch(std::runtime_error &e){
 205                 std::string identifier = __FILE__ +  std::string(", Line: ") + std::to_string(__LINE__) ;
 206                 std::string error_string = identifier = " Error freeing RMR message ";
 207                 mdclog_write(MDCLOG_ERR, error_string.c_str(), "");
 208         }
 209
 210         return;
 211 }
 212
 213 #endif /* XAPP_RMR_XAPP_RMR_H_ */