test/app/ad_xapp.cpp

   1 // vi: ts=4 sw=4 noet:
   2 /*
   3 ==================================================================================
   4         Copyright (c) 2021 AT&T Intellectual Property.
   5         Copyright (c) 2021 Alexandre Huff.
   6
   7    Licensed under the Apache License, Version 2.0 (the "License");
   8    you may not use this file except in compliance with the License.
   9    You may obtain a copy of the License at
  10
  11        http://www.apache.org/licenses/LICENSE-2.0
  12
  13    Unless required by applicable law or agreed to in writing, software
  14    distributed under the License is distributed on an "AS IS" BASIS,
  15    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16    See the License for the specific language governing permissions and
  17    limitations under the License.
  18 ==================================================================================
  19 */
  20
  21 /*
  22         Mnemonic:       ad_xapp.cpp
  23         Abstract:   Simulates the AD xApp sending an anomaly dectection message to
  24                 the TS xApp. It sends one message and exits.
  25
  26         Date:           20 May 2021
  27         Author:         Alexandre Huff
  28 */
  29
  30 #include <iostream>
  31 #include <memory>
  32 #include <thread>
  33 #include <unistd.h>
  34 #include <string.h>
  35
  36 #include <rmr/RIC_message_types.h>
  37 #include "ricxfcpp/xapp.hpp"
  38
  39 using namespace std;
  40 using namespace xapp;
  41
  42 unique_ptr<Xapp> xfw;
  43
  44 void ts_callback( Message& mbuf, int mtype, int subid, int len, Msg_component payload,  void* data ) {
  45     cout << "[AD] TS Callback got a message, type=" << mtype << ", length=" << len << "\n";
  46     cout << "[AD] Payload  is  " << payload.get() << endl;
  47
  48     // we only send one message, so we expect to receive only one as well
  49     xfw->Stop();
  50 }
  51
  52 // this thread just sends out one anomaly message to the TS xApp
  53 void ad_loop() {
  54     std::unique_ptr<Message> msg;
  55     Msg_component payload;
  56     int size;
  57     int plen;
  58
  59     cout << "[AD] In Anomaly Detection ad_loop()\n";
  60     sleep( 1 ); // just wait receiver thread starting up
  61
  62     msg = xfw->Alloc_msg(2048);
  63     size = msg->Get_available_size();
  64     if ( size < 2048 ) {
  65         cout << "[ERROR] Message returned does not have enough size: " << size << " < 2048" << endl;
  66         exit(1);
  67     }
  68
  69     // the message we are sending
  70     const char *ad_msg = "[{\"du-id\": 1010, \"ue-id\": \"Train passenger 2\", \"measTimeStampRf\": 1620835470108, \"Degradation\": \"RSRP RSSINR\"}]";
  71
  72     payload = msg->Get_payload();
  73     snprintf( (char *) payload.get(), 2048, "%s", ad_msg );
  74
  75     /*
  76         we are sending a string, so we have to include the nil byte to send with RMR and keep
  77         things simple in the receiver side
  78    */
  79     plen = strlen( (char *) payload.get() ) + 1;
  80     cout << "[AD] Sending a message to TS, length: " << plen << "\n";
  81     cout << "[AD] Message body " << payload.get() << endl;
  82
  83     // payload updated in place, nothing to copy from, so payload parm is nil
  84     if ( ! msg->Send_msg( TS_ANOMALY_UPDATE, Message::NO_SUBID, plen, nullptr ) ) // msg type 30003
  85         cout << "[ERROR] Unable to send a message to TS xApp, state: " << msg->Get_state() << endl;
  86 }
  87
  88 int main(int argc, char const *argv[]) {
  89     int nthreads = 1;
  90
  91     char* port = (char *) "4570";
  92
  93     cout << "[AD] Listening on port " << port << endl;
  94     xfw = std::unique_ptr<Xapp>( new Xapp( port, true ) );
  95
  96     xfw->Add_msg_cb( TS_ANOMALY_ACK, ts_callback, NULL ); /*Register a callback function for msg type 30004*/
  97
  98     std::thread ad_thread;
  99     ad_thread = std::thread(&ad_loop);
 100
 101     xfw->Run( nthreads );
 102
 103     ad_thread.join();
 104
 105     return 0;
 106 }