PagingPredicate.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_QUERY_PAGINGPREDICATE_H_
#define HAZELCAST_CLIENT_QUERY_PAGINGPREDICATE_H_

#include <string>
#include <memory>

#include "hazelcast/client/exception/IllegalStateException.h"
#include "hazelcast/client/exception/IllegalArgumentException.h"
#include "hazelcast/util/Util.h"
#include "hazelcast/util/Comparator.h"
#include "hazelcast/client/query/Predicate.h"
#include "hazelcast/client/serialization/ObjectDataOutput.h"
#include "hazelcast/client/serialization/ObjectDataInput.h"
#include "hazelcast/client/exception/IException.h"
#include "hazelcast/client/query/impl/predicates/PredicateDataSerializerHook.h"
#include "hazelcast/client/query/EntryComparator.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 query {
            enum HAZELCAST_API IterationType {
                        KEY = 0,
                        VALUE = 1,
                        ENTRY = 2
            };

            template<typename K, typename V>
            class InvalidComparator :  public query::EntryComparator<K, V> {
            public:
                int compare(const std::pair<const K *, const V *> &lhs, const std::pair<const K *, const V *> &rhs) const {
                    assert(0);
                    return -1;
                }

                int getFactoryId() const {
                    assert(0);
                    return -1;
                }

                int getClassId() const {
                    assert(0);
                    return -1;
                }

                void writeData(serialization::ObjectDataOutput &writer) const {
                    assert(0);
                }

                void readData(serialization::ObjectDataInput &reader) {
                    assert(0);
                }
            };

            static const std::string IterationNames[] = {"KEY", "VALUE", "ENTRY"};

            template<typename K, typename V>
            class PagingPredicate : public Predicate {
            public:
                PagingPredicate(size_t predicatePageSize) : pageSize(predicatePageSize), page(0),iterationType(VALUE) {
                }

                PagingPredicate(std::auto_ptr<Predicate> predicate, size_t predicatePageSize) : innerPredicate(predicate),
                                                                                             pageSize(predicatePageSize),
                                                                                             page(0),
                                                                                             iterationType(VALUE) {
                }

                PagingPredicate(std::auto_ptr<query::EntryComparator<K, V> > comparatorObj, size_t predicatePageSize) : comparator(
                        comparatorObj), pageSize(predicatePageSize), page(0), iterationType(VALUE) {
                }

                PagingPredicate(std::auto_ptr<Predicate> predicate, std::auto_ptr<query::EntryComparator<K, V> > comparatorObj,
                                size_t predicatePageSize) : innerPredicate(predicate), comparator(comparatorObj),
                                                         pageSize(predicatePageSize), page(0), iterationType(VALUE) {

                }

                ~PagingPredicate() {
                    for (typename std::vector<std::pair<size_t, std::pair<K *, V *> > >::const_iterator it = anchorList.begin();
                         it != anchorList.end(); ++it) {
                        delete it->second.first;
                        delete it->second.second;
                    }
                }

                void reset() {
                    iterationType = VALUE;
                    for (typename std::vector<std::pair<size_t, std::pair<K *, V *> > >::const_iterator it = anchorList.begin();
                         it != anchorList.end(); ++it) {
                        delete it->second.first;
                        delete it->second.second;
                    }
                    anchorList.clear();
                    page = 0;
                }

                void nextPage() {
                    ++page;
                }

                void previousPage() {
                    if (page != 0) {
                        --page;
                    }
                }

                IterationType getIterationType() const {
                    return iterationType;
                }

                void setIterationType(IterationType type) {
                    iterationType = type;
                }

                size_t getPage() const {
                    return page;
                }

                void setPage(size_t pageNumber) {
                    page = pageNumber;
                }

                size_t getPageSize() const {
                    return pageSize;
                }

                const Predicate *getPredicate() const {
                    return innerPredicate.get();
                }

                const query::EntryComparator<K, V> *getComparator() const {
                    return comparator.get();
                }


                const std::pair<K *, V *> *getAnchor() const {
                    if (0 == anchorList.size()) {
                        return (const std::pair<K *, V *> *)NULL;
                    }

                    return &anchorList[page].second;
                }

                const std::pair<size_t, std::pair<K *, V *> > *getNearestAnchorEntry() {
                    size_t anchorCount = anchorList.size();

                    if (page == 0 || anchorCount == 0) {
                        return (const std::pair<size_t, std::pair<K *, V *> > *) NULL;
                    }

                    if (page < anchorCount) {
                        return &anchorList[page - 1];
                    } else {
                        return &anchorList[anchorCount - 1];
                    }
                }

                int getFactoryId() const {
                    return impl::predicates::F_ID;
                }

                int getClassId() const {
                    return impl::predicates::PAGING_PREDICATE;
                }

                void writeData(serialization::ObjectDataOutput &out) const {
                    out.writeObject<serialization::IdentifiedDataSerializable>(innerPredicate.get());
                    out.writeObject<serialization::IdentifiedDataSerializable>(comparator.get());
                    out.writeInt((int)page);
                    out.writeInt((int)pageSize);
                    out.writeUTF(&IterationNames[iterationType]);
                    out.writeInt((int) anchorList.size());
                    for (typename std::vector<std::pair<size_t, std::pair<K *, V *> > >::const_iterator it = anchorList.begin();
                         it != anchorList.end(); ++it) {
                        out.writeInt((int)it->first);
                        out.writeObject<K>(it->second.first);
                        out.writeObject<V>(it->second.second);
                    }
                }

                void readData(serialization::ObjectDataInput &in) {
                    // Not need to read at the client side
                    throw exception::IException("PagingPredicate::readData",
                                                "Client should not need to use readData method!!!");
                }

                void setAnchor(size_t page, const std::pair<K *, V *> &anchorEntry) {
                    size_t anchorCount = anchorList.size();
                    if (page < anchorCount) {
                        // release the previous anchoe entry
                        delete anchorList[page].second.first;
                        delete anchorList[page].second.second;
                        anchorList[page] = std::pair<size_t, std::pair<K *, V *> >(page, anchorEntry);
                    } else if (page == anchorCount) {
                        anchorList.push_back(std::pair<size_t, std::pair<K *, V *> >(page, anchorEntry));
                    } else {
                        char msg[200];
                        util::snprintf(msg, 200, "Anchor index is not correct, expected: %d but found: %d", page,
                                       anchorCount);
                        throw exception::IllegalArgumentException("PagingPredicate::setAnchor", msg);
                    }
                }

            private:
                std::auto_ptr<Predicate> innerPredicate;
                // key is the page number, the value is the map entry as the anchor
                std::vector<std::pair<size_t, std::pair<K *, V *> > > anchorList;
                std::auto_ptr<query::EntryComparator<K, V> > comparator;
                size_t pageSize;
                size_t page;
                IterationType iterationType;
            };
        }
    }
}

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

#endif /* HAZELCAST_CLIENT_QUERY_PAGINGPREDICATE_H_ */