src/xapp-mgmt/subs_mgmt.hpp

   1 /*
   2 ==================================================================================
   3         Copyright (c) 2019-2020 AT&T Intellectual Property.
   4
   5    Licensed under the Apache License, Version 2.0 (the "License");
   6    you may not use this file except in compliance with the License.
   7    You may obtain a copy of the License at
   8
   9        http://www.apache.org/licenses/LICENSE-2.0
  10
  11    Unless required by applicable law or agreed to in writing, software
  12    distributed under the License is distributed on an "AS IS" BASIS,
  13    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14    See the License for the specific language governing permissions and
  15    limitations under the License.
  16 ==================================================================================
  17 */
  18 /*
  19  * subs_mgmt.hpp
  20  * Created on: 2019
  21  * Author: Ashwin Shridharan, Shraboni Jana
  22  */
  23
  24 #pragma once
  25
  26 #ifndef SUBSCRIPTION_HANDLER
  27 #define SUBSCRIPTION_HANDLER
  28
  29 #include <functional>
  30 #include <mdclog/mdclog.h>
  31 #include <mutex>
  32 #include <condition_variable>
  33 #include <unordered_map>
  34 #include <algorithm>
  35 #include <ctime>
  36 #include <unistd.h>
  37 #include <chrono>
  38 #include <tuple>
  39 #include <rmr/RIC_message_types.h>
  40 #include "e2ap_action.hpp"
  41 #include "e2ap_subscription_request.hpp"
  42 #include "e2sm_subscription.hpp"
  43
  44 #define SUBSCR_SUCCESS 1
  45 #define SUBSCR_ERR_TX -1
  46 #define SUBSCR_ERR_TIMEOUT -2
  47 #define SUBSCR_ERR_FAIL -3
  48 #define SUBSCR_ERR_UNKNOWN -4
  49 #define SUBSCR_ERR_DUPLICATE -5
  50
  51 using namespace std;
  52
  53 class TransmitterBase
  54 {
  55 public:
  56     virtual ~TransmitterBase() {}
  57
  58     template<class T>
  59     const T& getParam() const; //to be implemented after Parameter
  60
  61     template<class T, class U>
  62     void setParam(const U& rhs); //to be implemented after Parameter
  63 };
  64
  65 template <typename T>
  66 class Transmitter : public TransmitterBase
  67 {
  68 public:
  69         Transmitter(const T& tx) :obj(tx) {}
  70     const T& getParam() const {return obj;}
  71     void setParam(const T& tx) {obj=tx;}
  72 private:
  73     T obj;
  74 };
  75
  76 //Here's the trick: dynamic_cast rather than virtual
  77 template<class T> const T& TransmitterBase::getParam() const
  78 {
  79         return dynamic_cast<const Transmitter<T>&>(*this).getParam();
  80 }
  81 template<class T, class U> void TransmitterBase::setParam(const U& rhs)
  82 {
  83         dynamic_cast<Transmitter<T>&>(*this).setParam(rhs);
  84         return;
  85 }
  86
  87 typedef enum {
  88     request_pending = 1,
  89     request_success,
  90     request_failed,
  91     request_duplicate
  92 }Subscription_Status_Types;
  93
  94
  95 using transaction_identifier = std::string;
  96 using transaction_status = Subscription_Status_Types;
  97
  98 class SubscriptionHandler {
  99
 100 public:
 101
 102   SubscriptionHandler(unsigned int timeout_seconds = 30);
 103
 104   template <typename AppTransmitter>
 105   int manage_subscription_request(transaction_identifier, AppTransmitter &&);
 106
 107   template <typename AppTransmitter>
 108   int manage_subscription_delete_request(transaction_identifier, AppTransmitter &&);
 109
 110   void manage_subscription_response(int message_type, transaction_identifier id);
 111
 112   int  get_request_status(transaction_identifier);
 113   bool set_request_status(transaction_identifier, transaction_status);
 114   bool is_request_entry(transaction_identifier);
 115   void set_timeout(unsigned int);
 116   void clear(void);
 117   void set_ignore_subs_resp(bool b){_ignore_subs_resp = b;};
 118
 119   void print_subscription_status(){ for(auto it:status_table){std::cout << it.first << "::" << it.second << std::endl;}};
 120
 121 private:
 122
 123   bool add_request_entry(transaction_identifier, transaction_status);
 124   bool delete_request_entry(transaction_identifier);
 125
 126   template <typename AppTransmitter>
 127   bool add_transmitter_entry(transaction_identifier, AppTransmitter&&);
 128
 129   std::unordered_map<transaction_identifier, TransmitterBase> trans_table;
 130   std::unordered_map<transaction_identifier, transaction_status> status_table;
 131
 132   std::unique_ptr<std::mutex> _data_lock;
 133   std::unique_ptr<std::condition_variable> _cv;
 134
 135   std::chrono::seconds _time_out;
 136
 137   bool _ignore_subs_resp = false;
 138 };
 139
 140 template <typename AppTransmitter>
 141 bool SubscriptionHandler::add_transmitter_entry(transaction_identifier id, AppTransmitter &&trans){
 142           mdclog_write(MDCLOG_INFO,"Entry added for Transaction ID: %s",id.c_str());
 143
 144   // add entry in hash table if it does not exist
 145   auto search = trans_table.find(id);
 146   if(search != trans_table.end()){
 147     return false;
 148   }
 149
 150   Transmitter<AppTransmitter> tptr(trans);
 151   trans_table[id] = tptr;
 152   return true;
 153
 154 };
 155
 156 //this will work for both sending subscription request and subscription delete request.
 157 //The handler is oblivious of the message content and follows the transaction id.
 158 template<typename AppTransmitter>
 159 int SubscriptionHandler::manage_subscription_request(transaction_identifier rmr_trans_id, AppTransmitter && tx){
 160         int res;
 161   // put entry in request table
 162   {
 163     std::lock_guard<std::mutex> lock(*(_data_lock.get()));
 164
 165     res = add_request_entry(rmr_trans_id, request_pending);
 166     if(! res){
 167       mdclog_write(MDCLOG_ERR, "%s, %d : Error adding new subscription request %s to queue because request with identical key already present",  __FILE__, __LINE__, rmr_trans_id);
 168       return SUBSCR_ERR_DUPLICATE;
 169     }
 170   }
 171
 172
 173   // acquire lock ...
 174   std::unique_lock<std::mutex> _local_lock(*(_data_lock.get()));
 175
 176   // Send the message
 177   bool flg = tx();
 178
 179   if (!flg){
 180     // clear state
 181     delete_request_entry(rmr_trans_id);
 182     mdclog_write(MDCLOG_ERR, "%s, %d :: Error transmitting subscription request %s", __FILE__, __LINE__, rmr_trans_id.c_str() );
 183     return SUBSCR_ERR_TX;
 184   } else {
 185           mdclog_write(MDCLOG_INFO, "%s, %d :: Transmitted subscription request for trans_id %s", __FILE__, __LINE__, rmr_trans_id.c_str() );
 186           add_transmitter_entry(rmr_trans_id, tx);
 187
 188   }
 189
 190   // record time stamp ..
 191   auto start = std::chrono::system_clock::now();
 192   std::chrono::milliseconds t_out(_time_out);
 193
 194   //the wait functionality has been removed.
 195
 196
 197   _local_lock.unlock();
 198   // std::cout <<"Returning  res = " << res << " for request = " << rmr_trans_id  << std::endl;
 199    return res;
 200 };
 201
 202 #endif