replicated_map.h

/*
 * Copyright (c) 2008-2021, 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_;
            };
        };
    }
}
 
#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif