replicated_map.h

/*
 * 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.
 */
#pragma once
 
#include <string>
#include <memory>
 
#include "hazelcast/client/proxy/ReplicatedMapImpl.h"
#include "hazelcast/client/entry_listener.h"
#include "hazelcast/client/entry_event.h"
#include "hazelcast/client/map_event.h"
#include "hazelcast/client/query/predicates.h"
#include "hazelcast/logger.h"
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(push)
#pragma warning(disable : 4251) // for dll export
#endif
 
namespace hazelcast {
namespace client {
class HAZELCAST_API replicated_map : public proxy::ReplicatedMapImpl
{
    friend class spi::ProxyManager;
 
public:
    static constexpr const char* SERVICE_NAME = "hz:impl:replicatedMapService";
    template<typename K, typename V, typename R = V>
    boost::future<boost::optional<R>> put(const K& key,
                                          const V& value,
                                          std::chrono::milliseconds ttl)
    {
        return to_object<R>(put_data(to_data(key), to_data(value), ttl));
    }
 
    template<typename K, typename V>
    boost::future<void> put_all(const std::unordered_map<K, V>& entries)
    {
        return put_all_data(to_data_entries(entries));
    }
 
    boost::future<boost::uuids::uuid> add_entry_listener(
      entry_listener&& listener)
    {
        return proxy::ReplicatedMapImpl::add_entry_listener(
          std::shared_ptr<impl::BaseEventHandler>(
            new EntryEventHandler<
              protocol::codec::replicatedmap_addentrylistener_handler>(
              get_name(),
              get_context().get_client_cluster_service(),
              get_context().get_serialization_service(),
              std::move(listener),
              get_context().get_logger())));
    }
 
    template<typename K>
    typename std::enable_if<!std::is_base_of<query::predicate, K>::value,
                            boost::future<boost::uuids::uuid>>::type
    add_entry_listener(entry_listener&& listener, const K& key)
    {
        return proxy::ReplicatedMapImpl::add_entry_listener_to_key(
          std::shared_ptr<impl::BaseEventHandler>(
            new EntryEventHandler<
              protocol::codec::replicatedmap_addentrylistenertokey_handler>(
              get_name(),
              get_context().get_client_cluster_service(),
              get_context().get_serialization_service(),
              std::move(listener),
              get_context().get_logger())),
          to_data(key));
    }
 
    template<typename P>
    typename std::enable_if<std::is_base_of<query::predicate, P>::value,
                            boost::future<boost::uuids::uuid>>::type
    add_entry_listener(entry_listener&& listener, const P& predicate)
    {
        return proxy::ReplicatedMapImpl::add_entry_listener(
          std::shared_ptr<impl::BaseEventHandler>(
            new EntryEventHandler<
              protocol::codec::
                replicatedmap_addentrylistenerwithpredicate_handler>(
              get_name(),
              get_context().get_client_cluster_service(),
              get_context().get_serialization_service(),
              std::move(listener),
              get_context().get_logger())),
          to_data(predicate));
    }
 
    template<typename K, typename P>
    typename std::enable_if<std::is_base_of<query::predicate, P>::value,
                            boost::future<boost::uuids::uuid>>::type
    add_entry_listener(entry_listener&& listener,
                       const P& predicate,
                       const K& key)
    {
        return proxy::ReplicatedMapImpl::add_entry_listener(
          std::shared_ptr<impl::BaseEventHandler>(
            new EntryEventHandler<
              protocol::codec::
                replicatedmap_addentrylistenertokeywithpredicate_handler>(
              get_name(),
              get_context().get_client_cluster_service(),
              get_context().get_serialization_service(),
              std::move(listener),
              get_context().get_logger())),
          to_data(key),
          to_data(predicate));
    }
 
    template<typename V>
    boost::future<std::vector<V>> values()
    {
        return to_object_vector<V>(values_data());
    }
 
    template<typename K, typename V>
    boost::future<std::vector<std::pair<K, V>>> entry_set()
    {
        return to_entry_object_vector<K, V>(entry_set_data());
    }
 
    template<typename K>
    boost::future<std::vector<K>> key_set()
    {
        return to_object_vector<K>(key_set_data());
    }
 
    template<typename K>
    boost::future<bool> contains_key(const K& key)
    {
        return contains_key_data(to_data(key));
    }
 
    template<typename V>
    boost::future<bool> contains_value(const V& value)
    {
        return contains_value_data(to_data(value));
    }
 
    template<typename K, typename V>
    boost::future<boost::optional<V>> get(const K& key)
    {
        return to_object<V>(get_data(to_data(key)));
    }
 
    template<typename K, typename V, typename R = V>
    boost::future<boost::optional<R>> put(const K& key, const V& value)
    {
        return put<K, V, R>(key, value, std::chrono::milliseconds(0));
    }
 
