--- /dev/null
+/*==================================================================================
+
+ Copyright (c) 2018-2019 AT&T Intellectual Property.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+==================================================================================
+*/
+
+/* Author : Ashwin Sridharan
+ Date : Feb 2019
+*/
+
+
+/*
+ Unit testing of subscription handler
+*/
+
+#define CATCH_CONFIG_MAIN
+#include <catch2/catch.hpp>
+
+
+#include <string.h>
+#include <stdio.h>
+#include <iostream>
+#include <csignal>
+#include <chrono>
+#include <subscription_handler.hpp>
+#include <e2sm.hpp>
+#include <queue>
+#include <mutex>
+#include <rmr/RIC_message_types.h>
+#include <thread>
+
+
+// global queue for testing
+std::queue<std::string> message_bus;
+
+// global lock for testing
+std::mutex get_object ;
+
+bool is_running = true;
+
+bool mock_fail(int mtype, size_t len, void * payload, int mode){
+ return false;
+}
+
+bool mock_silent(int mtype, size_t len, void * payload, int mode){
+ return true;
+}
+
+
+bool mock_tx(int mytpe, size_t len, void *payload, int mode){
+
+ bool res;
+ int i;
+ subscription_helper he;
+ subscription_response_helper he_resp;
+
+ subscription_request sub_req;
+ subscription_response sub_resp;
+
+ subscription_delete sub_del_req;
+ subscription_delete_response sub_del_resp;
+ asn_dec_rval_t retval;
+
+ E2N_E2AP_PDU_t * e2ap_pdu_recv;
+ unsigned char buffer[256];
+ size_t buf_size = 256;
+ bool msg_ok = false;
+
+ e2ap_pdu_recv = 0;
+ retval = asn_decode(0, ATS_ALIGNED_BASIC_PER, &asn_DEF_E2N_E2AP_PDU, (void**)&(e2ap_pdu_recv), payload, len);
+ if(retval.code != RC_OK){
+ std::cerr <<"Error decoding E2N_E2AP Subscription response PDU. Reason = " << strerror(errno) << std::endl;
+ ASN_STRUCT_FREE(asn_DEF_E2N_E2AP_PDU, e2ap_pdu_recv);
+ return false;
+ }
+
+ int procedure_code = e2ap_pdu_recv->choice.initiatingMessage->procedureCode;
+
+ if(procedure_code == E2N_ProcedureCode_id_ricSubscription){
+
+ he.clear();
+ sub_req.get_fields(e2ap_pdu_recv->choice.initiatingMessage, he);
+
+
+ // set up response object
+ he_resp.set_request(he.get_request_id(), he.get_req_seq());
+ he_resp.set_function_id(he.get_function_id());
+ i = 0;
+
+ // we simply copy over actions to both admitted and not
+ // admitted list for now ..
+ // in future, may need to be more selective
+ for(auto &e : *(he.get_list())){
+ he_resp.add_action(e.get_id());
+ he_resp.add_action(e.get_id(), 1, 2);
+ i++;
+ }
+
+ if(mode == 0){
+ res = sub_resp.encode_e2ap_subscription_response(buffer, &buf_size, he_resp, true);
+ if (!res){
+ std::cerr << "Error encoding subscription response successful. Reason = " << sub_resp.get_error() << std::endl;
+ }
+ else{
+ msg_ok = true;
+ }
+ }
+ else{
+ res = sub_resp.encode_e2ap_subscription_response(buffer, &buf_size, he_resp, false);
+ if (!res){
+ std::cerr << "Error encoding subscription response failure . Reason = " << sub_resp.get_error() << std::endl;
+ }
+ else{
+ msg_ok = true;
+ }
+
+ };
+
+ }
+
+ else if (procedure_code == E2N_ProcedureCode_id_ricSubscriptionDelete){
+
+ he.clear();
+ sub_del_req.get_fields(e2ap_pdu_recv->choice.initiatingMessage, he);
+
+ // set up response object
+ he_resp.clear();
+ he_resp.set_request(he.get_request_id(), he.get_req_seq());
+ he_resp.set_function_id(he.get_function_id());
+ if(mode == 0){
+ res = sub_del_resp.encode_e2ap_subscription_delete_response(buffer, &buf_size, he_resp, true);
+ if (!res){
+ std::cerr << "Error encoding subscription delete response failure . Reason = " << sub_resp.get_error() << std::endl;
+ }
+ else{
+ msg_ok = true;
+ }
+ }
+ else{
+ res = sub_del_resp.encode_e2ap_subscription_delete_response(buffer, &buf_size, he_resp, false);
+ if (!res){
+ std::cerr << "Error encoding subscription delete response failure . Reason = " << sub_resp.get_error() << std::endl;
+ }
+ else{
+ msg_ok = true;
+ std::cout <<"Sending delete failures ..." << std::endl;
+ }
+
+ }
+ }
+ else{
+ std::cout <<"Illegal request" << std::endl;
+ }
+
+
+ // push to queue
+ if(msg_ok){
+ std::lock_guard<std::mutex> guard(get_object);
+ std::string msg((char *)buffer, buf_size);
+ //std::cout <<"Pushed to queue" << std::endl;
+ message_bus.push(msg);
+ }
+
+
+ ASN_STRUCT_FREE(asn_DEF_E2N_E2AP_PDU, e2ap_pdu_recv);
+ if(msg_ok)
+ return true;
+
+ else
+ return false;
+
+
+}
+
+
+
+// Randomly generate number of subscription response and delete
+// response packets and push to queue
+void random_tx(int num_packets){
+ subscription_response_helper he_resp;
+ subscription_response sub_resp;
+ subscription_delete sub_del_req;
+ subscription_delete_response sub_del_resp;
+ bool res;
+ unsigned char buffer[256];
+ size_t buf_size = 256;
+
+ he_resp.add_action(10);
+
+ // generate subscription responses
+ for(int i = 0; i < num_packets; i++){
+
+ // set up response object
+ he_resp.set_request(i, 1);
+ he_resp.set_function_id(0);
+ buf_size = 256;
+ res = sub_resp.encode_e2ap_subscription_response(buffer, &buf_size, he_resp, true);
+ {
+ std::lock_guard<std::mutex> guard(get_object);
+ std::string msg((char *)buffer, buf_size);
+ message_bus.push(msg);
+ }
+
+ buf_size = 256;
+ res = sub_resp.encode_e2ap_subscription_response(buffer, &buf_size, he_resp, false);
+ {
+ std::lock_guard<std::mutex> guard(get_object);
+ std::string msg((char *)buffer, buf_size);
+ message_bus.push(msg);
+ }
+
+ buf_size = 256;
+ res = sub_del_resp.encode_e2ap_subscription_delete_response(buffer, &buf_size, he_resp, true);
+ {
+ std::lock_guard<std::mutex> guard(get_object);
+ std::string msg((char *)buffer, buf_size);
+ message_bus.push(msg);
+ }
+
+ buf_size = 256;
+ res = sub_del_resp.encode_e2ap_subscription_delete_response(buffer, &buf_size, he_resp, false);
+ {
+ std::lock_guard<std::mutex> guard(get_object);
+ std::string msg((char *)buffer, buf_size);
+ message_bus.push(msg);
+ }
+
+
+ }
+}
+
+
+void mock_RAN (subscription_handler &_ref_sub_handler){
+ // Behaviour :
+
+ unsigned char incorrect_e2ap[128];
+ size_t incorrect_e2ap_size = 128;
+ for(int i = 0; i < 128; i++){
+ incorrect_e2ap[i] = 'b';
+ }
+
+ FILE *pfile = fopen("test-data/e2ap_indication_test.per", "r");
+ if(pfile == NULL){
+ std::cout << "Error opening e2ap_indication_test.per" << std::endl;
+ exit(-1);
+ }
+ unsigned char e2ap_msg[512];
+ size_t e2ap_msg_size = fread((char *)e2ap_msg, sizeof(char), 512, pfile);
+ fclose(pfile);
+
+ unsigned char message_buf[512];
+ std::string pdu;
+
+ while(is_running){
+ // send some random data, i.e incorrect E2AP
+ _ref_sub_handler.Response(RIC_INDICATION, incorrect_e2ap, incorrect_e2ap_size);
+ //std::cout <<"Sent random data to subscription handler" << std::endl;
+
+ // send an E2AP which is not subscription request
+ _ref_sub_handler.Response(RIC_INDICATION, e2ap_msg, e2ap_msg_size);
+ //std::cout <<"Sent incorrect e2ap to subscription handler" << std::endl;
+
+ // now look in the queue, pop it and send the data
+ // finally send correct payload
+ {
+ std::lock_guard<std::mutex> guard(get_object);
+ if(! message_bus.empty()){
+ pdu = message_bus.front();
+ memcpy(message_buf, pdu.c_str(), pdu.length());
+ message_bus.pop();
+ }
+ }
+
+ _ref_sub_handler.Response(RIC_SUB_RESP, message_buf, pdu.length());
+ //std::cout <<"Sent response to subscription handler" << std::endl;
+
+
+
+ sleep(1);
+ }
+
+}
+
+
+void send_request(subscription_handler &subscription_manager, std::vector<int> & status_vector, int index, bool (*tx)(int, size_t, void *, int), int mode ){
+ subscription_helper subscription_info;
+ subscription_request sub_req;
+ subscription_response_helper subscription_response_info;
+
+ int function_id = 0;
+ int action_id = 4;
+ int action_type = 0;
+
+ int request_id = 1;
+ int req_seq = 1;
+ int message_type = 1;
+ int procedure_code = 27;
+ std::string egnb_id = "Testgnb";
+ std::string plmn_id = "Testplmn";
+
+ unsigned char event_buf[128];
+ size_t event_buf_len = 128;
+ int res;
+
+
+ e2sm_event_trigger_helper trigger_data;
+ e2sm_event_trigger event_trigger;
+
+ trigger_data.egNB_id = egnb_id;
+ trigger_data.plmn_id = plmn_id;
+ trigger_data.egNB_id_type = 2;
+ trigger_data.interface_direction = 1;
+ trigger_data.procedure_code = procedure_code;
+ trigger_data.message_type = message_type;
+
+ res = event_trigger.encode_event_trigger(&event_buf[0], &event_buf_len, trigger_data);
+
+ subscription_info.clear();
+ subscription_info.set_request(request_id, req_seq);
+ subscription_info.set_function_id(function_id);
+ subscription_info.add_action(action_id, action_type);
+ subscription_info.set_event_def(&event_buf[0], event_buf_len);
+
+ auto transmitter = std::bind(tx, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, mode);
+ res = subscription_manager.RequestSubscription(subscription_info, subscription_response_info , RIC_SUB_REQ, transmitter);
+
+ if (res == SUBSCR_SUCCESS ){
+ // store -ve of request id
+ status_vector[index] = -1 * subscription_info.get_request_id();
+ }
+ else{
+ status_vector[index] = res;
+ }
+
+ std::cout <<"Subscription = " << subscription_info.get_request_id() << " Result = " << res << std::endl;
+}
+
+void delete_request(subscription_handler &subscription_manager, std::vector<int> & status_vector, int index, int request_id, bool ( *tx)(int, size_t, void *, int), int mode ){
+
+ subscription_helper subscription_info;
+ subscription_response_helper subscription_response_info;
+
+
+ //verify subscription deleted
+ subscription_info.set_request(request_id, 1);
+ subscription_info.set_function_id(0);
+
+ auto transmitter = std::bind(tx, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, mode);
+ status_vector[index] = subscription_manager.RequestSubscriptionDelete(subscription_info, subscription_response_info, RIC_SUB_DEL_REQ, transmitter);
+
+
+};
+
+
+TEST_CASE("Test subscription work flow", "E2AP Subscription"){
+
+ subscription_handler subscription_manager;
+
+ mdclog_attr_t *attr;
+ mdclog_attr_init(&attr);
+ mdclog_attr_set_ident(attr, "MOCK TEST SUBSCRIPTION WORK FLOW ");
+ mdclog_init(attr);
+ mdclog_level_set(MDCLOG_DEBUG);
+ mdclog_attr_destroy(attr);
+
+ unsigned char node_buffer[32];
+ std::string gNodeB = "TEST_GNOBDE";
+
+ std::copy(gNodeB.begin(), gNodeB.end(), node_buffer);
+ node_buffer[gNodeB.length()] = '\0';
+
+
+ //====================================
+
+ SECTION("Verify behaviour if no listener "){
+ std::cout <<"+++++++++" << std::endl << "TEST WITH NO LISTENER " << std::endl;
+
+ int num_sources = 10;
+ std::vector<int> status_vector(num_sources, 0);
+ subscription_manager.clear();
+
+ std::vector<std::thread> source_list;
+
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_fail), 0));
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_ERR_TX);
+ }
+
+ REQUIRE(subscription_manager.num_complete() == 0);
+ REQUIRE(subscription_manager.num_pending() == 0);
+ }
+
+ SECTION("Verify behaviour if listener does not respond"){
+ std::cout <<"+++++++++" << std::endl << "TEST WHEN LISTENER DOES NOT RESPOND " << std::endl;
+
+ int num_sources = 10;
+ std::vector<int> status_vector(num_sources, 0);
+
+ subscription_manager.clear();
+
+ std::vector<std::thread> source_list;
+
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_silent), 0));
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ REQUIRE(status_vector[i] == SUBSCR_ERR_TIMEOUT);
+ }
+
+ REQUIRE(subscription_manager.num_complete() == 0);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ }
+
+ SECTION("Verify timeout behaviour if listener does not response"){
+ std::cout <<"+++++++++" << std::endl << "TEST TIMEOUT BEHAVIOUR " << std::endl;
+
+ int res;
+ int num_sources = 1;
+ int timeout_val = 2;
+ int num_tries = 1;
+ std::vector<int> status_vector(num_sources, 0);
+
+ subscription_manager.clear();
+ subscription_manager.set_timeout(timeout_val);
+ subscription_manager.set_num_retries(num_tries);
+
+ auto start = std::chrono::steady_clock::now();
+ send_request(subscription_manager, status_vector, 0, mock_silent, 0);
+ auto end = std::chrono::steady_clock::now();
+
+ auto diff = end - start;
+
+ REQUIRE(status_vector[0] == SUBSCR_ERR_TIMEOUT);
+ REQUIRE(diff.count() >= num_tries * timeout_val);
+
+ num_tries = 2;
+ subscription_manager.set_num_retries(num_tries);
+ status_vector.clear();
+
+ start = std::chrono::steady_clock::now();
+ send_request(subscription_manager, status_vector, 0, mock_silent, 0);
+ end = std::chrono::steady_clock::now();
+
+ diff = end - start;
+
+ REQUIRE(status_vector[0] == SUBSCR_ERR_TIMEOUT);
+ REQUIRE(diff.count() >= num_tries * timeout_val);
+
+ }
+
+ SECTION("Verify rejection of illegal pdus"){
+ std::cout <<"+++++++++" << std::endl <<"TEST WITH ILLEGAL PDU PARAMS" << std::endl;
+ subscription_helper subscription_info;
+ subscription_response_helper subscription_response_info;
+
+ subscription_manager.clear();
+ int res;
+ int function_id = 60000;
+ int action_id = 4;
+ int action_type = 0;
+
+ int request_id = 1;
+ int req_seq = 1;
+ int message_type = 1;
+ int procedure_code = 27;
+
+ unsigned char event_buf[] = "Hello world";
+ size_t event_buf_len = strlen((const char *)event_buf);
+
+ subscription_info.clear();
+ subscription_info.set_request(request_id, req_seq);
+ subscription_info.set_function_id(function_id);
+ subscription_info.add_action(action_id, action_type);
+ subscription_info.set_event_def(&event_buf[0], event_buf_len);
+
+
+ auto transmitter = std::bind(mock_silent, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3, 0);
+ res = subscription_manager.RequestSubscription(subscription_info, subscription_response_info , RIC_SUB_REQ, transmitter);
+ REQUIRE(res == SUBSCR_ERR_ENCODE);
+
+
+
+ }
+
+ SECTION("Verify subscription request/response fail"){
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SUBSCRIPTION FAILURE " << std::endl;
+
+ subscription_manager.clear();
+
+ int num_sources = 20;
+ int num_sinks = 5;
+
+ std::vector<std::thread> source_list;
+ std::vector<std::thread> sink_list;
+
+ std::vector<int> status_vector(num_sources, 0);
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_tx), -1));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_ERR_FAIL);
+ }
+
+ // stop the sinks
+ is_running =false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+
+ REQUIRE(subscription_manager.num_complete() == 0);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ }
+
+ SECTION("Verify subscription request/response success"){
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SUBSCRIPTION SUCCESS " << std::endl;
+
+
+ subscription_manager.clear();
+
+ int num_sources = 10;
+ int num_sinks = 5;
+
+ std::vector<std::thread> source_list;
+ std::vector<std::thread> sink_list;
+
+ std::vector<int> status_vector(num_sources, 0);
+
+ // Test null cases in queries
+ REQUIRE(subscription_manager.is_subscription_entry(10) == false);
+ REQUIRE(subscription_manager.is_request_entry(1) == false);
+ REQUIRE(subscription_manager.get_request_status(1) == -1);
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] < 0);
+ REQUIRE(subscription_manager.is_subscription_entry(-1 * status_vector[i]) == true);
+ }
+
+ // stop the sinks
+ is_running =false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ const subscription_response_helper * sub_info = subscription_manager.get_subscription(-1);
+ REQUIRE(sub_info == NULL);
+
+ sub_info = subscription_manager.get_subscription(-1 * status_vector[0]);
+ REQUIRE(sub_info != NULL);
+ REQUIRE(sub_info->get_request_id() == -1 * status_vector[0]);
+
+ }
+
+ SECTION("Delete requests for non-existent subscription requests"){
+ std::cout <<"+++++++++" << std::endl << "TEST SUBSCRIPTION DELETE WITH NO CORRESPONDING REQUEST" << std::endl;
+
+ subscription_manager.clear();
+ REQUIRE(subscription_manager.get_request_status(0) == -1);
+ REQUIRE(subscription_manager.is_subscription_entry(0) == false);
+ REQUIRE(subscription_manager.is_request_entry(0) == false);
+
+ int num_sources = 10;
+
+ std::vector<std::thread> source_list;
+ std::vector<int> status_vector(num_sources, 0);
+ srand(100);
+ for(int i = 0; i < num_sources; i++){
+ int req_id = rand()%1000;
+ source_list.push_back(std::thread(delete_request, std::ref(subscription_manager), std::ref(status_vector), i,req_id , std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_ERR_MISSING);
+ }
+
+
+ }
+
+
+
+ SECTION("Delete requests that have succeeeded"){
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SUBSCRIPTION DELETES " << std::endl;
+
+ subscription_manager.clear();
+
+ int num_sources = 10;
+ int num_sinks = 5;
+ const subscription_response_helper * sub_resp_info;
+
+ std::vector<std::thread> source_list;
+ std::vector<std::thread> sink_list;
+
+ std::vector<int> status_vector(num_sources, 0);
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ // First do subscriptions ...
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] < 0 );
+ REQUIRE(subscription_manager.is_subscription_entry(-1 * status_vector[i]) == true);
+ sub_resp_info = subscription_manager.get_subscription(-1 * status_vector[i]);
+ REQUIRE(sub_resp_info != NULL);
+ REQUIRE(sub_resp_info->get_request_id() == -1 * status_vector[i]);
+
+ }
+
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+
+ // Store ids ..
+ std::vector<int> completed_requests = status_vector;
+
+ // Delete successes
+ source_list.clear();
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(delete_request, std::ref(subscription_manager), std::ref(status_vector), i, -1 * completed_requests[i], std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_SUCCESS);
+ }
+
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+
+ // stop the sinks
+ is_running =false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+ REQUIRE(subscription_manager.num_complete() == 0);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ }
+
+ SECTION("Deletes that fail"){
+
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SUBSCRIPTION DELETES THAT FAIL " << std::endl;
+
+ subscription_manager.clear();
+
+ int num_sources = 10;
+ int num_sinks = 5;
+ const subscription_response_helper * sub_resp_info;
+
+ std::vector<std::thread> source_list;
+ std::vector<std::thread> sink_list;
+ std::vector<int> status_vector(num_sources, 0);
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ // First do subscriptions ...
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] < 0 );
+ REQUIRE(subscription_manager.is_subscription_entry(-1 * status_vector[i]) == true);
+ sub_resp_info = subscription_manager.get_subscription(-1 * status_vector[i]);
+ REQUIRE(sub_resp_info != NULL);
+ REQUIRE(sub_resp_info->get_request_id() == -1 * status_vector[i]);
+
+ }
+
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+
+ // Store ids ..
+ std::vector<int> completed_requests = status_vector;
+
+ // Delete failures
+ source_list.clear();
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(delete_request, std::ref(subscription_manager), std::ref(status_vector), i, -1 * completed_requests[i], std::ref(mock_tx), 1));
+ }
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_ERR_FAIL);
+ }
+
+
+ // stop the sinks
+ is_running = false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+
+ // subscriptions are still there
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+
+ }
+
+ SECTION("Deletes that timed out "){
+
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SUBSCRIPTION DELETES THAT TIMEOUT " << std::endl;
+
+ subscription_manager.clear();
+
+ int num_sources = 10;
+ int num_sinks = 5;
+ const subscription_response_helper * sub_resp_info;
+
+ std::vector<std::thread> source_list;
+ std::vector<std::thread> sink_list;
+
+ std::vector<int> status_vector(num_sources, 0);
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ // First do subscriptions ...
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(send_request, std::ref(subscription_manager), std::ref(status_vector), i, std::ref(mock_tx), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] < 0 );
+ REQUIRE(subscription_manager.is_subscription_entry(-1 * status_vector[i]) == true);
+ sub_resp_info = subscription_manager.get_subscription(-1 * status_vector[i]);
+ REQUIRE(sub_resp_info != NULL);
+ REQUIRE(sub_resp_info->get_request_id() == -1 * status_vector[i]);
+
+ }
+
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+
+ // Store ids ..
+ std::vector<int> completed_requests = status_vector;
+
+ // Delete with time-outs
+ source_list.clear();
+ for(int i = 0; i < num_sources; i++){
+ source_list.push_back(std::thread(delete_request, std::ref(subscription_manager), std::ref(status_vector), i, -1 * completed_requests[i], std::ref(mock_silent), 0));
+ }
+
+
+ for(int i = 0; i < num_sources; i++){
+ source_list[i].join();
+ REQUIRE(status_vector[i] == SUBSCR_ERR_TIMEOUT);
+ }
+
+ // stop the sinks
+ is_running = false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+
+ REQUIRE(subscription_manager.num_complete() == num_sources);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ }
+
+ SECTION("Spurious messages"){
+ std::cout <<"+++++++++" << std::endl << "TEST WITH SPURIOUS RESPONSES" << std::endl;
+
+ // In this section, we basically inject
+ // spurious messages to subscription handler.
+ // There are no outcomes. basically
+ // handler should be able to ignore these messages
+
+ int num_packets = 50;
+ int num_sinks = 10;
+ std::vector<std::thread> sink_list;
+
+ subscription_manager.clear();
+ std::cout <<"Message queue size prior to fill = " << message_bus.size() << std::endl;
+ random_tx(num_packets);
+ std::cout <<"Message queue size post fill = " << message_bus.size() << std::endl;
+
+
+ // start up the sinks
+ is_running = true;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list.push_back(std::thread(mock_RAN, std::ref(subscription_manager)));
+ }
+
+ // wait for queue to drain out
+ while(! message_bus.empty()){
+ sleep(2);
+ }
+
+ // stop the sinks
+ is_running =false;
+ for(int i = 0; i < num_sinks; i++){
+ sink_list[i].join();
+ }
+ REQUIRE(subscription_manager.num_complete() == 0);
+ REQUIRE(subscription_manager.num_pending() == 0);
+
+ std::cout <<"Message queue size at end = " << message_bus.size() << std::endl;
+
+ }
+
+};
+
+
+
Code Review / ric-app / admin.git / blobdiff