| 2134 | } |
| 2135 | |
| 2136 | void routing_manager_impl::on_subscribe_ack(client_t _client, service_t _service, instance_t _instance, eventgroup_t _eventgroup, |
| 2137 | event_t _event, remote_subscription_id_t _id) { |
| 2138 | // have to lock event_registration_mutex_ to prevent parallel execution with subscribe |
| 2139 | // as to not miss any acknowledgement for a subscription that is currently being processed |
| 2140 | // (insert_subscription/handle_subscription_state) |
| 2141 | std::scoped_lock its_lock{event_registration_mutex_}; |
| 2142 | auto its_eventgroup = find_eventgroup(_service, _instance, _eventgroup); |
| 2143 | if (its_eventgroup) { |
| 2144 | auto its_subscription = its_eventgroup->get_remote_subscription(_id); |
| 2145 | if (its_subscription) { |
| 2146 | its_subscription->set_client_state(_client, remote_subscription_state_e::SUBSCRIPTION_ACKED); |
| 2147 | |
| 2148 | auto its_parent = its_subscription->get_parent(); |
| 2149 | if (its_parent) { |
| 2150 | its_parent->set_client_state(_client, remote_subscription_state_e::SUBSCRIPTION_ACKED); |
| 2151 | if (!its_subscription->is_pending()) { |
| 2152 | its_eventgroup->remove_remote_subscription(_id); |
| 2153 | } |
| 2154 | } |
| 2155 | |
| 2156 | if (discovery_) { |
| 2157 | discovery_->update_remote_subscription(its_subscription); |
| 2158 | |
| 2159 | VSOMEIP_INFO << "REMOTE SUBSCRIBE(" << hex4(_client) << "): [" << hex4(_service) << "." << hex4(_instance) << "." |
| 2160 | << hex4(_eventgroup) << "] from " << its_subscription->get_subscriber()->get_address() << ":" |
| 2161 | << its_subscription->get_subscriber()->get_port() |
| 2162 | << (its_subscription->get_subscriber()->is_reliable() ? " reliable" : " unreliable") |
| 2163 | << " was accepted. id=" << hex4(_id); |
| 2164 | |
| 2165 | return; |
| 2166 | } |
| 2167 | } else { |
| 2168 | std::scoped_lock its_inner_lock{remote_subscription_state_mutex_}; |
| 2169 | const auto its_tuple = std::make_tuple(_service, _instance, _eventgroup, _client); |
| 2170 | const auto its_state = remote_subscription_state_.find(its_tuple); |
| 2171 | if (its_state != remote_subscription_state_.end()) { |
| 2172 | if (its_state->second == subscription_state_e::SUBSCRIPTION_ACKNOWLEDGED) { |
| 2173 | // Already notified! |
| 2174 | return; |
| 2175 | } |
| 2176 | } |
| 2177 | remote_subscription_state_[its_tuple] = subscription_state_e::SUBSCRIPTION_ACKNOWLEDGED; |
| 2178 | } |
| 2179 | |
| 2180 | std::set<client_t> subscribed_clients; |
| 2181 | if (_client == VSOMEIP_ROUTING_CLIENT) { |
| 2182 | for (const auto& its_event : its_eventgroup->get_events()) { |
| 2183 | if (_event == ANY_EVENT || _event == its_event->get_event()) { |
| 2184 | const auto& its_subscribers = its_event->get_subscribers(); |
| 2185 | subscribed_clients.insert(its_subscribers.begin(), its_subscribers.end()); |
| 2186 | } |
| 2187 | } |
| 2188 | } else { |
| 2189 | subscribed_clients.insert(_client); |
| 2190 | } |
| 2191 | |
| 2192 | for (const auto& its_subscriber : subscribed_clients) { |
| 2193 | stub_->send_subscribe_ack(its_subscriber, _service, _instance, _eventgroup, _event); |
no test coverage detected