Collaboration diagram for kcenon::pacs::client::remote_node_manager::impl:

Classes
struct	pooled_association

Public Member Functions
void	notify_status_change (std::string_view node_id, node_status status)

void	update_node_status (const std::string &node_id, node_status status, const std::string &error_msg="")

kcenon::pacs::VoidResult	perform_echo (const remote_node &node)

void	health_check_loop ()

void	load_nodes_from_repo ()

Public Attributes
std::shared_ptr< storage::node_repository >	repo

std::shared_ptr< di::ILogger >	logger

node_manager_config	config

node_status_callback	status_callback

std::mutex	callback_mutex

std::unordered_map< std::string, remote_node >	node_cache

std::mutex	cache_mutex

std::unordered_map< std::string, node_statistics >	statistics

std::mutex	stats_mutex

std::unordered_map< std::string, std::deque< pooled_association > >	connection_pool

std::mutex	pool_mutex

std::atomic< bool >	health_check_running {false}

std::thread	health_check_thread

std::condition_variable	health_check_cv

std::mutex	health_check_mutex

Detailed Description

Definition at line 47 of file remote_node_manager.cpp.

Member Function Documentation

◆ health_check_loop()

void kcenon::pacs::client::remote_node_manager::impl::health_check_loop ( )

inline

Definition at line 211 of file remote_node_manager.cpp.

                             {
        while (health_check_running.load()) {
            // Verify all nodes
            std::vector<std::string> node_ids;
            {
                std::lock_guard<std::mutex> lock(cache_mutex);
                for (const auto& [id, _] : node_cache) {
                    node_ids.push_back(id);
                }
            }
 
            for (const auto& id : node_ids) {
                if (!health_check_running.load()) break;
 
                std::optional<remote_node> node;
                {
                    std::lock_guard<std::mutex> lock(cache_mutex);
                    auto it = node_cache.find(id);
                    if (it != node_cache.end()) {
                        node = it->second;
                    }
                }
 
                if (node) {
                    update_node_status(id, node_status::verifying);
 
                    auto result = perform_echo(*node);
                    if (result.is_ok()) {
                        update_node_status(id, node_status::online);
 
                        // Update statistics
                        std::lock_guard<std::mutex> lock(stats_mutex);
                        statistics[id].successful_operations++;
                        statistics[id].last_activity = std::chrono::system_clock::now();
                    } else {
                        update_node_status(id, node_status::offline, result.error().message);
 
                        // Update statistics
                        std::lock_guard<std::mutex> lock(stats_mutex);
                        statistics[id].failed_operations++;
                    }
                }
            }
 
            // Wait for next interval
            std::unique_lock<std::mutex> lock(health_check_mutex);
            health_check_cv.wait_for(lock, config.health_check_interval, [this] {
                return !health_check_running.load();
            });
        }
    }

References cache_mutex, config, health_check_cv, kcenon::pacs::client::node_manager_config::health_check_interval, health_check_mutex, health_check_running, node_cache, kcenon::pacs::client::offline, kcenon::pacs::client::online, perform_echo(), stats_mutex, update_node_status(), and kcenon::pacs::client::verifying.

Referenced by kcenon::pacs::client::remote_node_manager::start_health_check().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ load_nodes_from_repo()

void kcenon::pacs::client::remote_node_manager::impl::load_nodes_from_repo ( )

inline

Definition at line 263 of file remote_node_manager.cpp.

                                {
        if (!repo) return;
 
#ifdef PACS_WITH_DATABASE_SYSTEM
        auto nodes_result = repo->find_all();
        if (nodes_result.is_err()) return;
 
        std::lock_guard<std::mutex> lock(cache_mutex);
        for (auto& node : nodes_result.value()) {
            node_cache[node.node_id] = std::move(node);
        }
#else
        auto nodes = repo->find_all();
        std::lock_guard<std::mutex> lock(cache_mutex);
        for (auto& node : nodes) {
            node_cache[node.node_id] = std::move(node);
        }
#endif
    }

References cache_mutex, node_cache, and repo.

Referenced by kcenon::pacs::client::remote_node_manager::remote_node_manager().

Here is the caller graph for this function:

◆ notify_status_change()

void kcenon::pacs::client::remote_node_manager::impl::notify_status_change	(	std::string_view	node_id,
		node_status	status )

inline

Definition at line 87 of file remote_node_manager.cpp.

                                                                          {
        std::lock_guard<std::mutex> lock(callback_mutex);
        if (status_callback) {
            status_callback(node_id, status);
        }
    }

References callback_mutex, and status_callback.

