Real-time data stream processing. More...

#include <async_operations.h>

Collaboration diagram for database::async::stream_processor:

Classes
struct	stream_event

Public Types
enum class	stream_type { postgresql_notify , mongodb_change_stream , redis_pubsub , custom }

Public Member Functions
	stream_processor (std::shared_ptr< core::database_backend > db)

	~stream_processor ()

bool	start_stream (stream_type type, const std::string &channel)

bool	stop_stream (const std::string &channel)

void	stop_all_streams ()

template<concepts::StreamEventHandler< stream_event > Handler>
void	register_event_handler (const std::string &channel, Handler &&handler)

template<concepts::StreamEventHandler< stream_event > Handler>
void	register_global_handler (Handler &&handler)

void	register_event_handler (const std::string &channel, std::function< void(const stream_event &)> handler)

void	register_global_handler (std::function< void(const stream_event &)> handler)

template<concepts::StreamEventFilter< stream_event > Filter>
void	add_event_filter (const std::string &channel, Filter &&filter)

void	add_event_filter (const std::string &channel, std::function< bool(const stream_event &)> filter)

Private Member Functions
void	stream_thread (const std::string &channel, stream_type type)

void	process_event (const stream_event &event)

Private Attributes
std::shared_ptr< core::database_backend >	db_

std::mutex	threads_mutex_

std::unordered_map< std::string, std::thread >	stream_threads_

std::mutex	handlers_mutex_

std::unordered_map< std::string, std::function< void(const stream_event &)> >	event_handlers_

std::vector< std::function< void(const stream_event &)> >	global_handlers_

std::unordered_map< std::string, std::function< bool(const stream_event &)> >	event_filters_

std::atomic< bool >	running_ {true}

Detailed Description

Real-time data stream processing.

Thread-safety: All public methods are thread-safe. Event handlers are called from dedicated stream threads, so handlers must be thread-safe if they access shared state. The handlers_mutex_ protects all handler registration and filter operations.

Definition at line 563 of file async_operations.h.

Member Enumeration Documentation

◆ stream_type

enum class database::async::stream_processor::stream_type

strong

Enumerator
postgresql_notify
mongodb_change_stream
redis_pubsub
custom

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 566 of file async_operations.h.

                               {
            postgresql_notify,
            mongodb_change_stream,
            redis_pubsub,
            custom
        };

Constructor & Destructor Documentation

◆ stream_processor()

database::async::stream_processor::stream_processor ( std::shared_ptr< core::database_backend > db )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1142 of file async_operations.h.

        : db_(std::move(db))
    {
    }

◆ ~stream_processor()

database::async::stream_processor::~stream_processor ( )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1148 of file async_operations.h.

    {
        stop_all_streams();
    }

References stop_all_streams().

Here is the call graph for this function:

Member Function Documentation

◆ add_event_filter() [1/2]

template<concepts::StreamEventFilter< stream_event > Filter>

void database::async::stream_processor::add_event_filter	(	const std::string &	channel,
		Filter &&	filter )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 613 of file async_operations.h.

        {
            std::lock_guard<std::mutex> lock(handlers_mutex_);
            event_filters_[channel] = std::forward<Filter>(filter);
        }

References event_filters_, and handlers_mutex_.

◆ add_event_filter() [2/2]

void database::async::stream_processor::add_event_filter	(	const std::string &	channel,
		std::function< bool(const stream_event &)>	filter )

inline

Definition at line 1216 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        event_filters_[channel] = std::move(filter);
    }

References event_filters_, and handlers_mutex_.

◆ process_event()

void database::async::stream_processor::process_event ( const stream_event & event )

inlineprivate

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1245 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
 
        auto filter_it = event_filters_.find(event.channel);
        if (filter_it != event_filters_.end()) {
            if (!filter_it->second(event)) {
                return;
            }
        }
 
        auto handler_it = event_handlers_.find(event.channel);
        if (handler_it != event_handlers_.end()) {
            handler_it->second(event);
        }
 
        for (const auto& handler : global_handlers_) {
            handler(event);
        }
    }

