Asynchronous wrapper for log writers. More...

#include <async_writer.h>

Inheritance diagram for kcenon::logger::async_writer:

Collaboration diagram for kcenon::logger::async_writer:

Public Member Functions
	async_writer (std::unique_ptr< log_writer_interface > wrapped_writer, std::size_t queue_size=10000, std::chrono::seconds flush_timeout=std::chrono::seconds(5))
	Constructor accepting log_writer_interface (Decorator pattern)

	~async_writer () override
	Destructor.

void	start ()
	Start the async writer thread.

void	stop (bool force_flush=true)
	Stop the async writer thread.

common::VoidResult	write (const log_entry &entry) override
	Write a log entry asynchronously.

common::VoidResult	flush () override
	Flush all pending messages.

bool	is_healthy () const override
	Check if the writer is healthy.

std::string	get_name () const override
	Get the name of this writer.

Public Member Functions inherited from kcenon::logger::queued_writer_base< std::queue< log_entry > >
	queued_writer_base (std::unique_ptr< log_writer_interface > wrapped_writer, std::size_t max_queue_size, std::string_view decorator_name)
	Constructor for queued writer base.

	~queued_writer_base () override=default
	Virtual destructor.

bool	is_healthy () const override
	Check if the writer is healthy.

std::size_t	get_queue_size () const
	Get the current queue size.

std::size_t	get_max_queue_size () const
	Get maximum queue size.

Public Member Functions inherited from kcenon::logger::decorator_writer_base
	decorator_writer_base (std::unique_ptr< log_writer_interface > wrapped, std::string_view decorator_name)
	Construct a decorator writer base.

	~decorator_writer_base () override=default
	Virtual destructor for proper cleanup.

	decorator_writer_base (const decorator_writer_base &)=delete

decorator_writer_base &	operator= (const decorator_writer_base &)=delete

	decorator_writer_base (decorator_writer_base &&) noexcept=default

decorator_writer_base &	operator= (decorator_writer_base &&) noexcept=default

common::VoidResult	flush () override
	Flush the wrapped writer.

std::string	get_name () const override
	Get the name of this writer.

const log_writer_interface *	get_wrapped_writer () const noexcept
	Get the wrapped writer (const version)

Public Member Functions inherited from kcenon::logger::log_writer_interface
virtual	~log_writer_interface ()=default

virtual common::VoidResult	close ()
	Close the writer and release resources.

virtual auto	is_open () const -> bool
	Check if writer is open and ready.

Protected Member Functions
common::VoidResult	handle_overflow (const log_entry &) override
	Handle queue overflow.

void	on_entry_enqueued () override
	Called after entry is enqueued - notifies worker thread.

Protected Member Functions inherited from kcenon::logger::queued_writer_base< std::queue< log_entry > >
common::VoidResult	try_enqueue (const log_entry &entry)
	Try to enqueue an entry with overflow protection.

Protected Member Functions inherited from kcenon::logger::decorator_writer_base
log_writer_interface &	wrapped () noexcept
	Access the wrapped writer (non-const)

const log_writer_interface &	wrapped () const noexcept
	Access the wrapped writer (const)

const std::string &	decorator_name () const noexcept
	Get the decorator name.

Private Types
using	base_type = queued_writer_base<std::queue<log_entry>>

Private Member Functions
void	process_messages ()
	Process messages from the queue.

void	flush_remaining ()
	Flush any remaining messages after stopping.

Private Attributes
std::chrono::seconds	flush_timeout_

std::condition_variable	queue_cv_

std::condition_variable	flush_cv_

std::atomic< bool >	running_

std::thread	worker_thread_

Additional Inherited Members
Static Public Attributes inherited from kcenon::logger::decorator_writer_tag
static constexpr writer_category	category = writer_category::decorator

Static Public Attributes inherited from kcenon::logger::async_writer_tag
static constexpr writer_category	category = writer_category::asynchronous

Static Protected Member Functions inherited from kcenon::logger::queued_writer_base< std::queue< log_entry > >
static std::size_t	get_container_size (const std::queue< T > &container)
	Get container size (specialization for std::queue)

static std::size_t	get_container_size (const std::vector< T > &container)
	Get container size (specialization for std::vector)

static void	enqueue_entry (std::queue< T > &container, const log_entry &entry)
	Enqueue entry (specialization for std::queue)

static void	enqueue_entry (std::vector< T > &container, const log_entry &entry)
	Enqueue entry (specialization for std::vector)

Protected Attributes inherited from kcenon::logger::queued_writer_base< std::queue< log_entry > >
std::size_t	max_queue_size_

std::mutex	queue_mutex_

std::queue< log_entry >	queue_

std::atomic< bool >	shutting_down_

Detailed Description

Asynchronous wrapper for log writers.

This class wraps any base_writer implementation and provides asynchronous writing capabilities using a background thread. It extends queued_writer_base to share common queue management logic with batch_writer.

Category: Asynchronous (non-blocking), Decorator (wraps another writer)

Since: 1.4.0 Added async_writer_tag and decorator_writer_tag for classification; 4.0.0 Refactored to use queued_writer_base

