queued_writer_base.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "decorator_writer_base.h"
#include "../interfaces/log_entry.h"
#include "../interfaces/writer_category.h"
#include <mutex>
#include <memory>
#include <atomic>
#include <queue>
#include <vector>
 
namespace kcenon::logger {
 
inline log_entry copy_log_entry(const log_entry& entry) {
    if (entry.location) {
        return log_entry(entry.level,
                        entry.message.to_string(),
                        entry.location->file.to_string(),
                        entry.location->line,
                        entry.location->function.to_string(),
                        entry.timestamp);
    }
    return log_entry(entry.level,
                    entry.message.to_string(),
                    entry.timestamp);
}
 
template <typename Container>
class queued_writer_base : public decorator_writer_base, public async_writer_tag {
public:
    explicit queued_writer_base(std::unique_ptr<log_writer_interface> wrapped_writer,
                                std::size_t max_queue_size,
                                std::string_view decorator_name)
        : decorator_writer_base(std::move(wrapped_writer), decorator_name)
        , max_queue_size_(max_queue_size)
        , shutting_down_(false) {
    }
 
    ~queued_writer_base() override = default;
 
    bool is_healthy() const override {
        return !shutting_down_ && wrapped().is_healthy();
    }
 
    std::size_t get_queue_size() const {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        return get_container_size(queue_);
    }
 
    std::size_t get_max_queue_size() const {
        return max_queue_size_;
    }
 
protected:
    common::VoidResult try_enqueue(const log_entry& entry) {
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        if (get_container_size(queue_) >= max_queue_size_) {
            return handle_overflow(entry);
        }
 
        enqueue_entry(queue_, entry);
        on_entry_enqueued();
        return common::ok();
    }
 
    virtual common::VoidResult handle_overflow(const log_entry& /*entry*/) {
        return make_logger_void_result(logger_error_code::queue_full, "Queue is full");
    }
 
    virtual void on_entry_enqueued() {}
 
    template <typename T>
    static std::size_t get_container_size(const std::queue<T>& container) {
        return container.size();
    }
 
    template <typename T>
    static std::size_t get_container_size(const std::vector<T>& container) {
        return container.size();
    }
 
    template <typename T>
    static void enqueue_entry(std::queue<T>& container, const log_entry& entry) {
        container.push(copy_log_entry(entry));
    }
 
    template <typename T>
    static void enqueue_entry(std::vector<T>& container, const log_entry& entry) {
        container.push_back(copy_log_entry(entry));
    }
 
    std::size_t max_queue_size_;
 
    mutable std::mutex queue_mutex_;
    Container queue_;
 
    std::atomic<bool> shutting_down_;
};
 
} // namespace kcenon::logger