circularBufferConcurrent.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2020-2025 European Southern Observatory (ESO)
//
// SPDX-License-Identifier: LGPL-3.0-only
 
/*
 * circularBufferConcurrent.hpp
 *
 *  Created on: Jul 7, 2021
 *      Author: hsommer
 */
#ifndef HLCC_OLDBMUX_CIRCULARBUFFERCONCURRENT_HPP
#define HLCC_OLDBMUX_CIRCULARBUFFERCONCURRENT_HPP
 
#include <mutex>
#include <condition_variable>
#include <functional>
#include <optional>
 
#include <log4cplus/logger.h>
#include <rad/logger.hpp>
 
#include <boost/circular_buffer.hpp>
 
 
// HSO TODO: Check async calls in ProcessRequest in ~/git-rtc/rtctk/componentFramework/services/oldb/src/oldbAdapter.cpp
 
 
namespace hlcc::oldbmux {

template<typename T>
class CircularBufferConcurrent
{
public:
 
    explicit CircularBufferConcurrent(const std::string& name, std::size_t capacity,
            const log4cplus::Logger& logger);
 
    ~CircularBufferConcurrent();
 
    CircularBufferConcurrent(const CircularBufferConcurrent&) = delete;
    CircularBufferConcurrent& operator=(const CircularBufferConcurrent&) = delete;

    void Push(T&& data);

    std::optional<T> Poll(std::chrono::milliseconds timeout = std::chrono::milliseconds::zero());
 
    void Clear();
 
    int Size() const;

    const std::unique_ptr<std::scoped_lock<std::recursive_mutex>> Lock() const;

    void SetDiscardListener(std::function<void(T&)> discard_listener);

    boost::circular_buffer<T>& GetCb();
 
    /*
     * Closes the buffer. This notifies (releases) any thread in the Poll method
     * (if the thread called Poll with a timeout)
     * and turns future invocations of methods Push and Poll into no-ops.
     * Existing data in the buffer gets discarded, with notification of client.
     * 
     * This method can be used to prepare a CircularBufferConcurrent for destruction,
     * for which no other thread must be in its code. 
     * 
     * The current implementation returns already while another thread may still be inside Poll.
     * This is OK for the use of the buffer inside CiiOlbdDataPointAsync, which has its own sync mechanism 
     * to wait for that thread.
     */
    void Close();
 
private:
    const log4cplus::Logger m_logger;
 
    /*
     * See Close().
     */
    std::atomic<bool> m_closed;

    boost::circular_buffer<T> cb;

    mutable std::recursive_mutex mutex;

    std::condition_variable_any buffer_not_empty;

    std::function<void(T&)> discard_listener;

    const std::string name;
 
};
 
} // end namespace hlcc::oldbmux
 
// include template class definition from separate header file.
#include <hlcc/oldbmux/circularBufferConcurrent.ipp>
 
#endif /* HLCC_OLDBMUX_CIRCULARBUFFERCONCURRENT_HPP */