ringbuffer.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 "hazelcast/client/proxy/RingbufferImpl.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 ringbuffer : public proxy::RingbufferImpl {
            friend class spi::ProxyManager;
        public:
            template<typename E>
            boost::future<int64_t> add(const E &item) {
                return add_data(to_data(item));
            }
 
            template<typename E>
            boost::future<boost::optional<E>> read_one(int64_t sequence) {
                return to_object<E>(read_one_data(sequence));
            }
 
            template<typename E>
            boost::future<int64_t> add(const E &item, rb::overflow_policy overflow_policy) {
                return addData(to_data(item), overflow_policy);
            }
 
            template<typename E>
            boost::future<int64_t>
            add_all(const std::vector<E> &items, rb::overflow_policy overflow_policy) {
                return add_all_data(to_data_collection(items), overflow_policy);
            }
 
            template<typename IFUNCTION>
            boost::future<rb::read_result_set>
            read_many(int64_t start_sequence, int32_t min_count, int32_t max_count, const IFUNCTION *filter = nullptr) {
                auto filter_data = to_data<IFUNCTION>(filter);
                return read_many_data(start_sequence, min_count, max_count, &filter_data).then(boost::launch::sync, [=] (boost::future<protocol::ClientMessage> f) {
                    return get_result_set(std::move(f));
                });
            }
 
            rb::read_result_set get_result_set(boost::future<protocol::ClientMessage> f) {
                auto msg = f.get();
                auto *initial_frame = reinterpret_cast<protocol::ClientMessage::frame_header_t *>(msg.rd_ptr(
                        protocol::ClientMessage::RESPONSE_HEADER_LEN));
                auto read_count = msg.get<int32_t>();
                auto next_seq = msg.get<int64_t>();
                msg.rd_ptr(
                        static_cast<int32_t>(initial_frame->frame_len) - protocol::ClientMessage::RESPONSE_HEADER_LEN -
                        protocol::ClientMessage::INT32_SIZE -
                        protocol::ClientMessage::INT64_SIZE);
 
                auto datas = msg.get<std::vector<serialization::pimpl::data>>();
                auto item_seqs = msg.get_nullable<std::vector<int64_t>>();
                return rb::read_result_set(read_count, std::move(datas), get_serialization_service(), item_seqs,
                                           next_seq);
            }
 
            boost::future<rb::read_result_set>
            read_many(int64_t start_sequence, int32_t min_count, int32_t max_count) {
                return read_many_data(start_sequence, min_count, max_count, nullptr).then(boost::launch::sync, [=] (boost::future<protocol::ClientMessage> f) {
                    return get_result_set(std::move(f));
                });
            }
 
        private:
            ringbuffer(const std::string &object_name, spi::ClientContext *context) : RingbufferImpl(object_name,
                                                                                                     context) {}
        };
    }
}
 
#if  defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif