ClientMapProxy.h

/*
 * Copyright (c) 2008-2017, 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.
 */
#ifndef HAZELCAST_CLIENT_MIXEDTYPE_MAP_H_
#define HAZELCAST_CLIENT_MIXEDTYPE_MAP_H_

#include <string>
#include <map>
#include <set>
#include <vector>
#include <stdexcept>
#include <climits>
#include <assert.h>

#include "hazelcast/client/TypedData.h"
#include "hazelcast/client/monitor/LocalMapStats.h"
#include "hazelcast/client/monitor/impl/NearCacheStatsImpl.h"
#include "hazelcast/client/monitor/impl/LocalMapStatsImpl.h"
#include "hazelcast/client/protocol/codec/MapAddEntryListenerWithPredicateCodec.h"
#include "hazelcast/client/impl/EntryArrayImpl.h"
#include "hazelcast/client/proxy/IMapImpl.h"
#include "hazelcast/client/impl/EntryEventHandler.h"
#include "hazelcast/client/EntryListener.h"
#include "hazelcast/client/EntryView.h"
#include "hazelcast/client/serialization/pimpl/SerializationService.h"
#include "hazelcast/client/Future.h"
#include "hazelcast/client/protocol/codec/MapSubmitToKeyCodec.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 {
        namespace mixedtype {
            class HAZELCAST_API ClientMapProxy : public proxy::IMapImpl {
            public:
                typedef std::map<int, std::vector<boost::shared_ptr<serialization::pimpl::Data> > > PID_TO_KEY_MAP;

                ClientMapProxy(const std::string &instanceName, spi::ClientContext *context);

                template <typename K>
                bool containsKey(const K &key) {
                    serialization::pimpl::Data keyData = toData(key);
                    return containsKeyInternal(keyData);
                }

                template <typename V>
                bool containsValue(const V &value) {
                    return proxy::IMapImpl::containsValue(toData(value));
                }

                template <typename K>
                TypedData get(const K &key) {
                    serialization::pimpl::Data keyData = toData(key);
                    boost::shared_ptr<TypedData> value = getInternal(keyData);
                    TypedData result = *value;
                    return result;
                }

                template <typename K, typename V>
                TypedData put(const K &key, const V &value) {
                    return put(key, value, -1);
                }

                template <typename K, typename V>
                TypedData put(const K &key, const V &value, long ttlInMillis) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    return TypedData(putInternal(keyData, valueData, ttlInMillis), context->getSerializationService());
                }

                template <typename K>
                TypedData remove(const K &key) {
                    serialization::pimpl::Data keyData = toData(key);

                    std::auto_ptr<serialization::pimpl::Data> response = removeInternal(keyData);
                    return TypedData(response, context->getSerializationService());
                }

                template <typename K, typename V>
                bool remove(const K &key, const V &value) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    return removeInternal(keyData, valueData);
                }

                void removeAll(const query::Predicate &predicate);

                template <typename K>
                void deleteEntry(const K &key) {
                    serialization::pimpl::Data keyData = toData(key);

                    deleteInternal(keyData);
                }

                void flush();

                template <typename K>
                bool tryRemove(const K &key, long timeoutInMillis) {
                    serialization::pimpl::Data keyData = toData(key);

                    return tryRemoveInternal(keyData, timeoutInMillis);
                }

                template <typename K, typename V>
                bool tryPut(const K &key, const V &value, long timeoutInMillis) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    return tryPutInternal(keyData, valueData, timeoutInMillis);
                }

                template <typename K, typename V>
                void putTransient(const K &key, const V &value, long ttlInMillis) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    tryPutTransientInternal(keyData, valueData, ttlInMillis);
                }

                template <typename K, typename V>
                TypedData putIfAbsent(const K &key, const V &value) {
                    return putIfAbsent(key, value, -1);
                }

                template <typename K, typename V>
                TypedData putIfAbsent(const K &key, const V &value, long ttlInMillis) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    std::auto_ptr<serialization::pimpl::Data> response = putIfAbsentInternal(keyData, valueData, ttlInMillis);
                    return TypedData(response, context->getSerializationService());
                }

                template <typename K, typename V, typename NEWTYPE>
                bool replace(const K &key, const V &oldValue, const NEWTYPE &newValue) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data oldValueData = toData(oldValue);
                    serialization::pimpl::Data newValueData = toData(newValue);

                    return replaceIfSameInternal(keyData, oldValueData, newValueData);
                }

                template <typename K, typename V>
                TypedData replace(const K &key, const V &value) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    return TypedData(replaceInternal(keyData, valueData), context->getSerializationService());
                }

                template <typename K, typename V>
                void set(const K &key, const V &value) {
                    set(key, value, -1);
                }

                template <typename K, typename V>
                void set(const K &key, const V &value, long ttl) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data valueData = toData(value);

                    setInternal(keyData, valueData, ttl);
                }

                template <typename K>
                void lock(const K &key) {
                    lock(key, -1);
                }

                template <typename K>
                void lock(const K &key, long leaseTime) {
                    serialization::pimpl::Data keyData = toData(key);

                    proxy::IMapImpl::lock(toData(key), leaseTime);
                }

                template <typename K>
                bool isLocked(const K &key) {
                    return proxy::IMapImpl::isLocked(toData(key));
                }

                template <typename K>
                bool tryLock(const K &key) {
                    return tryLock(key, 0);
                }

                template <typename K>
                bool tryLock(const K &key, long timeInMillis) {
                    return proxy::IMapImpl::tryLock(toData(key), timeInMillis);
                }

                template <typename K>
                void unlock(const K &key) {
                    proxy::IMapImpl::unlock(toData(key));
                }

                template <typename K>
                void forceUnlock(const K &key) {
                    proxy::IMapImpl::forceUnlock(toData(key));
                }

                template<typename MapInterceptor>
                std::string addInterceptor(MapInterceptor &interceptor) {
                    return proxy::IMapImpl::addInterceptor(interceptor);
                }

                void removeInterceptor(const std::string &id);

                std::string addEntryListener(MixedEntryListener &listener, bool includeValue);

                std::string
                addEntryListener(MixedEntryListener &listener, const query::Predicate &predicate, bool includeValue);

                bool removeEntryListener(const std::string &registrationId);

                template <typename K>
                std::string addEntryListener(const K &key, MixedEntryListener &listener, bool includeValue) {
                    serialization::pimpl::Data keyData = toData(key);
                    impl::MixedEntryEventHandler<protocol::codec::MapAddEntryListenerCodec::AbstractEventHandler> *entryEventHandler =
                            new impl::MixedEntryEventHandler<protocol::codec::MapAddEntryListenerCodec::AbstractEventHandler>(
                                    getName(), context->getClusterService(), context->getSerializationService(),
                                    listener,
                                    includeValue);
                    return proxy::IMapImpl::addEntryListener(entryEventHandler, keyData, includeValue);
                }

                template <typename K>
                std::auto_ptr<EntryView<TypedData, TypedData> > getEntryView(const K &key) {
                    std::auto_ptr<serialization::pimpl::Data>keyData(new serialization::pimpl::Data(toData(key)));
                    std::auto_ptr<map::DataEntryView> dataEntryView = proxy::IMapImpl::getEntryViewData(*keyData);
                    if ((map::DataEntryView *)NULL == dataEntryView.get()) {
                        return std::auto_ptr<EntryView<TypedData, TypedData> >();
                    }
                    TypedData value(std::auto_ptr<serialization::pimpl::Data>(
                            new serialization::pimpl::Data(std::auto_ptr<std::vector<byte> >(
                                    new std::vector<byte>(dataEntryView->getValue().toByteArray())))),
                                    context->getSerializationService());
                    const TypedData &keyTypedData = TypedData(keyData, context->getSerializationService());
                    std::auto_ptr<EntryView<TypedData, TypedData> > view(new EntryView<TypedData, TypedData>(
                            keyTypedData, value, *dataEntryView));
                    return view;
                }

                template <typename K>
                bool evict(const K &key) {
                    serialization::pimpl::Data keyData = toData(key);

                    return evictInternal(keyData);
                }

                void evictAll();

                template <typename K>
                std::vector<std::pair<TypedData, TypedData> > getAll(const std::set<K> &keys) {
                    if (keys.empty()) {
                        return std::vector<std::pair<TypedData, TypedData> >();
                    }

                    PID_TO_KEY_MAP partitionToKeyData;

                    for (typename std::set<K>::const_iterator it = keys.begin();it != keys.end();++it) {
                        const K  &key = *it;
                        serialization::pimpl::Data keyData = toData<K>(key);

                        int partitionId = getPartitionId(keyData);

                        partitionToKeyData[partitionId].push_back(toShared(keyData));
                    }

                    return toTypedDataEntrySet(getAllInternal(partitionToKeyData));
                }

                std::vector<TypedData> keySet();

                std::vector<TypedData> keySet(const serialization::IdentifiedDataSerializable &predicate);

                std::vector<TypedData> keySet(const query::Predicate &predicate);

                template <typename K, typename V>
                std::vector<K> keySet(query::PagingPredicate <K, V> &predicate) {
                    predicate.setIterationType(query::KEY);

                    std::vector<serialization::pimpl::Data> dataResult = keySetForPagingPredicateData(predicate);

                    EntryVector entryResult;
                    for (std::vector<serialization::pimpl::Data>::iterator it = dataResult.begin();
                         it != dataResult.end(); ++it) {
                        entryResult.push_back(std::pair<serialization::pimpl::Data, serialization::pimpl::Data>(*it,
                                                                                                                serialization::pimpl::Data()));
                    }

                    client::impl::EntryArrayImpl<K, V> entries(entryResult, context->getSerializationService());
                    entries.sort(query::KEY, predicate.getComparator());

                    std::pair<size_t, size_t> range = updateAnchor<K, V>(entries, predicate, query::KEY);

                    std::vector<K> result;
                    for (size_t i = range.first; i < range.second; ++i) {
                        result.push_back(*entries.getKey(i));
                    }

                    return result;
                }

                std::vector<TypedData> values();

                std::vector<TypedData> values(const serialization::IdentifiedDataSerializable &predicate);

                std::vector<TypedData> values(const query::Predicate &predicate);

                template <typename K, typename V>
                std::vector<V> values(query::PagingPredicate <K, V> &predicate) {
                    predicate.setIterationType(query::VALUE);

                    EntryVector dataResult = proxy::IMapImpl::valuesForPagingPredicateData(predicate);

                    client::impl::EntryArrayImpl<K, V> entries(dataResult, context->getSerializationService());

                    entries.sort(query::VALUE, predicate.getComparator());

                    std::pair<size_t, size_t> range = updateAnchor<K, V>(entries, predicate, query::VALUE);

                    std::vector<V> result;
                    for (size_t i = range.first; i < range.second; ++i) {
                        result.push_back(*entries.getValue(i));
                    }
                    return result;
                }

                std::vector<std::pair<TypedData, TypedData> > entrySet();

                std::vector<std::pair<TypedData, TypedData> > entrySet(const serialization::IdentifiedDataSerializable &predicate);

                std::vector<std::pair<TypedData, TypedData> > entrySet(const query::Predicate &predicate);

                template <typename K, typename V>
                std::vector<std::pair<K, V> > entrySet(query::PagingPredicate <K, V> &predicate) {
                    std::vector<std::pair<serialization::pimpl::Data, serialization::pimpl::Data> > dataResult =
                            proxy::IMapImpl::entrySetForPagingPredicateData(predicate);

                    client::impl::EntryArrayImpl<K, V> entries(dataResult, context->getSerializationService());
                    entries.sort(query::ENTRY, predicate.getComparator());

                    std::pair<size_t, size_t> range = updateAnchor<K, V>(entries, predicate, query::ENTRY);

                    std::vector<std::pair<K, V> > result;
                    for (size_t i = range.first; i < range.second; ++i) {
                        std::pair<const K *, const V *> entry = entries[i];
                        result.push_back(std::pair<K, V>(*entry.first, *entry.second));
                    }
                    return result;
                }

                void addIndex(const std::string &attribute, bool ordered);

                template<typename K, typename EntryProcessor>
                TypedData executeOnKey(const K &key, const EntryProcessor &entryProcessor) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data processorData = toData(entryProcessor);

                    std::auto_ptr<serialization::pimpl::Data> response = executeOnKeyInternal(keyData, processorData);

                    return TypedData(response, getSerializationService());
                }

                template<typename K, typename EntryProcessor>
                Future<TypedData> submitToKey(const K &key, const EntryProcessor &entryProcessor) {
                    serialization::pimpl::Data keyData = toData(key);
                    serialization::pimpl::Data processorData = toData(entryProcessor);

                    return submitToKeyInternal<TypedData>(keyData, processorData);
                }

                template<typename K, typename EntryProcessor>
                std::map<K, TypedData> executeOnKeys(const std::set<K> &keys, const EntryProcessor &entryProcessor) {
                    EntryVector entries = executeOnKeysInternal<K, EntryProcessor>(keys, entryProcessor);

                    std::map<K, TypedData> result;
                    serialization::pimpl::SerializationService &serializationService = getSerializationService();
                    for (size_t i = 0; i < entries.size(); ++i) {
                        std::auto_ptr<K> keyObj = toObject<K>(entries[i].first);
                        result[*keyObj] = TypedData(std::auto_ptr<serialization::pimpl::Data>(
                                new serialization::pimpl::Data(entries[i].second)), serializationService);
                    }
                    return result;
                }

                template<typename EntryProcessor>
                std::map<TypedData, TypedData > executeOnEntries(const EntryProcessor &entryProcessor) {
                    EntryVector entries = proxy::IMapImpl::executeOnEntriesData<EntryProcessor>(entryProcessor);
                    std::map<TypedData, TypedData> result;
                    for (size_t i = 0; i < entries.size(); ++i) {
                        std::auto_ptr<serialization::pimpl::Data> keyData(new serialization::pimpl::Data(entries[i].first));
                        std::auto_ptr<serialization::pimpl::Data> valueData(new serialization::pimpl::Data(entries[i].second));
                        serialization::pimpl::SerializationService &serializationService = context->getSerializationService();
                        result[TypedData(keyData, serializationService)] = TypedData(valueData, serializationService);
                    }
                    return result;
                }

                template<typename EntryProcessor>
                std::map<TypedData, TypedData>
                executeOnEntries(const EntryProcessor &entryProcessor, const query::Predicate &predicate) {
                    EntryVector entries = proxy::IMapImpl::executeOnEntriesData<EntryProcessor>(entryProcessor,
                                                                                                predicate);
                    std::map<TypedData, TypedData> result;
                    for (size_t i = 0; i < entries.size(); ++i) {
                        std::auto_ptr<serialization::pimpl::Data> keyData(new serialization::pimpl::Data(entries[i].first));
                        std::auto_ptr<serialization::pimpl::Data> valueData(new serialization::pimpl::Data(entries[i].second));
                        serialization::pimpl::SerializationService &serializationService = context->getSerializationService();
                        result[TypedData(keyData, serializationService)] = TypedData(valueData, serializationService);
                    }
                    return result;
                }

                int size();

                bool isEmpty();


                template<typename K, typename V>
                void putAll(const std::map<K, V> &entries) {
                    std::map<int, EntryVector> entryMap;

                    for (typename std::map<K, V>::const_iterator it = entries.begin(); it != entries.end(); ++it) {
                        serialization::pimpl::Data keyData = toData(it->first);
                        serialization::pimpl::Data valueData = toData(it->second);

                        int partitionId = getPartitionId(keyData);

                        entryMap[partitionId].push_back(std::make_pair(keyData, valueData));
                    }

                    putAllInternal(entryMap);
                }

                void clear();

                serialization::pimpl::SerializationService &getSerializationService() const;

                virtual monitor::LocalMapStats &getLocalMapStats();
            protected:
                virtual boost::shared_ptr<TypedData> getInternal(serialization::pimpl::Data &keyData);

                virtual bool containsKeyInternal(const serialization::pimpl::Data &keyData);

                virtual std::auto_ptr<serialization::pimpl::Data> removeInternal(
                        const serialization::pimpl::Data &keyData);

                virtual bool removeInternal(
                        const serialization::pimpl::Data &keyData, const serialization::pimpl::Data &valueData);

                virtual void removeAllInternal(const serialization::pimpl::Data &predicateData);

                virtual void deleteInternal(const serialization::pimpl::Data &keyData);

                virtual bool tryRemoveInternal(const serialization::pimpl::Data &keyData, long timeoutInMillis);

                virtual bool tryPutInternal(const serialization::pimpl::Data &keyData,
                                            const serialization::pimpl::Data &valueData, long timeoutInMillis);

                virtual std::auto_ptr<serialization::pimpl::Data> putInternal(const serialization::pimpl::Data &keyData,
                                                                              const serialization::pimpl::Data &valueData, long timeoutInMillis);

                virtual void tryPutTransientInternal(const serialization::pimpl::Data &keyData,
                                                     const serialization::pimpl::Data &valueData, int ttlInMillis);

                virtual std::auto_ptr<serialization::pimpl::Data> putIfAbsentInternal(const serialization::pimpl::Data &keyData,
                                                                                      const serialization::pimpl::Data &valueData,
                                                                                      int ttlInMillis);

                virtual bool replaceIfSameInternal(const serialization::pimpl::Data &keyData,
                                                   const serialization::pimpl::Data &valueData,
                                                   const serialization::pimpl::Data &newValueData);

                virtual std::auto_ptr<serialization::pimpl::Data> replaceInternal(const serialization::pimpl::Data &keyData,
                                                                                  const serialization::pimpl::Data &valueData);

                virtual void setInternal(const serialization::pimpl::Data &keyData, const serialization::pimpl::Data &valueData,
                                         int ttlInMillis);

                virtual bool evictInternal(const serialization::pimpl::Data &keyData);

                EntryVector getAllInternal(const PID_TO_KEY_MAP &partitionToKeyData) {
                    std::map<int, std::vector<serialization::pimpl::Data> > datas;
                    for (PID_TO_KEY_MAP::const_iterator it = partitionToKeyData.begin();
                         it != partitionToKeyData.end(); ++it) {
                        const std::vector<boost::shared_ptr<serialization::pimpl::Data> > &valueDatas = it->second;
                        for (std::vector<boost::shared_ptr<serialization::pimpl::Data> >::const_iterator valueIt = valueDatas.begin();
                             valueIt != valueDatas.end(); ++valueIt) {
                            datas[it->first].push_back(*(*valueIt));
                        }
                    }
                    return proxy::IMapImpl::getAllData(datas);
                }

                virtual std::auto_ptr<serialization::pimpl::Data>
                executeOnKeyInternal(const serialization::pimpl::Data &keyData,
                                     const serialization::pimpl::Data &processor);

                template <typename ResultType>
                Future<ResultType>
                submitToKeyInternal(const serialization::pimpl::Data &keyData,
                                    const serialization::pimpl::Data &processor) {
                    int partitionId = getPartitionId(keyData);

                    std::auto_ptr<protocol::ClientMessage> request =
                            protocol::codec::MapSubmitToKeyCodec::RequestParameters::encode(getName(),
                                                                                            processor,
                                                                                            keyData,
                                                                                            util::getThreadId());

                    connection::CallFuture callFuture = invokeAndGetFuture(request, partitionId);

                    return client::Future<ResultType>(callFuture, getSerializationService(), submitToKeyDecoder);
                }

                static std::auto_ptr<serialization::pimpl::Data> submitToKeyDecoder(protocol::ClientMessage &response);

                template<typename K, typename EntryProcessor>
                EntryVector executeOnKeysInternal(const std::set<K> &keys, const EntryProcessor &entryProcessor) {
                    if (keys.empty()) {
                        return EntryVector();
                    }

                    std::vector<serialization::pimpl::Data> keysData;
                    for (typename std::set<K>::const_iterator it = keys.begin(); it != keys.end(); ++it) {
                        keysData.push_back(toData<K>(*it));
                    }

                    serialization::pimpl::Data entryProcessorData = toData<EntryProcessor>(entryProcessor);

                    return proxy::IMapImpl::executeOnKeysData(keysData, entryProcessorData);
                }

                virtual void
                putAllInternal(const std::map<int, EntryVector> &entries);

            private:
                monitor::impl::LocalMapStatsImpl stats;
            };
        }
    }
}

#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif

#endif /* HAZELCAST_CLIENT_MIXEDTYPE_MAP_H_ */