Examples: decorator_usage.cpp, and writer_builder_example.cpp.

Definition at line 37 of file async_writer.h.

Member Typedef Documentation

◆ base_type

using kcenon::logger::async_writer::base_type = queued_writer_base<std::queue<log_entry>>

private

Definition at line 38 of file async_writer.h.

Constructor & Destructor Documentation

◆ async_writer()

kcenon::logger::async_writer::async_writer	(	std::unique_ptr< log_writer_interface >	wrapped_writer,
		std::size_t	queue_size = 10000,
		std::chrono::seconds	flush_timeout = std::chrono::seconds(5) )

inlineexplicit

Constructor accepting log_writer_interface (Decorator pattern)

Parameters

wrapped_writer	The writer to wrap with async functionality
queue_size	Maximum queue size for pending messages
flush_timeout	Maximum time to wait for flush operation (in seconds)

Definition at line 47 of file async_writer.h.

        : base_type(std::move(wrapped_writer), queue_size, "async")
        , flush_timeout_(flush_timeout)
        , running_(false) {
    }

◆ ~async_writer()

kcenon::logger::async_writer::~async_writer ( )

inlineoverride

Destructor.

Definition at line 58 of file async_writer.h.

                             {
        stop();
    }

References stop().

Here is the call graph for this function:

Member Function Documentation

◆ flush()

common::VoidResult kcenon::logger::async_writer::flush ( )

inlineoverridevirtual

Flush all pending messages.

Returns: common::VoidResult indicating success or error

Implements kcenon::logger::log_writer_interface.

Definition at line 140 of file async_writer.h.

                                      {
        if (!running_) {
            return wrapped().flush();
        }
 
        // Wait for the queue to be empty with a timeout to prevent indefinite blocking
        std::unique_lock<std::mutex> lock(queue_mutex_);
        bool flushed = flush_cv_.wait_for(lock, flush_timeout_, [this]() {
            return queue_.empty();
        });
 
        if (!flushed) {
            // Timeout occurred - worker thread may have exited or is blocked
            return make_logger_void_result(logger_error_code::flush_timeout,
                                           "Flush operation timed out after " +
                                           std::to_string(flush_timeout_.count()) + " seconds");
        }
 
        // Flush the wrapped writer
        return wrapped().flush();
    }

References kcenon::logger::log_writer_interface::flush(), flush_cv_, kcenon::logger::flush_timeout, flush_timeout_, kcenon::logger::make_logger_void_result(), kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_mutex_, running_, and kcenon::logger::decorator_writer_base::wrapped().

Here is the call graph for this function:

◆ flush_remaining()

void kcenon::logger::async_writer::flush_remaining ( )

inlineprivate

Flush any remaining messages after stopping.

Definition at line 227 of file async_writer.h.

                           {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        while (!queue_.empty()) {
            auto entry = std::move(queue_.front());
            queue_.pop();
            wrapped().write(entry);
        }
        wrapped().flush();
    }

References kcenon::logger::log_writer_interface::flush(), kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_mutex_, kcenon::logger::decorator_writer_base::wrapped(), and kcenon::logger::log_writer_interface::write().

Here is the call graph for this function:

◆ get_name()

std::string kcenon::logger::async_writer::get_name ( ) const

inlineoverridevirtual

Get the name of this writer.

Returns: Writer name

Implements kcenon::logger::log_writer_interface.

Definition at line 174 of file async_writer.h.

                                        {
        return "async_" + wrapped().get_name();
    }

References kcenon::logger::log_writer_interface::get_name(), and kcenon::logger::decorator_writer_base::wrapped().

Here is the call graph for this function:

◆ handle_overflow()

common::VoidResult kcenon::logger::async_writer::handle_overflow ( const log_entry & )

inlineoverrideprotectedvirtual

Handle queue overflow.

Parameters

entry The entry that couldn't be enqueued

Returns: Error result indicating queue full

Reimplemented from kcenon::logger::queued_writer_base< std::queue< log_entry > >.

Definition at line 184 of file async_writer.h.

                                                                  {
        return make_logger_void_result(logger_error_code::queue_full, "Async writer queue is full");
    }

References kcenon::logger::make_logger_void_result(), and kcenon::logger::queue_full.

Here is the call graph for this function:

◆ is_healthy()

bool kcenon::logger::async_writer::is_healthy ( ) const

inlineoverridevirtual

Check if the writer is healthy.

Returns: true if healthy, false otherwise

Implements kcenon::logger::log_writer_interface.

Definition at line 166 of file async_writer.h.

                                     {
        return wrapped().is_healthy() && running_;
    }

References kcenon::logger::log_writer_interface::is_healthy(), running_, and kcenon::logger::decorator_writer_base::wrapped().

Here is the call graph for this function:

◆ on_entry_enqueued()

void kcenon::logger::async_writer::on_entry_enqueued ( )

inlineoverrideprotectedvirtual

Called after entry is enqueued - notifies worker thread.

Reimplemented from kcenon::logger::queued_writer_base< std::queue< log_entry > >.

Definition at line 191 of file async_writer.h.

                                      {
        queue_cv_.notify_one();
    }