References database::async::stream_processor::stream_event::channel, event_filters_, event_handlers_, global_handlers_, and handlers_mutex_.

Referenced by stream_thread().

Here is the caller graph for this function:

◆ register_event_handler() [1/2]

template<concepts::StreamEventHandler< stream_event > Handler>

void database::async::stream_processor::register_event_handler	(	const std::string &	channel,
		Handler &&	handler )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 592 of file async_operations.h.

        {
            std::lock_guard<std::mutex> lock(handlers_mutex_);
            event_handlers_[channel] = std::forward<Handler>(handler);
        }

References event_handlers_, and handlers_mutex_.

◆ register_event_handler() [2/2]

void database::async::stream_processor::register_event_handler	(	const std::string &	channel,
		std::function< void(const stream_event &)>	handler )

inline

Definition at line 1201 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        event_handlers_[channel] = std::move(handler);
    }

References event_handlers_, and handlers_mutex_.

◆ register_global_handler() [1/2]

template<concepts::StreamEventHandler< stream_event > Handler>

void database::async::stream_processor::register_global_handler ( Handler && handler )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 599 of file async_operations.h.

        {
            std::lock_guard<std::mutex> lock(handlers_mutex_);
            global_handlers_.push_back(std::forward<Handler>(handler));
        }

References global_handlers_, and handlers_mutex_.

◆ register_global_handler() [2/2]

void database::async::stream_processor::register_global_handler ( std::function< void(const stream_event &)> handler )

inline

Definition at line 1209 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(handlers_mutex_);
        global_handlers_.push_back(std::move(handler));
    }

References global_handlers_, and handlers_mutex_.

◆ start_stream()

bool database::async::stream_processor::start_stream	(	stream_type	type,
		const std::string &	channel )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1153 of file async_operations.h.

    {
        std::lock_guard<std::mutex> lock(threads_mutex_);
        if (stream_threads_.count(channel) > 0) {
            return false;
        }
        stream_threads_.emplace(
            channel,
            std::thread(&stream_processor::stream_thread, this, channel, type));
        return true;
    }

References stream_thread(), stream_threads_, and threads_mutex_.

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ stop_all_streams()

void database::async::stream_processor::stop_all_streams ( )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1185 of file async_operations.h.

    {
        running_.store(false);
        std::unordered_map<std::string, std::thread> threads;
        {
            std::lock_guard<std::mutex> lock(threads_mutex_);
            threads.swap(stream_threads_);
        }
        for (auto& [channel, t] : threads) {
            if (t.joinable()) {
                t.join();
            }
        }
        running_.store(true);
    }

References running_, stream_threads_, and threads_mutex_.

Referenced by ~stream_processor().

Here is the caller graph for this function:

◆ stop_stream()

bool database::async::stream_processor::stop_stream ( const std::string & channel )

inline

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1166 of file async_operations.h.

    {
        std::thread t;
        {
            std::lock_guard<std::mutex> lock(threads_mutex_);
            auto it = stream_threads_.find(channel);
            if (it == stream_threads_.end()) {
                return false;
            }
            t = std::move(it->second);
            stream_threads_.erase(it);
        }
        // Thread detects its channel removal via map-check and exits
        if (t.joinable()) {
            t.join();
        }
        return true;
    }

References stream_threads_, and threads_mutex_.

◆ stream_thread()

void database::async::stream_processor::stream_thread	(	const std::string &	channel,
		stream_type	type )

inlineprivate

Examples: /home/runner/work/database_system/database_system/database/async/async_operations.h.

Definition at line 1224 of file async_operations.h.

    {
        stream_event connected_event;
        connected_event.type = type;
        connected_event.channel = channel;
        connected_event.payload = "stream_connected";
        connected_event.timestamp = std::chrono::system_clock::now();
        process_event(connected_event);
 
        while (running_.load()) {
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
            {
                std::lock_guard<std::mutex> lock(threads_mutex_);
                if (stream_threads_.find(channel) == stream_threads_.end()) {
                    return;
                }
            }
        }
    }