diff options
Diffstat (limited to 'test/network_tests/subscribe_notify_tests/subscribe_notify_test_one_event_two_eventgroups_client.cpp')
| -rw-r--r-- | test/network_tests/subscribe_notify_tests/subscribe_notify_test_one_event_two_eventgroups_client.cpp | 395 |
1 files changed, 395 insertions, 0 deletions
diff --git a/test/network_tests/subscribe_notify_tests/subscribe_notify_test_one_event_two_eventgroups_client.cpp b/test/network_tests/subscribe_notify_tests/subscribe_notify_test_one_event_two_eventgroups_client.cpp new file mode 100644 index 0000000..c3b37a1 --- /dev/null +++ b/test/network_tests/subscribe_notify_tests/subscribe_notify_test_one_event_two_eventgroups_client.cpp @@ -0,0 +1,395 @@ +// Copyright (C) 2014-2017 Bayerische Motoren Werke Aktiengesellschaft (BMW AG) +// This Source Code Form is subject to the terms of the Mozilla Public +// License, v. 2.0. If a copy of the MPL was not distributed with this +// file, You can obtain one at http://mozilla.org/MPL/2.0/. +#ifndef VSOMEIP_ENABLE_SIGNAL_HANDLING +#include <csignal> +#endif +#include <chrono> +#include <condition_variable> +#include <iomanip> +#include <iostream> +#include <sstream> +#include <thread> +#include <utility> + +#include <vsomeip/vsomeip.hpp> +#include <vsomeip/internal/logger.hpp> + +#include <gtest/gtest.h> + +#include "subscribe_notify_test_globals.hpp" + +class subscribe_notify_test_one_event_two_eventgroups_client { +public: + subscribe_notify_test_one_event_two_eventgroups_client( + struct subscribe_notify_test::service_info _info, bool _use_tcp) : + app_( + vsomeip::runtime::get()->create_application( + "subscribe_notify_test_client")), + info_(_info), + use_tcp_(_use_tcp), + wait_availability_(true), + wait_set_value_(true), + wait_shutdown_response_(true), + run_thread_(std::bind(&subscribe_notify_test_one_event_two_eventgroups_client::run, this)) { + } + ~subscribe_notify_test_one_event_two_eventgroups_client() { + run_thread_.join(); + } + + bool init() { + if (!app_->init()) { + ADD_FAILURE() << "Couldn't initialize application"; + return false; + } + + app_->register_state_handler( + std::bind( + &subscribe_notify_test_one_event_two_eventgroups_client::on_state, + this, std::placeholders::_1)); + + app_->register_message_handler(vsomeip::ANY_SERVICE, + vsomeip::ANY_INSTANCE, vsomeip::ANY_METHOD, + std::bind( + &subscribe_notify_test_one_event_two_eventgroups_client::on_message, + this, std::placeholders::_1)); + + app_->register_availability_handler(info_.service_id, info_.instance_id, + std::bind( + &subscribe_notify_test_one_event_two_eventgroups_client::on_availability, + this, std::placeholders::_1, std::placeholders::_2, + std::placeholders::_3)); + + std::set<vsomeip::eventgroup_t> its_groups; + // the service offers three events in two eventgroups + // one of the events is in both eventgroups (info_.event_id + 2) + its_groups.insert(info_.eventgroup_id); + app_->request_event(info_.service_id, info_.instance_id, + info_.event_id, its_groups, + vsomeip::event_type_e::ET_FIELD, + (use_tcp_ ? vsomeip::reliability_type_e::RT_RELIABLE : vsomeip::reliability_type_e::RT_UNRELIABLE)); + app_->request_event(info_.service_id, info_.instance_id, + static_cast<vsomeip::event_t>(info_.event_id + 2), + its_groups, vsomeip::event_type_e::ET_FIELD, + (use_tcp_ ? vsomeip::reliability_type_e::RT_RELIABLE : vsomeip::reliability_type_e::RT_UNRELIABLE)); + its_groups.erase(info_.eventgroup_id); + its_groups.insert(static_cast<vsomeip::eventgroup_t>(info_.eventgroup_id +1)); + app_->request_event(info_.service_id, info_.instance_id, + static_cast<vsomeip::event_t>(info_.event_id+1), + its_groups, vsomeip::event_type_e::ET_FIELD, + (use_tcp_ ? vsomeip::reliability_type_e::RT_RELIABLE : vsomeip::reliability_type_e::RT_UNRELIABLE)); + app_->request_event(info_.service_id, info_.instance_id, + static_cast<vsomeip::event_t>(info_.event_id+2), + its_groups, vsomeip::event_type_e::ET_FIELD, + (use_tcp_ ? vsomeip::reliability_type_e::RT_RELIABLE : vsomeip::reliability_type_e::RT_UNRELIABLE)); + + return true; + } + + void start() { + app_->start(); + } + + void stop() { + app_->clear_all_handler(); + app_->unsubscribe(info_.service_id, info_.instance_id, info_.eventgroup_id); + app_->unsubscribe(info_.service_id, info_.instance_id, static_cast<vsomeip::eventgroup_t>(info_.eventgroup_id+1)); + app_->release_event(info_.service_id, info_.instance_id, info_.event_id); + app_->release_event(info_.service_id, info_.instance_id, static_cast<vsomeip::event_t>(info_.event_id+1)); + app_->release_event(info_.service_id, info_.instance_id, static_cast<vsomeip::event_t>(info_.event_id+2)); + app_->release_service(info_.service_id, info_.instance_id); + app_->stop(); + } + + void on_state(vsomeip::state_type_e _state) { + if (_state == vsomeip::state_type_e::ST_REGISTERED) { + app_->request_service(info_.service_id, info_.instance_id); + } + } + + void on_availability(vsomeip::service_t _service, + vsomeip::instance_t _instance, + bool _is_available) { + VSOMEIP_DEBUG << "Service [" + << std::setw(4) << std::setfill('0') << std::hex << _service + << "." << _instance << "] is " + << (_is_available ? "available." : "NOT available."); + if (_service == info_.service_id && _instance == info_.instance_id && _is_available) { + std::lock_guard<std::mutex> its_lock(availability_mutex_); + wait_availability_ = false; + availability_condition_.notify_one(); + } + } + + void on_message(const std::shared_ptr<vsomeip::message> &_response) { + std::stringstream its_message; + its_message << "Received a message [" + << std::setw(4) << std::setfill('0') << std::hex + << _response->get_service() << "." + << std::setw(4) << std::setfill('0') << std::hex + << _response->get_instance() << "." + << std::setw(4) << std::setfill('0') << std::hex + << _response->get_method() << "] from Client/Session [" + << std::setw(4) << std::setfill('0') << std::hex + << _response->get_client() << "/" + << std::setw(4) << std::setfill('0') << std::hex + << _response->get_session() + << "] = "; + std::shared_ptr<vsomeip::payload> its_payload = + _response->get_payload(); + its_message << "(" << std::dec << its_payload->get_length() << ") "; + for (uint32_t i = 0; i < its_payload->get_length(); ++i) + its_message << std::hex << std::setw(2) << std::setfill('0') + << (int) its_payload->get_data()[i] << " "; + VSOMEIP_DEBUG << its_message.str(); + ASSERT_EQ(info_.service_id, _response->get_service()); + + if (_response->get_method() == info_.method_id + || _response->get_method() == subscribe_notify_test::shutdown_method_id) { + ASSERT_EQ(vsomeip::message_type_e::MT_RESPONSE, _response->get_message_type()); + ASSERT_EQ(vsomeip::return_code_e::E_OK, _response->get_return_code()); + std::lock_guard<std::mutex> its_lock(shutdown_response_mutex_); + wait_shutdown_response_ = false; + shutdown_response_condition_.notify_one(); + } else if (_response->get_method() == subscribe_notify_test::set_method_id) { + std::lock_guard<std::mutex> its_lock(set_value_mutex_); + wait_set_value_ = false; + set_value_condition_.notify_one(); + } else if (_response->get_method() >= info_.event_id + && _response->get_method() <= static_cast<vsomeip::event_t>(info_.event_id + 3)) { + std::lock_guard<std::mutex> its_lock(events_mutex_); + received_events_.push_back(_response->get_payload()); + if (received_events_.size() > 4) { + ADD_FAILURE() << "Received too many events [" + << std::hex << _response->get_method() + << " (" << std::dec << received_events_.size() << ")"; + } + number_received_events_[_response->get_method()]++; + events_condition_.notify_one(); + } else { + ADD_FAILURE() << "Received unknown method id: " << std::setw(4) + << std::setfill('0') << std::hex << _response->get_method(); + } + + } + + void set_field_at_service(vsomeip::byte_t _value) { + std::shared_ptr<vsomeip::runtime> its_runtime = vsomeip::runtime::get(); + std::shared_ptr<vsomeip::message> its_request = its_runtime->create_request(false); + its_request->set_service(info_.service_id); + its_request->set_instance(info_.instance_id); + its_request->set_method(subscribe_notify_test::set_method_id); + its_request->set_reliable(use_tcp_); + std::shared_ptr<vsomeip::payload> its_payload = its_runtime->create_payload(&_value, sizeof(_value)); + its_request->set_payload(its_payload); + app_->send(its_request); + } + + void call_method_at_service(vsomeip::method_t _method) { + std::shared_ptr<vsomeip::runtime> its_runtime = vsomeip::runtime::get(); + std::shared_ptr<vsomeip::message> its_request = its_runtime->create_request(false); + its_request->set_service(info_.service_id); + its_request->set_instance(info_.instance_id); + its_request->set_method(_method); + its_request->set_reliable(use_tcp_); + app_->send(its_request); + } + + void wait_on_condition(std::unique_lock<std::mutex>&& _lock, bool *_predicate, std::condition_variable&& _condition, std::uint32_t _timeout) { + while (*_predicate) { + if (std::cv_status::timeout == _condition.wait_for(_lock, std::chrono::seconds(_timeout))) { + ADD_FAILURE() << "Condition variable wasn't notified within time (" + << _timeout << "sec)"; + } + } + *_predicate = true; + } + + void subscribe_at_service() { + // subscribe to both eventgroups + app_->subscribe(info_.service_id, info_.instance_id, info_.eventgroup_id); + app_->subscribe(info_.service_id, info_.instance_id, static_cast<vsomeip::eventgroup_t>(info_.eventgroup_id+1)); + } + + void unsubscribe_at_service() { + app_->unsubscribe(info_.service_id, info_.instance_id, info_.eventgroup_id); + app_->unsubscribe(info_.service_id, info_.instance_id, static_cast<vsomeip::eventgroup_t>(info_.eventgroup_id+1)); + } + + void wait_for_events(std::unique_lock<std::mutex>&& _lock, + std::uint32_t _expected_number_received_events, + std::condition_variable&& _condition) { + std::cv_status its_status(std::cv_status::no_timeout); + while (received_events_.size() != _expected_number_received_events + && its_status != std::cv_status::timeout) { + its_status = _condition.wait_for(_lock, std::chrono::seconds(15)); + if (std::cv_status::timeout == its_status) { + ADD_FAILURE() << "Didn't receive expected number of events: " + << _expected_number_received_events + << " within time. Instead received: " << received_events_.size(); + } + } + ASSERT_EQ(size_t(_expected_number_received_events), received_events_.size()); + } + + void check_received_events_payload(vsomeip::byte_t _value) { + for (const auto &p : received_events_) { + ASSERT_EQ(vsomeip::length_t(1), p->get_length()); + ASSERT_EQ(vsomeip::byte_t(_value), *p->get_data()); + } + received_events_.clear(); + } + + void check_received_events_number(std::set<std::pair<vsomeip::event_t, std::uint32_t>> _expected) { + for (const auto &e : _expected) { + auto event = number_received_events_.find(e.first); + ASSERT_NE(number_received_events_.end(), event); + ASSERT_EQ(e.second, event->second); + } + number_received_events_.clear(); + } + + void run() { + std::unique_lock<std::mutex> its_availability_lock(availability_mutex_); + wait_on_condition(std::move(its_availability_lock), &wait_availability_, std::move(availability_condition_), 300); + // service is available now + + for (int i = 0; i < 3; i++) { + // set value + set_field_at_service(0x1); + { + std::unique_lock<std::mutex> its_set_value_lock(set_value_mutex_); + wait_on_condition(std::move(its_set_value_lock), &wait_set_value_, std::move(set_value_condition_), 30); + } + + // subscribe + std::unique_lock<std::mutex> its_events_lock(events_mutex_); + subscribe_at_service(); + wait_for_events(std::move(its_events_lock), 4, std::move(events_condition_)); + check_received_events_payload(0x1); + + std::set<std::pair<vsomeip::event_t, std::uint32_t>> its_expected; + its_expected.insert({info_.event_id, 1}); + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+1), 1}); + // Initial event for the event which is member of both eventgroups has to be sent twice + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+2), 2}); + + check_received_events_number(its_expected); + its_expected.clear(); + // set value again + set_field_at_service(0x2); + { + std::unique_lock<std::mutex> its_set_value_lock(set_value_mutex_); + wait_on_condition(std::move(its_set_value_lock), &wait_set_value_, std::move(set_value_condition_), 30); + } + + wait_for_events(std::move(its_events_lock), 3, std::move(events_condition_)); + check_received_events_payload(0x2); + its_expected.insert({info_.event_id, 1}); + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+1), 1}); + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+2), 1}); + + check_received_events_number(its_expected); + its_expected.clear(); + + // set value again + set_field_at_service(0x3); + { + std::unique_lock<std::mutex> its_set_value_lock(set_value_mutex_); + wait_on_condition(std::move(its_set_value_lock), &wait_set_value_, std::move(set_value_condition_), 30); + } + wait_for_events(std::move(its_events_lock), 3, std::move(events_condition_)); + check_received_events_payload(0x3); + its_expected.insert({info_.event_id, 1}); + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+1), 1}); + its_expected.insert({static_cast<vsomeip::event_t>(info_.event_id+2), 1}); + check_received_events_number(its_expected); + + unsubscribe_at_service(); + // sleep some time to ensure the unsubscription was processed by the + // remote routing_manager before setting the field again in the next + // loop. + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + } + std::unique_lock<std::mutex> its_shutdown_lock(shutdown_response_mutex_); + call_method_at_service(subscribe_notify_test::shutdown_method_id); + wait_on_condition(std::move(its_shutdown_lock), &wait_shutdown_response_, std::move(shutdown_response_condition_), 30); + stop(); + } + +private: + std::shared_ptr<vsomeip::application> app_; + struct subscribe_notify_test::service_info info_; + bool use_tcp_; + + bool wait_availability_; + std::mutex availability_mutex_; + std::condition_variable availability_condition_; + + bool wait_set_value_; + std::mutex set_value_mutex_; + std::condition_variable set_value_condition_; + + bool wait_shutdown_response_; + std::mutex shutdown_response_mutex_; + std::condition_variable shutdown_response_condition_; + + std::mutex events_mutex_; + std::condition_variable events_condition_; + + std::vector<std::shared_ptr<vsomeip::payload>> received_events_; + std::map<vsomeip::event_t, std::uint32_t> number_received_events_; + std::thread run_thread_; +}; + +#ifndef VSOMEIP_ENABLE_SIGNAL_HANDLING + subscribe_notify_test_one_event_two_eventgroups_client *its_client_ptr(nullptr); + void handle_signal(int _signal) { + if (its_client_ptr != nullptr && + (_signal == SIGINT || _signal == SIGTERM)) + its_client_ptr->stop(); + } +#endif + +static bool use_tcp; + +TEST(someip_subscribe_notify_test_one_event_two_eventgroups, subscribe_to_service) +{ + subscribe_notify_test_one_event_two_eventgroups_client its_client( + subscribe_notify_test::service_info_subscriber_based_notification, use_tcp); +#ifndef VSOMEIP_ENABLE_SIGNAL_HANDLING + its_client_ptr = &its_client; + signal(SIGINT, handle_signal); + signal(SIGTERM, handle_signal); +#endif + if (its_client.init()) { + its_client.start(); + } +} + +#if defined(__linux__) || defined(ANDROID) +int main(int argc, char** argv) +{ + ::testing::InitGoogleTest(&argc, argv); + if(argc < 2) { + std::cerr << "Please specify a offer type of the service, like: " << argv[0] << " UDP" << std::endl; + std::cerr << "Valid offer types include:" << std::endl; + std::cerr << "[UDP, TCP]" << std::endl; + return 1; + } + + if(std::string("TCP") == std::string(argv[1])) { + use_tcp = true; + } else if(std::string("UDP") == std::string(argv[1])) { + use_tcp = false; + } else { + std::cerr << "Wrong subscription type passed, exiting" << std::endl; + std::cerr << "Valid subscription types include:" << std::endl; + std::cerr << "[UDP, TCP]" << std::endl; + return 1; + } + + return RUN_ALL_TESTS(); +} +#endif |