#include <future>
#include <thread>
#include <iomanip>
#include <iostream>
#include <boost/asio/ip/host_name.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/exception/diagnostic_information.hpp>
#if defined(__linux__) || defined(ANDROID)
#include <dlfcn.h>
#include <sys/syscall.h>
#endif
#include <vsomeip/defines.hpp>
#include <vsomeip/runtime.hpp>
#include <vsomeip/plugins/application_plugin.hpp>
#include <vsomeip/plugins/pre_configuration_plugin.hpp>
#include <vsomeip/internal/logger.hpp>
#include "../include/application_impl.hpp"
#ifdef VSOMEIP_ENABLE_MULTIPLE_ROUTING_MANAGERS
#include "../../configuration/include/configuration_impl.hpp"
#else
#include "../../configuration/include/configuration.hpp"
#include "../../configuration/include/configuration_plugin.hpp"
#endif #include "../../endpoints/include/endpoint.hpp"
#include "../../message/include/serializer.hpp"
#include "../../plugin/include/plugin_manager_impl.hpp"
#include "../../routing/include/routing_manager_impl.hpp"
#include "../../routing/include/routing_manager_client.hpp"
#include "../../security/include/security.hpp"
#include "../../tracing/include/connector_impl.hpp"
#include "../../utility/include/utility.hpp"
namespace vsomeip_v3 {
#ifdef ANDROID
configuration::~configuration() {}
#endif
uint32_t application_impl::app_counter__ = 0;
std::mutex application_impl::app_counter_mutex__;
application_impl::application_impl(const std::string &_name, const std::string &_path)
: runtime_(runtime::get()),
client_(VSOMEIP_CLIENT_UNSET),
session_(0),
is_initialized_(false),
name_(_name),
path_(_path),
#if VSOMEIP_BOOST_VERSION >= 106600
work_(std::make_shared<boost::asio::executor_work_guard<
boost::asio::io_context::executor_type> >(io_.get_executor())),
#else
work_(std::make_shared<boost::asio::io_context::work>(io_)),
#endif
routing_(0),
state_(state_type_e::ST_DEREGISTERED),
security_mode_(security_mode_e::SM_ON),
#ifdef VSOMEIP_ENABLE_SIGNAL_HANDLING
signals_(io_, SIGINT, SIGTERM),
catched_signal_(false),
#endif
is_dispatching_(false),
max_dispatchers_(VSOMEIP_MAX_DISPATCHERS),
max_dispatch_time_(VSOMEIP_MAX_DISPATCH_TIME),
stopped_(false),
block_stopping_(false),
is_routing_manager_host_(false),
stopped_called_(false),
watchdog_timer_(io_),
client_side_logging_(false),
has_session_handling_(true)
{
}
application_impl::~application_impl() {
runtime_->remove_application(name_);
try {
if (stop_thread_.joinable()) {
stop_thread_.detach();
}
} catch (const std::exception& e) {
std::cerr << __func__ << " catched exception (shutdown): " << e.what() << std::endl;
}
try {
std::lock_guard<std::mutex> its_lock_start_stop(start_stop_mutex_);
for (const auto& t : io_threads_) {
if (t->joinable()) {
t->detach();
}
}
io_threads_.clear();
} catch (const std::exception& e) {
std::cerr << __func__ << " catched exception (io threads): " << e.what() << std::endl;
}
try {
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
for (const auto& its_dispatcher : dispatchers_) {
if (its_dispatcher.second->joinable()) {
its_dispatcher.second->detach();
}
}
dispatchers_.clear();
} catch (const std::exception& e) {
std::cerr << __func__ << " catched exception (dispatchers): " << e.what() << std::endl;
}
}
bool application_impl::init() {
std::lock_guard<std::mutex> its_initialized_lock(initialize_mutex_);
if(is_initialized_) {
VSOMEIP_WARNING << "Trying to initialize an already initialized application.";
return true;
}
if (name_ == "") {
const char *its_name = getenv(VSOMEIP_ENV_APPLICATION_NAME);
if (nullptr != its_name) {
name_ = its_name;
}
}
std::string configuration_path;
std::string config_module = "";
const char *its_config_module = getenv(VSOMEIP_ENV_CONFIGURATION_MODULE);
if (nullptr != its_config_module) {
} else { #ifndef VSOMEIP_ENABLE_MULTIPLE_ROUTING_MANAGERS
auto its_plugin = plugin_manager::get()->get_plugin(
plugin_type_e::CONFIGURATION_PLUGIN, VSOMEIP_CFG_LIBRARY);
if (its_plugin) {
auto its_configuration_plugin
= std::dynamic_pointer_cast<configuration_plugin>(its_plugin);
if (its_configuration_plugin) {
configuration_ = its_configuration_plugin->get_configuration(name_, path_);
VSOMEIP_INFO << "Configuration module loaded.";
} else {
std::cerr << "Invalid configuration module!" << std::endl;
std::exit(EXIT_FAILURE);
}
} else {
std::cerr << "1 Configuration module could not be loaded!" << std::endl;
std::exit(EXIT_FAILURE);
}
#else
configuration_ = std::dynamic_pointer_cast<configuration>(
std::make_shared<vsomeip_v3::cfg::configuration_impl>(configuration_path));
if (configuration_path.length()) {
configuration_->set_configuration_path(configuration_path);
}
configuration_->load(name_);
#endif }
if (configuration_->is_local_routing()) {
sec_client_.port = VSOMEIP_SEC_PORT_UNUSED;
#ifdef __unix__
sec_client_.user = getuid();
sec_client_.group = getgid();
#else
sec_client_.user = ANY_UID;
sec_client_.group = ANY_GID;
#endif
} else {
auto its_guest_address = configuration_->get_routing_guest_address();
if (its_guest_address.is_v4()) {
#if VSOMEIP_BOOST_VERSION < 106600
sec_client_.host = htonl(static_cast<std::uint32_t>(its_guest_address.to_v4().to_ulong()));
#else
sec_client_.host = htonl(its_guest_address.to_v4().to_uint());
#endif
}
sec_client_.port = VSOMEIP_SEC_PORT_UNSET;
}
if (configuration_->is_security_enabled()) {
if (configuration_->is_security_external()) {
if (security::load()) {
VSOMEIP_INFO << "Using external security implementation!";
auto its_result = security::initialize();
if (VSOMEIP_SEC_POLICY_OK != its_result)
VSOMEIP_ERROR << "Intializing external security implementation failed ("
<< std::dec << its_result << ')';
}
} else {
VSOMEIP_INFO << "Using internal security implementation!";
if (configuration_->is_security_audit())
security_mode_ = security_mode_e::SM_AUDIT;
}
} else {
security_mode_ = security_mode_e::SM_OFF;
VSOMEIP_INFO << "Security disabled!";
}
const char *client_side_logging = getenv(VSOMEIP_ENV_CLIENTSIDELOGGING);
if (client_side_logging != nullptr) {
client_side_logging_ = true;
VSOMEIP_INFO << "Client side logging for application: " << name_
<< " is enabled";
if ('\0' != *client_side_logging) {
std::stringstream its_converter(client_side_logging);
if ('"' == its_converter.peek()) {
its_converter.get(); }
uint16_t val(0xffffu);
bool stop_parsing(false);
do {
const uint16_t prev_val(val);
its_converter >> std::hex >> std::setw(4) >> val;
const std::stringstream::int_type c = its_converter.eof()?'\0':its_converter.get();
switch (c) {
case '"':
case '.':
case ':':
case ' ':
case '\0': {
if ('.' != c) {
if (0xffffu == prev_val) {
VSOMEIP_INFO << "+filter "
<< std::hex << std::setfill('0')
<< std::setw(4) << val;
client_side_logging_filter_.insert(std::make_tuple(val, ANY_INSTANCE));
} else {
VSOMEIP_INFO << "+filter "
<< std::hex << std::setfill('0')
<< std::setw(4) << prev_val << "." << std::setw(4) << val;
client_side_logging_filter_.insert(std::make_tuple(prev_val, val));
}
val = 0xffffu;
}
}
break;
default:
stop_parsing = true;
break;
}
}
while (!stop_parsing && its_converter.good());
}
}
std::shared_ptr<configuration> its_configuration = get_configuration();
if (its_configuration) {
VSOMEIP_INFO << "Initializing vsomeip (" VSOMEIP_VERSION ") application \"" << name_ << "\".";
client_ = its_configuration->get_id(name_);
max_dispatchers_ = its_configuration->get_max_dispatchers(name_) + 1;
max_dispatch_time_ = its_configuration->get_max_dispatch_time(name_);
has_session_handling_ = its_configuration->has_session_handling(name_);
if (!has_session_handling_)
VSOMEIP_INFO << "application: " << name_
<< " has session handling switched off!";
std::string its_routing_host = its_configuration->get_routing_host_name();
if (its_routing_host != "") {
is_routing_manager_host_ = (its_routing_host == name_);
if (is_routing_manager_host_ &&
!utility::is_routing_manager(configuration_->get_network())) {
#ifndef VSOMEIP_ENABLE_MULTIPLE_ROUTING_MANAGERS
VSOMEIP_ERROR << "application: " << name_ << " configured as "
"routing but other routing manager present. Won't "
"instantiate routing";
is_routing_manager_host_ = false;
return false;
#else
is_routing_manager_host_ = true;
#endif }
} else {
auto its_routing_address = its_configuration->get_routing_host_address();
auto its_routing_port = its_configuration->get_routing_host_port();
if (its_routing_address.is_unspecified()
|| is_local_endpoint(its_routing_address, its_routing_port))
is_routing_manager_host_ = utility::is_routing_manager(configuration_->get_network());
}
if (is_routing_manager_host_) {
VSOMEIP_INFO << "Instantiating routing manager [Host].";
if (client_ == VSOMEIP_CLIENT_UNSET) {
client_ = static_cast<client_t>(
(configuration_->get_diagnosis_address() << 8)
& configuration_->get_diagnosis_mask());
utility::request_client_id(configuration_, name_, client_);
}
routing_ = std::make_shared<routing_manager_impl>(this);
} else {
VSOMEIP_INFO << "Instantiating routing manager [Proxy].";
routing_ = std::make_shared<routing_manager_client>(this, client_side_logging_, client_side_logging_filter_);
}
routing_->init();
#ifdef USE_DLT
std::shared_ptr<trace::connector_impl> its_connector
= trace::connector_impl::get();
std::shared_ptr<cfg::trace> its_trace_configuration
= its_configuration->get_trace();
its_connector->configure(its_trace_configuration);
#endif
VSOMEIP_INFO << "Application(" << (name_ != "" ? name_ : "unnamed")
<< ", " << std::hex << std::setfill('0') << std::setw(4) << client_
<< ") is initialized ("
<< std::dec << max_dispatchers_ << ", " << max_dispatch_time_ << ").";
is_initialized_ = true;
}
#ifdef VSOMEIP_ENABLE_SIGNAL_HANDLING
if (is_initialized_) {
signals_.add(SIGINT);
signals_.add(SIGTERM);
auto its_signal_handler =
[this] (boost::system::error_code const &_error, int _signal) {
if (!_error) {
switch (_signal) {
case SIGTERM:
case SIGINT:
catched_signal_ = true;
stop();
break;
default:
break;
}
}
};
signals_.async_wait(its_signal_handler);
}
#endif
if (configuration_) {
auto its_plugins = configuration_->get_plugins(name_);
auto its_app_plugin_info = its_plugins.find(plugin_type_e::APPLICATION_PLUGIN);
if (its_app_plugin_info != its_plugins.end()) {
for (auto its_library : its_app_plugin_info->second) {
auto its_application_plugin = plugin_manager::get()->get_plugin(
plugin_type_e::APPLICATION_PLUGIN, its_library);
if (its_application_plugin) {
VSOMEIP_INFO << "Client 0x" << std::hex << get_client()
<< " Loading plug-in library: " << its_library << " succeeded!";
std::dynamic_pointer_cast<application_plugin>(its_application_plugin)->
on_application_state_change(name_, application_plugin_state_e::STATE_INITIALIZED);
}
}
}
} else {
std::cerr << "Configuration module could not be loaded!" << std::endl;
std::exit(EXIT_FAILURE);
}
return is_initialized_;
}
void application_impl::start() {
#if defined(__linux__) || defined(ANDROID)
if (getpid() != static_cast<pid_t>(syscall(SYS_gettid))) {
std::stringstream s;
s << std::hex << std::setfill('0')
<< std::setw(4) << client_ << "_io" << std::setw(2) << 0;
pthread_setname_np(pthread_self(),s.str().c_str());
}
#endif
{
std::lock_guard<std::mutex> its_initialized_lock(initialize_mutex_);
if (!is_initialized_) {
VSOMEIP_ERROR << "Trying to start an unintialized application.";
return;
}
}
const size_t io_thread_count = configuration_->get_io_thread_count(name_);
const int io_thread_nice_level = configuration_->get_io_thread_nice_level(name_);
{
std::lock_guard<std::mutex> its_lock(start_stop_mutex_);
if (io_.stopped()) {
io_.reset();
} else if(stop_thread_.joinable()) {
VSOMEIP_ERROR << "Trying to start an already started application.";
return;
}
if (stopped_) {
{
std::lock_guard<std::mutex> its_lock_start_stop(block_stop_mutex_);
block_stopping_ = true;
block_stop_cv_.notify_all();
}
stopped_ = false;
return;
}
stopped_ = false;
stopped_called_ = false;
VSOMEIP_INFO << "Starting vsomeip application \"" << name_ << "\" ("
<< std::hex << std::setw(4) << std::setfill('0') << client_
<< ") using " << std::dec << io_thread_count << " threads"
#if defined(__linux__) || defined(ANDROID) || defined(__QNX__)
<< " I/O nice " << io_thread_nice_level
#endif
;
start_caller_id_ = std::this_thread::get_id();
{
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
is_dispatching_ = true;
auto its_main_dispatcher = std::make_shared<std::thread>(
std::bind(&application_impl::main_dispatch, shared_from_this()));
dispatchers_[its_main_dispatcher->get_id()] = its_main_dispatcher;
}
if (stop_thread_.joinable()) {
stop_thread_.join();
}
stop_thread_= std::thread(&application_impl::shutdown, shared_from_this());
if (routing_)
routing_->start();
for (size_t i = 0; i < io_thread_count - 1; i++) {
std::shared_ptr<std::thread> its_thread
= std::make_shared<std::thread>([this, i, io_thread_nice_level] {
VSOMEIP_INFO << "io thread id from application: "
<< std::hex << std::setw(4) << std::setfill('0')
<< client_ << " (" << name_ << ") is: " << std::hex
<< std::this_thread::get_id()
#if defined(__linux__) || defined(ANDROID)
<< " TID: " << std::dec << static_cast<int>(syscall(SYS_gettid))
#endif
;
#if defined(__linux__) || defined(ANDROID)
{
std::stringstream s;
s << std::hex << std::setw(4) << std::setfill('0')
<< client_ << "_io" << std::setw(2) << i+1;
pthread_setname_np(pthread_self(),s.str().c_str());
}
if ((VSOMEIP_IO_THREAD_NICE_LEVEL != io_thread_nice_level) && (io_thread_nice_level != nice(io_thread_nice_level))) {
VSOMEIP_WARNING << "nice(" << io_thread_nice_level << ") failed " << errno << " for " << std::this_thread::get_id();
}
#endif
while(true) {
try {
io_.run();
break;
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::start() "
"caught exception: " << e.what();
}
}
});
io_threads_.insert(its_thread);
}
}
auto its_plugins = configuration_->get_plugins(name_);
auto its_app_plugin_info = its_plugins.find(plugin_type_e::APPLICATION_PLUGIN);
if (its_app_plugin_info != its_plugins.end()) {
for (const auto& its_library : its_app_plugin_info->second) {
auto its_application_plugin = plugin_manager::get()->get_plugin(
plugin_type_e::APPLICATION_PLUGIN, its_library);
if (its_application_plugin) {
std::dynamic_pointer_cast<application_plugin>(its_application_plugin)->
on_application_state_change(name_, application_plugin_state_e::STATE_STARTED);
}
}
}
app_counter_mutex__.lock();
app_counter__++;
app_counter_mutex__.unlock();
VSOMEIP_INFO << "io thread id from application: "
<< std::hex << std::setw(4) << std::setfill('0') << client_ << " ("
<< name_ << ") is: " << std::this_thread::get_id()
#if defined(__linux__) || defined(ANDROID)
<< " TID: " << std::dec << static_cast<int>(syscall(SYS_gettid))
#endif
;
#if defined(__linux__) || defined(ANDROID)
if ((VSOMEIP_IO_THREAD_NICE_LEVEL != io_thread_nice_level) && (io_thread_nice_level != nice(io_thread_nice_level))) {
VSOMEIP_WARNING << "nice(" << io_thread_nice_level << ") failed " << errno << " for " << std::this_thread::get_id();
}
#endif
while(true) {
try {
io_.run();
if (stop_thread_.joinable()) {
stop_thread_.join();
}
break;
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::start() caught exception: " << e.what();
}
}
{
std::lock_guard<std::mutex> its_lock_start_stop(block_stop_mutex_);
block_stopping_ = true;
block_stop_cv_.notify_all();
}
{
std::lock_guard<std::mutex> its_lock(start_stop_mutex_);
stopped_ = false;
}
app_counter_mutex__.lock();
app_counter__--;
#ifdef VSOMEIP_ENABLE_SIGNAL_HANDLING
if (catched_signal_ && !app_counter__) {
app_counter_mutex__.unlock();
VSOMEIP_INFO << "Exiting vsomeip application...";
exit(0);
}
#endif
app_counter_mutex__.unlock();
}
void application_impl::stop() {
VSOMEIP_INFO << "Stopping vsomeip application \"" << name_ << "\" ("
<< std::hex << std::setw(4) << std::setfill('0') << client_ << ").";
bool block = true;
{
std::lock_guard<std::mutex> its_lock_start_stop(start_stop_mutex_);
if (stopped_ || stopped_called_) {
return;
}
stop_caller_id_ = std::this_thread::get_id();
stopped_ = true;
stopped_called_ = true;
for (const auto& thread : io_threads_) {
if (thread->get_id() == std::this_thread::get_id()) {
block = false;
}
}
if (start_caller_id_ == stop_caller_id_) {
block = false;
}
}
auto its_plugins = configuration_->get_plugins(name_);
auto its_app_plugin_info = its_plugins.find(plugin_type_e::APPLICATION_PLUGIN);
if (its_app_plugin_info != its_plugins.end()) {
for (const auto& its_library : its_app_plugin_info->second) {
auto its_application_plugin = plugin_manager::get()->get_plugin(
plugin_type_e::APPLICATION_PLUGIN, its_library);
if (its_application_plugin) {
std::dynamic_pointer_cast<application_plugin>(its_application_plugin)->
on_application_state_change(name_, application_plugin_state_e::STATE_STOPPED);
}
}
}
{
std::lock_guard<std::mutex> its_lock_start_stop(start_stop_mutex_);
stop_cv_.notify_one();
}
if (block) {
std::unique_lock<std::mutex> block_stop_lock(block_stop_mutex_);
while (!block_stopping_) {
block_stop_cv_.wait(block_stop_lock);
}
block_stopping_ = false;
}
}
void application_impl::process(int _number) {
(void)_number;
VSOMEIP_ERROR << "application::process is not (yet) implemented.";
}
security_mode_e application_impl::get_security_mode() const {
return security_mode_;
}
void application_impl::offer_service(service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) {
if (routing_)
routing_->offer_service(client_, _service, _instance, _major, _minor);
}
void application_impl::stop_offer_service(service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) {
if (routing_)
routing_->stop_offer_service(client_, _service, _instance, _major, _minor);
}
void application_impl::request_service(service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) {
if (routing_)
routing_->request_service(client_, _service, _instance, _major, _minor);
}
void application_impl::release_service(service_t _service,
instance_t _instance) {
if (routing_)
routing_->release_service(client_, _service, _instance);
}
void application_impl::subscribe(service_t _service, instance_t _instance,
eventgroup_t _eventgroup,
major_version_t _major,
event_t _event) {
if (routing_) {
bool send_back_cached(false);
bool send_back_cached_group(false);
check_send_back_cached_event(_service, _instance, _event, _eventgroup,
&send_back_cached, &send_back_cached_group);
if (send_back_cached) {
send_back_cached_event(_service, _instance, _event);
} else if(send_back_cached_group) {
send_back_cached_eventgroup(_service, _instance, _eventgroup);
}
if (check_subscription_state(_service, _instance, _eventgroup, _event)) {
routing_->subscribe(client_, &sec_client_,
_service, _instance, _eventgroup, _major,
_event, nullptr);
}
}
}
void application_impl::unsubscribe(service_t _service, instance_t _instance,
eventgroup_t _eventgroup) {
remove_subscription(_service, _instance, _eventgroup, ANY_EVENT);
if (routing_)
routing_->unsubscribe(client_, &sec_client_, _service, _instance, _eventgroup, ANY_EVENT);
}
void application_impl::unsubscribe(service_t _service, instance_t _instance,
eventgroup_t _eventgroup, event_t _event) {
remove_subscription(_service, _instance, _eventgroup, _event);
if (routing_)
routing_->unsubscribe(client_, &sec_client_, _service, _instance, _eventgroup, _event);
}
bool application_impl::is_available(
service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) const {
std::lock_guard<std::mutex> its_lock(availability_mutex_);
return (is_available_unlocked(_service, _instance, _major, _minor)
== availability_state_e::AS_AVAILABLE);
}
availability_state_e
application_impl::is_available_unlocked(
service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) const {
availability_state_e its_state(availability_state_e::AS_UNKNOWN);
auto check_major_minor = [&](const available_instance_t::const_iterator &_found_instance) {
auto found_major = _found_instance->second.find(_major);
if (found_major != _found_instance->second.end()) {
if (_minor <= found_major->second.first || _minor == ANY_MINOR
|| _minor == DEFAULT_MINOR) {
its_state = found_major->second.second;
}
} else if (_major == DEFAULT_MAJOR || _major == ANY_MAJOR) {
for (const auto &found_major : _found_instance->second) {
if (_minor == DEFAULT_MINOR || _minor == ANY_MINOR) {
its_state = found_major.second.second;
break;
} else if (_minor <= found_major.second.first) {
its_state = found_major.second.second;
break;
}
}
}
};
auto found_service = available_.find(_service);
if (found_service != available_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
check_major_minor(found_instance);
} else if (_instance == ANY_INSTANCE) {
for (auto it = found_service->second.cbegin();
it != found_service->second.cend(); it++) {
check_major_minor(it);
if (its_state != availability_state_e::AS_UNKNOWN) {
break;
}
}
}
} else if (_service == ANY_SERVICE) {
for (const auto &found_service : available_) {
auto found_instance = found_service.second.find(_instance);
if (found_instance != found_service.second.end()) {
check_major_minor(found_instance);
if (its_state != availability_state_e::AS_UNKNOWN) {
break;
}
} else if (_instance == ANY_INSTANCE) {
for (auto it = found_service.second.cbegin();
it != found_service.second.cend(); it++) {
check_major_minor(it);
if (its_state != availability_state_e::AS_UNKNOWN) {
break;
}
}
}
if (its_state != availability_state_e::AS_UNKNOWN) {
break;
}
}
}
return its_state;
}
bool application_impl::are_available(
available_t &_available,
service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) const {
std::lock_guard<std::mutex> its_lock(availability_mutex_);
return (are_available_unlocked(_available, _service, _instance, _major, _minor)
== availability_state_e::AS_AVAILABLE);
}
availability_state_e
application_impl::are_available_unlocked(available_t &_available,
service_t _service, instance_t _instance,
major_version_t _major, minor_version_t _minor) const {
if(_service == ANY_SERVICE) {
for(auto its_available_services_it = available_.begin();
its_available_services_it != available_.end();
++its_available_services_it) {
_available[its_available_services_it->first];
}
} else {
if(available_.find(_service) != available_.end()) {
_available[_service];
}
}
for(auto its_available_services_it = _available.begin();
its_available_services_it != _available.end();
++its_available_services_it) {
auto found_available_service = available_.find(its_available_services_it->first);
if (found_available_service != available_.end()) {
if(_instance == ANY_INSTANCE) {
for(auto its_available_instances_it = found_available_service->second.begin();
its_available_instances_it != found_available_service->second.end();
++its_available_instances_it) {
_available[its_available_services_it->first][its_available_instances_it->first];
}
} else {
if(found_available_service->second.find(_instance) != found_available_service->second.end()) {
_available[its_available_services_it->first][_instance];
}
}
}
}
for(auto its_available_services_it = _available.begin();
its_available_services_it != _available.end();
++its_available_services_it) {
auto found_available_service = available_.find(its_available_services_it->first);
if (found_available_service != available_.end()) {
for(auto its_available_instances_it = found_available_service->second.begin();
its_available_instances_it != found_available_service->second.end();
++its_available_instances_it) {
auto found_available_instance = found_available_service->second.find(its_available_instances_it->first);
if(found_available_instance != found_available_service->second.end()) {
if(_major == ANY_MAJOR || _major == DEFAULT_MAJOR) {
for(auto its_available_major_it = found_available_instance->second.begin();
its_available_major_it != found_available_instance->second.end();
++its_available_major_it) {
_available[its_available_services_it->first][its_available_instances_it->first][its_available_major_it->first];
}
} else {
if(found_available_instance->second.find(_major) != found_available_instance->second.end()) {
_available[its_available_services_it->first][its_available_instances_it->first][_major];
}
}
}
}
}
}
auto its_available_services_it = _available.begin();
while(its_available_services_it != _available.end()) {
bool found_minor(false);
auto found_available_service = available_.find(its_available_services_it->first);
if (found_available_service != available_.end()) {
for(auto its_available_instances_it = found_available_service->second.begin();
its_available_instances_it != found_available_service->second.end();
++its_available_instances_it) {
auto found_available_instance = found_available_service->second.find(its_available_instances_it->first);
if(found_available_instance != found_available_service->second.end()) {
for(auto its_available_major_it = found_available_instance->second.begin();
its_available_major_it != found_available_instance->second.end();
++its_available_major_it) {
auto found_available_major = found_available_instance->second.find(its_available_major_it->first);
if (found_available_major != found_available_instance->second.end()) {
if ((_minor == ANY_MINOR || _minor == DEFAULT_MINOR
|| _minor <= found_available_major->second.first)
&& availability_state_e::AS_AVAILABLE == found_available_major->second.second) {
_available[its_available_services_it->first][its_available_instances_it->first][its_available_major_it->first]
= found_available_major->second.first;
found_minor = true;
}
}
}
}
}
}
if(found_minor)
++its_available_services_it;
else
its_available_services_it = _available.erase(its_available_services_it);
}
if (_available.empty()) {
_available[_service][_instance][_major] = _minor ;
return availability_state_e::AS_UNAVAILABLE;
}
return availability_state_e::AS_AVAILABLE;
}
void application_impl::send(std::shared_ptr<message> _message) {
bool is_request = utility::is_request(_message);
if (client_side_logging_
&& (client_side_logging_filter_.empty()
|| (1 == client_side_logging_filter_.count(std::make_tuple(_message->get_service(), ANY_INSTANCE)))
|| (1 == client_side_logging_filter_.count(std::make_tuple(_message->get_service(), _message->get_instance()))))) {
VSOMEIP_INFO << "application_impl::send: ("
<< std::hex << std::setfill('0')
<< std::setw(4) << client_ << "): ["
<< std::setw(4) << _message->get_service() << "."
<< std::setw(4) << _message->get_instance() << "."
<< std::setw(4) << _message->get_method() << ":"
<< std::setw(4) << (is_request ? session_ : _message->get_session()) << ":"
<< std::setw(4) << (is_request ? client_.load() : _message->get_client()) << "] "
<< "type=" << static_cast<std::uint32_t>(_message->get_message_type())
<< " thread=" << std::this_thread::get_id();
}
if (routing_) {
if (is_request) {
_message->set_client(client_);
_message->set_session(get_session(true));
}
(void)routing_->send(client_, _message, false);
}
}
void application_impl::notify(service_t _service, instance_t _instance,
event_t _event, std::shared_ptr<payload> _payload, bool _force) const {
if (routing_)
routing_->notify(_service, _instance, _event, _payload, _force);
}
void application_impl::notify_one(service_t _service, instance_t _instance,
event_t _event, std::shared_ptr<payload> _payload,
client_t _client, bool _force) const {
if (routing_) {
routing_->notify_one(_service, _instance, _event, _payload, _client,
_force
#ifdef VSOMEIP_ENABLE_COMPAT
, false
#endif
);
}
}
void application_impl::register_state_handler(const state_handler_t &_handler) {
std::lock_guard<std::mutex> its_lock(state_handler_mutex_);
handler_ = _handler;
}
void application_impl::unregister_state_handler() {
std::lock_guard<std::mutex> its_lock(state_handler_mutex_);
handler_ = nullptr;
}
void application_impl::register_availability_handler(service_t _service,
instance_t _instance, const availability_handler_t &_handler,
major_version_t _major, minor_version_t _minor) {
std::lock_guard<std::mutex> availability_lock(availability_mutex_);
auto its_handler_ext = [_handler](service_t _service, instance_t _instance,
availability_state_e _state) {
_handler(_service, _instance,
(_state == availability_state_e::AS_AVAILABLE));
};
register_availability_handler_unlocked(_service, _instance,
its_handler_ext, _major, _minor);
}
void application_impl::register_availability_handler(service_t _service,
instance_t _instance, const availability_state_handler_t &_handler,
major_version_t _major, minor_version_t _minor) {
std::lock_guard<std::mutex> availability_lock(availability_mutex_);
register_availability_handler_unlocked(_service, _instance,
_handler, _major, _minor);
}
void application_impl::register_availability_handler_unlocked(service_t _service,
instance_t _instance, availability_state_handler_t _handler,
major_version_t _major, minor_version_t _minor) {
if (state_ == state_type_e::ST_REGISTERED) {
available_t its_available;
auto are_available = are_available_unlocked(its_available, _service, _instance, _major, _minor);
availability_[_service][_instance][_major][_minor]
= std::make_pair(_handler, true);
std::lock_guard<std::mutex> handlers_lock(handlers_mutex_);
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>([_handler, are_available, its_available]() {
for(const auto& available_services_it : its_available)
for(const auto& available_instances_it : available_services_it.second)
_handler(available_services_it.first, available_instances_it.first, are_available);
});
its_sync_handler->handler_type_ = handler_type_e::AVAILABILITY;
its_sync_handler->service_id_ = _service;
its_sync_handler->instance_id_ = _instance;
handlers_.push_back(its_sync_handler);
dispatcher_condition_.notify_one();
} else {
availability_[_service][_instance][_major][_minor]
= std::make_pair(_handler, false);
}
}
void application_impl::unregister_availability_handler(service_t _service,
instance_t _instance, major_version_t _major, minor_version_t _minor) {
std::lock_guard<std::mutex> its_lock(availability_mutex_);
auto found_service = availability_.find(_service);
if (found_service != availability_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_major = found_instance->second.find(_major);
if (found_major != found_instance->second.end()) {
auto found_minor = found_major->second.find(_minor);
if (found_minor != found_major->second.end()) {
found_major->second.erase(_minor);
if (!found_major->second.size()) {
found_instance->second.erase(_major);
if (!found_instance->second.size()) {
found_service->second.erase(_instance);
if (!found_service->second.size()) {
availability_.erase(_service);
}
}
}
}
}
}
}
}
void application_impl::on_subscription(
service_t _service, instance_t _instance, eventgroup_t _eventgroup,
client_t _client, const vsomeip_sec_client_t *_sec_client,
const std::string &_env, bool _subscribed,
const std::function<void(bool)> &_accepted_cb) {
bool handler_found = false;
std::pair<subscription_handler_sec_t, async_subscription_handler_sec_t> its_handlers;
{
std::lock_guard<std::mutex> its_lock(subscription_mutex_);
auto found_service = subscription_.find(_service);
if (found_service != subscription_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
its_handlers = found_eventgroup->second;
handler_found = true;
}
}
}
}
if (handler_found) {
if(auto its_handler = its_handlers.first) {
_accepted_cb(its_handler(_client, _sec_client, _env, _subscribed));
} else if(auto its_handler = its_handlers.second) {
its_handler(_client, _sec_client, _env, _subscribed, _accepted_cb);
}
} else {
_accepted_cb(true);
}
}
void application_impl::register_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
const subscription_handler_t &_handler) {
subscription_handler_ext_t its_handler_ext
= [_handler](client_t _client, uid_t _uid, gid_t _gid,
const std::string &_env, bool _is_subscribed) {
(void)_env; return _handler(_client, _uid, _gid, _is_subscribed);
};
register_subscription_handler(_service, _instance, _eventgroup,
its_handler_ext);
}
void application_impl::register_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
const subscription_handler_ext_t &_handler) {
subscription_handler_sec_t its_handler_sec = [_handler](
client_t _client,
const vsomeip_sec_client_t* _sec_client,
const std::string& _env,
bool _is_subscribed
){
uid_t its_uid {_sec_client->user};
gid_t its_gid {_sec_client->group};
return _handler(
_client,
its_uid,
its_gid,
_env,
_is_subscribed
);
};
register_subscription_handler(_service, _instance, _eventgroup, its_handler_sec);
}
void application_impl::register_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
const subscription_handler_sec_t &_handler) {
std::lock_guard<std::mutex> its_lock(subscription_mutex_);
subscription_[_service][_instance][_eventgroup] = std::make_pair(_handler, nullptr);
}
void application_impl::unregister_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup) {
std::lock_guard<std::mutex> its_lock(subscription_mutex_);
auto found_service = subscription_.find(_service);
if (found_service != subscription_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
found_instance->second.erase(_eventgroup);
}
}
}
}
void application_impl::on_subscription_status(
service_t _service, instance_t _instance,
eventgroup_t _eventgroup, event_t _event, uint16_t _error) {
bool entry_found(false);
{
std::lock_guard<std::mutex> its_lock(subscriptions_state_mutex_);
auto its_service = subscription_state_.find(_service);
if (its_service == subscription_state_.end())
its_service = subscription_state_.find(ANY_SERVICE);
if (its_service != subscription_state_.end()) {
auto its_instance = its_service->second.find(_instance);
if (its_instance == its_service->second.end())
its_instance = its_service->second.find(ANY_INSTANCE);
if (its_instance != its_service->second.end()) {
auto its_eventgroup = its_instance->second.find(_eventgroup);
if (its_eventgroup == its_instance->second.end())
its_eventgroup = its_instance->second.find(ANY_EVENTGROUP);
if (its_eventgroup != its_instance->second.end()) {
auto its_event = its_eventgroup->second.find(_event);
if (its_event == its_eventgroup->second.end())
its_event = its_eventgroup->second.find(ANY_EVENT);
if (its_event != its_eventgroup->second.end()) {
entry_found = true;
its_event->second = (_error ?
subscription_state_e::SUBSCRIPTION_NOT_ACKNOWLEDGED :
subscription_state_e::SUBSCRIPTION_ACKNOWLEDGED);
}
auto its_any_event = its_eventgroup->second.find(ANY_EVENT);
if (its_any_event != its_eventgroup->second.end()) {
entry_found = true;
its_any_event->second = (_error ?
subscription_state_e::SUBSCRIPTION_NOT_ACKNOWLEDGED :
subscription_state_e::SUBSCRIPTION_ACKNOWLEDGED);
}
}
}
}
}
if (entry_found) {
deliver_subscription_state(_service, _instance, _eventgroup, _event, _error);
}
}
void application_impl::deliver_subscription_state(service_t _service, instance_t _instance,
eventgroup_t _eventgroup, event_t _event, uint16_t _error) {
std::vector<subscription_status_handler_t> handlers;
{
std::lock_guard<std::mutex> its_lock(subscription_status_handlers_mutex_);
auto found_service = subscription_status_handlers_.find(_service);
if (found_service != subscription_status_handlers_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
auto found_event = found_eventgroup->second.find(_event);
if (found_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
auto found_any_eventgroup = found_instance->second.find(ANY_EVENTGROUP);
if (found_any_eventgroup != found_instance->second.end()) {
auto found_event = found_any_eventgroup->second.find(_event);
if (found_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_any_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
}
found_instance = found_service->second.find(ANY_INSTANCE);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
auto found_event = found_eventgroup->second.find(_event);
if (found_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
auto found_any_eventgroup = found_instance->second.find(ANY_EVENTGROUP);
if (found_any_eventgroup != found_instance->second.end()) {
auto found_event = found_any_eventgroup->second.find(_event);
if (found_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_any_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
}
}
found_service = subscription_status_handlers_.find(ANY_SERVICE);
if (found_service != subscription_status_handlers_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
auto found_event = found_eventgroup->second.find(_event);
if (found_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
auto found_any_eventgroup = found_instance->second.find(ANY_EVENTGROUP);
if (found_any_eventgroup != found_instance->second.end()) {
auto found_event = found_any_eventgroup->second.find(_event);
if (found_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_any_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
}
found_instance = found_service->second.find(ANY_INSTANCE);
if (found_instance != found_service->second.end()) {
auto found_eventgroup = found_instance->second.find(_eventgroup);
if (found_eventgroup != found_instance->second.end()) {
auto found_event = found_eventgroup->second.find(_event);
if (found_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
auto found_any_eventgroup = found_instance->second.find(ANY_EVENTGROUP);
if (found_any_eventgroup != found_instance->second.end()) {
auto found_event = found_any_eventgroup->second.find(_event);
if (found_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_event->second.second)) {
handlers.push_back(found_event->second.first);
}
}
auto found_any_event = found_any_eventgroup->second.find(ANY_EVENT);
if (found_any_event != found_any_eventgroup->second.end()) {
if (!_error || (_error && found_any_event->second.second)) {
handlers.push_back(found_any_event->second.first);
}
}
}
}
}
}
{
std::unique_lock<std::mutex> handlers_lock(handlers_mutex_);
for (auto &handler : handlers) {
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>([handler, _service,
_instance, _eventgroup,
_event, _error]() {
handler(_service, _instance,
_eventgroup, _event, _error);
});
its_sync_handler->handler_type_ = handler_type_e::SUBSCRIPTION;
its_sync_handler->service_id_ = _service;
its_sync_handler->instance_id_ = _instance;
its_sync_handler->method_id_ = _event;
its_sync_handler->eventgroup_id_ = _eventgroup;
handlers_.push_back(its_sync_handler);
}
if (handlers.size()) {
dispatcher_condition_.notify_one();
}
}
}
void application_impl::register_subscription_status_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup, event_t _event,
subscription_status_handler_t _handler, bool _is_selective) {
std::lock_guard<std::mutex> its_lock(subscription_status_handlers_mutex_);
if (_handler) {
subscription_status_handlers_[_service][_instance][_eventgroup][_event] =
std::make_pair(_handler, _is_selective);
} else {
VSOMEIP_WARNING <<
"application_impl::register_subscription_status_handler: "
"_handler is null, for unregistration please use "
"application_impl::unregister_subscription_status_handler ["
<< std::hex << std::setfill('0')
<< std::setw(4) << _service << "."
<< std::setw(4) << _instance << "."
<< std::setw(4) << _eventgroup << "."
<< std::setw(4) << _event << "]";
}
}
void application_impl::unregister_subscription_status_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup, event_t _event) {
std::lock_guard<std::mutex> its_lock(subscription_status_handlers_mutex_);
auto its_service = subscription_status_handlers_.find(_service);
if (its_service != subscription_status_handlers_.end()) {
auto its_instance = its_service->second.find(_instance);
if (its_instance != its_service->second.end()) {
auto its_eventgroup = its_instance->second.find(_eventgroup);
if (its_eventgroup != its_instance->second.end()) {
its_eventgroup->second.erase(_event);
if (its_eventgroup->second.empty()) {
its_instance->second.erase(_eventgroup);
if (its_instance->second.empty()) {
its_service->second.erase(_instance);
if (its_service->second.empty()) {
subscription_status_handlers_.erase(_service);
}
}
}
}
}
}
}
void application_impl::register_message_handler(service_t _service,
instance_t _instance, method_t _method, const message_handler_t &_handler) {
register_message_handler_ext(_service, _instance, _method, _handler,
handler_registration_type_e::HRT_REPLACE);
}
void application_impl::unregister_message_handler(service_t _service,
instance_t _instance, method_t _method) {
std::lock_guard<std::mutex> its_lock(members_mutex_);
auto found_service = members_.find(_service);
if (found_service != members_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_method = found_instance->second.find(_method);
if (found_method != found_instance->second.end()) {
found_instance->second.erase(_method);
}
}
}
}
void application_impl::offer_event(service_t _service, instance_t _instance,
event_t _notifier, const std::set<eventgroup_t> &_eventgroups,
event_type_e _type,
std::chrono::milliseconds _cycle, bool _change_resets_cycle,
bool _update_on_change,
const epsilon_change_func_t &_epsilon_change_func,
reliability_type_e _reliability) {
if (routing_) {
if (_cycle == std::chrono::milliseconds::zero()
&& _change_resets_cycle == false
&& _update_on_change == true) {
configuration_->get_event_update_properties(
_service, _instance, _notifier,
_cycle, _change_resets_cycle, _update_on_change);
VSOMEIP_INFO << __func__
<< std::hex << std::setfill('0')
<< ": Event [" << std::setw(4) << _service << "."
<< std::setw(4) << _instance << "."
<< std::setw(4) << _notifier
<< "] uses configured cycle time "
<< std::dec << _cycle.count() << "ms";
}
routing_->register_event(client_,
_service, _instance,
_notifier, _eventgroups, _type, _reliability,
_cycle, _change_resets_cycle, _update_on_change,
_epsilon_change_func, true);
}
}
void application_impl::stop_offer_event(service_t _service, instance_t _instance,
event_t _event) {
if (routing_)
routing_->unregister_event(client_, _service, _instance, _event, true);
}
void application_impl::request_event(service_t _service, instance_t _instance,
event_t _event, const std::set<eventgroup_t> &_eventgroups,
event_type_e _type, reliability_type_e _reliability) {
if (routing_)
routing_->register_event(client_,
_service, _instance,
_event, _eventgroups, _type, _reliability,
std::chrono::milliseconds::zero(), false, true,
nullptr,
false);
}
void application_impl::release_event(service_t _service, instance_t _instance,
event_t _event) {
if (routing_)
routing_->unregister_event(client_, _service, _instance, _event, false);
}
const std::string & application_impl::get_name() const {
return name_;
}
client_t application_impl::get_client() const {
return client_;
}
void application_impl::set_client(const client_t &_client) {
client_ = _client;
}
session_t application_impl::get_session(bool _is_request) {
if (!has_session_handling_ && !_is_request)
return 0;
std::lock_guard<std::mutex> its_lock(session_mutex_);
if (0 == ++session_) {
session_ = 1;
}
return session_;
}
const vsomeip_sec_client_t *application_impl::get_sec_client() const {
return &sec_client_;
}
void application_impl::set_sec_client_port(port_t _port) {
sec_client_.port = htons(_port);
}
std::shared_ptr<configuration> application_impl::get_configuration() const {
return configuration_;
}
diagnosis_t application_impl::get_diagnosis() const {
return configuration_->get_diagnosis_address();
}
boost::asio::io_context &application_impl::get_io() {
return io_;
}
void application_impl::on_state(state_type_e _state) {
{
std::lock_guard<std::mutex> availability_lock(availability_mutex_);
if (state_ != _state) {
state_ = _state;
if (state_ == state_type_e::ST_REGISTERED) {
for (const auto &its_service : availability_) {
for (const auto &its_instance : its_service.second) {
for (const auto &its_major : its_instance.second) {
for (const auto &its_minor : its_major.second) {
if (!its_minor.second.second) {
register_availability_handler_unlocked(
its_service.first,
its_instance.first,
its_minor.second.first,
its_major.first, its_minor.first);
}
}
}
}
}
}
}
}
bool has_state_handler(false);
state_handler_t handler = nullptr;
{
std::lock_guard<std::mutex> its_lock(state_handler_mutex_);
if (handler_) {
has_state_handler = true;
handler = handler_;
}
}
if (has_state_handler) {
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>([handler, _state]() {
handler(_state);
});
its_sync_handler->handler_type_ = handler_type_e::STATE;
handlers_.push_back(its_sync_handler);
dispatcher_condition_.notify_one();
}
}
void application_impl::on_availability(service_t _service, instance_t _instance,
availability_state_e _state, major_version_t _major, minor_version_t _minor) {
std::vector<availability_state_handler_t> its_handlers;
{
std::lock_guard<std::mutex> availability_lock(availability_mutex_);
if (_state == is_available_unlocked(_service, _instance, _major, _minor)) {
return;
}
if (_state != availability_state_e::AS_UNAVAILABLE) {
available_[_service][_instance][_major] = std::make_pair(_minor, _state);
} else {
auto found_available_service = available_.find(_service);
if (found_available_service != available_.end()) {
auto found_instance = found_available_service->second.find(_instance);
if( found_instance != found_available_service->second.end()) {
auto found_major = found_instance->second.find(_major);
if( found_major != found_instance->second.end() ){
if(_minor == found_major->second.first)
found_available_service->second.erase(_instance);
}
}
}
}
auto find_matching_handler =
[&](const availability_major_minor_t& _av_ma_mi_it) {
auto found_major = _av_ma_mi_it.find(_major);
if (found_major != _av_ma_mi_it.end()) {
for (std::int32_t mi = static_cast<std::int32_t>(_minor); mi >= 0; mi--) {
const auto found_minor = found_major->second.find(static_cast<minor_version_t>(mi));
if (found_minor != found_major->second.end()) {
its_handlers.push_back(found_minor->second.first);
}
}
const auto found_any_minor = found_major->second.find(ANY_MINOR);
if (found_any_minor != found_major->second.end()) {
its_handlers.push_back(found_any_minor->second.first);
}
}
found_major = _av_ma_mi_it.find(ANY_MAJOR);
if (found_major != _av_ma_mi_it.end()) {
for (std::int32_t mi = static_cast<std::int32_t>(_minor); mi >= 0; mi--) {
const auto found_minor = found_major->second.find(static_cast<minor_version_t>(mi));
if (found_minor != found_major->second.end()) {
its_handlers.push_back(found_minor->second.first);
}
}
const auto found_any_minor = found_major->second.find(ANY_MINOR);
if (found_any_minor != found_major->second.end()) {
its_handlers.push_back(found_any_minor->second.first);
}
}
};
auto found_service = availability_.find(_service);
if (found_service != availability_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
find_matching_handler(found_instance->second);
}
found_instance = found_service->second.find(ANY_INSTANCE);
if (found_instance != found_service->second.end()) {
find_matching_handler(found_instance->second);
}
}
found_service = availability_.find(ANY_SERVICE);
if (found_service != availability_.end()) {
auto found_instance = found_service->second.find(_instance);
if( found_instance != found_service->second.end()) {
find_matching_handler(found_instance->second);
}
found_instance = found_service->second.find(ANY_INSTANCE);
if( found_instance != found_service->second.end()) {
find_matching_handler(found_instance->second);
}
}
{
std::lock_guard<std::mutex> handlers_lock(handlers_mutex_);
for (const auto &handler : its_handlers) {
std::shared_ptr<sync_handler> its_sync_handler =
std::make_shared<sync_handler>(
[handler, _service, _instance, _state]()
{
handler(_service, _instance, _state);
});
its_sync_handler->handler_type_ = handler_type_e::AVAILABILITY;
its_sync_handler->service_id_ = _service;
its_sync_handler->instance_id_ = _instance;
handlers_.push_back(its_sync_handler);
}
}
}
if (_state == availability_state_e::AS_UNAVAILABLE) {
{
std::lock_guard<std::mutex> its_lock(subscriptions_mutex_);
auto found_service = subscriptions_.find(_service);
if (found_service != subscriptions_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
for (auto &event : found_instance->second) {
for (auto &eventgroup : event.second) {
eventgroup.second = false;
}
}
}
}
}
{
std::lock_guard<std::mutex> its_lock(subscriptions_state_mutex_);
auto its_service = subscription_state_.find(_service);
if (its_service != subscription_state_.end()) {
auto its_instance = its_service->second.find(_instance);
if (its_instance != its_service->second.end()) {
for (auto &its_eventgroup : its_instance->second) {
for (auto &its_event : its_eventgroup.second) {
its_event.second
= subscription_state_e::SUBSCRIPTION_NOT_ACKNOWLEDGED;
}
}
}
}
}
}
if (its_handlers.size()) {
std::lock_guard<std::mutex> handlers_lock(handlers_mutex_);
dispatcher_condition_.notify_one();
}
}
void application_impl::find_service_handlers(
std::deque<message_handler_t> &_handlers,
service_t _service, instance_t _instance, method_t _method) const {
auto its_service_it = members_.find(_service);
if (its_service_it != members_.end()) {
find_instance_handlers(_handlers, its_service_it,
_instance, _method);
if (_handlers.empty()) {
find_instance_handlers(_handlers, its_service_it,
ANY_INSTANCE, _method);
}
}
}
void application_impl::find_instance_handlers(
std::deque<message_handler_t> &_handlers,
const members_iterator_t &_it,
instance_t _instance, method_t _method) const {
auto its_instance_it = _it->second.find(_instance);
if (its_instance_it != _it->second.end()) {
find_method_handlers(_handlers, its_instance_it, _method);
if (_handlers.empty()) {
find_method_handlers(_handlers, its_instance_it, ANY_METHOD);
}
}
}
void application_impl::find_method_handlers(
std::deque<message_handler_t> &_handlers,
const members_instances_iterator_t &_it,
method_t _method) const {
auto its_method_it = _it->second.find(_method);
if (its_method_it != _it->second.end()) {
_handlers = its_method_it->second;
}
}
void application_impl::on_message(std::shared_ptr<message> &&_message) {
const service_t its_service = _message->get_service();
const instance_t its_instance = _message->get_instance();
const method_t its_method = _message->get_method();
if (_message->get_message_type() == message_type_e::MT_NOTIFICATION) {
if (!check_for_active_subscription(its_service, its_instance,
static_cast<event_t>(its_method))) {
VSOMEIP_INFO << "application_impl::on_message ["
<< std::hex << std::setfill('0')
<< std::setw(4) << its_service << "."
<< std::setw(4) << its_instance << "."
<< std::setw(4) << its_method << "]"
<< ": blocked as the subscription is already inactive.";
return;
}
}
{
std::lock_guard<std::mutex> its_lock(members_mutex_);
std::deque<message_handler_t> its_handlers;
find_service_handlers(its_handlers, its_service, its_instance, its_method);
if (its_handlers.empty())
find_service_handlers(its_handlers, ANY_SERVICE, its_instance, its_method);
if (its_handlers.size()) {
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
for (const auto &handler : its_handlers) {
std::shared_ptr<sync_handler> its_sync_handler =
std::make_shared<sync_handler>([handler, _message]() {
handler(_message);
});
its_sync_handler->handler_type_ = handler_type_e::MESSAGE;
its_sync_handler->service_id_ = _message->get_service();
its_sync_handler->instance_id_ = _message->get_instance();
its_sync_handler->method_id_ = _message->get_method();
its_sync_handler->session_id_ = _message->get_session();
handlers_.push_back(its_sync_handler);
}
dispatcher_condition_.notify_one();
}
}
}
routing_manager * application_impl::get_routing_manager() const {
return routing_.get();
}
void application_impl::main_dispatch() {
#if defined(__linux__) || defined(ANDROID) || defined(__QNX__)
{
std::stringstream s;
s << std::hex << std::setw(4) << std::setfill('0')
<< client_ << "_m_dispatch";
pthread_setname_np(pthread_self(),s.str().c_str());
}
#endif
const std::thread::id its_id = std::this_thread::get_id();
VSOMEIP_INFO << "main dispatch thread id from application: "
<< std::hex << std::setw(4) << std::setfill('0') << client_ << " ("
<< name_ << ") is: " << std::hex << its_id
#if defined(__linux__) || defined(ANDROID)
<< " TID: " << std::dec << static_cast<int>(syscall(SYS_gettid))
#endif
;
std::unique_lock<std::mutex> its_lock(handlers_mutex_);
while (is_dispatching_) {
if (handlers_.empty() || !is_active_dispatcher(its_id)) {
dispatcher_condition_.notify_all();
while (is_dispatching_ && (handlers_.empty() || !is_active_dispatcher(its_id))) {
dispatcher_condition_.wait(its_lock);
}
} else {
std::shared_ptr<sync_handler> its_handler;
while (is_dispatching_ && is_active_dispatcher(its_id)
&& (its_handler = get_next_handler())) {
its_lock.unlock();
invoke_handler(its_handler);
if (!is_dispatching_)
return;
its_lock.lock();
reschedule_availability_handler(its_handler);
remove_elapsed_dispatchers();
#ifdef _WIN32
if(!is_dispatching_) {
its_lock.unlock();
return;
}
#endif
}
}
}
its_lock.unlock();
}
void application_impl::dispatch() {
#if defined(__linux__) || defined(ANDROID)
{
std::stringstream s;
s << std::hex << std::setw(4) << std::setfill('0')
<< client_ << "_dispatch";
pthread_setname_np(pthread_self(),s.str().c_str());
}
#endif
const std::thread::id its_id = std::this_thread::get_id();
VSOMEIP_INFO << "dispatch thread id from application: "
<< std::hex << std::setw(4) << std::setfill('0') << client_ << " ("
<< name_ << ") is: " << std::hex << its_id
#if defined(__linux__) || defined(ANDROID)
<< " TID: " << std::dec << static_cast<int>(syscall(SYS_gettid))
#endif
;
std::unique_lock<std::mutex> its_lock(handlers_mutex_);
while (is_active_dispatcher(its_id)) {
if (is_dispatching_ && handlers_.empty()) {
dispatcher_condition_.wait(its_lock);
if (handlers_.empty() && !is_active_dispatcher(its_id)) {
if (!is_dispatching_) {
return;
}
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
elapsed_dispatchers_.insert(its_id);
return;
}
} else {
std::shared_ptr<sync_handler> its_handler;
while (is_dispatching_ && is_active_dispatcher(its_id)
&& (its_handler = get_next_handler())) {
its_lock.unlock();
invoke_handler(its_handler);
if (!is_dispatching_)
return;
its_lock.lock();
reschedule_availability_handler(its_handler);
remove_elapsed_dispatchers();
}
}
}
if (is_dispatching_) {
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
elapsed_dispatchers_.insert(its_id);
}
dispatcher_condition_.notify_all();
}
std::shared_ptr<application_impl::sync_handler> application_impl::get_next_handler() {
std::shared_ptr<sync_handler> its_next_handler;
while (!handlers_.empty() && !its_next_handler) {
its_next_handler = handlers_.front();
handlers_.pop_front();
if (its_next_handler->handler_type_ == handler_type_e::AVAILABILITY) {
const std::pair<service_t, instance_t> its_si_pair = std::make_pair(
its_next_handler->service_id_,
its_next_handler->instance_id_);
auto found_si = availability_handlers_.find(its_si_pair);
if (found_si != availability_handlers_.end()
&& !found_si->second.empty()
&& found_si->second.front() != its_next_handler) {
found_si->second.push_back(its_next_handler);
its_next_handler = nullptr;
} else {
availability_handlers_[its_si_pair].push_back(its_next_handler);
}
} else if (its_next_handler->handler_type_ == handler_type_e::MESSAGE) {
const std::pair<service_t, instance_t> its_si_pair = std::make_pair(
its_next_handler->service_id_,
its_next_handler->instance_id_);
auto found_si = availability_handlers_.find(its_si_pair);
if (found_si != availability_handlers_.end()
&& found_si->second.size() > 1) {
found_si->second.push_back(its_next_handler);
its_next_handler = nullptr;
}
}
}
return its_next_handler;
}
void application_impl::reschedule_availability_handler(
const std::shared_ptr<sync_handler> &_handler) {
if (_handler->handler_type_ == handler_type_e::AVAILABILITY) {
const std::pair<service_t, instance_t> its_si_pair = std::make_pair(
_handler->service_id_, _handler->instance_id_);
auto found_si = availability_handlers_.find(its_si_pair);
if (found_si != availability_handlers_.end()) {
if (!found_si->second.empty()
&& found_si->second.front() == _handler) {
found_si->second.pop_front();
for (auto it = found_si->second.rbegin();
it != found_si->second.rend(); it++) {
handlers_.push_front(*it);
}
availability_handlers_.erase(found_si);
}
return;
}
VSOMEIP_WARNING << __func__
<< ": An unknown availability handler returned!";
}
}
void application_impl::invoke_handler(std::shared_ptr<sync_handler> &_handler) {
const std::thread::id its_id = std::this_thread::get_id();
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>(_handler->service_id_,
_handler->instance_id_, _handler->method_id_,
_handler->session_id_, _handler->eventgroup_id_,
_handler->handler_type_);
boost::asio::steady_timer its_dispatcher_timer(io_);
its_dispatcher_timer.expires_from_now(std::chrono::milliseconds(max_dispatch_time_));
its_dispatcher_timer.async_wait([this, its_sync_handler](const boost::system::error_code &_error) {
if (!_error) {
print_blocking_call(its_sync_handler);
if (has_active_dispatcher()) {
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
dispatcher_condition_.notify_all();
} else {
while (is_dispatching_) {
if (dispatcher_mutex_.try_lock()) {
if (dispatchers_.size() < max_dispatchers_) {
if (is_dispatching_) {
auto its_dispatcher = std::make_shared<std::thread>(
std::bind(&application_impl::dispatch, shared_from_this()));
dispatchers_[its_dispatcher->get_id()] = its_dispatcher;
} else {
VSOMEIP_INFO << "Won't start new dispatcher "
"thread as Client=" << std::hex
<< get_client() << " is shutting down";
}
} else {
VSOMEIP_ERROR << "Maximum number of dispatchers exceeded. Configuration: "
<< " Max dispatchers: " << std::dec << max_dispatchers_
<< " Max dispatch time: " << std::dec << max_dispatch_time_;
}
dispatcher_mutex_.unlock();
break;
} else {
std::this_thread::yield();
}
}
}
}
});
if (client_side_logging_
&& (client_side_logging_filter_.empty()
|| (1 == client_side_logging_filter_.count(std::make_tuple(its_sync_handler->service_id_, ANY_INSTANCE)))
|| (1 == client_side_logging_filter_.count(std::make_tuple(its_sync_handler->service_id_, its_sync_handler->instance_id_))))) {
VSOMEIP_INFO << "Invoking handler: ("
<< std::hex << std::setfill('0')
<< std::setw(4) << client_ << "): ["
<< std::setw(4) << its_sync_handler->service_id_ << "."
<< std::setw(4) << its_sync_handler->instance_id_ << "."
<< std::setw(4) << its_sync_handler->method_id_ << ":"
<< std::setw(4) << its_sync_handler->session_id_ << "] "
<< "type=" << static_cast<std::uint32_t>(its_sync_handler->handler_type_)
<< " thread=" << std::hex << its_id;
}
while (is_dispatching_ ) {
if (dispatcher_mutex_.try_lock()) {
running_dispatchers_.insert(its_id);
dispatcher_mutex_.unlock();
break;
}
std::this_thread::yield();
}
if (is_dispatching_) {
try {
_handler->handler_();
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::invoke_handler caught exception: "
<< e.what();
print_blocking_call(its_sync_handler);
}
}
boost::system::error_code ec;
its_dispatcher_timer.cancel(ec);
while (is_dispatching_ ) {
if (dispatcher_mutex_.try_lock()) {
running_dispatchers_.erase(its_id);
dispatcher_mutex_.unlock();
return;
}
std::this_thread::yield();
}
}
bool application_impl::has_active_dispatcher() {
while (is_dispatching_) {
if (dispatcher_mutex_.try_lock()) {
for (const auto &d : dispatchers_) {
if (running_dispatchers_.find(d.first) == running_dispatchers_.end() &&
elapsed_dispatchers_.find(d.first) == elapsed_dispatchers_.end()) {
dispatcher_mutex_.unlock();
return true;
}
}
dispatcher_mutex_.unlock();
return false;
}
std::this_thread::yield();
}
return false;
}
bool application_impl::is_active_dispatcher(const std::thread::id &_id) {
while (is_dispatching_) {
if (dispatcher_mutex_.try_lock()) {
for (const auto &d : dispatchers_) {
if (d.first != _id &&
running_dispatchers_.find(d.first) == running_dispatchers_.end() &&
elapsed_dispatchers_.find(d.first) == elapsed_dispatchers_.end()) {
dispatcher_mutex_.unlock();
return false;
}
}
dispatcher_mutex_.unlock();
return true;
}
std::this_thread::yield();
}
return false;
}
void application_impl::remove_elapsed_dispatchers() {
if (is_dispatching_) {
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
for (auto id : elapsed_dispatchers_) {
auto its_dispatcher = dispatchers_.find(id);
if (its_dispatcher->second->joinable())
its_dispatcher->second->join();
dispatchers_.erase(id);
}
elapsed_dispatchers_.clear();
}
}
void application_impl::clear_all_handler() {
unregister_state_handler();
{
std::lock_guard<std::mutex> its_lock(offered_services_handler_mutex_);
offered_services_handler_ = nullptr;
}
{
std::lock_guard<std::mutex> availability_lock(availability_mutex_);
availability_.clear();
}
{
std::lock_guard<std::mutex> its_lock(subscription_mutex_);
subscription_.clear();
}
{
std::lock_guard<std::mutex> its_lock(subscription_error_mutex_);
eventgroup_error_handlers_.clear();
}
{
std::lock_guard<std::mutex> its_lock(members_mutex_);
members_.clear();
}
{
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
handlers_.clear();
}
}
void application_impl::shutdown() {
VSOMEIP_INFO << "shutdown thread id from application: "
<< std::hex << std::setw(4) << std::setfill('0') << client_ << " ("
<< name_ << ") is: " << std::hex << std::this_thread::get_id()
#if defined(__linux__) || defined(ANDROID)
<< " TID: " << std::dec << static_cast<int>(syscall(SYS_gettid))
#endif
;
#if defined(__linux__) || defined(ANDROID) || defined(__QNX__)
boost::asio::detail::posix_signal_blocker blocker;
{
std::stringstream s;
s << std::hex << std::setw(4) << std::setfill('0')
<< client_ << "_shutdown";
pthread_setname_np(pthread_self(),s.str().c_str());
}
#endif
{
std::unique_lock<std::mutex> its_lock(start_stop_mutex_);
while(!stopped_) {
stop_cv_.wait(its_lock);
}
}
{
std::lock_guard<std::mutex> its_handler_lock(handlers_mutex_);
is_dispatching_ = false;
dispatcher_condition_.notify_all();
}
try {
std::lock_guard<std::mutex> its_lock(dispatcher_mutex_);
for (const auto& its_dispatcher : dispatchers_) {
if (its_dispatcher.second->get_id() != stop_caller_id_) {
if (its_dispatcher.second->joinable()) {
its_dispatcher.second->join();
}
} else {
its_dispatcher.second->detach();
}
}
availability_handlers_.clear();
running_dispatchers_.clear();
elapsed_dispatchers_.clear();
dispatchers_.clear();
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::" << __func__ << ": stopping dispatchers, "
<< " catched exception: " << e.what();
}
try {
if (routing_)
routing_->stop();
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::" << __func__ << ": stopping routing, "
<< " catched exception: " << e.what();
}
try {
work_.reset();
io_.stop();
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::" << __func__ << ": stopping io, "
<< " catched exception: " << e.what();
}
try {
std::lock_guard<std::mutex> its_lock_start_stop(start_stop_mutex_);
for (const auto& t : io_threads_) {
if (t->joinable()) {
t->join();
}
}
io_threads_.clear();
} catch (const std::exception &e) {
VSOMEIP_ERROR << "application_impl::" << __func__ << ": joining threads, "
<< " catched exception: " << e.what();
}
}
bool application_impl::is_routing() const {
return is_routing_manager_host_;
}
void application_impl::send_back_cached_event(service_t _service,
instance_t _instance,
event_t _event) {
std::shared_ptr<event> its_event = routing_->find_event(_service,
_instance, _event);
if (its_event && its_event->is_field() && its_event->is_set()) {
std::shared_ptr<message> its_message = runtime_->create_notification();
its_message->set_service(_service);
its_message->set_method(_event);
its_message->set_instance(_instance);
its_message->set_payload(its_event->get_payload());
its_message->set_initial(true);
on_message(std::move(its_message));
VSOMEIP_INFO << "Sending back cached event ("
<< std::hex << std::setfill('0')
<< std::setw(4) << client_ << "): ["
<< std::setw(4) << _service << "."
<< std::setw(4) << _instance << "."
<< std::setw(4) << _event << "]";
}
}
void application_impl::send_back_cached_eventgroup(service_t _service,
instance_t _instance,
eventgroup_t _eventgroup) {
std::set<std::shared_ptr<event>> its_events = routing_->find_events(_service, _instance,
_eventgroup);
for(const auto &its_event : its_events) {
if (its_event && its_event->is_field() && its_event->is_set()) {
std::shared_ptr<message> its_message = runtime_->create_notification();
const event_t its_event_id(its_event->get_event());
its_message->set_service(_service);
its_message->set_method(its_event_id);
its_message->set_instance(_instance);
its_message->set_payload(its_event->get_payload());
its_message->set_initial(true);
on_message(std::move(its_message));
VSOMEIP_INFO << "Sending back cached event ("
<< std::hex << std::setfill('0')
<< std::setw(4) << client_ << "): ["
<< std::setw(4) << _service << "."
<< std::setw(4) << _instance << "."
<< std::setw(4) << its_event_id
<< "] from eventgroup "
<< std::setw(4) << _eventgroup;
}
}
}
void application_impl::set_routing_state(routing_state_e _routing_state) {
if (routing_)
routing_->set_routing_state(_routing_state);
}
void application_impl::check_send_back_cached_event(
service_t _service, instance_t _instance, event_t _event,
eventgroup_t _eventgroup, bool *_send_back_cached_event,
bool *_send_back_cached_eventgroup) {
std::lock_guard<std::mutex> its_lock(subscriptions_mutex_);
*_send_back_cached_event = false;
*_send_back_cached_eventgroup = false;
bool already_subscribed(false);
auto found_service = subscriptions_.find(_service);
if(found_service != subscriptions_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_event = found_instance->second.find(_event);
if (found_event != found_instance->second.end()) {
auto found_eventgroup = found_event->second.find(_eventgroup);
if (found_eventgroup != found_event->second.end()) {
already_subscribed = true;
if (found_eventgroup->second) {
if(_event == ANY_EVENT) {
*_send_back_cached_eventgroup = true;
} else {
*_send_back_cached_event = true;
}
}
}
}
}
}
if (!already_subscribed) {
subscriptions_[_service][_instance][_event][_eventgroup] = false;
}
}
void application_impl::remove_subscription(service_t _service,
instance_t _instance,
eventgroup_t _eventgroup,
event_t _event) {
{
std::lock_guard<std::mutex> its_lock(subscriptions_state_mutex_);
auto its_service = subscription_state_.find(_service);
if (its_service != subscription_state_.end()) {
auto its_instance = its_service->second.find(_instance);
if (its_instance != its_service->second.end()) {
if (_event == ANY_EVENT) {
its_instance->second.erase(_eventgroup);
} else {
auto its_eventgroup = its_instance->second.find(_eventgroup);
if (its_eventgroup != its_instance->second.end()) {
its_eventgroup->second.erase(_event);
if (its_eventgroup->second.empty()) {
its_instance->second.erase(_eventgroup);
}
}
}
if (its_instance->second.empty())
its_service->second.erase(its_instance);
}
if (its_service->second.empty())
subscription_state_.erase(its_service);
}
}
std::lock_guard<std::mutex> its_lock(subscriptions_mutex_);
auto found_service = subscriptions_.find(_service);
if(found_service != subscriptions_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_event = found_instance->second.find(_event);
if (found_event != found_instance->second.end()) {
if (found_event->second.erase(_eventgroup)) {
if (!found_event->second.size()) {
found_instance->second.erase(_event);
if (!found_instance->second.size()) {
found_service->second.erase(_instance);
if (!found_service->second.size()) {
subscriptions_.erase(_service);
}
}
}
}
}
}
}
}
bool application_impl::check_for_active_subscription(service_t _service,
instance_t _instance,
event_t _event) {
std::lock_guard<std::mutex> its_lock(subscriptions_mutex_);
auto found_service = subscriptions_.find(_service);
if(found_service != subscriptions_.end()) {
auto found_instance = found_service->second.find(_instance);
if (found_instance != found_service->second.end()) {
auto found_event = found_instance->second.find(_event);
if (found_event != found_instance->second.end()) {
if (found_event->second.size()) {
for (auto &eventgroup : found_event->second) {
eventgroup.second = true;
}
return true;
}
} else {
auto found_any_event = found_instance->second.find(ANY_EVENT);
if (found_any_event != found_instance->second.end()) {
if (routing_) {
std::shared_ptr<event> its_event = routing_->find_event(
_service, _instance, _event);
if (its_event) {
for (const auto eg : its_event->get_eventgroups()) {
auto found_eventgroup = found_any_event->second.find(eg);
if (found_eventgroup != found_any_event->second.end()) {
found_eventgroup->second = true;
return true;
}
}
}
}
}
}
}
}
return false;
}
bool application_impl::check_subscription_state(service_t _service, instance_t _instance,
eventgroup_t _eventgroup, event_t _event) {
bool is_acknowledged(false);
bool should_subscribe(true);
{
bool has_found(false);
std::lock_guard<std::mutex> its_lock(subscriptions_state_mutex_);
auto its_service = subscription_state_.find(_service);
if (its_service != subscription_state_.end()) {
auto its_instance = its_service->second.find(_instance);
if (its_instance != its_service->second.end()) {
auto its_eventgroup = its_instance->second.find(_eventgroup);
if (its_eventgroup != its_instance->second.end()) {
auto its_event = its_eventgroup->second.find(_event);
if (its_event != its_eventgroup->second.end()) {
if (its_event->second != subscription_state_e::SUBSCRIPTION_NOT_ACKNOWLEDGED) {
has_found = true;
should_subscribe = false;
if (its_event->second == subscription_state_e::SUBSCRIPTION_ACKNOWLEDGED) {
is_acknowledged = true;
}
}
}
}
}
}
if (!has_found) {
subscription_state_[_service][_instance][_eventgroup][_event]
= subscription_state_e::IS_SUBSCRIBING;
}
}
if (!should_subscribe && is_acknowledged) {
deliver_subscription_state(_service, _instance, _eventgroup, _event, 0 );
}
return should_subscribe;
}
void application_impl::print_blocking_call(const std::shared_ptr<sync_handler>& _handler) {
switch (_handler->handler_type_) {
case handler_type_e::AVAILABILITY:
VSOMEIP_WARNING << "BLOCKING CALL AVAILABILITY("
<< std::hex << std::setfill('0')
<< std::setw(4) << get_client() << "): ["
<< std::setw(4) << _handler->service_id_ << "."
<< std::setw(4) << _handler->instance_id_ << "]";
break;
case handler_type_e::MESSAGE:
VSOMEIP_WARNING << "BLOCKING CALL MESSAGE("
<< std::hex << std::setfill('0')
<< std::setw(4) << get_client() << "): ["
<< std::setw(4) << _handler->service_id_ << "."
<< std::setw(4) << _handler->instance_id_ << "."
<< std::setw(4) << _handler->method_id_ << ":"
<< std::setw(4) << _handler->session_id_ << "]";
break;
case handler_type_e::STATE:
VSOMEIP_WARNING << "BLOCKING CALL STATE("
<< std::hex << std::setw(4) << std::setfill('0') << get_client() << ")";
break;
case handler_type_e::SUBSCRIPTION:
VSOMEIP_WARNING << "BLOCKING CALL SUBSCRIPTION("
<< std::hex << std::setfill('0')
<< std::setw(4) << get_client() << "): ["
<< std::setw(4) << _handler->service_id_ << "."
<< std::setw(4) << _handler->instance_id_ << "."
<< std::setw(4) << _handler->eventgroup_id_ << ":"
<< std::setw(4) << _handler->method_id_ << "]";
break;
case handler_type_e::OFFERED_SERVICES_INFO:
VSOMEIP_WARNING << "BLOCKING CALL OFFERED_SERVICES_INFO("
<< std::hex << std::setw(4) << std::setfill('0') << get_client() << ")";
break;
case handler_type_e::WATCHDOG:
VSOMEIP_WARNING << "BLOCKING CALL WATCHDOG("
<< std::hex << std::setw(4) << std::setfill('0') << get_client() << ")";
break;
case handler_type_e::UNKNOWN:
VSOMEIP_WARNING << "BLOCKING CALL UNKNOWN("
<< std::hex << std::setw(4) << std::setfill('0') << get_client() << ")";
break;
}
}
void application_impl::get_offered_services_async(offer_type_e _offer_type,
const offered_services_handler_t &_handler) {
{
std::lock_guard<std::mutex> its_lock(offered_services_handler_mutex_);
offered_services_handler_ = _handler;
}
if (!is_routing_manager_host_) {
routing_->send_get_offered_services_info(get_client(), _offer_type);
} else {
std::vector<std::pair<service_t, instance_t>> its_services;
auto its_routing_manager_host = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
for (const auto& s : its_routing_manager_host->get_offered_services()) {
for (const auto& i : s.second) {
auto its_unreliable_endpoint = i.second->get_endpoint(false);
auto its_reliable_endpoint = i.second->get_endpoint(true);
if (_offer_type == offer_type_e::OT_LOCAL) {
if ( ((its_unreliable_endpoint && (its_unreliable_endpoint->get_local_port() == ILLEGAL_PORT))
&& (its_reliable_endpoint && (its_reliable_endpoint->get_local_port() == ILLEGAL_PORT)))
|| (!its_reliable_endpoint && !its_unreliable_endpoint)) {
its_services.push_back(std::make_pair(s.first, i.first));
}
} else if (_offer_type == offer_type_e::OT_REMOTE) {
if ((its_unreliable_endpoint && its_unreliable_endpoint->get_local_port() != ILLEGAL_PORT)
|| (its_reliable_endpoint && its_reliable_endpoint->get_local_port() != ILLEGAL_PORT)) {
its_services.push_back(std::make_pair(s.first, i.first));
}
} else if (_offer_type == offer_type_e::OT_ALL) {
its_services.push_back(std::make_pair(s.first, i.first));
}
}
}
on_offered_services_info(its_services);
}
return;
}
void application_impl::on_offered_services_info(std::vector<std::pair<service_t, instance_t>> &_services) {
bool has_offered_services_handler(false);
offered_services_handler_t handler = nullptr;
{
std::lock_guard<std::mutex> its_lock(offered_services_handler_mutex_);
if (offered_services_handler_) {
has_offered_services_handler = true;
handler = offered_services_handler_;
}
}
if (has_offered_services_handler) {
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>([handler, _services]() {
handler(_services);
});
its_sync_handler->handler_type_ = handler_type_e::OFFERED_SERVICES_INFO;
handlers_.push_back(its_sync_handler);
dispatcher_condition_.notify_one();
}
}
void application_impl::watchdog_cbk(boost::system::error_code const &_error) {
if (!_error) {
watchdog_handler_t handler = nullptr;
{
std::lock_guard<std::mutex> its_lock(watchdog_timer_mutex_);
handler = watchdog_handler_;
if (handler && std::chrono::seconds::zero() != watchdog_interval_) {
watchdog_timer_.expires_from_now(watchdog_interval_);
watchdog_timer_.async_wait(std::bind(&application_impl::watchdog_cbk,
this, std::placeholders::_1));
}
}
if (handler) {
std::lock_guard<std::mutex> its_lock(handlers_mutex_);
std::shared_ptr<sync_handler> its_sync_handler
= std::make_shared<sync_handler>([handler]() { handler(); });
its_sync_handler->handler_type_ = handler_type_e::WATCHDOG;
handlers_.push_back(its_sync_handler);
dispatcher_condition_.notify_one();
}
}
}
void application_impl::set_watchdog_handler(const watchdog_handler_t &_handler,
std::chrono::seconds _interval) {
if (_handler && std::chrono::seconds::zero() != _interval) {
std::lock_guard<std::mutex> its_lock(watchdog_timer_mutex_);
watchdog_handler_ = _handler;
watchdog_interval_ = _interval;
watchdog_timer_.expires_from_now(_interval);
watchdog_timer_.async_wait(std::bind(&application_impl::watchdog_cbk,
this, std::placeholders::_1));
} else {
std::lock_guard<std::mutex> its_lock(watchdog_timer_mutex_);
watchdog_timer_.cancel();
watchdog_handler_ = nullptr;
watchdog_interval_ = std::chrono::seconds::zero();
}
}
void application_impl::register_async_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
const async_subscription_handler_t &_handler) {
async_subscription_handler_ext_t its_handler_ext
= [_handler](client_t _client, uid_t _uid, gid_t _gid,
const std::string &_env, bool _is_subscribed,
const std::function< void (const bool) > &_cb) {
(void)_env; _handler(_client, _uid, _gid, _is_subscribed, _cb);
};
register_async_subscription_handler(_service, _instance, _eventgroup,
its_handler_ext);
}
void application_impl::register_async_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
const async_subscription_handler_ext_t &_handler) {
async_subscription_handler_sec_t its_handler_sec = [_handler](
client_t _client,
const vsomeip_sec_client_t* _sec_client,
const std::string& _env,
bool _is_subscribed,
const std::function<void(bool)> &_cb
){
uid_t its_uid {_sec_client->user};
gid_t its_gid {_sec_client->group};
_handler(
_client,
its_uid,
its_gid,
_env,
_is_subscribed,
_cb
);
};
register_async_subscription_handler(_service, _instance, _eventgroup, its_handler_sec);
}
void application_impl::register_async_subscription_handler(service_t _service,
instance_t _instance, eventgroup_t _eventgroup,
async_subscription_handler_sec_t _handler) {
std::lock_guard<std::mutex> its_lock(subscription_mutex_);
subscription_[_service][_instance][_eventgroup] = std::make_pair(nullptr, _handler);
}
void application_impl::register_sd_acceptance_handler(
const sd_acceptance_handler_t &_handler) {
if (is_routing() && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->register_sd_acceptance_handler(_handler);
}
}
void application_impl::register_reboot_notification_handler(
const reboot_notification_handler_t &_handler) {
if (is_routing() && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->register_reboot_notification_handler(_handler);
}
}
void application_impl::set_sd_acceptance_required(
const remote_info_t &_remote, const std::string &_path, bool _enable) {
if (!is_routing()) {
return;
}
const boost::asio::ip::address its_address(_remote.ip_.is_v4_ ?
static_cast<boost::asio::ip::address>(boost::asio::ip::address_v4(
_remote.ip_.address_.v4_)) :
static_cast<boost::asio::ip::address>(boost::asio::ip::address_v6(
_remote.ip_.address_.v6_)));
if (_remote.first_ == std::numeric_limits<std::uint16_t>::max()
&& _remote.last_ == 0) {
configuration_->set_sd_acceptance_rules_active(its_address, _enable);
return;
}
configuration::port_range_t its_range { _remote.first_, _remote.last_ };
configuration_->set_sd_acceptance_rule(its_address,
its_range, port_type_e::PT_UNKNOWN,
_path, _remote.is_reliable_, _enable, true);
if (_enable && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->sd_acceptance_enabled(its_address, its_range,
_remote.is_reliable_);
}
}
void application_impl::set_sd_acceptance_required(
const sd_acceptance_map_type_t& _remotes, bool _enable) {
(void)_remotes;
(void)_enable;
#if 0#endif
}
application::sd_acceptance_map_type_t
application_impl::get_sd_acceptance_required() {
sd_acceptance_map_type_t its_ret;
if (is_routing()) {
for (const auto& e : configuration_->get_sd_acceptance_rules()) {
remote_info_t its_remote_info;
its_remote_info.ip_.is_v4_ = e.first.is_v4();
if (its_remote_info.ip_.is_v4_) {
its_remote_info.ip_.address_.v4_ = e.first.to_v4().to_bytes();
} else {
its_remote_info.ip_.address_.v6_ = e.first.to_v6().to_bytes();
}
for (const auto& reliability : e.second.second) {
its_remote_info.is_reliable_ = reliability.first;
for (const auto& port_range : reliability.second.first) {
if (port_range.lower() == port_range.upper()) {
its_remote_info.first_ = port_range.lower();
its_remote_info.last_ = port_range.lower();
its_remote_info.is_range_ = false;
} else {
its_remote_info.first_ = port_range.lower();
its_remote_info.last_ = port_range.upper();
its_remote_info.is_range_ = true;
}
its_ret[its_remote_info] = e.second.first;
}
for (const auto& port_range : reliability.second.second) {
if (port_range.lower() == port_range.upper()) {
its_remote_info.first_ = port_range.lower();
its_remote_info.last_ = port_range.lower();
its_remote_info.is_range_ = false;
} else {
its_remote_info.first_ = port_range.lower();
its_remote_info.last_ = port_range.upper();
its_remote_info.is_range_ = true;
}
its_ret[its_remote_info] = e.second.first;
}
}
}
}
return its_ret;
}
void application_impl::register_routing_ready_handler(
const routing_ready_handler_t &_handler) {
if (is_routing() && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->register_routing_ready_handler(_handler);
}
}
void application_impl::register_routing_state_handler(
const routing_state_handler_t &_handler) {
if (is_routing() && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->register_routing_state_handler(_handler);
}
}
bool application_impl::update_service_configuration(service_t _service,
instance_t _instance,
std::uint16_t _port,
bool _reliable,
bool _magic_cookies_enabled,
bool _offer) {
bool ret = false;
if (!is_routing_manager_host_) {
VSOMEIP_ERROR << __func__ << " is only intended to be called by "
"application acting as routing manager host";
} else if (!routing_) {
VSOMEIP_ERROR << __func__ << " routing is zero";
} else {
auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
if (rm_impl) {
if (_offer) {
ret = rm_impl->offer_service_remotely(_service, _instance,
_port, _reliable, _magic_cookies_enabled);
} else {
ret = rm_impl->stop_offer_service_remotely(_service, _instance,
_port, _reliable, _magic_cookies_enabled);
}
}
}
return ret;
}
void application_impl::update_security_policy_configuration(uint32_t _uid,
uint32_t _gid,
::std::shared_ptr<policy> _policy,
std::shared_ptr<payload> _payload,
const security_update_handler_t &_handler) {
#ifdef VSOMEIP_DISABLE_SECURITY
(void)_uid;
(void)_gid;
(void)_policy;
(void)_payload;
(void)_handler;
#else
if (!is_routing()) {
VSOMEIP_ERROR << __func__ << " is only intended to be called by "
"application acting as routing manager host";
} else if (!routing_) {
VSOMEIP_ERROR << __func__ << " routing is zero";
} else {
auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
if (rm_impl) {
rm_impl->update_security_policy_configuration(_uid, _gid, _policy, _payload, _handler);
}
}
#endif }
void application_impl::remove_security_policy_configuration(uint32_t _uid,
uint32_t _gid,
const security_update_handler_t &_handler) {
#ifdef VSOMEIP_DISABLE_SECURITY
(void)_uid;
(void)_gid;
(void)_handler;
#else
if (!is_routing()) {
VSOMEIP_ERROR << __func__ << " is only intended to be called by "
"application acting as routing manager host";
} else if (!routing_) {
VSOMEIP_ERROR << __func__ << " routing is zero";
} else {
auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
if (rm_impl) {
rm_impl->remove_security_policy_configuration(_uid, _gid, _handler);
}
}
#endif }
void application_impl::subscribe_with_debounce(service_t _service, instance_t _instance,
eventgroup_t _eventgroup, major_version_t _major,
event_t _event, const debounce_filter_t &_filter) {
if (routing_) {
bool send_back_cached(false);
bool send_back_cached_group(false);
check_send_back_cached_event(_service, _instance, _event, _eventgroup,
&send_back_cached, &send_back_cached_group);
if (send_back_cached) {
send_back_cached_event(_service, _instance, _event);
} else if(send_back_cached_group) {
send_back_cached_eventgroup(_service, _instance, _eventgroup);
}
if (check_subscription_state(_service, _instance, _eventgroup, _event)) {
auto its_filter = std::make_shared<debounce_filter_impl_t>(_filter);
routing_->subscribe(client_, get_sec_client(),
_service, _instance, _eventgroup, _major,
_event, its_filter);
}
}
}
bool
application_impl::is_local_endpoint(const boost::asio::ip::address &_unicast,
port_t _port) {
try {
boost::asio::ip::tcp::endpoint its_endpoint(_unicast, _port);
boost::asio::ip::tcp::socket its_socket(io_, its_endpoint);
its_socket.close();
return true;
} catch (...) {
}
return false;
}
void application_impl::register_message_acceptance_handler(
const message_acceptance_handler_t &_handler) {
if (is_routing() && routing_) {
const auto rm_impl = std::dynamic_pointer_cast<routing_manager_impl>(routing_);
rm_impl->register_message_acceptance_handler(_handler);
}
}
std::map<std::string, std::string>
application_impl::get_additional_data(const std::string &_plugin_name) {
if (configuration_) {
return configuration_->get_additional_data(name_, _plugin_name);
}
return std::map<std::string, std::string>();
}
void application_impl::register_message_handler_ext(
service_t _service, instance_t _instance, method_t _method,
const message_handler_t &_handler,
handler_registration_type_e _type) {
std::lock_guard<std::mutex> its_lock(members_mutex_);
switch (_type) {
case handler_registration_type_e::HRT_REPLACE:
members_[_service][_instance][_method].clear();
[[gnu::fallthrough]];
case handler_registration_type_e::HRT_APPEND:
members_[_service][_instance][_method].push_back(_handler);
break;
case handler_registration_type_e::HRT_PREPEND:
members_[_service][_instance][_method].push_front(_handler);
break;
default:
;
}
}
}