pipeline_adapter.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/pacs/network/pipeline/pipeline_coordinator.h>
#include <kcenon/pacs/network/pipeline/pipeline_job_types.h>
#include <kcenon/pacs/network/pipeline/jobs/receive_network_io_job.h>
#include <kcenon/pacs/network/pipeline/jobs/send_network_io_job.h>
#include <kcenon/pacs/network/pipeline/jobs/storage_query_exec_job.h>
#include <kcenon/pacs/core/result.h>
 
#include <chrono>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
 
namespace kcenon::pacs::network::pipeline {
 
struct session_context {
    uint64_t session_id;
 
    std::string remote_ae_title;
 
    std::string local_ae_title;
 
    uint32_t max_pdu_size{16384};
 
    std::chrono::steady_clock::time_point created_at;
 
    std::chrono::steady_clock::time_point last_activity;
};
 
class pipeline_adapter {
public:
    using send_callback = std::function<VoidResult(uint64_t session_id,
                                                   const std::vector<uint8_t>& data)>;
 
    using association_callback = std::function<void(uint64_t session_id,
                                                    kcenon::pacs::network::pdu_type type,
                                                    const std::vector<uint8_t>& data)>;
 
    using session_event_callback = std::function<void(uint64_t session_id,
                                                      const std::string& event)>;
 
    // =========================================================================
    // Construction
    // =========================================================================
 
    pipeline_adapter();
 
    explicit pipeline_adapter(const pipeline_config& config);
 
    ~pipeline_adapter();
 
    // Non-copyable, non-movable
    pipeline_adapter(const pipeline_adapter&) = delete;
    pipeline_adapter& operator=(const pipeline_adapter&) = delete;
    pipeline_adapter(pipeline_adapter&&) = delete;
    pipeline_adapter& operator=(pipeline_adapter&&) = delete;
 
    // =========================================================================
    // Lifecycle Management
    // =========================================================================
 
    [[nodiscard]] auto start() -> VoidResult;
 
    [[nodiscard]] auto stop() -> VoidResult;
 
    [[nodiscard]] auto is_running() const noexcept -> bool;
 
    // =========================================================================
    // Session Management
    // =========================================================================
 
    void register_session(uint64_t session_id, session_context context);
 
    void unregister_session(uint64_t session_id);
 
    [[nodiscard]] auto get_session(uint64_t session_id)
        -> std::optional<session_context>;
 
    [[nodiscard]] auto get_active_session_count() const noexcept -> size_t;
 
    // =========================================================================
    // Data Handling
    // =========================================================================
 
    [[nodiscard]] auto on_data_received(uint64_t session_id,
                                        std::vector<uint8_t> data) -> VoidResult;
 
    void on_connection_closed(uint64_t session_id);
 
    // =========================================================================
    // Service Handler Registration
    // =========================================================================
 
    using service_handler = storage_query_exec_job::service_handler;
 
    void register_c_store_handler(service_handler handler);
 
    void register_c_find_handler(service_handler handler);
 
    void register_c_get_handler(service_handler handler);
 
    void register_c_move_handler(service_handler handler);
 
    void register_c_echo_handler(service_handler handler);
 
    // =========================================================================
    // Callback Registration
    // =========================================================================
 
    void set_send_callback(send_callback callback);
 
    void set_association_callback(association_callback callback);
 
    void set_session_event_callback(session_event_callback callback);
 
    // =========================================================================
    // Metrics & Statistics
    // =========================================================================
 
    [[nodiscard]] auto get_metrics() noexcept -> pipeline_metrics&;
 
    [[nodiscard]] auto get_metrics() const noexcept -> const pipeline_metrics&;
 
    [[nodiscard]] auto get_coordinator() noexcept -> pipeline_coordinator&;
 
    [[nodiscard]] auto get_coordinator() const noexcept
        -> const pipeline_coordinator&;
 
private:
    std::unique_ptr<pipeline_coordinator> coordinator_;
 
    std::unordered_map<uint64_t, session_context> sessions_;
    mutable std::mutex sessions_mutex_;
 
    service_handler c_store_handler_;
    service_handler c_find_handler_;
    service_handler c_get_handler_;
    service_handler c_move_handler_;
    service_handler c_echo_handler_;
    mutable std::mutex handlers_mutex_;
 
    send_callback send_callback_;
    association_callback association_callback_;
    session_event_callback session_event_callback_;
    mutable std::mutex callbacks_mutex_;
 
    [[nodiscard]] auto get_handler_for_command(dimse_command_type type)
        -> service_handler;
 
    void on_job_completed(const job_context& ctx, bool success);
 
    void on_backpressure(pipeline_stage stage, size_t queue_depth);
};
 
}  // namespace kcenon::pacs::network::pipeline