Referenced by update_node_status().

Here is the caller graph for this function:

◆ perform_echo()

kcenon::pacs::VoidResult kcenon::pacs::client::remote_node_manager::impl::perform_echo ( const remote_node & node )

inline

Definition at line 123 of file remote_node_manager.cpp.

                                                               {
        using namespace network;
 
        // Build association config
        association_config assoc_config;
        assoc_config.calling_ae_title = config.local_ae_title;
        assoc_config.called_ae_title = node.ae_title;
        assoc_config.proposed_contexts.push_back({
            1,
            std::string(verification_sop_class_uid),
            default_transfer_syntaxes
        });
 
        // Connect
        auto connect_result = association::connect(
            node.host,
            node.port,
            assoc_config,
            std::chrono::duration_cast<association::duration>(node.connection_timeout)
        );
 
        if (connect_result.is_err()) {
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::connection_failed,
                "Failed to connect to node: " + node.node_id,
                connect_result.error().message);
        }
 
        auto& assoc = connect_result.value();
 
        // Get accepted context
        auto context_id = assoc.accepted_context_id(verification_sop_class_uid);
        if (!context_id) {
            assoc.abort();
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::no_acceptable_context,
                "Verification SOP Class not accepted by " + node.node_id);
        }
 
        // Build C-ECHO-RQ
        auto echo_rq = dimse::make_c_echo_rq(1);
 
        // Send request
        auto send_result = assoc.send_dimse(*context_id, echo_rq);
        if (send_result.is_err()) {
            assoc.abort();
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::send_failed,
                "Failed to send C-ECHO-RQ to " + node.node_id);
        }
 
        // Receive response
        auto recv_result = assoc.receive_dimse(
            std::chrono::duration_cast<association::duration>(node.dimse_timeout)
        );
 
        if (recv_result.is_err()) {
            assoc.abort();
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::receive_failed,
                "Failed to receive C-ECHO-RSP from " + node.node_id);
        }
 
        const auto& [recv_context_id, response] = recv_result.value();
 
        // Verify response
        if (response.command() != dimse::command_field::c_echo_rsp) {
            assoc.abort();
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::dimse_error,
                "Unexpected response from " + node.node_id);
        }
 
        // Check status (status_success is a constexpr, not enum member)
        if (response.status() != dimse::status_success) {
            [[maybe_unused]] auto release_result = assoc.release();
            return kcenon::pacs::pacs_void_error(
                kcenon::pacs::error_codes::dimse_error,
                "C-ECHO failed with status: " +
                std::to_string(static_cast<uint16_t>(response.status())));
        }
 
        // Release association
        [[maybe_unused]] auto release_result = assoc.release();
 
        return kcenon::pacs::ok();
    }

References kcenon::pacs::client::remote_node::ae_title, config, kcenon::pacs::error_codes::connection_failed, kcenon::pacs::client::remote_node::connection_timeout, kcenon::pacs::error_codes::dimse_error, kcenon::pacs::client::remote_node::dimse_timeout, kcenon::pacs::client::remote_node::host, kcenon::pacs::client::node_manager_config::local_ae_title, kcenon::pacs::error_codes::no_acceptable_context, kcenon::pacs::client::remote_node::node_id, kcenon::pacs::pacs_void_error(), kcenon::pacs::client::remote_node::port, kcenon::pacs::error_codes::receive_failed, and kcenon::pacs::error_codes::send_failed.

Referenced by health_check_loop(), and kcenon::pacs::client::remote_node_manager::verify_node().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ update_node_status()

void kcenon::pacs::client::remote_node_manager::impl::update_node_status	(	const std::string &	node_id,
		node_status	status,
		const std::string &	error_msg = "" )

inline

Definition at line 94 of file remote_node_manager.cpp.

                                                             {
        // Update cache
        {
            std::lock_guard<std::mutex> lock(cache_mutex);
            auto it = node_cache.find(node_id);
            if (it != node_cache.end()) {
                auto old_status = it->second.status;
                it->second.status = status;
                if (status == node_status::online) {
                    it->second.last_verified = std::chrono::system_clock::now();
                } else if (status == node_status::error || status == node_status::offline) {
                    it->second.last_error = std::chrono::system_clock::now();
                    it->second.last_error_message = error_msg;
                }
 
                // Notify if status changed
                if (old_status != status) {
                    notify_status_change(node_id, status);
                }
            }
        }
 
        // Update repository (ignore result as this is best-effort)
        if (repo) {
            [[maybe_unused]] auto result = repo->update_status(node_id, status, error_msg);
        }
    }