ringbuffer.h

/*
 * Copyright (c) 2008-2025, 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"
#include "hazelcast/client/ringbuffer/read_result_set.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_type*>(
            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)
    {}
};
} // namespace client
} // namespace hazelcast
 
#if defined(WIN32) || defined(_WIN32) || defined(WIN64) || defined(_WIN64)
#pragma warning(pop)
#endif