spi.cpp

/*
 * Copyright (c) 2008-2022, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#include <utility>
 
#include <boost/uuid/uuid_hash.hpp>
#include <boost/functional/hash.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
 
#include "hazelcast/client/hazelcast_client.h"
#include <hazelcast/client/protocol/codec/ErrorCodec.h>
#include <hazelcast/client/spi/impl/ListenerMessageCodec.h>
#include <hazelcast/client/spi/impl/ClientClusterServiceImpl.h>
#include <hazelcast/client/spi/impl/listener/cluster_view_listener.h>
#include <hazelcast/client/spi/impl/listener/listener_service_impl.h>
#include <hazelcast/client/spi/impl/discovery/remote_address_provider.h>
#include <hazelcast/client/spi/impl/discovery/cloud_discovery.h>
#include <hazelcast/util/AddressUtil.h>
#include "hazelcast/client/member_selectors.h"
#include "hazelcast/client/lifecycle_event.h"
#include "hazelcast/client/initial_membership_event.h"
#include "hazelcast/client/membership_event.h"
#include "hazelcast/client/lifecycle_listener.h"
#include "hazelcast/client/spi/ProxyManager.h"
#include "hazelcast/client/spi/ClientProxy.h"
#include "hazelcast/client/spi/ClientContext.h"
#include "hazelcast/client/spi/impl/ClientInvocation.h"
#include "hazelcast/client/spi/impl/ClientInvocationServiceImpl.h"
#include "hazelcast/client/impl/hazelcast_client_instance_impl.h"
#include "hazelcast/client/spi/impl/ClientPartitionServiceImpl.h"
#include "hazelcast/client/spi/impl/DefaultAddressProvider.h"
#include "hazelcast/client/spi/impl/sequence/CallIdSequenceWithBackpressure.h"
#include "hazelcast/client/spi/impl/sequence/CallIdSequenceWithoutBackpressure.h"
#include "hazelcast/client/spi/impl/sequence/FailFastCallIdSequence.h"
#include "hazelcast/client/spi/impl/sequence/CallIdFactory.h"
#include "hazelcast/client/connection/ClientConnectionManagerImpl.h"
#include "hazelcast/util/AddressHelper.h"
#include "hazelcast/util/HashUtil.h"
#include "hazelcast/util/concurrent/BackoffIdleStrategy.h"
#ifdef HZ_BUILD_WITH_SSL
#include <hazelcast/util/SyncHttpsClient.h>
#endif // HZ_BUILD_WITH_SSL
 
namespace hazelcast {
namespace client {
const std::unordered_set<member>&
initial_membership_event::get_members() const
{
    return members_;
}
 
cluster&
initial_membership_event::get_cluster()
{
    return cluster_;
}
 
initial_membership_event::initial_membership_event(
  cluster& cluster,
  std::unordered_set<member> members)
  : cluster_(cluster)
  , members_(std::move(members))
{}
 
lifecycle_event::lifecycle_event(lifecycle_state state)
  : state_(state)
{}
 
lifecycle_event::lifecycle_state
lifecycle_event::get_state() const
{
    return state_;
}
 
namespace spi {
ProxyManager::ProxyManager(ClientContext& context)
  : client_(context)
{}
 
void
ProxyManager::init()
{}
 
void
ProxyManager::destroy()
{
    std::lock_guard<std::recursive_mutex> guard(lock_);
    for (auto& p : proxies_) {
        try {
            auto proxy = p.second.get();
            p.second.get()->on_shutdown();
        } catch (std::exception& se) {
            auto& lg = client_.get_logger();
            HZ_LOG(
              lg,
              finest,
              boost::str(boost::format(
                           "Proxy was not created, "
                           "hence onShutdown can be called. Exception: %1%") %
                         se.what()));
        }
    }
    proxies_.clear();
}
 
boost::future<void>
ProxyManager::initialize(const std::shared_ptr<ClientProxy>& client_proxy)
{
    auto clientMessage = protocol::codec::client_createproxy_encode(
      client_proxy->get_name(), client_proxy->get_service_name());
    return spi::impl::ClientInvocation::create(
             client_, clientMessage, client_proxy->get_service_name())
      ->invoke()
      .then(boost::launch::sync, [=](boost::future<protocol::ClientMessage> f) {
          f.get();
          client_proxy->on_initialize();
      });
}
 
boost::future<void>
ProxyManager::destroy_proxy(ClientProxy& proxy)
{
    DefaultObjectNamespace objectNamespace(proxy.get_service_name(),
                                           proxy.get_name());
    std::shared_ptr<ClientProxy> registeredProxy;
    {
        std::lock_guard<std::recursive_mutex> guard(lock_);
        auto it = proxies_.find(objectNamespace);
        registeredProxy = it == proxies_.end() ? nullptr : it->second.get();
        if (it != proxies_.end()) {
            proxies_.erase(it);
        }
    }
 
    try {
        if (registeredProxy) {
            try {
                proxy.destroy_locally();
                return proxy.destroy_remotely();
            } catch (exception::iexception&) {
                proxy.destroy_remotely();
                throw;
            }
        }
        if (&proxy != registeredProxy.get()) {
            // The given proxy is stale and was already destroyed, but the
            // caller may have allocated local resources in the context of this
            // stale proxy instance after it was destroyed, so we have to
            // cleanup it locally one more time to make sure there are no
            // leaking local resources.
            proxy.destroy_locally();
        }
    } catch (...) {
        if (&proxy != registeredProxy.get()) {
            // The given proxy is stale and was already destroyed, but the
            // caller may have allocated local resources in the context of this
            // stale proxy instance after it was destroyed, so we have to
            // cleanup it locally one more time to make sure there are no
            // leaking local resources.
            proxy.destroy_locally();
        }
        throw;
    }
    return boost::make_ready_future();
}
 
ClientContext::ClientContext(const client::hazelcast_client& hazelcast_client)
  : hazelcast_client_(*hazelcast_client.client_impl_)
{}
 
ClientContext::ClientContext(
  client::impl::hazelcast_client_instance_impl& hazelcast_client)
  : hazelcast_client_(hazelcast_client)
{}
 
serialization::pimpl::SerializationService&
ClientContext::get_serialization_service()
{
    return hazelcast_client_.serialization_service_;
}
 
impl::ClientClusterServiceImpl&
ClientContext::get_client_cluster_service()
{
    return hazelcast_client_.cluster_service_;
}
 
impl::ClientInvocationServiceImpl&
ClientContext::get_invocation_service()
{
    return *hazelcast_client_.invocation_service_;
}
 
client_config&
ClientContext::get_client_config()
{
    return hazelcast_client_.client_config_;
}
 
impl::ClientPartitionServiceImpl&
ClientContext::get_partition_service()
{
    return *hazelcast_client_.partition_service_;
}
 
lifecycle_service&
ClientContext::get_lifecycle_service()
{
    return hazelcast_client_.lifecycle_service_;
}
 
spi::impl::listener::listener_service_impl&
ClientContext::get_client_listener_service()
{
    return *hazelcast_client_.listener_service_;
}
 
connection::ClientConnectionManagerImpl&
ClientContext::get_connection_manager()
{
    return *hazelcast_client_.connection_manager_;
}
 
internal::nearcache::NearCacheManager&
ClientContext::get_near_cache_manager()
{
    return *hazelcast_client_.near_cache_manager_;
}
 
client_properties&
ClientContext::get_client_properties()
{
    return hazelcast_client_.client_properties_;
}
 
cluster&
ClientContext::get_cluster()
{
    return hazelcast_client_.cluster_;
}
 
std::shared_ptr<impl::sequence::CallIdSequence>&
ClientContext::get_call_id_sequence() const
{
    return hazelcast_client_.call_id_sequence_;
}
 
const protocol::ClientExceptionFactory&
ClientContext::get_client_exception_factory() const
{
    return hazelcast_client_.get_exception_factory();
}
 
const std::string&
ClientContext::get_name() const
{
    return hazelcast_client_.get_name();
}
 
impl::ClientExecutionServiceImpl&
ClientContext::get_client_execution_service() const
{
    return *hazelcast_client_.execution_service_;
}
 
const std::shared_ptr<client::impl::ClientLockReferenceIdGenerator>&
ClientContext::get_lock_reference_id_generator()
{
    return hazelcast_client_.get_lock_reference_id_generator();
}
 
std::shared_ptr<client::impl::hazelcast_client_instance_impl>
ClientContext::get_hazelcast_client_implementation()
{
    return hazelcast_client_.shared_from_this();
}
 
spi::ProxyManager&
ClientContext::get_proxy_manager()
{
    return hazelcast_client_.get_proxy_manager();
}
 
logger&
ClientContext::get_logger()
{
    return *hazelcast_client_.logger_;
}
 
client::impl::statistics::Statistics&
ClientContext::get_clientstatistics()
{
    return *hazelcast_client_.statistics_;
}
 
spi::impl::listener::cluster_view_listener&
ClientContext::get_cluster_view_listener()
{
    return *hazelcast_client_.cluster_listener_;
}
 
boost::uuids::uuid
ClientContext::random_uuid()
{
    return hazelcast_client_.random_uuid();
}
 
cp::internal::session::proxy_session_manager&
ClientContext::get_proxy_session_manager()
{
    return hazelcast_client_.proxy_session_manager_;
}
 
lifecycle_service::lifecycle_service(
  ClientContext& client_context,
  const std::vector<lifecycle_listener>& listeners)
  : client_context_(client_context)
  , listeners_()
  , shutdown_completed_latch_(1)
{
    for (const auto& listener : listeners) {
        add_listener(lifecycle_listener(listener));
    }
}
 
bool
lifecycle_service::start()
{
    bool expected = false;
    if (!active_.compare_exchange_strong(expected, true)) {
        return false;
    }
 
    fire_lifecycle_event(lifecycle_event::STARTED);
 
    client_context_.get_client_execution_service().start();
 
    client_context_.get_client_listener_service().start();
 
    client_context_.get_invocation_service().start();
 
    client_context_.get_client_cluster_service().start();
 
    client_context_.get_cluster_view_listener().start();
 
    if (!client_context_.get_connection_manager().start()) {
        return false;
    }
 
    auto& connectionStrategyConfig =
      client_context_.get_client_config().get_connection_strategy_config();
    if (!connectionStrategyConfig.is_async_start()) {
        // The client needs to open connections to all members before any
        // services requiring internal listeners start
        wait_for_initial_membership_event();
        client_context_.get_connection_manager()
          .connect_to_all_cluster_members();
    }
 
    client_context_.get_invocation_service().add_backup_listener();
 
    client_context_.get_clientstatistics().start();
 
    return true;
}
 
void
lifecycle_service::shutdown()
{
    bool expected = true;
    if (!active_.compare_exchange_strong(expected, false)) {
        shutdown_completed_latch_.wait();
        return;
    }
    try {
        fire_lifecycle_event(lifecycle_event::SHUTTING_DOWN);
        client_context_.get_proxy_session_manager().shutdown();
        client_context_.get_clientstatistics().shutdown();
        client_context_.get_proxy_manager().destroy();
        client_context_.get_connection_manager().shutdown();
        client_context_.get_client_cluster_service().shutdown();
        client_context_.get_invocation_service().shutdown();
        client_context_.get_client_listener_service().shutdown();
        client_context_.get_near_cache_manager().destroy_all_near_caches();
        fire_lifecycle_event(lifecycle_event::SHUTDOWN);
        client_context_.get_client_execution_service().shutdown();
        client_context_.get_serialization_service().dispose();
        shutdown_completed_latch_.count_down();
    } catch (std::exception& e) {
        HZ_LOG(
          client_context_.get_logger(),
          info,
          boost::str(
            boost::format(
              "An exception occured during LifecycleService shutdown. %1%") %
            e.what()));
        shutdown_completed_latch_.count_down();
    }
}
 
boost::uuids::uuid
lifecycle_service::add_listener(lifecycle_listener&& lifecycle_listener)
{
    std::lock_guard<std::mutex> lg(listener_lock_);
    const auto id = uuid_generator_();
    listeners_.emplace(id, std::move(lifecycle_listener));
    return id;
}
 
bool
lifecycle_service::remove_listener(const boost::uuids::uuid& registration_id)
{
    std::lock_guard<std::mutex> guard(listener_lock_);
    return listeners_.erase(registration_id) == 1;
}
 
void
lifecycle_service::fire_lifecycle_event(const lifecycle_event& lifecycle_event)
{
    std::lock_guard<std::mutex> guard(listener_lock_);
    logger& lg = client_context_.get_logger();
 
    std::function<void(lifecycle_listener&)> fire_one;
 
    switch (lifecycle_event.get_state()) {
        case lifecycle_event::STARTING: {
            // convert the date string from "2016-04-20" to 20160420
            std::string date(HAZELCAST_STRINGIZE(HAZELCAST_GIT_COMMIT_DATE));
            util::git_date_to_hazelcast_log_date(date);
            std::string commitId(HAZELCAST_STRINGIZE(HAZELCAST_GIT_COMMIT_ID));
            commitId.erase(std::remove(commitId.begin(), commitId.end(), '"'),
                           commitId.end());
 
            HZ_LOG(lg,
                   info,
                   (boost::format("(%1%:%2%) LifecycleService::LifecycleEvent "
                                  "Client (%3%) is STARTING") %
                    date % commitId %
                    client_context_.get_connection_manager().get_client_uuid())
                     .str());
            char msg[100];
            util::hz_snprintf(
              msg,
              100,
              "(%s:%s) LifecycleService::LifecycleEvent STARTING",
              date.c_str(),
              commitId.c_str());
            HZ_LOG(lg, info, msg);
 
            fire_one = [](lifecycle_listener& listener) {
                listener.starting_();
            };
            break;
        }
        case lifecycle_event::STARTED: {
            HZ_LOG(lg, info, "LifecycleService::LifecycleEvent STARTED");
 
            fire_one = [](lifecycle_listener& listener) {
                listener.started_();
            };
            break;
        }
        case lifecycle_event::SHUTTING_DOWN: {
            HZ_LOG(lg, info, "LifecycleService::LifecycleEvent SHUTTING_DOWN");
 
            fire_one = [](lifecycle_listener& listener) {
                listener.shutting_down_();
            };
            break;
        }
        case lifecycle_event::SHUTDOWN: {
            HZ_LOG(lg, info, "LifecycleService::LifecycleEvent SHUTDOWN");
 
            fire_one = [](lifecycle_listener& listener) {
                listener.shutdown_();
            };
            break;
        }
        case lifecycle_event::CLIENT_CONNECTED: {
            HZ_LOG(
              lg, info, "LifecycleService::LifecycleEvent CLIENT_CONNECTED");
 
            fire_one = [](lifecycle_listener& listener) {
                listener.connected_();
            };
            break;
        }
        case lifecycle_event::CLIENT_DISCONNECTED: {
            HZ_LOG(
              lg, info, "LifecycleService::LifecycleEvent CLIENT_DISCONNECTED");
 
            fire_one = [](lifecycle_listener& listener) {
                listener.disconnected_();
            };
            break;
        }
    }
 
    for (auto& item : listeners_) {
        fire_one(item.second);
    }
}
 
bool
lifecycle_service::is_running()
{
    return active_;
}
 
lifecycle_service::~lifecycle_service()
{
    if (active_) {
        shutdown();
    }
}
 
void
lifecycle_service::wait_for_initial_membership_event() const
{
    client_context_.get_client_cluster_service()
      .wait_initial_member_list_fetched();
}
 
DefaultObjectNamespace::DefaultObjectNamespace(const std::string& service,
                                               const std::string& object)
  : service_name_(service)
  , object_name_(object)
{}
 
const std::string&
DefaultObjectNamespace::get_service_name() const
{
    return service_name_;
}
 
const std::string&
DefaultObjectNamespace::get_object_name() const
{
    return object_name_;
}
 
bool
DefaultObjectNamespace::operator==(const DefaultObjectNamespace& rhs) const
{
    return service_name_ == rhs.service_name_ &&
           object_name_ == rhs.object_name_;
}
 
ClientProxy::ClientProxy(const std::string& name,
                         const std::string& service_name,
                         ClientContext& context)
  : name_(name)
  , service_name_(service_name)
  , context_(context)
{}
 
ClientProxy::~ClientProxy() = default;
 
const std::string&
ClientProxy::get_name() const
{
    return name_;
}
 
const std::string&
ClientProxy::get_service_name() const
{
    return service_name_;
}
 
ClientContext&
ClientProxy::get_context()
{
    return context_;
}
 
void
ClientProxy::on_destroy()
{}
 
boost::future<void>
ClientProxy::destroy()
{
    return get_context().get_proxy_manager().destroy_proxy(*this);
}
 
void
ClientProxy::destroy_locally()
{
    if (pre_destroy()) {
        try {
            on_destroy();
            post_destroy();
        } catch (exception::iexception&) {
            post_destroy();
            throw;
        }
    }
}
 
bool
ClientProxy::pre_destroy()
{
    return true;
}
 
void
ClientProxy::post_destroy()
{}
 
void
ClientProxy::on_initialize()
{}
 
void
ClientProxy::on_shutdown()
{}
 
serialization::pimpl::SerializationService&
ClientProxy::get_serialization_service()
{
    return context_.get_serialization_service();
}
 
boost::future<void>
ClientProxy::destroy_remotely()
{
    auto clientMessage = protocol::codec::client_destroyproxy_encode(
      get_name(), get_service_name());
    return spi::impl::ClientInvocation::create(
             get_context(),
             std::make_shared<protocol::ClientMessage>(
               std::move(clientMessage)),
             get_name())
      ->invoke()
      .then(boost::launch::sync,
            [](boost::future<protocol::ClientMessage> f) { f.get(); });
}
 
boost::future<boost::uuids::uuid>
ClientProxy::register_listener(
  std::shared_ptr<impl::ListenerMessageCodec> listener_message_codec,
  std::shared_ptr<client::impl::BaseEventHandler> handler)
{
    handler->set_logger(&get_context().get_logger());
    return get_context().get_client_listener_service().register_listener(
      listener_message_codec, handler);
}
 
boost::future<bool>
ClientProxy::deregister_listener(boost::uuids::uuid registration_id)
{
    return get_context().get_client_listener_service().deregister_listener(
      registration_id);
}
 
namespace impl {
boost::uuids::uuid
ListenerMessageCodec::decode_add_response(protocol::ClientMessage& msg) const
{
    return msg.get_first_uuid();
}
 
bool
ListenerMessageCodec::decode_remove_response(protocol::ClientMessage& msg) const
{
    return msg.get_first_fixed_sized_field<bool>();
}
 
ClientInvocationServiceImpl::ClientInvocationServiceImpl(ClientContext& client)
  : client_(client)
  , logger_(client.get_logger())
  , invocation_timeout_(
      std::chrono::seconds(client.get_client_properties().get_integer(
        client.get_client_properties().get_invocation_timeout_seconds())))
  , invocation_retry_pause_(
      std::chrono::milliseconds(client.get_client_properties().get_long(
        client.get_client_properties().get_invocation_retry_pause_millis())))
  , smart_routing_(
      client.get_client_config().get_network_config().is_smart_routing())
  , backup_acks_enabled_(smart_routing_ &&
                         client.get_client_config().backup_acks_enabled())
  , fail_on_indeterminate_operation_state_(
      client.get_client_properties().get_boolean(
        client.get_client_properties().fail_on_indeterminate_state()))
  , backup_timeout_(
      std::chrono::milliseconds(client.get_client_properties().get_integer(
        client.get_client_properties().backup_timeout_millis())))
{}
 
void
ClientInvocationServiceImpl::start()
{}
 
void
ClientInvocationServiceImpl::add_backup_listener()
{
    if (this->backup_acks_enabled_) {
        auto& listener_service = this->client_.get_client_listener_service();
        listener_service
          .register_listener(std::make_shared<BackupListenerMessageCodec>(),
                             std::make_shared<noop_backup_event_handler>())
          .get();
    }
}
 
void
ClientInvocationServiceImpl::shutdown()
{
    is_shutdown_.store(true);
}
 
std::chrono::milliseconds
ClientInvocationServiceImpl::get_invocation_timeout() const
{
    return invocation_timeout_;
}
 
std::chrono::milliseconds
ClientInvocationServiceImpl::get_invocation_retry_pause() const
{
    return invocation_retry_pause_;
}
 
bool
ClientInvocationServiceImpl::is_redo_operation()
{
    return client_.get_client_config().is_redo_operation();
}
 
void
ClientInvocationServiceImpl::handle_client_message(
  const std::shared_ptr<ClientInvocation>& invocation,
  const std::shared_ptr<protocol::ClientMessage>& response)
{
    try {
        if (protocol::codec::ErrorCodec::EXCEPTION_MESSAGE_TYPE ==
            response->get_message_type()) {
            auto error_holder = protocol::codec::ErrorCodec::decode(*response);
            invocation->notify_exception(
              client_.get_client_exception_factory().create_exception(
                error_holder));
        } else {
            invocation->notify(response);
        }
    } catch (std::exception& e) {
        HZ_LOG(
          logger_,
          severe,
          boost::str(boost::format("Failed to process response for %1%. %2%") %
                     *invocation % e.what()));
    }
}
 
bool
ClientInvocationServiceImpl::send(
  const std::shared_ptr<impl::ClientInvocation>& invocation,
  const std::shared_ptr<connection::Connection>& connection)
{
    if (is_shutdown_) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_client_not_active(
          "ClientInvocationServiceImpl::send", "Client is shut down"));
    }
 
    if (backup_acks_enabled_) {
        invocation->get_client_message()->add_flag(
          protocol::ClientMessage::BACKUP_AWARE_FLAG);
    }
 
    write_to_connection(*connection, invocation);
    invocation->set_send_connection(connection);
    return true;
}
 
void
ClientInvocationServiceImpl::write_to_connection(
  connection::Connection& connection,
  const std::shared_ptr<ClientInvocation>& client_invocation)
{
    auto clientMessage = client_invocation->get_client_message();
    connection.write(client_invocation);
}
 
void
ClientInvocationServiceImpl::check_invocation_allowed()
{
    client_.get_connection_manager().check_invocation_allowed();
}
 
bool
ClientInvocationServiceImpl::invoke(
  std::shared_ptr<ClientInvocation> invocation)
{
    auto connection = client_.get_connection_manager().get_random_connection();
    if (!connection) {
        HZ_LOG(logger_, finest, "No connection found to invoke");
        return false;
    }
    return send(invocation, connection);
}
 
DefaultAddressProvider::DefaultAddressProvider(
  config::client_network_config& network_config)
  : network_config_(network_config)
{}
 
std::vector<address>
DefaultAddressProvider::load_addresses()
{
    std::vector<address> addresses = network_config_.get_addresses();
    if (addresses.empty()) {
        addresses.emplace_back("127.0.0.1", 5701);
    }
 
    // TODO Implement AddressHelper to add alternative ports for the same
    // address
 
    return addresses;
}
 
boost::optional<address>
DefaultAddressProvider::translate(const address& addr)
{
    return addr;
}
 
bool
DefaultAddressProvider::is_default_provider()
{
    return true;
}
 
const boost::shared_ptr<ClientClusterServiceImpl::member_list_snapshot>
  ClientClusterServiceImpl::EMPTY_SNAPSHOT(
    new ClientClusterServiceImpl::member_list_snapshot{ -1 });
 
constexpr boost::chrono::milliseconds
  ClientClusterServiceImpl::INITIAL_MEMBERS_TIMEOUT;
const endpoint_qualifier ClientClusterServiceImpl::CLIENT{ 1, "" };
const endpoint_qualifier ClientClusterServiceImpl::MEMBER{ 0, "" };
 
ClientClusterServiceImpl::ClientClusterServiceImpl(
  hazelcast::client::spi::ClientContext& client)
  : client_(client)
  , member_list_snapshot_(EMPTY_SNAPSHOT)
  , labels_(client.get_client_config().get_labels())
  , initial_list_fetched_latch_(1)
{}
 
boost::uuids::uuid
ClientClusterServiceImpl::add_membership_listener_without_init(
  membership_listener&& listener)
{
    std::lock_guard<std::mutex> g(listeners_lock_);
    auto id = client_.random_uuid();
    listeners_.emplace(id, std::move(listener));
    return id;
}
 
boost::optional<member>
ClientClusterServiceImpl::get_member(boost::uuids::uuid uuid) const
{
    assert(!uuid.is_nil());
    auto members_view_ptr = member_list_snapshot_.load();
    const auto it = members_view_ptr->members.find(uuid);
    if (it == members_view_ptr->members.end()) {
        return boost::none;
    }
    return { it->second };
}
 
std::vector<member>
ClientClusterServiceImpl::get_member_list() const
{
    auto members_view_ptr = member_list_snapshot_.load();
    std::vector<member> result;
    result.reserve(members_view_ptr->members.size());
    for (const auto& e : members_view_ptr->members) {
        result.emplace_back(e.second);
    }
    return result;
}
 
void
ClientClusterServiceImpl::start()
{
    for (auto& listener :
         client_.get_client_config().get_membership_listeners()) {
        add_membership_listener(membership_listener(listener));
    }
}
 
void
ClientClusterServiceImpl::fire_initial_membership_event(
  const initial_membership_event& event)
{
    std::lock_guard<std::mutex> g(listeners_lock_);
 
    for (auto& item : listeners_) {
        membership_listener& listener = item.second;
        if (listener.init_) {
            listener.init_(event);
        }
    }
}
 
void
ClientClusterServiceImpl::shutdown()
{
    initial_list_fetched_latch_.try_count_down();
}
 
boost::uuids::uuid
ClientClusterServiceImpl::add_membership_listener(
  membership_listener&& listener)
{
    std::lock_guard<std::mutex> cluster_view_g(cluster_view_lock_);
 
    auto id = add_membership_listener_without_init(std::move(listener));
 
    std::lock_guard<std::mutex> listeners_g(listeners_lock_);
    auto added_listener = listeners_[id];
 
    if (added_listener.init_) {
        auto& cluster = client_.get_cluster();
        auto members_ptr = member_list_snapshot_.load();
        if (!members_ptr->members.empty()) {
            std::unordered_set<member> members;
            for (const auto& e : members_ptr->members) {
                members.insert(e.second);
            }
            added_listener.init_(initial_membership_event(cluster, members));
        }
    }
 
    return id;
}
 
bool
ClientClusterServiceImpl::remove_membership_listener(
  boost::uuids::uuid registration_id)
{
    std::lock_guard<std::mutex> g(listeners_lock_);
    return listeners_.erase(registration_id) == 1;
}
 
std::vector<member>
ClientClusterServiceImpl::get_members(const member_selector& selector) const
{
    std::vector<member> result;
    for (auto&& member : get_member_list()) {
        if (selector.select(member)) {
            result.emplace_back(std::move(member));
        }
    }
 
    return result;
}
 
local_endpoint
ClientClusterServiceImpl::get_local_client() const
{
    connection::ClientConnectionManagerImpl& cm =
      client_.get_connection_manager();
    auto connection = cm.get_random_connection();
    auto inetSocketAddress =
      connection ? connection->get_local_socket_address() : boost::none;
    auto uuid = cm.get_client_uuid();
    return local_endpoint(
      uuid, std::move(inetSocketAddress), client_.get_name(), labels_);
}
 
void
ClientClusterServiceImpl::clear_member_list_version()
{
    std::lock_guard<std::mutex> g(cluster_view_lock_);
    auto& lg = client_.get_logger();
    HZ_LOG(lg, finest, "Resetting the member list version ");
    auto cluster_view_snapshot = member_list_snapshot_.load();
    // This check is necessary so that `clear_member_list_version` when handling
    // auth response will not intervene with client failover logic
    if (cluster_view_snapshot != EMPTY_SNAPSHOT) {
        member_list_snapshot_.store(boost::shared_ptr<member_list_snapshot>(
          new member_list_snapshot{ 0, cluster_view_snapshot->members }));
    }
}
 
std::vector<membership_event>
ClientClusterServiceImpl::clear_member_list_and_return_events()
{
    std::lock_guard<std::mutex> g(cluster_view_lock_);
 
    auto& lg = client_.get_logger();
    HZ_LOG(lg, finest, "Resetting the member list");
 
    auto previous_list = member_list_snapshot_.load()->members;
 
    member_list_snapshot_.store(
      boost::shared_ptr<member_list_snapshot>(new member_list_snapshot{ 0 }));
 
    return detect_membership_events(
      previous_list,
      std::unordered_map<boost::uuids::uuid,
                         member,
                         boost::hash<boost::uuids::uuid>>());
}
 
void
ClientClusterServiceImpl::clear_member_list()
{
    auto events = clear_member_list_and_return_events();
    fire_events(std::move(events));
}
 
void
ClientClusterServiceImpl::handle_event(int32_t version,
                                       const std::vector<member>& member_infos)
{
    auto& lg = client_.get_logger();
    HZ_LOG(
      lg,
      finest,
      boost::str(
        boost::format("Handling new snapshot with membership version: %1%, "
                      "membersString %2%") %
        version % members_string(create_snapshot(version, member_infos))));
    auto cluster_view_snapshot = member_list_snapshot_.load();
    if (cluster_view_snapshot == EMPTY_SNAPSHOT) {
        std::lock_guard<std::mutex> g(cluster_view_lock_);
        cluster_view_snapshot = member_list_snapshot_.load();
        if (cluster_view_snapshot == EMPTY_SNAPSHOT) {
            // this means this is the first time client connected to cluster
            apply_initial_state(version, member_infos);
            initial_list_fetched_latch_.count_down();
            return;
        }
    }
 
    std::vector<membership_event> events;
    if (version >= cluster_view_snapshot->version) {
        std::lock_guard<std::mutex> g(cluster_view_lock_);
        cluster_view_snapshot = member_list_snapshot_.load();
        if (version >= cluster_view_snapshot->version) {
            auto prev_members = cluster_view_snapshot->members;
            auto snapshot = boost::make_shared<member_list_snapshot>(
              create_snapshot(version, member_infos));
            member_list_snapshot_.store(snapshot);
            events = detect_membership_events(prev_members, snapshot->members);
        }
    }
 
    fire_events(std::move(events));
}
 
ClientClusterServiceImpl::member_list_snapshot
ClientClusterServiceImpl::create_snapshot(int32_t version,
                                          const std::vector<member>& members)
{
    member_list_snapshot result;
    result.version = version;
    for (auto& m : members) {
        auto const& address_map = m.address_map();
        if (address_map.empty()) {
            result.members.insert({ m.get_uuid(), m });
        } else {
            auto found = address_map.find(CLIENT);
            address member_address;
            if (found != address_map.end()) {
                member_address = found->second;
            } else {
                found = address_map.find(MEMBER);
                assert(found != address_map.end());
                member_address = found->second;
            }
            member new_member(member_address,
                              m.get_uuid(),
                              m.is_lite_member(),
                              m.get_attributes(),
                              m.address_map(),
                              m.get_version());
            result.members.emplace(new_member.get_uuid(),
                                   std::move(new_member));
        }
    }
 
    return result;
}
 
std::string
ClientClusterServiceImpl::members_string(
  const ClientClusterServiceImpl::member_list_snapshot& snapshot)
{
    std::stringstream out;
    auto const& members = snapshot.members;
    out << std::endl << std::endl << "Members [" << members.size() << "]  {";
    for (auto const& e : members) {
        out << std::endl << "\t" << e.second;
    }
    out << std::endl << "}" << std::endl;
    return out.str();
}
 
void
ClientClusterServiceImpl::apply_initial_state(
  int32_t version,
  const std::vector<member>& member_infos)
{
    auto snapshot = boost::make_shared<member_list_snapshot>(
      create_snapshot(version, member_infos));
    member_list_snapshot_.store(snapshot);
    HZ_LOG(client_.get_logger(), info, members_string(*snapshot));
    std::unordered_set<member> members;
    for (auto const& e : snapshot->members) {
        members.insert(e.second);
    }
    std::lock_guard<std::mutex> g(listeners_lock_);
    for (auto& item : listeners_) {
        membership_listener& listener = item.second;
        if (listener.init_) {
            listener.init_(
              initial_membership_event(client_.get_cluster(), members));
        }
    }
}
 
std::vector<membership_event>
ClientClusterServiceImpl::detect_membership_events(
  std::unordered_map<boost::uuids::uuid,
                     member,
                     boost::hash<boost::uuids::uuid>> previous_members,
  const std::unordered_map<boost::uuids::uuid,
                           member,
                           boost::hash<boost::uuids::uuid>>& current_members)
{
    std::vector<member> new_members;
 
    for (auto const& e : current_members) {
        if (!previous_members.erase(e.first)) {
            new_members.emplace_back(e.second);
        }
    }
 
    std::vector<membership_event> events;
 
    // removal events should be added before added events
    for (auto const& e : previous_members) {
        events.emplace_back(
          client_.get_cluster(),
          e.second,
          membership_event::membership_event_type::MEMBER_LEFT,
          current_members);
        auto connection =
          client_.get_connection_manager().get_connection(e.second.get_uuid());
        if (connection) {
            connection->close(
              "",
              std::make_exception_ptr(exception::target_disconnected(
                "ClientClusterServiceImpl::detect_membership_events",
                (boost::format(
                   "The client has closed the connection to this member, after "
                   "receiving a member left event from the cluster. %1%") %
                 *connection)
                  .str())));
        }
    }
    for (auto const& member : new_members) {
        events.emplace_back(
          client_.get_cluster(),
          member,
          membership_event::membership_event_type::MEMBER_JOINED,
          current_members);
    }
 
    if (!events.empty()) {
        auto snapshot = member_list_snapshot_.load();
        if (!snapshot->members.empty()) {
            HZ_LOG(client_.get_logger(), info, members_string(*snapshot));
        }
    }
    return events;
}
 
void
ClientClusterServiceImpl::fire_events(std::vector<membership_event> events)
{
    std::lock_guard<std::mutex> g(listeners_lock_);
 
    for (auto const& event : events) {
        for (auto& item : listeners_) {
            membership_listener& listener = item.second;
            if (event.get_event_type() ==
                membership_event::membership_event_type::MEMBER_JOINED) {
                listener.joined_(event);
            } else {
                listener.left_(event);
            }
        }
    }
}
 
void
ClientClusterServiceImpl::wait_initial_member_list_fetched() const
{
    // safe to const cast here since latch operations are already thread safe
    // ops.
    if ((const_cast<boost::latch&>(initial_list_fetched_latch_))
          .wait_for(INITIAL_MEMBERS_TIMEOUT) == boost::cv_status::timeout) {
        BOOST_THROW_EXCEPTION(exception::illegal_state(
          "ClientClusterServiceImpl::wait_initial_member_list_fetched",
          "Could not get initial member list from cluster!"));
    }
}
 
bool
ClientInvocationServiceImpl::invoke_on_connection(
  const std::shared_ptr<ClientInvocation>& invocation,
  const std::shared_ptr<connection::Connection>& connection)
{
    return send(invocation, connection);
}
 
bool
ClientInvocationServiceImpl::invoke_on_partition_owner(
  const std::shared_ptr<ClientInvocation>& invocation,
  int partition_id)
{
    auto partition_owner =
      client_.get_partition_service().get_partition_owner(partition_id);
    if (partition_owner.is_nil()) {
        HZ_LOG(logger_,
               finest,
               boost::str(
                 boost::format(
                   "Partition owner is not assigned yet for partition %1%") %
                 partition_id));
        return false;
    }
    return invoke_on_target(invocation, partition_owner);
}
 
bool
ClientInvocationServiceImpl::invoke_on_target(
  const std::shared_ptr<ClientInvocation>& invocation,
  boost::uuids::uuid uuid)
{
    assert(!uuid.is_nil());
    auto connection = client_.get_connection_manager().get_connection(uuid);
    if (!connection) {
        HZ_LOG(
          logger_,
          finest,
          boost::str(boost::format("Client is not connected to target : %1%") %
                     uuid));
        return false;
    }
    return send(invocation, connection);
}
 
bool
ClientInvocationServiceImpl::is_smart_routing() const
{
    return smart_routing_;
}
 
const std::chrono::milliseconds&
ClientInvocationServiceImpl::get_backup_timeout() const
{
    return backup_timeout_;
}
 
bool
ClientInvocationServiceImpl::fail_on_indeterminate_state() const
{
    return fail_on_indeterminate_operation_state_;
}
 
ClientExecutionServiceImpl::ClientExecutionServiceImpl(
  const std::string& name,
  const client_properties& properties,
  int32_t pool_size,
  spi::lifecycle_service& service)
  : lifecycle_service_(service)
  , client_properties_(properties)
{}
 
void
ClientExecutionServiceImpl::start()
{
    int internalPoolSize = client_properties_.get_integer(
      client_properties_.get_internal_executor_pool_size());
    if (internalPoolSize <= 0) {
        internalPoolSize = util::IOUtil::to_value<int>(
          client_properties::INTERNAL_EXECUTOR_POOL_SIZE_DEFAULT);
    }
 
    internal_executor_.reset(
      new hazelcast::util::hz_thread_pool(internalPoolSize));
 
    user_executor_.reset(new hazelcast::util::hz_thread_pool());
}
 
void
ClientExecutionServiceImpl::shutdown()
{
    shutdown_thread_pool(internal_executor_.get());
    shutdown_thread_pool(user_executor_.get());
}
 
util::hz_thread_pool&
ClientExecutionServiceImpl::get_user_executor()
{
    return *user_executor_;
}
 
void
ClientExecutionServiceImpl::shutdown_thread_pool(
  hazelcast::util::hz_thread_pool* pool)
{
    if (!pool) {
        return;
    }
    pool->close();
}
 
constexpr int ClientInvocation::MAX_FAST_INVOCATION_COUNT;
constexpr int ClientInvocation::UNASSIGNED_PARTITION;
 
ClientInvocation::ClientInvocation(
  spi::ClientContext& client_context,
  std::shared_ptr<protocol::ClientMessage>&& message,
  const std::string& name,
  int partition,
  const std::shared_ptr<connection::Connection>& conn,
  boost::uuids::uuid uuid)
  : logger_(client_context.get_logger())
  , lifecycle_service_(client_context.get_lifecycle_service())
  , invocation_service_(client_context.get_invocation_service())
  , execution_service_(
      client_context.get_client_execution_service().shared_from_this())
  , call_id_sequence_(client_context.get_call_id_sequence())
  , uuid_(uuid)
  , partition_id_(partition)
  , start_time_(std::chrono::steady_clock::now())
  , retry_pause_(invocation_service_.get_invocation_retry_pause())
  , object_name_(name)
  , connection_(conn)
  , bound_to_single_connection_(conn != nullptr)
  , invoke_count_(0)
  , urgent_(false)
  , smart_routing_(invocation_service_.is_smart_routing())
{
    message->set_partition_id(partition_id_);
    client_message_ =
      boost::make_shared<std::shared_ptr<protocol::ClientMessage>>(message);
    set_send_connection(nullptr);
}
 
ClientInvocation::~ClientInvocation() = default;
 
boost::future<protocol::ClientMessage>
ClientInvocation::invoke()
{
    assert(client_message_.load());
    // for back pressure
    call_id_sequence_->next();
    invoke_on_selection();
    if (!lifecycle_service_.is_running()) {
        return invocation_promise_.get_future().then(
          [](boost::future<protocol::ClientMessage> f) { return f.get(); });
    }
    auto id_seq = call_id_sequence_;
    return invocation_promise_.get_future().then(
      execution_service_->get_user_executor(),
      [=](boost::future<protocol::ClientMessage> f) {
          id_seq->complete();
          return f.get();
      });
}
 
boost::future<protocol::ClientMessage>
ClientInvocation::invoke_urgent()
{
    assert(client_message_.load());
    urgent_ = true;
    // for back pressure
    call_id_sequence_->force_next();
    invoke_on_selection();
    if (!lifecycle_service_.is_running()) {
        return invocation_promise_.get_future().then(
          [](boost::future<protocol::ClientMessage> f) { return f.get(); });
    }
    auto id_seq = call_id_sequence_;
    return invocation_promise_.get_future().then(
      execution_service_->get_user_executor(),
      [=](boost::future<protocol::ClientMessage> f) {
          id_seq->complete();
          return f.get();
      });
}
 
void
ClientInvocation::invoke_on_selection()
{
    try {
        invoke_count_++;
        if (!urgent_) {
            invocation_service_.check_invocation_allowed();
        }
 
        if (is_bind_to_single_connection()) {
            bool invoked = false;
            auto conn = connection_.lock();
            if (conn) {
                invoked = invocation_service_.invoke_on_connection(
                  shared_from_this(), conn);
            }
            if (!invoked) {
                std::string message;
                if (conn) {
                    message =
                      (boost::format("Could not invoke on connection %1%") %
                       *conn)
                        .str();
                } else {
                    message = "Could not invoke. Bound to a connection that is "
                              "deleted already.";
                }
                notify_exception(std::make_exception_ptr(exception::io(
                  "ClientInvocation::invoke_on_selection", message)));
            }
            return;
        }
 
        bool invoked = false;
        if (smart_routing_) {
            if (partition_id_ != -1) {
                invoked = invocation_service_.invoke_on_partition_owner(
                  shared_from_this(), partition_id_);
            } else if (!uuid_.is_nil()) {
                invoked = invocation_service_.invoke_on_target(
                  shared_from_this(), uuid_);
            } else {
                invoked = invocation_service_.invoke(shared_from_this());
            }
            if (!invoked) {
                invoked = invocation_service_.invoke(shared_from_this());
            }
        } else {
            invoked = invocation_service_.invoke(shared_from_this());
        }
        if (!invoked) {
            notify_exception(std::make_exception_ptr(
              exception::io("No connection found to invoke")));
        }
    } catch (exception::iexception&) {
        notify_exception(std::current_exception());
    } catch (std::exception&) {
        assert(false);
    }
}
 
bool
ClientInvocation::is_bind_to_single_connection() const
{
    return bound_to_single_connection_;
}
 
void
ClientInvocation::run()
{
    retry();
}
 
void
ClientInvocation::retry()
{
    // retry modifies the client message and should not reuse the client
    // message. It could be the case that it is in write queue of the
    // connection.
    client_message_ =
      boost::make_shared<std::shared_ptr<protocol::ClientMessage>>(
        copy_message());
 
    try {
        invoke_on_selection();
    } catch (exception::iexception& e) {
        set_exception(e, boost::current_exception());
    } catch (std::exception&) {
        assert(false);
    }
}
 
void
ClientInvocation::set_exception(const std::exception& e,
                                boost::exception_ptr exception_ptr)
{
    invoked_or_exception_set_.store(true);
    try {
        auto send_conn = send_connection_.load();
        if (send_conn) {
            auto connection = send_conn->lock();
            if (connection) {
                auto call_id =
                  client_message_.load()->get()->get_correlation_id();
                boost::asio::post(
                  connection->get_socket().get_executor(),
                  [=]() { connection->deregister_invocation(call_id); });
            }
        }
        invocation_promise_.set_exception(std::move(exception_ptr));
    } catch (boost::promise_already_satisfied& se) {
        if (!event_handler_) {
            HZ_LOG(logger_,
                   finest,
                   boost::str(boost::format(
                                "Failed to set the exception for invocation. "
                                "%1%, %2% Exception to be set: %3%") %
                              se.what() % *this % e.what()));
        }
    }
}
 
void
ClientInvocation::notify_exception(std::exception_ptr exception)
{
    erase_invocation();
    try {
        std::rethrow_exception(exception);
    } catch (exception::iexception& iex) {
        log_exception(iex);
 
        if (!lifecycle_service_.is_running()) {
            try {
                std::throw_with_nested(boost::enable_current_exception(
                  exception::hazelcast_client_not_active(
                    iex.get_source(), "Client is shutting down")));
            } catch (exception::iexception& e) {
                set_exception(e, boost::current_exception());
            }
            return;
        }
 
        if (!should_retry(iex)) {
            set_exception(iex, boost::current_exception());
            return;
        }
 
        auto timePassed = std::chrono::steady_clock::now() - start_time_;
        if (timePassed > invocation_service_.get_invocation_timeout()) {
            HZ_LOG(
              logger_,
              finest,
              boost::str(boost::format("Exception will not be retried because "
                                       "invocation timed out. %1%") %
                         iex.what()));
 
            auto now = std::chrono::steady_clock::now();
 
            auto timeoutException =
              (exception::exception_builder<exception::operation_timeout>(
                 "ClientInvocation::newoperation_timeout_exception")
               << *this
               << " timed out because exception occurred after client "
                  "invocation timeout "
               << std::chrono::duration_cast<std::chrono::milliseconds>(
                    invocation_service_.get_invocation_timeout())
                    .count()
               << "msecs. Last exception:" << iex << " Current time :"
               << util::StringUtil::time_to_string(now) << ". "
               << "Start time: "
               << util::StringUtil::time_to_string(start_time_)
               << ". Total elapsed time: "
               << std::chrono::duration_cast<std::chrono::milliseconds>(
                    now - start_time_)
                    .count()
               << " ms. ")
                .build();
            try {
                BOOST_THROW_EXCEPTION(timeoutException);
            } catch (...) {
                set_exception(timeoutException, boost::current_exception());
            }
 
            return;
        }
 
        try {
            execute();
        } catch (std::exception& e) {
            set_exception(e, boost::current_exception());
        }
    } catch (...) {
        assert(false);
    }
}
 
void
ClientInvocation::erase_invocation() const
{
    if (!this->event_handler_) {
        auto sent_connection = get_send_connection();
        if (sent_connection) {
            auto this_invocation = shared_from_this();
            boost::asio::post(sent_connection->get_socket().get_executor(),
                              [=]() {
                                  sent_connection->invocations.erase(
                                    this_invocation->get_client_message()
                                      ->get_correlation_id());
                              });
        }
    }
}
 
bool
ClientInvocation::should_retry(exception::iexception& exception)
{
    auto errorCode = exception.get_error_code();
    if (is_bind_to_single_connection() &&
        (errorCode == protocol::IO ||
         errorCode == protocol::TARGET_DISCONNECTED)) {
        return false;
    }
 
    if (!uuid_.is_nil() && errorCode == protocol::TARGET_NOT_MEMBER) {
        // when invocation send to a specific member
        // if target is no longer a member, we should not retry
        // note that this exception could come from the server
        return false;
    }
 
    if (errorCode == protocol::IO ||
        errorCode == protocol::HAZELCAST_INSTANCE_NOT_ACTIVE ||
        exception.is_retryable()) {
        return true;
    }
    if (errorCode == protocol::TARGET_DISCONNECTED) {
        return client_message_.load()->get()->is_retryable() ||
               invocation_service_.is_redo_operation();
    }
    return false;
}
 
std::ostream&
operator<<(std::ostream& os, const ClientInvocation& invocation)
{
    std::ostringstream target;
    if (invocation.is_bind_to_single_connection()) {
        auto conn = invocation.connection_.lock();
        if (conn) {
            target << "connection " << *conn;
        }
    } else if (invocation.partition_id_ != -1) {
        target << "partition " << invocation.partition_id_;
    } else if (!invocation.uuid_.is_nil()) {
        target << "uuid " << boost::to_string(invocation.uuid_);
    } else {
        target << "random";
    }
    os << "ClientInvocation{"
       << "requestMessage = " << *invocation.client_message_.load()->get()
       << ", objectName = " << invocation.object_name_
       << ", target = " << target.str() << ", sendConnection = ";
    auto sendConnection = invocation.get_send_connection();
    if (sendConnection) {
        os << *sendConnection;
    } else {
        os << "nullptr";
    }
    os << ", backup_acks_expected_ = "
       << static_cast<int>(invocation.backup_acks_expected_)
       << ", backup_acks_received = " << invocation.backup_acks_received_;
 
    if (invocation.pending_response_) {
        os << ", pending_response: " << *invocation.pending_response_;
    }
 
    os << '}';
 
    return os;
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(
  spi::ClientContext& client_context,
  std::shared_ptr<protocol::ClientMessage>&& client_message,
  const std::string& object_name,
  int partition_id)
{
    return std::shared_ptr<ClientInvocation>(new ClientInvocation(
      client_context, std::move(client_message), object_name, partition_id));
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(
  spi::ClientContext& client_context,
  std::shared_ptr<protocol::ClientMessage>&& client_message,
  const std::string& object_name,
  const std::shared_ptr<connection::Connection>& connection)
{
    return std::shared_ptr<ClientInvocation>(
      new ClientInvocation(client_context,
                           std::move(client_message),
                           object_name,
                           UNASSIGNED_PARTITION,
                           connection));
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(
  spi::ClientContext& client_context,
  std::shared_ptr<protocol::ClientMessage>&& client_message,
  const std::string& object_name,
  boost::uuids::uuid uuid)
{
    return std::shared_ptr<ClientInvocation>(
      new ClientInvocation(client_context,
                           std::move(client_message),
                           object_name,
                           UNASSIGNED_PARTITION,
                           nullptr,
                           uuid));
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(spi::ClientContext& client_context,
                         protocol::ClientMessage& client_message,
                         const std::string& object_name,
                         int partition_id)
{
    return create(
      client_context,
      std::make_shared<protocol::ClientMessage>(std::move(client_message)),
      object_name,
      partition_id);
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(
  spi::ClientContext& client_context,
  protocol::ClientMessage& client_message,
  const std::string& object_name,
  const std::shared_ptr<connection::Connection>& connection)
{
    return create(
      client_context,
      std::make_shared<protocol::ClientMessage>(std::move(client_message)),
      object_name,
      connection);
}
 
std::shared_ptr<ClientInvocation>
ClientInvocation::create(spi::ClientContext& client_context,
                         protocol::ClientMessage& client_message,
                         const std::string& object_name,
                         boost::uuids::uuid uuid)
{
    return create(
      client_context,
      std::make_shared<protocol::ClientMessage>(std::move(client_message)),
      object_name,
      uuid);
}
 
std::shared_ptr<connection::Connection>
ClientInvocation::get_send_connection() const
{
    return send_connection_.load()->lock();
}
 
void
ClientInvocation::wait_invoked() const
{
    // it could be either invoked or cancelled before invoked
    while (!invoked_or_exception_set_) {
        std::this_thread::sleep_for(retry_pause_);
    }
}
 
void
ClientInvocation::set_send_connection(
  const std::shared_ptr<connection::Connection>& conn)
{
    send_connection_.store(
      boost::make_shared<std::weak_ptr<connection::Connection>>(conn));
    invoked_or_exception_set_.store(true);
}
 
void
ClientInvocation::notify(const std::shared_ptr<protocol::ClientMessage>& msg)
{
    if (!msg) {
        BOOST_THROW_EXCEPTION(
          exception::illegal_argument("response can't be null"));
    }
 
    int8_t expected_backups = msg->get_number_of_backups();
 
    // if a regular response comes and there are backups, we need to wait for
    // the backups when the backups complete, the response will be send by the
    // last backup or backup-timeout-handle mechanism kicks on
    if (expected_backups > backup_acks_received_) {
        // so the invocation has backups and since not all backups have
        // completed, we need to wait (it could be that backups arrive earlier
        // than the response)
 
        pending_response_received_time_ = std::chrono::steady_clock::now();
 
        backup_acks_expected_ = expected_backups;
 
        // it is very important that the response is set after the
        // backupsAcksExpected is set, else the system can assume the invocation
        // is complete because there is a response and no backups need to
        // respond
        pending_response_ = msg;
 
        // we are done since not all backups have completed. Therefore we should
        // not notify the future
        return;
    }
 
    // we are going to notify the future that a response is available; this can
    // happen when:
    // - we had a regular operation (so no backups we need to wait for) that
    // completed
    // - we had a backup-aware operation that has completed, but also all its
    // backups have completed
    complete(msg);
}
 
void
ClientInvocation::complete(const std::shared_ptr<protocol::ClientMessage>& msg)
{
    try {
        // TODO: move msg content here?
        this->invocation_promise_.set_value(*msg);
    } catch (std::exception& e) {
        HZ_LOG(logger_,
               warning,
               boost::str(boost::format(
                            "Failed to set the response for invocation. "
                            "Dropping the response. %1%, %2% Response: %3%") %
                          e.what() % *this % *msg));
    }
    this->erase_invocation();
}
 
std::shared_ptr<protocol::ClientMessage>
ClientInvocation::get_client_message() const
{
    return *client_message_.load();
}
 
const std::shared_ptr<EventHandler<protocol::ClientMessage>>&
ClientInvocation::get_event_handler() const
{
    return event_handler_;
}
 
void
ClientInvocation::set_event_handler(
  const std::shared_ptr<EventHandler<protocol::ClientMessage>>& handler)
{
    ClientInvocation::event_handler_ = handler;
}
 
void
ClientInvocation::execute()
{
    auto this_invocation = shared_from_this();
    auto command = [=]() { this_invocation->run(); };
 
    // first we force a new invocation slot because we are going to return our
    // old invocation slot immediately after It is important that we first
    // 'force' taking a new slot; otherwise it could be that a sneaky invocation
    // gets through that takes our slot!
    int64_t callId = call_id_sequence_->force_next();
    client_message_.load()->get()->set_correlation_id(callId);
 
    // we release the old slot
    call_id_sequence_->complete();
 
    if (invoke_count_ < MAX_FAST_INVOCATION_COUNT) {
        // fast retry for the first few invocations
        execution_service_->execute(command);
    } else {
        // progressive retry delay
        int64_t delayMillis = util::min<int64_t>(
          static_cast<int64_t>(1)
            << (invoke_count_ - MAX_FAST_INVOCATION_COUNT),
          std::chrono::duration_cast<std::chrono::milliseconds>(retry_pause_)
            .count());
        retry_timer_ = execution_service_->schedule(
          command, std::chrono::milliseconds(delayMillis));
    }
}
 
const std::string
ClientInvocation::get_name() const
{
    return "ClientInvocation";
}
 
std::shared_ptr<protocol::ClientMessage>
ClientInvocation::copy_message()
{
    return std::make_shared<protocol::ClientMessage>(**client_message_.load());
}
 
boost::promise<protocol::ClientMessage>&
ClientInvocation::get_promise()
{
    return invocation_promise_;
}
 
void
ClientInvocation::log_exception(exception::iexception& e)
{
    HZ_LOG(logger_,
           finest,
           boost::str(boost::format(
                        "Invocation got an exception %1%, invoke count : %2%, "
                        "exception : %3%") %
                      *this % invoke_count_.load() % e));
}
 
void
ClientInvocation::notify_backup()
{
    ++backup_acks_received_;
 
    if (!pending_response_) {
        // no pendingResponse has been set, so we are done since the invocation
        // on the primary needs to complete first
        return;
    }
 
    // if a pendingResponse is set, then the backupsAcksExpected has been set
    // (so we can now safely read backupsAcksExpected)
    if (backup_acks_expected_ != backup_acks_received_) {
        // we managed to complete a backup, but we were not the one completing
        // the last backup, so we are done
        return;
    }
 
    // we are the lucky one since we just managed to complete the last backup
    // for this invocation and since the pendingResponse is set, we can set it
    // on the future
    complete_with_pending_response();
}
 
void
ClientInvocation::detect_and_handle_backup_timeout(
  const std::chrono::milliseconds& backup_timeout)
{
    // if the backups have completed, we are done; this also filters out all non
    // backup-aware operations since the backupsAcksExpected will always be
    // equal to the backupsAcksReceived
    if (backup_acks_expected_ == backup_acks_received_) {
        return;
    }
 
    // if no response has yet been received, we we are done; we are only going
    // to re-invoke an operation if the response of the primary has been
    // received, but the backups have not replied
    if (!pending_response_) {
        return;
    }
 
    // if this has not yet expired (so has not been in the system for a too long
    // period) we ignore it
    if (pending_response_received_time_ + backup_timeout >=
        std::chrono::steady_clock::now()) {
        return;
    }
 
    if (invocation_service_.fail_on_indeterminate_state()) {
        auto exception = boost::enable_current_exception(
          (exception::exception_builder<
             exception::indeterminate_operation_state>(
             "ClientInvocation::detect_and_handle_backup_timeout")
           << *this << " failed because backup acks missed.")
            .build());
        notify_exception(std::make_exception_ptr(exception));
        return;
    }
 
    // the backups have not yet completed, but we are going to release the
    // future anyway if a pendingResponse has been set
    complete_with_pending_response();
}
 
void
ClientInvocation::complete_with_pending_response()
{
    complete(pending_response_);
}
 
ClientContext&
impl::ClientTransactionManagerServiceImpl::get_client() const
{
    return client_;
}
 
ClientTransactionManagerServiceImpl::ClientTransactionManagerServiceImpl(
  ClientContext& client)
  : client_(client)
{}
 
std::shared_ptr<connection::Connection>
ClientTransactionManagerServiceImpl::connect()
{
    auto& invocationService = client_.get_invocation_service();
    auto startTime = std::chrono::steady_clock::now();
    auto invocationTimeout = invocationService.get_invocation_timeout();
    client_config& clientConfig = client_.get_client_config();
    bool smartRouting = clientConfig.get_network_config().is_smart_routing();
 
    while (client_.get_lifecycle_service().is_running()) {
        try {
            auto connection =
              client_.get_connection_manager().get_random_connection();
            if (!connection) {
                throw_exception(smartRouting);
            }
            return connection;
        } catch (exception::hazelcast_client_offline&) {
            throw;
        } catch (exception::iexception&) {
            if (std::chrono::steady_clock::now() - startTime >
                invocationTimeout) {
                std::rethrow_exception(new_operation_timeout_exception(
                  std::current_exception(), invocationTimeout, startTime));
            }
        }
        std::this_thread::sleep_for(
          invocationService.get_invocation_retry_pause());
    }
    BOOST_THROW_EXCEPTION(exception::hazelcast_client_not_active(
      "ClientTransactionManagerServiceImpl::connect", "Client is shutdown"));
}
 
std::exception_ptr
ClientTransactionManagerServiceImpl::new_operation_timeout_exception(
  std::exception_ptr cause,
  std::chrono::milliseconds invocation_timeout,
  std::chrono::steady_clock::time_point start_time)
{
    std::ostringstream sb;
    auto now = std::chrono::steady_clock::now();
    sb << "Creating transaction context timed out because exception occurred "
          "after client invocation timeout "
       << std::chrono::duration_cast<std::chrono::milliseconds>(
            invocation_timeout)
            .count()
       << " ms. "
       << "Current time: "
       << util::StringUtil::time_to_string(std::chrono::steady_clock::now())
       << ". "
       << "Start time: " << util::StringUtil::time_to_string(start_time)
       << ". Total elapsed time: "
       << std::chrono::duration_cast<std::chrono::milliseconds>(now -
                                                                start_time)
            .count()
       << " ms. ";
    try {
        std::rethrow_exception(cause);
    } catch (...) {
        try {
            std::throw_with_nested(boost::enable_current_exception(
              exception::operation_timeout("ClientTransactionManagerServiceImpl"
                                           "::newoperation_timeout_exception",
                                           sb.str())));
        } catch (...) {
            return std::current_exception();
        }
    }
    return nullptr;
}
 
void
ClientTransactionManagerServiceImpl::throw_exception(bool smart_routing)
{
    auto& client_config = client_.get_client_config();
    auto& connection_strategy_Config =
      client_config.get_connection_strategy_config();
    auto reconnect_mode = connection_strategy_Config.get_reconnect_mode();
    if (reconnect_mode ==
        config::client_connection_strategy_config::reconnect_mode::ASYNC) {
        BOOST_THROW_EXCEPTION(exception::hazelcast_client_offline(
          "ClientTransactionManagerServiceImpl::throw_exception", ""));
    }
    if (smart_routing) {
        auto members = client_.get_cluster().get_members();
        std::ostringstream msg;
        if (members.empty()) {
            msg << "No address was return by the LoadBalancer since there are "
                   "no members in the cluster";
        } else {
            msg << "No address was return by the LoadBalancer. "
                   "But the cluster contains the following members:{\n";
            for (auto const& m : members) {
                msg << '\t' << m << '\n';
            }
            msg << "}";
        }
        BOOST_THROW_EXCEPTION(exception::illegal_state(
          "ClientTransactionManagerServiceImpl::throw_exception", msg.str()));
    }
    BOOST_THROW_EXCEPTION(exception::illegal_state(
      "ClientTransactionManagerServiceImpl::throw_exception",
      "No active connection is found"));
}
 
ClientPartitionServiceImpl::ClientPartitionServiceImpl(ClientContext& client)
  : client_(client)
  , logger_(client.get_logger())
  , partition_count_(0)
  , partition_table_(
      boost::shared_ptr<partition_table>(new partition_table{ 0, -1 }))
{}
 
void
ClientPartitionServiceImpl::handle_event(
  int32_t connection_id,
  int32_t version,
  const std::vector<std::pair<boost::uuids::uuid, std::vector<int>>>&
    partitions)
{
    HZ_LOG(logger_,
           finest,
           boost::str(
             boost::format(
               "Handling new partition table with partitionStateVersion: %1%") %
             version));
 
    while (true) {
        auto current = partition_table_.load();
        if (!should_be_applied(connection_id, version, partitions, *current)) {
            return;
        }
        if (partition_table_.compare_exchange_strong(
              current,
              boost::shared_ptr<partition_table>(new partition_table{
                connection_id, version, convert_to_map(partitions) }))) {
            HZ_LOG(
              logger_,
              finest,
              boost::str(
                boost::format(
                  "Applied partition table with partitionStateVersion : %1%") %
                version));
            return;
        }
    }
}
 
boost::uuids::uuid
ClientPartitionServiceImpl::get_partition_owner(int32_t partition_id)
{
    auto table_ptr = partition_table_.load();
    auto it = table_ptr->partitions.find(partition_id);
    if (it != table_ptr->partitions.end()) {
        return it->second;
    }
    return boost::uuids::nil_uuid();
}
 
int32_t
ClientPartitionServiceImpl::get_partition_id(
  const serialization::pimpl::data& key)
{
    int32_t pc = get_partition_count();
    if (pc <= 0) {
        return 0;
    }
    int hash = key.get_partition_hash();
    return util::HashUtil::hash_to_index(hash, pc);
}
 
int32_t
ClientPartitionServiceImpl::get_partition_count()
{
    return partition_count_.load();
}
 
std::shared_ptr<client::impl::Partition>
ClientPartitionServiceImpl::get_partition(int partition_id)
{
    return std::shared_ptr<client::impl::Partition>(
      new PartitionImpl(partition_id, client_, *this));
}
 
bool
ClientPartitionServiceImpl::check_and_set_partition_count(
  int32_t new_partition_count)
{
    int32_t expected = 0;
    if (partition_count_.compare_exchange_strong(expected,
                                                 new_partition_count)) {
        return true;
    }
    return partition_count_.load() == new_partition_count;
}
 
bool
ClientPartitionServiceImpl::should_be_applied(
  int32_t connection_id,
  int32_t version,
  const std::vector<std::pair<boost::uuids::uuid, std::vector<int>>>&
    partitions,
  const partition_table& current)
{
    auto& lg = client_.get_logger();
    if (partitions.empty()) {
        if (logger_.enabled(logger::level::finest)) {
            log_failure(connection_id, version, current, "response is empty");
        }
        return false;
    }
    if (!current.connection_id || connection_id != current.connection_id) {
        HZ_LOG(
          lg, finest, ([&current, connection_id]() {
              auto frmt = boost::format(
                "Event coming from a new connection. Old connection id: %1%, "
                "new connection %2%");
 
              if (current.connection_id) {
                  frmt = frmt % current.connection_id;
              } else {
                  frmt = frmt % "none";
              }
 
              return boost::str(frmt % connection_id);
          })());
 
        return true;
    }
    if (version <= current.version) {
        if (lg.enabled(logger::level::finest)) {
            log_failure(
              connection_id, version, current, "response state version is old");
        }
        return false;
    }
    return true;
}
 
void
ClientPartitionServiceImpl::log_failure(
  int32_t connection_id,
  int32_t version,
  const ClientPartitionServiceImpl::partition_table& current,
  const std::string& cause)
{
    HZ_LOG(logger_, finest, [&]() {
        auto frmt = boost::format(
          " We will not apply the response, since %1% ."
          " Response is from connection with id %2%. "
          "Current connection id is %3%, response state version:%4%. "
          "Current state version: %5%");
        if (current.connection_id) {
            return boost::str(frmt % cause % connection_id %
                              current.connection_id % version %
                              current.version);
        } else {
            return boost::str(frmt % cause % connection_id % "nullptr" %
                              version % current.version);
        }
    }());
}
 
void
ClientPartitionServiceImpl::reset()
{
    partition_table_.store(nullptr);
}
 
std::unordered_map<int32_t, boost::uuids::uuid>
ClientPartitionServiceImpl::convert_to_map(
  const std::vector<std::pair<boost::uuids::uuid, std::vector<int>>>&
    partitions)
{
    std::unordered_map<int32_t, boost::uuids::uuid> new_partitions;
    for (auto const& e : partitions) {
        for (auto pid : e.second) {
            new_partitions.insert({ pid, e.first });
        }
    }
    return new_partitions;
}
 
int
ClientPartitionServiceImpl::PartitionImpl::get_partition_id() const
{
    return partition_id_;
}
 
boost::optional<member>
ClientPartitionServiceImpl::PartitionImpl::get_owner() const
{
    auto owner = partition_service_.get_partition_owner(partition_id_);
    if (!owner.is_nil()) {
        return client_.get_client_cluster_service().get_member(owner);
    }
    return boost::none;
}
 
ClientPartitionServiceImpl::PartitionImpl::PartitionImpl(
  int partition_id,
  ClientContext& client,
  ClientPartitionServiceImpl& partition_service)
  : partition_id_(partition_id)
  , client_(client)
  , partition_service_(partition_service)
{}
 
namespace sequence {
CallIdSequenceWithoutBackpressure::CallIdSequenceWithoutBackpressure()
  : head_(0)
{}
 
CallIdSequenceWithoutBackpressure::~CallIdSequenceWithoutBackpressure() =
  default;
 
int32_t
CallIdSequenceWithoutBackpressure::get_max_concurrent_invocations() const
{
    return INT32_MAX;
}
 
int64_t
CallIdSequenceWithoutBackpressure::next()
{
    return force_next();
}
 
int64_t
CallIdSequenceWithoutBackpressure::force_next()
{
    return ++head_;
}
 
void
CallIdSequenceWithoutBackpressure::complete()
{
    // no-op
}
 
int64_t
CallIdSequenceWithoutBackpressure::get_last_call_id()
{
    return head_;
}
 
// TODO: see if we can utilize std::hardware_destructive_interference_size
AbstractCallIdSequence::AbstractCallIdSequence(
  int32_t max_concurrent_invocations)
{
    std::ostringstream out;
    out << "maxConcurrentInvocations should be a positive number. "
           "maxConcurrentInvocations="
        << max_concurrent_invocations;
    this->max_concurrent_invocations_ = util::Preconditions::check_positive(
      max_concurrent_invocations, out.str());
 
    for (size_t i = 0; i < longs_.size(); ++i) {
        longs_[i] = 0;
    }
}
 
AbstractCallIdSequence::~AbstractCallIdSequence() = default;
 
int32_t
AbstractCallIdSequence::get_max_concurrent_invocations() const
{
    return max_concurrent_invocations_;
}
 
int64_t
AbstractCallIdSequence::next()
{
    if (!has_space()) {
        handle_no_space_left();
    }
    return force_next();
}
 
int64_t
AbstractCallIdSequence::force_next()
{
    return ++longs_[INDEX_HEAD];
}
 
void
AbstractCallIdSequence::complete()
{
    ++longs_[INDEX_TAIL];
    assert(longs_[INDEX_TAIL] <= longs_[INDEX_HEAD]);
}
 
int64_t
AbstractCallIdSequence::get_last_call_id()
{
    return longs_[INDEX_HEAD];
}
 
bool
AbstractCallIdSequence::has_space()
{
    return longs_[INDEX_HEAD] - longs_[INDEX_TAIL] <
           max_concurrent_invocations_;
}
 
int64_t
AbstractCallIdSequence::get_tail()
{
    return longs_[INDEX_TAIL];
}
 
const std::unique_ptr<util::concurrent::IdleStrategy>
  CallIdSequenceWithBackpressure::IDLER(
    new util::concurrent::BackoffIdleStrategy(
      0,
      0,
      std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::microseconds(1000))
        .count(),
      std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::microseconds(MAX_DELAY_MS * 1000))
        .count()));
 
CallIdSequenceWithBackpressure::CallIdSequenceWithBackpressure(
  int32_t max_concurrent_invocations,
  int64_t backoff_timeout_ms)
  : AbstractCallIdSequence(max_concurrent_invocations)
{
    std::ostringstream out;
    out << "backoffTimeoutMs should be a positive number. backoffTimeoutMs="
        << backoff_timeout_ms;
    util::Preconditions::check_positive(backoff_timeout_ms, out.str());
 
    backoff_timeout_nanos_ =
      std::chrono::duration_cast<std::chrono::nanoseconds>(
        std::chrono::milliseconds(backoff_timeout_ms))
        .count();
}
 
void
CallIdSequenceWithBackpressure::handle_no_space_left()
{
    auto start = std::chrono::steady_clock::now();
    for (int64_t idleCount = 0;; idleCount++) {
        int64_t elapsedNanos =
          std::chrono::duration_cast<std::chrono::nanoseconds>(
            std::chrono::steady_clock::now() - start)
            .count();
        if (elapsedNanos > backoff_timeout_nanos_) {
            throw(exception::exception_builder<exception::hazelcast_overload>(
                    "CallIdSequenceWithBackpressure::handleNoSpaceLeft")
                  << "Timed out trying to acquire another call ID."
                  << " maxConcurrentInvocations = "
                  << get_max_concurrent_invocations() << ", backoffTimeout = "
                  << std::chrono::microseconds(backoff_timeout_nanos_ / 1000)
                       .count()
                  << " msecs, elapsed:"
                  << std::chrono::microseconds(elapsedNanos / 1000).count()
                  << " msecs")
              .build();
        }
        IDLER->idle(idleCount);
        if (has_space()) {
            return;
        }
    }
}
 
FailFastCallIdSequence::FailFastCallIdSequence(
  int32_t max_concurrent_invocations)
  : AbstractCallIdSequence(max_concurrent_invocations)
{}
 
void
FailFastCallIdSequence::handle_no_space_left()
{
    throw(exception::exception_builder<exception::hazelcast_overload>(
            "FailFastCallIdSequence::handleNoSpaceLeft")
          << "Maximum invocation count is reached. maxConcurrentInvocations = "
          << get_max_concurrent_invocations())
      .build();
}
 
std::unique_ptr<CallIdSequence>
CallIdFactory::new_call_id_sequence(bool is_back_pressure_enabled,
                                    int32_t max_allowed_concurrent_invocations,
                                    int64_t backoff_timeout_ms)
{
    if (!is_back_pressure_enabled) {
        return std::unique_ptr<CallIdSequence>(
          new CallIdSequenceWithoutBackpressure());
    } else if (backoff_timeout_ms <= 0) {
        return std::unique_ptr<CallIdSequence>(
          new FailFastCallIdSequence(max_allowed_concurrent_invocations));
    } else {
        return std::unique_ptr<CallIdSequence>(
          new CallIdSequenceWithBackpressure(max_allowed_concurrent_invocations,
                                             backoff_timeout_ms));
    }
}
} // namespace sequence
 
namespace listener {
listener_service_impl::listener_service_impl(ClientContext& client_context,
                                             int32_t event_thread_count)
  : client_context_(client_context)
  , serialization_service_(client_context.get_serialization_service())
  , logger_(client_context.get_logger())
  , client_connection_manager_(client_context.get_connection_manager())
  , number_of_event_threads_(event_thread_count)
  , smart_(client_context.get_client_config()
             .get_network_config()
             .is_smart_routing())
{
    auto& invocationService = client_context.get_invocation_service();
    invocation_timeout_ = invocationService.get_invocation_timeout();
    invocation_retry_pause_ = invocationService.get_invocation_retry_pause();
}
 
bool
listener_service_impl::registers_local_only() const
{
    return smart_;
}
 
boost::future<boost::uuids::uuid>
listener_service_impl::register_listener(
  std::shared_ptr<ListenerMessageCodec> listener_message_codec,
  std::shared_ptr<client::impl::BaseEventHandler> handler)
{
    auto task = boost::packaged_task<boost::uuids::uuid()>([=]() {
        return register_listener_internal(listener_message_codec, handler);
    });
    auto f = task.get_future();
    boost::asio::post(registration_executor_->get_executor(), std::move(task));
    return f;
}
 
boost::future<bool>
listener_service_impl::deregister_listener(boost::uuids::uuid registration_id)
{
    util::Preconditions::check_not_nill(
      registration_id, "Nil userRegistrationId is not allowed!");
 
    boost::packaged_task<bool()> task(
      [=]() { return deregister_listener_internal(registration_id); });
    auto f = task.get_future();
    boost::asio::post(registration_executor_->get_executor(), std::move(task));
    return f;
}
 
void
listener_service_impl::connection_added(
  const std::shared_ptr<connection::Connection> connection)
{
    boost::asio::post(registration_executor_->get_executor(),
                      [=]() { connection_added_internal(connection); });
}
 
void
listener_service_impl::connection_removed(
  const std::shared_ptr<connection::Connection> connection)
{
    boost::asio::post(registration_executor_->get_executor(),
                      [=]() { connection_removed_internal(connection); });
}
 
void
listener_service_impl::remove_event_handler(
  int64_t call_id,
  const std::shared_ptr<connection::Connection>& connection)
{
    boost::asio::post(connection->get_socket().get_executor(),
                      std::packaged_task<void()>(
                        [=]() { connection->deregister_invocation(call_id); }));
}
 
void
listener_service_impl::handle_client_message(
  const std::shared_ptr<ClientInvocation> invocation,
  const std::shared_ptr<protocol::ClientMessage> response)
{
    try {
        auto partitionId = response->get_partition_id();
        if (partitionId == -1) {
            // execute on random thread on the thread pool
            boost::asio::post(event_executor_->get_executor(), [=]() {
                process_event_message(invocation, response);
            });
            return;
        }
 
        // process on certain thread which is same for the partition id
        boost::asio::post(
          event_strands_[partitionId % event_strands_.size()],
          [=]() { process_event_message(invocation, response); });
 
    } catch (const std::exception& e) {
        if (client_context_.get_lifecycle_service().is_running()) {
            HZ_LOG(
              logger_,
              warning,
              boost::str(boost::format("Delivery of event message to event "
                                       "handler failed. %1%, %2%, %3%") %
                         e.what() % *response % *invocation));
        }
    }
}
 
void
listener_service_impl::shutdown()
{
    ClientExecutionServiceImpl::shutdown_thread_pool(event_executor_.get());
    ClientExecutionServiceImpl::shutdown_thread_pool(
      registration_executor_.get());
}
 
void
listener_service_impl::start()
{
    event_executor_.reset(
      new hazelcast::util::hz_thread_pool(number_of_event_threads_));
    registration_executor_.reset(new hazelcast::util::hz_thread_pool(1));
 
    for (int i = 0; i < number_of_event_threads_; ++i) {
        event_strands_.emplace_back(event_executor_->get_executor());
    }
 
    client_connection_manager_.add_connection_listener(shared_from_this());
}
 
boost::uuids::uuid
listener_service_impl::register_listener_internal(
  std::shared_ptr<ListenerMessageCodec> listener_message_codec,
  std::shared_ptr<client::impl::BaseEventHandler> handler)
{
    auto user_registration_id = client_context_.random_uuid();
 
    std::shared_ptr<listener_registration> registration(
      new listener_registration{ listener_message_codec, handler });
    registrations_.put(user_registration_id, registration);
    for (auto const& connection :
         client_connection_manager_.get_active_connections()) {
        try {
            invoke(registration, connection);
        } catch (exception::iexception& e) {
            if (connection->is_alive()) {
                deregister_listener_internal(user_registration_id);
                BOOST_THROW_EXCEPTION(
                  (exception::exception_builder<exception::hazelcast_>(
                     "ClientListenerService::RegisterListenerTask::call")
                   << "Listener can not be added " << e)
                    .build());
            }
        }
    }
    return user_registration_id;
}
 
bool
listener_service_impl::deregister_listener_internal(
  boost::uuids::uuid user_registration_id)
{
    auto listenerRegistration = registrations_.get(user_registration_id);
    if (!listenerRegistration) {
        return false;
    }
    bool successful = true;
 
    auto listener_registrations =
      listenerRegistration->registrations.entry_set();
    for (auto it = listener_registrations.begin();
         it != listener_registrations.end();) {
        const auto& registration = it->second;
        const auto& subscriber = it->first;
        try {
            const auto& listenerMessageCodec = listenerRegistration->codec;
            auto serverRegistrationId = registration->server_registration_id;
            auto request =
              listenerMessageCodec->encode_remove_request(serverRegistrationId);
            auto invocation = ClientInvocation::create(
              client_context_, request, "", subscriber);
            invocation->invoke().get();
 
            remove_event_handler(registration->call_id, subscriber);
 
            it = listener_registrations.erase(it);
        } catch (exception::iexception& e) {
            ++it;
            if (subscriber->is_alive()) {
                successful = false;
                std::ostringstream endpoint;
                if (subscriber->get_remote_address()) {
                    endpoint << *subscriber->get_remote_address();
                } else {
                    endpoint << "null";
                }
                HZ_LOG(logger_,
                       warning,
                       boost::str(
                         boost::format(
                           "ClientListenerService::deregisterListenerInternal "
                           "Deregistration of listener with ID %1% "
                           "has failed to address %2% %3%") %
                         user_registration_id %
                         subscriber->get_remote_address() % e));
            }
        }
    }
    if (successful) {
        registrations_.remove(user_registration_id);
    }
    return successful;
}
 
void
listener_service_impl::connection_added_internal(
  const std::shared_ptr<connection::Connection>& connection)
{
    for (const auto& listener_registration : registrations_.values()) {
        invoke_from_internal_thread(listener_registration, connection);
    }
}
 
void
listener_service_impl::connection_removed_internal(
  const std::shared_ptr<connection::Connection>& connection)
{
    for (auto& registry : registrations_.values()) {
        registry->registrations.remove(connection);
    }
}
 
void
listener_service_impl::invoke_from_internal_thread(
  const std::shared_ptr<listener_registration>& listener_registration,
  const std::shared_ptr<connection::Connection>& connection)
{
    try {
        invoke(listener_registration, connection);
    } catch (exception::iexception& e) {
        HZ_LOG(logger_,
               warning,
               boost::str(
                 boost::format("Listener with pointer %1% can not be added to "
                               "a new connection: %2%, reason: %3%") %
                 listener_registration.get() % *connection % e));
    }
}
 
void
listener_service_impl::invoke(
  const std::shared_ptr<listener_registration>& listener_registration,
  const std::shared_ptr<connection::Connection>& connection)
{
    if (listener_registration->registrations.contains_key(connection)) {
        return;
    }
 
    const auto& codec = listener_registration->codec;
    auto request = codec->encode_add_request(registers_local_only());
    const auto& handler = listener_registration->handler;
    handler->before_listener_register();
 
    auto invocation = ClientInvocation::create(
      client_context_,
      std::make_shared<protocol::ClientMessage>(std::move(request)),
      "",
      connection);
    invocation->set_event_handler(handler);
    auto clientMessage = invocation->invoke_urgent().get();
 
    auto serverRegistrationId = codec->decode_add_response(clientMessage);
    handler->on_listener_register();
    int64_t correlationId =
      invocation->get_client_message()->get_correlation_id();
 
    (*listener_registration)
      .registrations.put(
        connection,
        std::shared_ptr<connection_registration>(
          new connection_registration{ serverRegistrationId, correlationId }));
}
 
void
listener_service_impl::process_event_message(
  const std::shared_ptr<ClientInvocation> invocation,
  const std::shared_ptr<protocol::ClientMessage> response)
{
    auto eventHandler = invocation->get_event_handler();
    if (!eventHandler) {
        if (client_context_.get_lifecycle_service().is_running()) {
            HZ_LOG(logger_,
                   warning,
                   boost::str(
                     boost::format("No eventHandler for invocation. "
                                   "Ignoring this invocation response. %1%") %
                     *invocation));
        }
 
        return;
    }
 
    try {
        eventHandler->handle(*response);
    } catch (std::exception& e) {
        if (client_context_.get_lifecycle_service().is_running()) {
            HZ_LOG(
              logger_,
              warning,
              boost::str(boost::format("Delivery of event message to event "
                                       "handler failed. %1%, %2%, %3%") %
                         e.what() % *response % *invocation));
        }
    }
}
 
listener_service_impl::~listener_service_impl() = default;
 
void
cluster_view_listener::start()
{
    client_context_.get_connection_manager().add_connection_listener(
      shared_from_this());
}
 
void
cluster_view_listener::connection_added(
  const std::shared_ptr<connection::Connection> connection)
{
    try_register(connection);
}
 
void
cluster_view_listener::connection_removed(
  const std::shared_ptr<connection::Connection> connection)
{
    try_reregister_to_random_connection(connection->get_connection_id());
}
 
cluster_view_listener::cluster_view_listener(ClientContext& client_context)
  : client_context_(client_context)
{}
 
void
cluster_view_listener::try_register(
  std::shared_ptr<connection::Connection> connection)
{
    int32_t expected_id = -1;
    if (!listener_added_connection_id_.compare_exchange_strong(
          expected_id, connection->get_connection_id())) {
        // already registering/registered to another connection
        return;
    }
 
    auto invocation = ClientInvocation::create(
      client_context_,
      std::make_shared<protocol::ClientMessage>(
        protocol::codec::client_addclusterviewlistener_encode()),
      "",
      connection);
 
    auto handler =
      std::make_shared<event_handler>(connection->get_connection_id(), *this);
    invocation->set_event_handler(handler);
    handler->before_listener_register();
 
    std::weak_ptr<cluster_view_listener> weak_self = shared_from_this();
    auto conn_id = connection->get_connection_id();
 
    invocation->invoke_urgent().then(
      [weak_self, handler, conn_id](boost::future<protocol::ClientMessage> f) {
          auto self = weak_self.lock();
          if (!self)
              return;
 
          if (f.has_value()) {
              handler->on_listener_register();
              return;
          }
 
          // completes with exception, listener needs to be reregistered
          self->try_reregister_to_random_connection(conn_id);
      });
}
 
void
cluster_view_listener::try_reregister_to_random_connection(
  int32_t old_connection_id)
{
    if (!listener_added_connection_id_.compare_exchange_strong(
          old_connection_id, -1)) {
        // somebody else already trying to reregister
        return;
    }
    auto new_connection =
      client_context_.get_connection_manager().get_random_connection();
    if (new_connection) {
        try_register(new_connection);
    }
}
 
cluster_view_listener::~cluster_view_listener() = default;
 
void
cluster_view_listener::event_handler::handle_membersview(
  int32_t version,
  const std::vector<member>& member_infos)
{
    view_listener.client_context_.get_client_cluster_service().handle_event(
      version, member_infos);
}
 
void
cluster_view_listener::event_handler::handle_partitionsview(
  int32_t version,
  const std::vector<std::pair<boost::uuids::uuid, std::vector<int>>>&
    partitions)
{
    view_listener.client_context_.get_partition_service().handle_event(
      connection_id, version, partitions);
}
 
void
cluster_view_listener::event_handler::before_listener_register()
{
    view_listener.client_context_.get_client_cluster_service()
      .clear_member_list_version();
    auto& lg = view_listener.client_context_.get_logger();
    HZ_LOG(
      lg,
      finest,
      boost::str(boost::format(
                   "Register attempt of cluster_view_listener::event_handler "
                   "to connection with id %1%") %
                 connection_id));
}
 
void
cluster_view_listener::event_handler::on_listener_register()
{
    auto& lg = view_listener.client_context_.get_logger();
    HZ_LOG(lg,
           finest,
           boost::str(
             boost::format("Registered cluster_view_listener::event_handler to "
                           "connection with id %1%") %
             connection_id));
}
 
cluster_view_listener::event_handler::event_handler(
  int connectionId,
  cluster_view_listener& viewListener)
  : connection_id(connectionId)
  , view_listener(viewListener)
{}
} // namespace listener
 
protocol::ClientMessage
ClientInvocationServiceImpl::BackupListenerMessageCodec::encode_add_request(
  bool local_only) const
{
    return protocol::codec::client_localbackuplistener_encode();
}
 
protocol::ClientMessage
ClientInvocationServiceImpl::BackupListenerMessageCodec::encode_remove_request(
  boost::uuids::uuid real_registration_id) const
{
    assert(0);
    return protocol::ClientMessage(0);
}
 
void
ClientInvocationServiceImpl::noop_backup_event_handler::handle_backup(
  int64_t source_invocation_correlation_id)
{
    assert(0);
}
 
namespace discovery {
remote_address_provider::remote_address_provider(
  std::function<std::unordered_map<address, address>()> addr_map_method,
  bool use_public)
  : refresh_address_map_(std::move(addr_map_method))
  , use_public_(use_public)
{}
 
std::vector<address>
remote_address_provider::load_addresses()
{
    auto address_map = refresh_address_map_();
    std::lock_guard<std::mutex> guard(lock_);
    private_to_public_ = address_map;
    std::vector<address> addresses;
    addresses.reserve(address_map.size());
    for (const auto& addr_pair : address_map) {
        addresses.push_back(addr_pair.first);
    }
    return addresses;
}
 
boost::optional<address>
remote_address_provider::translate(const address& addr)
{
    // if it is inside cloud, return private address otherwise we need to
    // translate it.
    if (!use_public_) {
        return addr;
    }
 
    {
        std::lock_guard<std::mutex> guard(lock_);
        auto found = private_to_public_.find(addr);
        if (found != private_to_public_.end()) {
            return found->second;
        }
    }
 
    auto address_map = refresh_address_map_();
 
    std::lock_guard<std::mutex> guard(lock_);
    private_to_public_ = address_map;
 
    auto found = private_to_public_.find(addr);
    if (found != private_to_public_.end()) {
        return found->second;
    }
 
    return boost::none;
}
 
#ifdef HZ_BUILD_WITH_SSL
cloud_discovery::cloud_discovery(config::cloud_config& config,
                                 std::string cloud_base_url,
                                 std::chrono::steady_clock::duration timeout)
  : cloud_config_(config)
  , cloud_base_url_(cloud_base_url)
  , timeout_(timeout)
{}
#else
cloud_discovery::cloud_discovery(config::cloud_config& config,
                                 std::string cloud_base_url,
                                 std::chrono::steady_clock::duration timeout)
{}
#endif // HZ_BUILD_WITH_SSL
 
std::unordered_map<address, address>
cloud_discovery::get_addresses()
{
#ifdef HZ_BUILD_WITH_SSL
    try {
        util::SyncHttpsClient httpsConnection(cloud_base_url_,
                                              std::string(CLOUD_URL_PATH) +
                                                cloud_config_.discovery_token,
                                              timeout_,
                                              cloud_config_.discovery_token);
        auto& conn_stream = httpsConnection.connect_and_get_response();
        return parse_json_response(conn_stream);
    } catch (std::exception& e) {
        std::throw_with_nested(
          boost::enable_current_exception(exception::illegal_state(
            "cloud_discovery::get_addresses", e.what())));
    }
#else
    util::Preconditions::check_ssl("cloud_discovery::get_addresses");
    return std::unordered_map<address, address>();
#endif
}
 
std::unordered_map<address, address>
cloud_discovery::parse_json_response(std::istream& conn_stream)
{
    namespace pt = boost::property_tree;
 
    pt::ptree root;
    pt::read_json(conn_stream, root);
 
    std::unordered_map<address, address> addresses;
    for (const auto& item : root) {
        auto private_address =
          item.second.get<std::string>(PRIVATE_ADDRESS_PROPERTY);
        auto public_address =
          item.second.get<std::string>(PUBLIC_ADDRESS_PROPERTY);
 
        address public_addr = create_address(public_address, -1);
        // if it is not explicitly given, create the private address with public
        // addresses port
        auto private_addr =
          create_address(private_address, public_addr.get_port());
        addresses.emplace(std::move(private_addr), std::move(public_addr));
    }
 
    return addresses;
}
 
address
cloud_discovery::create_address(const std::string& hostname, int default_port)
{
    auto address_holder =
      util::AddressUtil::get_address_holder(hostname, default_port);
    auto scoped_hostname =
      util::AddressHelper::get_scoped_hostname(address_holder);
    return address(std::move(scoped_hostname), address_holder.get_port());
}
} // namespace discovery
} // namespace impl
} // namespace spi
} // namespace client
} // namespace hazelcast
 
namespace std {
bool
less<hazelcast::client::spi::DefaultObjectNamespace>::operator()(
  const hazelcast::client::spi::DefaultObjectNamespace& lhs,
  const hazelcast::client::spi::DefaultObjectNamespace& rhs) const
{
    int result = lhs.get_service_name().compare(rhs.get_service_name());
    if (result < 0) {
        return true;
    }
 
    if (result > 0) {
        return false;
    }
 
    return lhs.get_object_name().compare(rhs.get_object_name()) < 0;
}
 
std::size_t
hash<hazelcast::client::spi::DefaultObjectNamespace>::operator()(
  const hazelcast::client::spi::DefaultObjectNamespace& k) const noexcept
{
    return std::hash<std::string>()(k.get_service_name() + k.get_object_name());
}
} // namespace std