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 typedef struct{
  54         struct timespec ts;
  55         int32_t message_type;
  56         int32_t state;
  57         int32_t payload_length;
  58         unsigned char sid[RMR_MAX_SID]; //Subscription ID.
  59         unsigned char src[RMR_MAX_SRC]; //Xapp Name
  60         unsigned char meid[RMR_MAX_MEID]={};
  61
  62 }  xapp_rmr_header;
  63
  64
  65 class XappRmr{
  66 private:
  67         std::string _proto_port;
  68         int _nattempts;
  69         bool _rmr_is_ready;
  70     bool _listen;
  71         void* _xapp_rmr_ctx;
  72         rmr_mbuf_t*             _xapp_send_buff;                                        // send buffer
  73         rmr_mbuf_t*             _xapp_received_buff;                                    // received buffer
  74
  75
  76 public:
  77
  78         XappRmr(std::string, int rmrattempts=10);
  79         ~XappRmr(void);
  80         void xapp_rmr_init(bool);
  81
  82         template <class MessageProcessor>
  83         void xapp_rmr_receive(MessageProcessor&&, XappRmr *parent);
  84
  85         bool xapp_rmr_send(xapp_rmr_header*, void*);
  86
  87         bool rmr_header(xapp_rmr_header*);
  88         void set_listen(bool);
  89         bool get_listen(void);
  90         int get_is_ready(void);
  91         bool get_isRunning(void);
  92         void* get_rmr_context(void);
  93
  94 };
  95
  96
  97 // main workhorse thread which does the listen->process->respond loop
  98 template <class MsgHandler>
  99 void XappRmr::xapp_rmr_receive(MsgHandler&& msgproc, XappRmr *parent){
 100
 101         bool* resend = new bool(false);
 102         // Get the thread id
 103         std::thread::id my_id = std::this_thread::get_id();
 104         std::stringstream thread_id;
 105         std::stringstream ss;
 106         std::fstream io_file;
 107
 108         thread_id << my_id;
 109
 110         // Get the rmr context from parent (all threads and parent use same rmr context. rmr context is expected to be thread safe)
 111         if(!parent->get_is_ready()){
 112                         mdclog_write( MDCLOG_ERR, "RMR Shows Not Ready in RECEIVER, file= %s, line=%d ",__FILE__,__LINE__);
 113                         return;
 114         }
 115         void *rmr_context = parent->get_rmr_context();
 116         assert(rmr_context != NULL);
 117
 118         // Get buffer specific to this thread
 119         this->_xapp_received_buff = NULL;
 120         this->_xapp_received_buff = rmr_alloc_msg(rmr_context, RMR_DEF_SIZE);
 121         assert(this->_xapp_received_buff != NULL);
 122
 123         mdclog_write(MDCLOG_INFO, "Starting receiver thread %s",  thread_id.str().c_str());
 124         io_file.open("/tmp/timestamp.txt", std::ios::in|std::ios::out|std::ios::app);
 125         std::time_t sentMsg_time;
 126         std::time_t recvMsg_time;
 127         struct timeval ts_recv;
 128         struct timeval ts_sent;
 129
 130         while(parent->get_listen()) {
 131                 mdclog_write(MDCLOG_INFO, "Listening at Thread: %s",  thread_id.str().c_str());
 132
 133                 this->_xapp_received_buff = rmr_rcv_msg( rmr_context, this->_xapp_received_buff );
 134                 //this->_xapp_received_buff = rmr_call( rmr_context, this->_xapp_received_buff);
 135
 136                 if (io_file) {
 137                         gettimeofday(&ts_recv, NULL);
 138                         io_file << "Received Msg with msgType: " << this->_xapp_received_buff->mtype << " at time: " <<  (ts_recv.tv_sec * 1000000) + (ts_recv.tv_usec) << std::endl;
 139                 }
 140
 141                 if( this->_xapp_received_buff->mtype < 0 || this->_xapp_received_buff->state != RMR_OK ) {
 142                         mdclog_write(MDCLOG_ERR, "bad msg:  state=%d  errno=%d, file= %s, line=%d", this->_xapp_received_buff->state, errno, __FILE__,__LINE__ );
 143                         return;
 144                 }
 145                 else
 146                 {
 147                         mdclog_write(MDCLOG_INFO,"RMR Received Message of Type: %d",this->_xapp_received_buff->mtype);
 148                         mdclog_write(MDCLOG_INFO,"RMR Received Message: %s",(char*)this->_xapp_received_buff->payload);
 149
 150                     //in case message handler returns true, need to resend the message.
 151                         msgproc(this->_xapp_received_buff, resend);
 152
 153                         if(*resend){
 154                                 mdclog_write(MDCLOG_INFO,"RMR Return to Sender Message of Type: %d",this->_xapp_received_buff->mtype);
 155                                 mdclog_write(MDCLOG_INFO,"RMR Return to Sender Message: %s",(char*)this->_xapp_received_buff->payload);
 156
 157                                 if (io_file) {
 158                                         gettimeofday(&ts_sent, NULL);
 159                                         io_file << "Send Msg with msgType: " << this->_xapp_received_buff->mtype << " at time: " <<  (ts_sent.tv_sec * 1000000) + (ts_sent.tv_usec) << std::endl;
 160
 161                                         io_file << "Time diff: " << ((ts_sent.tv_sec - ts_recv.tv_sec)*1000000 + (ts_sent.tv_usec - ts_recv.tv_usec)) << std::endl;
 162                                 }
 163
 164                                 rmr_rts_msg(rmr_context, this->_xapp_received_buff );
 165                                 sleep(1);
 166                                 *resend = false;
 167                         }
 168                         continue;
 169
 170                 }
 171
 172         }
 173
 174         if (io_file) {
 175                 io_file.close();
 176         }
 177
 178         // Clean up
 179         try{
 180                 delete resend;
 181                 rmr_free_msg(this->_xapp_received_buff);
 182         }
 183         catch(std::runtime_error &e){
 184                 std::string identifier = __FILE__ +  std::string(", Line: ") + std::to_string(__LINE__) ;
 185                 std::string error_string = identifier = " Error freeing RMR message ";
 186                 mdclog_write(MDCLOG_ERR, error_string.c_str(), "");
 187         }
 188
 189         return;
 190 }
 191
 192
 193
 194
 195 #endif /* XAPP_RMR_XAPP_RMR_H_ */