paging_predicate.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 <cassert>
 
#include "hazelcast/client/exception/protocol_exceptions.h"
#include "hazelcast/util/Util.h"
#include "hazelcast/util/Comparator.h"
#include "hazelcast/client/query/predicates.h"
#include "hazelcast/client/query/entry_comparator.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 protocol {
            namespace codec {
                namespace holder {
                    struct paging_predicate_holder;
                }
            }
        }
        class imap;
        namespace query {
            class paging_predicate_marker {};
 
            enum struct HAZELCAST_API iteration_type {
                        KEY = 0,
                        VALUE = 1,
                        ENTRY = 2
            };
 
            static const std::string IterationNames[] = {"KEY", "VALUE", "ENTRY"};
 
            struct anchor_data_list {
                std::vector<int32_t> page_list;
                std::vector<std::pair<serialization::pimpl::data, boost::optional<serialization::pimpl::data>>> data_list;
            };
 
            template<typename K, typename V>
            class paging_predicate : public predicate, public paging_predicate_marker {
                friend imap;
                friend protocol::codec::holder::paging_predicate_holder;
                friend serialization::hz_serializer<query::paging_predicate<K, V>>;
            public:
                ~paging_predicate() = default;
 
                void reset() {
                    iteration_type_ = iteration_type::VALUE;
                    anchor_data_list_.page_list.clear();
                    anchor_data_list_.data_list.clear();
                    page_ = 0;
                }
 
                void next_page() {
                    ++page_;
                }
 
                void previous_page() {
                    if (page_ != 0) {
                        --page_;
                    }
                }
 
                iteration_type get_iteration_type() const {
                    return iteration_type_;
                }
 
                void set_iteration_type(iteration_type type) {
                    iteration_type_ = type;
                }
 
                size_t get_page() const {
                    return page_;
                }
 
                void set_page(size_t page_number) {
                    page_ = page_number;
                }
 
                size_t get_page_size() const {
                    return page_size_;
                }
 
                const query::entry_comparator<K, V> *get_comparator() const {
                    return comparator_.get();
                }
 
                void set_anchor_data_list(anchor_data_list anchor_data_list) {
                    anchor_data_list_ = std::move(anchor_data_list);
                }
 
            private:
                anchor_data_list anchor_data_list_;
                std::shared_ptr<query::entry_comparator<K, V>> comparator_;
                serialization::object_data_output out_stream_;
                size_t page_size_;
                size_t page_;
                iteration_type iteration_type_;
                boost::optional<serialization::pimpl::data> comparator_data_;
                boost::optional<serialization::pimpl::data> predicate_data_;
 
                paging_predicate(serialization::pimpl::SerializationService &serialization_service,
                                 size_t predicate_page_size) : out_stream_(serialization_service.new_output_stream()),
                        page_size_(predicate_page_size), page_(0), iteration_type_(iteration_type::VALUE) {
                    out_stream_.write_object<bool>(nullptr);
                    out_stream_.write_object<bool>(nullptr);
                }
 
                template<typename INNER_PREDICATE>
                paging_predicate(serialization::pimpl::SerializationService &serialization_service,
                                 size_t predicate_page_size, const INNER_PREDICATE &predicate)
                        : out_stream_(serialization_service.new_output_stream()), page_size_(predicate_page_size), page_(0),
                        iteration_type_(iteration_type::VALUE) {
                    out_stream_.write_object(predicate);
                    out_stream_.write_object<bool>(nullptr);
                    predicate_data_ = serialization_service.to_data<INNER_PREDICATE>(predicate);
                }
 
                template<typename COMPARATOR>
                paging_predicate(serialization::pimpl::SerializationService &serialization_service,
                                 COMPARATOR &&comp, size_t predicate_page_size)
                        : out_stream_(serialization_service.new_output_stream()), page_size_(predicate_page_size), page_(0),
                        iteration_type_(iteration_type::VALUE) {
                    out_stream_.write_object<bool>(nullptr);
                    out_stream_.write_object(comp);
                    comparator_data_ = serialization_service.to_data<COMPARATOR>(comp);
                    comparator_ = std::make_shared<COMPARATOR>(std::forward<COMPARATOR>(comp));
                }
 
                template<typename INNER_PREDICATE, typename COMPARATOR>
                paging_predicate(serialization::pimpl::SerializationService &serialization_service,
                                 const INNER_PREDICATE &predicate, COMPARATOR &&comp, size_t predicate_page_size)
                        : out_stream_(serialization_service.new_output_stream()), page_size_(predicate_page_size), page_(0),
                        iteration_type_(iteration_type::VALUE) {
                    out_stream_.write_object(predicate);
                    out_stream_.write_object(comp);
                    predicate_data_ = serialization_service.to_data<INNER_PREDICATE>(predicate);
                    comparator_data_ = serialization_service.to_data<COMPARATOR>(comp);
                    comparator_ = std::make_shared<COMPARATOR>(std::forward<COMPARATOR>(comp));
                }
            };
        }
 
        namespace serialization {
            template<typename K, typename V>
            struct hz_serializer<query::paging_predicate<K, V>> : public identified_data_serializer {
                static constexpr int32_t get_factory_id() noexcept {
                    return static_cast<int32_t>(query::predicate_data_serializer_hook::F_ID);
                }
 
                static constexpr int32_t get_class_id() noexcept {
                    return static_cast<int32_t>(query::predicate_data_serializer_hook::PAGING_PREDICATE);
                }
 
                static void write_data(const query::paging_predicate<K, V> &obj, object_data_output &out) {
                    out.write_bytes(obj.outStream.toByteArray());
                    out.write<int32_t>((int32_t) obj.page);
                    out.write<int32_t>((int32_t) obj.pageSize);
                    out.write<std::string>(obj.IterationNames[static_cast<int32_t>(obj.iterationType)]);
                    out.write<int32_t>((int32_t) obj.anchor_data_list_.data_list.size());
                    const auto &data_list = obj.anchor_data_list_.data_list;
                    const auto &page_list = obj.anchor_data_list_.page_list;
                    for (size_t i = 0; i < obj.anchor_data_list_.data_list.size(); ++i) {
                        out.write<int32_t>(page_list[i]);
                        out.write_object<K>(data_list[i].first);
                        out.write_object<V>(data_list[i].second);
                    }
                }
 
                static query::paging_predicate<K, V> read_data(object_data_input &in) {
                    // Not need to read at the client side
                    BOOST_THROW_EXCEPTION(exception::hazelcast_serialization("readData",
                                                                                       "Client should not need to use readdata method!!!"));
                }
            };
 
        }
    }
}
 
#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif