messaging_integration.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
// Internal implementation file - suppress deprecation warnings
#define CONTAINER_INTERNAL_INCLUDE
#include "core/container.h"
#undef CONTAINER_INTERNAL_INCLUDE
#include "core/concepts.h"
#include <functional>
#include <memory>
#include <concepts>
 
#ifdef HAS_PERFORMANCE_METRICS
#include <chrono>
#include <atomic>
#include <mutex>
#include <unordered_map>
#endif
 
namespace kcenon::container::integration {
 
class messaging_integration {
public:
    static std::shared_ptr<value_container> create_optimized_container(
        const std::string& message_type = "data_container"
    );
 
    static std::string serialize_for_messaging(
        const std::shared_ptr<value_container>& container,
        bool compress = false
    );
 
    static std::shared_ptr<value_container> deserialize_from_messaging(
        const std::string& data,
        bool decompress = false
    );
 
#ifdef HAS_PERFORMANCE_METRICS
    struct metrics {
        std::atomic<uint64_t> containers_created{0};
        std::atomic<uint64_t> serializations_performed{0};
        std::atomic<uint64_t> deserializations_performed{0};
        std::atomic<uint64_t> total_serialize_time_us{0};
        std::atomic<uint64_t> total_deserialize_time_us{0};
    };
 
    static metrics& get_metrics();
    static void reset_metrics();
    static std::string get_metrics_summary();
#endif
 
#ifdef HAS_EXTERNAL_INTEGRATION
    using container_callback_t = std::function<void(const std::shared_ptr<value_container>&)>;
 
    static void register_creation_callback(container_callback_t callback);
    static void register_serialization_callback(container_callback_t callback);
    static void unregister_callbacks();
#endif
 
private:
#ifdef HAS_PERFORMANCE_METRICS
    static metrics metrics_;
#endif
 
#ifdef HAS_EXTERNAL_INTEGRATION
    static std::vector<container_callback_t> creation_callbacks_;
    static std::vector<container_callback_t> serialization_callbacks_;
    static std::mutex callback_mutex_;
#endif
};
 
class messaging_container_builder {
public:
    messaging_container_builder();
 
    messaging_container_builder& source(const std::string& id, const std::string& sub_id = "");
    messaging_container_builder& target(const std::string& id, const std::string& sub_id = "");
    messaging_container_builder& message_type(const std::string& type);
 
    template<typename T>
    messaging_container_builder& set(const std::string& key, T&& value);
 
    messaging_container_builder& optimize_for_size();
    messaging_container_builder& optimize_for_speed();
 
    std::shared_ptr<value_container> build();
 
private:
    std::shared_ptr<value_container> container_;
    bool size_optimized_ = false;
    bool speed_optimized_ = false;
};
 
#ifdef HAS_PERFORMANCE_METRICS
class container_performance_monitor {
public:
    explicit container_performance_monitor(const std::string& operation_name);
    ~container_performance_monitor();
 
    void set_container_size(size_t size);
    void set_result_size(size_t size);
 
private:
    std::string operation_name_;
    std::chrono::high_resolution_clock::time_point start_time_;
    size_t container_size_ = 0;
    size_t result_size_ = 0;
};
#endif // HAS_PERFORMANCE_METRICS
 
// Template implementation for set (unified API)
// Uses concepts to provide clearer error messages for unsupported types
template<typename T>
messaging_container_builder& messaging_container_builder::set(const std::string& key, T&& value) {
    using DecayedT = std::decay_t<T>;
 
    if constexpr (std::is_same_v<DecayedT, std::shared_ptr<value_container>>) {
        // Handle nested containers by serializing and storing as bytes
        auto result = value->serialize_string(value_container::serialization_format::binary);
        if (result.is_ok()) {
            std::string serialized_data = result.value();
            std::vector<uint8_t> bytes(serialized_data.begin(), serialized_data.end());
            container_->set(key, bytes);
        }
    } else if constexpr (std::is_same_v<DecayedT, bool>) {
        container_->set(key, value);
    } else if constexpr (concepts::IntegralType<DecayedT>) {
        if constexpr (sizeof(DecayedT) <= 4) {
            container_->set(key, static_cast<int32_t>(value));
        } else {
            container_->set(key, static_cast<int64_t>(value));
        }
    } else if constexpr (concepts::FloatingPointType<DecayedT>) {
        if constexpr (std::is_same_v<DecayedT, float>) {
            container_->set(key, value);
        } else {
            container_->set(key, value);
        }
    } else if constexpr (std::is_same_v<DecayedT, std::string>) {
        container_->set(key, std::string(value));
    } else if constexpr (concepts::StringLike<DecayedT>) {
        container_->set(key, std::string(value));
    }
 
    return *this;
}
 
} // namespace kcenon::container::integration
 
// Convenience macros for performance monitoring
#ifdef HAS_PERFORMANCE_METRICS
#define CONTAINER_PERF_MONITOR(name) \
    kcenon::container::integration::container_performance_monitor _monitor(name)
 
#define CONTAINER_PERF_SET_SIZE(size) \
    _monitor.set_container_size(size)
 
#define CONTAINER_PERF_SET_RESULT(size) \
    _monitor.set_result_size(size)
#else
#define CONTAINER_PERF_MONITOR(name)
#define CONTAINER_PERF_SET_SIZE(size)
#define CONTAINER_PERF_SET_RESULT(size)
#endif