RawPointerMap.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_ADAPTOR_RAWPOINTERMAP_H_
#define HAZELCAST_CLIENT_ADAPTOR_RAWPOINTERMAP_H_

#include "hazelcast/client/IMap.h"
#include "hazelcast/client/adaptor/MapEntryView.h"
#include "hazelcast/client/impl/DataArrayImpl.h"
#include "hazelcast/client/impl/EntryArrayImpl.h"
#include "hazelcast/client/impl/EntryArrayKeyAdaptor.h"
#include "hazelcast/client/impl/EntryArrayValueAdaptor.h"
#include "hazelcast/client/map/ClientMapProxy.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 adaptor {
            template<typename K, typename V>
            class RawPointerMap {
            public:
                RawPointerMap(IMap<K, V> &mapToBeAdopted) : map(mapToBeAdopted),
                                                            mapProxy(*map.mapImpl) {
                    serializationService = &mapProxy.getSerializationService();
                }

                bool containsKey(const K &key) {
                    return map.containsKey(key);
                }

                bool containsValue(const V &value) {
                    return map.containsValue(value);
                }

                std::auto_ptr<V> get(const K &key) {
                    return serializationService->toObject<V>(
                            mapProxy.getData(serializationService->toData<K>(&key)).get());
                }

                std::auto_ptr<V> put(const K &key, const V &value) {
                    return put(key, value, -1);
                }

                std::auto_ptr<V> put(const K &key, const V &value, long ttlInMillis) {
                    return serializationService->toObject<V>(
                            mapProxy.putData(serializationService->toData<K>(&key), serializationService->toData<V>(&value),
                                        ttlInMillis).get());
                }

                std::auto_ptr<V> remove(const K &key) {
                    return serializationService->toObject<V>(
                            mapProxy.removeData(serializationService->toData<K>(&key)).get());
                }

                bool remove(const K &key, const V &value) {
                    return map.remove(key, value);
                }

                void removeAll(const query::Predicate &predicate) {
                    map.removeAll(predicate);
                }

                void deleteEntry(const K &key) {
                    map.deleteEntry(key);
                }

                void flush() {
                    map.flush();
                }

                bool tryRemove(const K &key, long timeoutInMillis) {
                    return map.tryRemove(key, timeoutInMillis);
                }

                bool tryPut(const K &key, const V &value, long timeoutInMillis) {
                    return map.tryPut(key, value, timeoutInMillis);
                }

                void putTransient(const K &key, const V &value, long ttlInMillis) {
                    map.putTransient(key, value, ttlInMillis);
                }

                std::auto_ptr<V> putIfAbsent(const K &key, const V &value) {
                    return serializationService->toObject<V>(mapProxy.putIfAbsentData(serializationService->toData<K>(&key),
                                                                                serializationService->toData<V>(&value),
                                                                                -1).get());
                }

                std::auto_ptr<V> putIfAbsent(const K &key, const V &value, long ttlInMillis) {
                    return serializationService->toObject<V>(
                            mapProxy.putIfAbsentData(serializationService->toData<K>(&key),
                                            serializationService->toData<V>(&value),
                                            ttlInMillis).get());
                }

                bool replace(const K &key, const V &oldValue, const V &newValue) {
                    return map.replace(key, oldValue, newValue);
                }

                std::auto_ptr<V> replace(const K &key, const V &value) {
                    return serializationService->toObject<V>(mapProxy.replaceData(
                            serializationService->toData<K>(&key), serializationService->toData<V>(&value)).get());
                }

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

                void set(const K &key, const V &value, long ttl) {
                    map.set(key, value, ttl);
                }

                void lock(const K &key) {
                    map.lock(key);
                }

                void lock(const K &key, long leaseTime) {
                    map.lock(key, leaseTime);
                }

                bool isLocked(const K &key) {
                    return map.isLocked(key);
                }

                bool tryLock(const K &key) {
                    return map.tryLock(key, 0);
                }

                bool tryLock(const K &key, long timeInMillis) {
                    return map.tryLock(key, timeInMillis);
                }

                void unlock(const K &key) {
                    map.unlock(key);
                }

                void forceUnlock(const K &key) {
                    map.forceUnlock(key);
                }

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

                void removeInterceptor(const std::string &id) {
                    map.removeInterceptor(id);
                }

                std::string addEntryListener(EntryListener<K, V> &listener, bool includeValue) {
                    return map.addEntryListener(listener, includeValue);
                }

                std::string addEntryListener(EntryListener<K, V> &listener, const query::Predicate &predicate, bool includeValue) {
                    return map.addEntryListener(listener, predicate, includeValue);
                }

                bool removeEntryListener(const std::string &registrationId) {
                    return map.removeEntryListener(registrationId);
                }


                std::string addEntryListener(EntryListener<K, V> &listener, const K &key, bool includeValue) {
                    return map.addEntryListener(listener, key, includeValue);
                }

                std::auto_ptr<MapEntryView<K, V> > getEntryView(const K &key) {
                    std::auto_ptr<map::DataEntryView> dataView = mapProxy.getEntryViewData(serializationService->toData<K>(&key));
                    if ((map::DataEntryView *)NULL == dataView.get()) {
                        return std::auto_ptr<MapEntryView<K, V> >();
                    }
                    return std::auto_ptr<MapEntryView<K, V> >(new MapEntryView<K, V>(dataView, *serializationService));
                }

                bool evict(const K &key) {
                    return map.evict(key);
                }

                void evictAll() {
                    map.evictAll();
                }

                std::auto_ptr<EntryArray<K, V> > getAll(const std::set<K> &keys) {
                    if (keys.empty()) {
                        return std::auto_ptr<EntryArray<K, V> >();
                    }

                    std::map<int, std::vector<serialization::pimpl::Data> > partitionToKeyData;
                    // group the request per parition id
                    for (typename std::set<K>::const_iterator it = keys.begin();it != keys.end(); ++it) {
                        serialization::pimpl::Data keyData = mapProxy.template toData<K>(*it);

                        int partitionId = mapProxy.getPartitionId(keyData);

                        partitionToKeyData[partitionId].push_back(keyData);
                    }

                    EntryVector entrySet = mapProxy.getAllData(partitionToKeyData);

                    return std::auto_ptr<EntryArray<K, V> >(
                            new client::impl::EntryArrayImpl<K, V>(entrySet, *serializationService));
                }

                std::auto_ptr<DataArray<K> > keySet() {
                    std::vector<serialization::pimpl::Data> dataResult = mapProxy.keySetData();
                    return std::auto_ptr<DataArray<K> >(new hazelcast::client::impl::DataArrayImpl<K>(
                            dataResult, *serializationService));
                }

                std::auto_ptr<DataArray<K> > keySet(const query::Predicate &predicate) {
                    std::vector<serialization::pimpl::Data> dataResult = mapProxy.keySetData(predicate);
                    return std::auto_ptr<DataArray<K> >(new hazelcast::client::impl::DataArrayImpl<K>(
                            dataResult, *serializationService));
                }

                std::auto_ptr<DataArray<K> > keySet(query::PagingPredicate<K, V> &predicate) {
                    predicate.setIterationType(query::KEY);

                    std::vector<serialization::pimpl::Data> dataResult = mapProxy.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, *serializationService);

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

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

                    std::auto_ptr<EntryArray<K, V> > subList(new client::impl::EntryArrayImpl<K, V>(
                            entries, range.first, range.second));

                    std::auto_ptr<DataArray<K> > result = std::auto_ptr<DataArray<K> >(
                            new impl::EntryArrayKeyAdaptor<K, V>(subList));

                    return result;
                }

                std::auto_ptr<DataArray<V> > values() {
                    std::vector<serialization::pimpl::Data> dataResult = mapProxy.valuesData();
                    return std::auto_ptr<DataArray<V> >(
                            new hazelcast::client::impl::DataArrayImpl<V>(dataResult, *serializationService));
                }

                std::auto_ptr<DataArray<V> > values(const query::Predicate &predicate) {
                    std::vector<serialization::pimpl::Data> dataResult = mapProxy.valuesData(predicate);
                    return std::auto_ptr<DataArray<V> >(
                            new hazelcast::client::impl::DataArrayImpl<V>(dataResult, *serializationService));
                }

                std::auto_ptr<DataArray<V> > values(query::PagingPredicate<K, V> &predicate) {
                    predicate.setIterationType(query::VALUE);

                    EntryVector entryResult = mapProxy.valuesForPagingPredicateData(predicate);

                    client::impl::EntryArrayImpl<K, V> entries(entryResult, *serializationService);

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

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

                    std::auto_ptr<EntryArray<K, V> > subList(new client::impl::EntryArrayImpl<K, V>(entries, range.first, range.second));
                    std::auto_ptr<DataArray<V> > result = std::auto_ptr<DataArray<V> >(
                            new impl::EntryArrayValueAdaptor<K, V>(subList));

                    return result;
                }

                std::auto_ptr<EntryArray<K, V> > entrySet() {
                    EntryVector entries = mapProxy.entrySetData();
                    return std::auto_ptr<EntryArray<K, V> >(
                            new client::impl::EntryArrayImpl<K, V>(entries, *serializationService));
                }

                std::auto_ptr<EntryArray<K, V> > entrySet(const query::Predicate &predicate) {
                    EntryVector entries = mapProxy.entrySetData(predicate);
                    return std::auto_ptr<EntryArray<K, V> >(
                            new client::impl::EntryArrayImpl<K, V>(entries, *serializationService));
                }

                std::auto_ptr<EntryArray<K, V> > entrySet(query::PagingPredicate<K, V> &predicate) {
                    std::vector<std::pair<serialization::pimpl::Data, serialization::pimpl::Data> > dataResult =
                            mapProxy.entrySetForPagingPredicateData(predicate);

                    client::impl::EntryArrayImpl<K, V> entries(dataResult, *serializationService);

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

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

                    return std::auto_ptr<EntryArray<K, V> >(
                            new client::impl::EntryArrayImpl<K, V>(entries, range.first, range.second));
                }

                void addIndex(const std::string &attribute, bool ordered) {
                    map.addIndex(attribute, ordered);
                }

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

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

                    return serializationService->toObject<ResultType>(resultData.get());
                }

                template<typename ResultType, typename EntryProcessor>
                Future<ResultType> submitToKey(const K &key, const EntryProcessor &entryProcessor) {
                    return map.template submitToKey<ResultType, EntryProcessor>(key, entryProcessor);
                }

                template<typename ResultType, typename EntryProcessor>
                std::auto_ptr<EntryArray<K, ResultType> > executeOnKeys(const std::set<K> &keys, const EntryProcessor &entryProcessor) {
                    EntryVector results = mapProxy.template executeOnKeysInternal<EntryProcessor>(keys, entryProcessor);

                    return std::auto_ptr<EntryArray<K, ResultType> >(
                            new client::impl::EntryArrayImpl<K, ResultType>(results, *serializationService));
                }

                template<typename ResultType, typename EntryProcessor>
                std::auto_ptr<EntryArray<K, ResultType> > executeOnEntries(const EntryProcessor &entryProcessor) {
                    EntryVector results = mapProxy.template executeOnEntriesData<EntryProcessor>(entryProcessor);

                    return std::auto_ptr<EntryArray<K, ResultType> >(
                            new client::impl::EntryArrayImpl<K, ResultType>(results, *serializationService));
                }

                template<typename ResultType, typename EntryProcessor>
                std::auto_ptr<EntryArray<K, ResultType> > executeOnEntries(const EntryProcessor &entryProcessor, const query::Predicate &predicate) {
                    EntryVector results = mapProxy.template executeOnEntriesData<EntryProcessor>(entryProcessor, predicate);

                    return std::auto_ptr<EntryArray<K, ResultType> >(
                            new client::impl::EntryArrayImpl<K, ResultType>(results, *serializationService));
                }

                int size() {
                    return map.size();
                }

                bool isEmpty() {
                    return map.isEmpty();
                }


                void putAll(const std::map<K, V> &entries) {
                    map.putAll(entries);
                }

                void clear() {
                    map.clear();
                }

            private:
                IMap<K, V> &map;
                map::ClientMapProxy<K, V> &mapProxy;
                serialization::pimpl::SerializationService *serializationService;
            };
        }
    }
}

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

#endif /* HAZELCAST_CLIENT_ADAPTOR_RAWPOINTERMAP_H_ */