References queue_cv_.

◆ process_messages()

void kcenon::logger::async_writer::process_messages ( )

inlineprivate

Process messages from the queue.

Definition at line 199 of file async_writer.h.

                            {
        while (running_) {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for messages or stop signal
            queue_cv_.wait(lock, [this]() {
                return !queue_.empty() || !running_;
            });
 
            // Process all available messages
            while (!queue_.empty()) {
                auto entry = std::move(queue_.front());
                queue_.pop();
 
                // Unlock while writing
                lock.unlock();
                wrapped().write(entry);
                lock.lock();
            }
 
            // Notify flush waiters
            flush_cv_.notify_all();
        }
    }

References flush_cv_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_, queue_cv_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_mutex_, running_, kcenon::logger::decorator_writer_base::wrapped(), and kcenon::logger::log_writer_interface::write().

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ start()

void kcenon::logger::async_writer::start ( )

inline

Start the async writer thread.

Definition at line 65 of file async_writer.h.

                 {
        // Use compare_exchange to safely check and set running_ flag
        bool expected = false;
        if (!running_.compare_exchange_strong(expected, true)) {
            return; // Already running
        }
 
        // Try to create the worker thread with proper error handling
        try {
            worker_thread_ = std::thread([this]() {
                process_messages();
            });
        } catch (const std::exception&) {
            // Thread creation failed, rollback the running_ flag
            running_.store(false);
            throw; // Re-throw to notify caller
        }
    }

References process_messages(), running_, and worker_thread_.

Here is the call graph for this function:

◆ stop()

void kcenon::logger::async_writer::stop ( bool force_flush = true )

inline

Stop the async writer thread.

Parameters

force_flush If true, process all remaining messages before stopping

Definition at line 88 of file async_writer.h.

                                       {
        if (!running_.exchange(false)) {
            return; // Already stopped
        }
 
        // Process remaining messages if requested
        if (force_flush) {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            size_t remaining_count = queue_.size();
            if (remaining_count > 0) {
                std::cerr << "[async_writer] Info: Processing " << remaining_count
                          << " remaining messages before shutdown.\n";
            }
        }
 
        // Signal the worker thread to stop
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            queue_cv_.notify_all();
        }
 
        // Wait for the worker thread to finish (infinite wait - shutdown prioritizes safety)
        if (worker_thread_.joinable()) {
            worker_thread_.join();
        }
 
        // Verify all messages were processed
        if (!queue_.empty()) {
            std::cerr << "[async_writer] Warning: " << queue_.size()
                      << " messages were not processed during shutdown.\n";
        }
    }

References kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_, queue_cv_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::queue_mutex_, running_, and worker_thread_.

Referenced by ~async_writer().

Here is the caller graph for this function:

◆ write()

common::VoidResult kcenon::logger::async_writer::write ( const log_entry & entry )

inlineoverridevirtual

Write a log entry asynchronously.

Parameters

entry The log entry to write

Returns: common::VoidResult indicating success or error

Since: 3.5.0 Changed to use log_entry directly

Implements kcenon::logger::decorator_writer_base.

Definition at line 127 of file async_writer.h.

                                                            {
        if (!running_) {
            // If not running, write directly
            return wrapped().write(entry);
        }
 
        return try_enqueue(entry);
    }

References running_, kcenon::logger::queued_writer_base< std::queue< log_entry > >::try_enqueue(), kcenon::logger::decorator_writer_base::wrapped(), and kcenon::logger::log_writer_interface::write().

Here is the call graph for this function:

Member Data Documentation

◆ flush_cv_

std::condition_variable kcenon::logger::async_writer::flush_cv_

private

Definition at line 240 of file async_writer.h.

Referenced by flush(), and process_messages().

◆ flush_timeout_

std::chrono::seconds kcenon::logger::async_writer::flush_timeout_

private

Definition at line 237 of file async_writer.h.

Referenced by flush().

◆ queue_cv_

std::condition_variable kcenon::logger::async_writer::queue_cv_

private

Definition at line 239 of file async_writer.h.

Referenced by on_entry_enqueued(), process_messages(), and stop().

◆ running_

std::atomic<bool> kcenon::logger::async_writer::running_

private

Definition at line 242 of file async_writer.h.

Referenced by flush(), is_healthy(), process_messages(), start(), stop(), and write().

◆ worker_thread_

std::thread kcenon::logger::async_writer::worker_thread_

private

Definition at line 243 of file async_writer.h.

Referenced by start(), and stop().

The documentation for this class was generated from the following file:

include/kcenon/logger/writers/async_writer.h

Public Member Functions

Protected Member Functions

Private Types

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Typedef Documentation

◆ base_type

Constructor & Destructor Documentation

◆ async_writer()

◆ ~async_writer()

Member Function Documentation

◆ flush()

◆ flush_remaining()

◆ get_name()

◆ handle_overflow()

◆ is_healthy()

◆ on_entry_enqueued()

◆ process_messages()

◆ start()

◆ stop()

◆ write()

Member Data Documentation

◆ flush_cv_

◆ flush_timeout_

◆ queue_cv_

◆ running_

◆ worker_thread_