test/mock_a1_mediator.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 /* Author : Ashwin Sridharan
  20
  21    A sample test client to demonstrate A1 functionality.
  22    Sends different kind of policy requests (valid/invalid), create/update/delete and prints out response
  23 */
  24
  25 #include <limits>
  26 #include <map>
  27 #include <getopt.h>
  28 #include <csignal>
  29 #include <time.h>
  30 #include <mutex>
  31 #include <condition_variable>
  32 #include <chrono>
  33 #include <atomic>
  34 #include <xapp_utils.hpp>
  35 #include <vector>
  36 #include <rmr/RIC_message_types.h>
  37
  38 #define MAX_TIMEOUTS 2
  39
  40 std::string gNodeB = "";
  41 std::mutex notify_lock;
  42 std::condition_variable notify_var;
  43
  44 bool rcv_message(rmr_mbuf_t *message){
  45   switch(message->mtype){
  46   case A1_POLICY_RESP:
  47     {
  48       std::lock_guard<std::mutex> lck(notify_lock);
  49       std::cout <<"A1 Mediator received response = " << (char *)message->payload << " of len = " << strlen((char *)message->payload) << std::endl;
  50     }
  51     // released the lock. notify the sleep thread (if any)
  52     notify_var.notify_all();
  53     break;
  54
  55   default:
  56     std::cout <<"Unknown RMR message of type " << message->mtype << " received" << std::endl;
  57   }
  58
  59   return false;
  60 }
  61
  62
  63 void usage(char *command){
  64     std::cout <<"Usage : " << command << " " << std::endl;
  65     std::cout <<" --name[-n] xapp_instance_name " << std::endl;
  66     std::cout <<" --port[-p] port to listen on (default is tcp:4561) " << std::endl;
  67     std::cout << " --op[-o] operation mode from {0, 1, 2} : 0 is CREATE, 1 is UPDATE, 2 is DELETE" << std::endl;
  68     std::cout << " --window [-w] window size in seconds (default is 60 seconds)" << std::endl;
  69     std::cout << " --blockrate [-b] blocking rate percentage (default is 90%)" << std::endl;
  70     std::cout << " --trigger[-t] trigger threshold (default is 40 requests in window)" << std::endl;
  71     std::cout << " --enforce  set policy to enforce if flag prvoided (default is 0 i.e not enforce)" << std::endl;
  72     std::cout << " --class [-c] subscriber profile id to which policy must be applied (default is 5)" << std::endl;
  73     std::cout << " --instance[-i]  policy instance id (default is ac-xapp-1)" << std::endl;
  74     std::cout << std::endl;
  75 }
  76
  77
  78 void msg_error(rmr_mbuf_t *message){
  79   mdclog_write(MDCLOG_ERR, "Error sending message of length %d and type %d, Reason %d",  message->len,  message->mtype, errno );
  80 };
  81
  82
  83 int main(int argc, char *argv[]){
  84
  85
  86   char name[128] = "test_a1_client";
  87   char port[16] = "tcp:9000";
  88   unsigned int num_threads = 1;
  89   std::unique_ptr<XaPP> my_xapp;
  90   std::string schema_file;
  91
  92   enum OPERATIONS{CREATE, UPDATE, DELETE};
  93   static const char * op_strings[] = {"CREATE", "UPDATE", "DELETE"};
  94
  95   OPERATIONS op = CREATE;
  96   std::string instance_id = "ac-xapp-1";
  97   int class_id = 5;
  98   int enforce = 0;
  99   int blocking_rate = 90; // percentage
 100   int window_length = 60; // seconds
 101   int trigger_threshold = 40;
 102
 103   std::chrono::seconds time_out(1);
 104
 105   // Parse command line options
 106   static struct option long_options[] =
 107     {
 108
 109      /* Thse options require arguments */
 110      {"name", required_argument, 0, 'n'},
 111      {"port", required_argument, 0, 'p'},
 112      {"window", required_argument, 0, 'w'},
 113      {"blockrate", required_argument, 0, 'b'},
 114      {"trigger", required_argument, 0, 't'},
 115      {"class", required_argument, 0, 'c'},
 116      {"op", required_argument, 0, 'o'},
 117      {"instance", required_argument, 0, 'i'},
 118      {"enforce", no_argument, &enforce, 1},
 119
 120     };
 121
 122
 123   while(1) {
 124
 125     int option_index = 0;
 126     char c = getopt_long(argc, argv, "n:p:w:b:t:c:o:i:", long_options, &option_index);
 127
 128     if(c == -1){
 129       break;
 130     }
 131
 132     switch(c)
 133       {
 134
 135       case 0:
 136         /* An option flag was set.
 137            Do nothing for now */
 138         break;
 139
 140       case 'n':
 141         strcpy(name, optarg);
 142         break;
 143
 144       case 'p':
 145         strcpy(port, optarg);
 146         break;
 147
 148       case 'w':
 149         window_length = atoi(optarg);
 150         break;
 151
 152       case 't':
 153         trigger_threshold = atoi(optarg);
 154         break;
 155
 156       case 'b':
 157         blocking_rate = atof(optarg);
 158         break;
 159
 160       case 'o':
 161         op = static_cast<OPERATIONS>(atoi(optarg));
 162         break;
 163
 164       case 'i':
 165         instance_id.assign(optarg);
 166         break;
 167
 168       case 'c':
 169         class_id = atoi(optarg);
 170         break;
 171
 172       case 'h':
 173         usage(argv[0]);
 174         exit(0);
 175
 176       default:
 177         usage(argv[0]);
 178         exit(1);
 179       }
 180   };
 181
 182   int log_level = MDCLOG_INFO;
 183   init_logger(name, static_cast<mdclog_severity_t>(log_level));
 184
 185   mdclog_write(MDCLOG_INFO, "XaPP name specified = %s", name);
 186   mdclog_write(MDCLOG_INFO, "XaPP port specified = %s", port);
 187
 188    init_logger(name, MDCLOG_INFO);
 189
 190    mdclog_write(MDCLOG_INFO, "XaPP name specified = %s", name);
 191    mdclog_write(MDCLOG_INFO, "XaPP port specified = %s", port);
 192
 193    mdclog_write(MDCLOG_INFO,"XaPP listener threads specified = auto");
 194    my_xapp = std::make_unique<XaPP>(name, port, 16384);
 195
 196
 197    // Start receiving loop ...
 198    std::vector<int> thread_ids(num_threads);
 199    for(unsigned int i = 0; i < num_threads; i++){
 200      thread_ids[i] = (*my_xapp).StartThread(rcv_message, msg_error);
 201      i++;
 202    };
 203
 204
 205    char buffer[1024];
 206    std::string message_string ;
 207    std::stringstream policy;
 208    std::stringstream msg;
 209    bool res = false;
 210    switch(op){
 211
 212    case CREATE:
 213    case UPDATE:
 214      policy <<"{ " << "\"enforce\":true, " << "\"window_length\":" << window_length << " , \"trigger_threshold\":" << trigger_threshold << ",  \"blocking_rate\":" << blocking_rate << ", \"class\":" << class_id << " }" ;
 215
 216      // Send a create/update
 217      msg << "{ " << "\"policy_type_id\":" << 21000 << "," << "\"policy_instance_id\":\"" << instance_id << "\", \"operation\":\"" << op_strings[op]  << "\", \"payload\" :" << policy.str() << "}";
 218      res = true;
 219      break;
 220
 221    case DELETE:
 222      // send a delete
 223      msg << "{ " << "\"policy_type_id\":" << 21000 << "," << "\"policy_instance_id\": \"" << instance_id << "\", \"operation\": \"" << op_strings[op]  <<  "\" }";
 224      res = true;
 225      break;
 226
 227    default:
 228      std::cerr <<"Not yet supported " << std::endl;
 229
 230    }
 231
 232    if(res){
 233      message_string = msg.str();
 234      std::cout <<"Sending message = " << message_string << std::endl;
 235      memcpy(buffer, message_string.c_str(), message_string.length());
 236      my_xapp.get()->Send(A1_POLICY_REQ,  message_string.length(),  buffer, link_types::HIGH_RELIABILITY);
 237    }
 238
 239    std::unique_lock<std::mutex> lck(notify_lock);
 240
 241    // release lock and got to sleep waiting to be notified
 242    notify_var.wait_for(lck,  std::chrono::seconds(5));
 243
 244    // finish
 245    (*my_xapp).Stop();
 246
 247 }