src/xapp-utils/xapp_rmr.cc

   1 /*
   2 ==================================================================================
   3
   4         Copyright (c) 2018-2019 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
  49 //Get RMR Context.
  50 void XappRmr::xapp_rmr_init(){
  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         return;
  66
  67 }
  68
  69
  70 //RMR Send with payload and header.
  71 bool XappRmr::xapp_rmr_send(xapp_rmr_header *hdr, void *payload){
  72
  73         // Get the thread id
  74         std::thread::id my_id = std::this_thread::get_id();
  75         std::stringstream thread_id;
  76         std::stringstream ss;
  77
  78         thread_id << my_id;
  79         mdclog_write(MDCLOG_INFO, "Sending thread %s",  thread_id.str().c_str());
  80
  81
  82         int rmr_attempts = _nattempts;
  83
  84         if( _xapp_send_buff == NULL ) {
  85                 _xapp_send_buff = rmr_alloc_msg(_xapp_rmr_ctx, RMR_DEF_SIZE);
  86         }
  87         _xapp_send_buff->mtype  = hdr->message_type;
  88         memcpy(_xapp_send_buff->payload, payload, hdr->payload_length);
  89         _xapp_send_buff->len = hdr->payload_length;
  90
  91         if(!_rmr_is_ready) {
  92                 mdclog_write(MDCLOG_ERR,"RMR Context is Not Ready in SENDER, file= %s, line=%d",__FILE__,__LINE__);
  93                 return false;
  94         }
  95
  96         while(rmr_attempts > 0){
  97                 _xapp_send_buff = rmr_send_msg(_xapp_rmr_ctx,_xapp_send_buff);
  98
  99                 if(!_xapp_send_buff) {
 100                         mdclog_write(MDCLOG_ERR,"Error In Sending Message , file= %s, line=%d, attempt=%d",__FILE__,__LINE__,rmr_attempts);
 101                         rmr_attempts--;
 102                 }
 103                 else if (_xapp_send_buff->state == RMR_OK){
 104                         mdclog_write(MDCLOG_INFO,"Message Sent: RMR State = RMR_OK");
 105                         rmr_attempts = 0;
 106                         _xapp_send_buff = NULL;
 107                         return true;
 108                 }
 109                 else
 110                 {
 111                         mdclog_write(MDCLOG_INFO,"Need to retry RMR: state=%d, attempt=%d, file=%s, line=%d",_xapp_send_buff->state, rmr_attempts,__FILE__,__LINE__);
 112                         if(_xapp_send_buff->state == RMR_ERR_RETRY){
 113                                 usleep(1);                      }
 114                                 rmr_attempts--;
 115                 }
 116                 sleep(1);
 117         }
 118         return false;
 119 }
 120
 121 //----------------------------------------
 122 // Some get/set methods
 123 //---------------------------------------
 124 bool XappRmr::get_listen(void){
 125   return _listen;
 126 }
 127
 128
 129 void XappRmr::set_listen(bool listen){
 130   _listen = listen;
 131 }
 132
 133 int XappRmr::get_is_ready(void){
 134   return _rmr_is_ready;
 135 }
 136
 137 bool XappRmr::get_isRunning(void){
 138   return _listen;
 139 }
 140
 141
 142 void * XappRmr::get_rmr_context(void){
 143   return _xapp_rmr_ctx;
 144 }
 145
 146
 147 void init_logger(const char  *AppName, mdclog_severity_t log_level)
 148 {
 149     mdclog_attr_t *attr;
 150     mdclog_attr_init(&attr);
 151     mdclog_attr_set_ident(attr, AppName);
 152     mdclog_init(attr);
 153     mdclog_level_set(log_level);
 154     mdclog_attr_destroy(attr);
 155 }
 156