Bouncer/src/xapp-utils/xapp_rmr.cc

   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
  21 #include "xapp_rmr.hpp"
  22
  23 XappRmr::XappRmr(std::string port, int rmrattempts){
  24
  25         _proto_port = port;
  26         _nattempts = rmrattempts;
  27         _xapp_rmr_ctx = NULL;
  28         _xapp_received_buff = NULL;
  29         _xapp_send_buff =NULL;
  30         _rmr_is_ready = false;
  31         _listen = false;
  32
  33 };
  34
  35 XappRmr::~XappRmr(void){
  36         // free memory
  37         if(_xapp_received_buff)
  38                 rmr_free_msg(_xapp_received_buff);
  39
  40         if(_xapp_send_buff)
  41                 rmr_free_msg(_xapp_send_buff);
  42
  43         if (_xapp_rmr_ctx){
  44                 rmr_close(_xapp_rmr_ctx);
  45         }
  46 };
  47
  48 //Get RMR Context.
  49 void XappRmr::xapp_rmr_init(bool rmr_listen){
  50
  51
  52         // Initialize the RMR context
  53         _xapp_rmr_ctx = rmr_init(const_cast<char*>(_proto_port.c_str()), RMR_MAX_RCV_BYTES, RMRFL_NONE);
  54
  55         if ( _xapp_rmr_ctx == NULL){
  56                 mdclog_write(MDCLOG_ERR,"Error Initializing RMR, file= %s, line=%d",__FILE__,__LINE__);
  57         }
  58         while( ! rmr_ready(_xapp_rmr_ctx) ) {
  59                 mdclog_write(MDCLOG_INFO,">>> waiting for RMR, file= %s, line=%d",__FILE__,__LINE__);
  60                 sleep(1);
  61         }
  62         _rmr_is_ready = true;
  63         mdclog_write(MDCLOG_INFO,"RMR Context is Ready, file= %s, line=%d",__FILE__,__LINE__);
  64
  65         //Set the listener requirement
  66         _listen = rmr_listen;
  67         return;
  68
  69 }
  70
  71 bool XappRmr::rmr_header(xapp_rmr_header *hdr){
  72
  73         _xapp_send_buff->mtype  = hdr->message_type;
  74         _xapp_send_buff->len = hdr->payload_length;
  75         _xapp_send_buff->sub_id = -1;
  76         rmr_str2meid(_xapp_send_buff, hdr->meid);
  77
  78
  79         return true;
  80 }
  81
  82 //RMR Send with payload and header.
  83 bool XappRmr::xapp_rmr_send(xapp_rmr_header *hdr, void *payload){
  84
  85         // Get the thread id
  86         std::thread::id my_id = std::this_thread::get_id();
  87         std::stringstream thread_id;
  88         std::stringstream ss;
  89
  90         thread_id << my_id;
  91         mdclog_write(MDCLOG_INFO, "Sending thread %s",  thread_id.str().c_str());
  92
  93
  94         int rmr_attempts = _nattempts;
  95
  96         if( _xapp_send_buff == NULL ) {
  97                 _xapp_send_buff = rmr_alloc_msg(_xapp_rmr_ctx, RMR_DEF_SIZE);
  98         }
  99
 100         bool res = rmr_header(hdr);
 101         if(!res){
 102                 mdclog_write(MDCLOG_ERR,"RMR HEADERS were incorrectly populated, file= %s, line=%d",__FILE__,__LINE__);
 103                 return false;
 104         }
 105
 106         memcpy(_xapp_send_buff->payload, payload, hdr->payload_length);
 107         _xapp_send_buff->len = hdr->payload_length;
 108
 109         if(!_rmr_is_ready) {
 110                 mdclog_write(MDCLOG_ERR,"RMR Context is Not Ready in SENDER, file= %s, line=%d",__FILE__,__LINE__);
 111                 return false;
 112         }
 113
 114         while(rmr_attempts > 0){
 115
 116                 _xapp_send_buff = rmr_send_msg(_xapp_rmr_ctx,_xapp_send_buff);
 117                 if(!_xapp_send_buff) {
 118                         mdclog_write(MDCLOG_ERR,"Error In Sending Message , file= %s, line=%d, attempt=%d",__FILE__,__LINE__,rmr_attempts);
 119                         rmr_attempts--;
 120                 }
 121                 else if (_xapp_send_buff->state == RMR_OK){
 122                         mdclog_write(MDCLOG_INFO,"Message Sent: RMR State = RMR_OK");
 123                         rmr_attempts = 0;
 124                         _xapp_send_buff = NULL;
 125                         return true;
 126                 }
 127                 else
 128                 {
 129                         mdclog_write(MDCLOG_INFO,"Need to retry RMR: state=%d, attempt=%d, file=%s, line=%d",_xapp_send_buff->state, rmr_attempts,__FILE__,__LINE__);
 130                         if(_xapp_send_buff->state == RMR_ERR_RETRY){
 131                                 usleep(1);                      }
 132                                 rmr_attempts--;
 133                 }
 134                 sleep(1);
 135         }
 136         return false;
 137 }
 138
 139 //----------------------------------------
 140 // Some get/set methods
 141 //---------------------------------------
 142 bool XappRmr::get_listen(void){
 143   return _listen;
 144 }
 145
 146
 147 void XappRmr::set_listen(bool listen){
 148   _listen = listen;
 149 }
 150
 151 int XappRmr::get_is_ready(void){
 152   return _rmr_is_ready;
 153 }
 154
 155 bool XappRmr::get_isRunning(void){
 156   return _listen;
 157 }
 158
 159
 160 void * XappRmr::get_rmr_context(void){
 161   return _xapp_rmr_ctx;
 162 }
 163
 164
 165 void init_logger(const char  *AppName, mdclog_severity_t log_level)
 166 {
 167     mdclog_attr_t *attr;
 168     mdclog_attr_init(&attr);
 169     mdclog_attr_set_ident(attr, AppName);
 170     mdclog_init(attr);
 171     mdclog_level_set(log_level);
 172     mdclog_attr_destroy(attr);
 173 }
 174