    template<typename K, typename V>
    boost::future<boost::optional<V>> remove(const K& key)
    {
        return to_object<V>(remove_data(to_data(key)));
    }
 
private:
    replicated_map(const std::string& object_name, spi::ClientContext* context)
      : proxy::ReplicatedMapImpl(SERVICE_NAME, object_name, context)
    {}
 
    template<typename HANDLER>
    class EntryEventHandler : public HANDLER
    {
    public:
        EntryEventHandler(
          const std::string& instance_name,
          spi::impl::ClientClusterServiceImpl& cluster_service,
          serialization::pimpl::SerializationService& serialization_service,
          entry_listener&& listener,
          logger& lg)
          : instance_name_(instance_name)
          , cluster_service_(cluster_service)
          , serialization_service_(serialization_service)
          , listener_(std::move(listener))
          , logger_(lg)
        {}
 
        void handle_entry(
          const boost::optional<serialization::pimpl::data>& key,
          const boost::optional<serialization::pimpl::data>& value,
          const boost::optional<serialization::pimpl::data>& old_value,
          const boost::optional<serialization::pimpl::data>& merging_value,
          int32_t event_type,
          boost::uuids::uuid uuid,
          int32_t number_of_affected_entries) override
        {
            if (event_type ==
                static_cast<int32_t>(entry_event::type::CLEAR_ALL)) {
                fire_map_wide_event(key,
                                    value,
                                    old_value,
                                    merging_value,
                                    event_type,
                                    uuid,
                                    number_of_affected_entries);
                return;
            }
 
            fire_entry_event(key,
                             value,
                             old_value,
                             merging_value,
                             event_type,
                             uuid,
                             number_of_affected_entries);
        }
 
    private:
        void fire_map_wide_event(
          const boost::optional<serialization::pimpl::data>& key,
          const boost::optional<serialization::pimpl::data>& value,
          const boost::optional<serialization::pimpl::data>& old_value,
          const boost::optional<serialization::pimpl::data>& merging_value,
          int32_t event_type,
          boost::uuids::uuid uuid,
          int32_t number_of_affected_entries)
        {
            auto member = cluster_service_.get_member(uuid);
            auto mapEventType = static_cast<entry_event::type>(event_type);
            map_event mapEvent(std::move(member).value(),
                               mapEventType,
                               instance_name_,
                               number_of_affected_entries);
            listener_.map_cleared_(std::move(mapEvent));
        }
 
        void fire_entry_event(
          const boost::optional<serialization::pimpl::data>& key,
          const boost::optional<serialization::pimpl::data>& value,
          const boost::optional<serialization::pimpl::data>& old_value,
          const boost::optional<serialization::pimpl::data>& merging_value,
          int32_t event_type,
          boost::uuids::uuid uuid,
          int32_t number_of_affected_entries)
        {
            typed_data eventKey, val, oldVal, mergingVal;
            if (value) {
                val = typed_data(std::move(*value), serialization_service_);
            }
            if (old_value) {
                oldVal =
                  typed_data(std::move(*old_value), serialization_service_);
            }
            if (merging_value) {
                mergingVal =
                  typed_data(std::move(*merging_value), serialization_service_);
            }
            if (key) {
                eventKey = typed_data(std::move(*key), serialization_service_);
            }
            auto m = cluster_service_.get_member(uuid);
            if (!m.has_value()) {
                m = member(uuid);
            }
            auto type = static_cast<entry_event::type>(event_type);
            entry_event entryEvent(instance_name_,
                                   std::move(m.value()),
                                   type,
                                   std::move(eventKey),
                                   std::move(val),
                                   std::move(oldVal),
                                   std::move(mergingVal));
            switch (type) {
                case entry_event::type::ADDED:
                    listener_.added_(std::move(entryEvent));
                    break;
                case entry_event::type::REMOVED:
                    listener_.removed_(std::move(entryEvent));
                    break;
                case entry_event::type::UPDATED:
                    listener_.updated_(std::move(entryEvent));
                    break;
                case entry_event::type::EVICTED:
                    listener_.evicted_(std::move(entryEvent));
                    break;
                default:
                    HZ_LOG(
                      logger_,
                      warning,
                      boost::str(boost::format(
                                   "Received unrecognized event with type: %1% "
                                   "Dropping the event!!!") %
                                 static_cast<int>(type)));
            }
        }
 
    private:
        const std::string& instance_name_;
        spi::impl::ClientClusterServiceImpl& cluster_service_;
        serialization::pimpl::SerializationService& serialization_service_;
        entry_listener listener_;
        logger& logger_;
    };
};
} // namespace client
} // namespace hazelcast
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif