udp_transport.h

Go to the documentation of this file.

#pragma once
 
// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
 
#include "../core/result_types.h"
#include "../core/error_codes.h"
#include <string>
#include <vector>
#include <cstdint>
#include <chrono>
#include <memory>
#include <atomic>
#include <span>
 
namespace kcenon { namespace monitoring {
 
struct udp_send_result {
    std::size_t bytes_sent{0};
    std::chrono::milliseconds elapsed{0};
};
 
struct udp_statistics {
    std::size_t packets_sent{0};
    std::size_t bytes_sent{0};
    std::size_t send_failures{0};
};
 
class udp_transport {
public:
    virtual ~udp_transport() = default;
 
    virtual common::VoidResult connect(const std::string& host, uint16_t port) = 0;
 
    virtual common::VoidResult send(std::span<const uint8_t> data) = 0;
 
    common::VoidResult send(const std::string& data) {
        return send(std::span<const uint8_t>(
            reinterpret_cast<const uint8_t*>(data.data()),
            data.size()));
    }
 
    virtual bool is_connected() const = 0;
 
    virtual void disconnect() = 0;
 
    virtual bool is_available() const = 0;
 
    virtual std::string name() const = 0;
 
    virtual udp_statistics get_statistics() const = 0;
 
    virtual void reset_statistics() = 0;
};
 
class stub_udp_transport : public udp_transport {
private:
    std::string host_;
    uint16_t port_{0};
    bool connected_{false};
    bool simulate_success_{true};
    mutable std::atomic<std::size_t> packets_sent_{0};
    mutable std::atomic<std::size_t> bytes_sent_{0};
    mutable std::atomic<std::size_t> send_failures_{0};
 
public:
    stub_udp_transport() = default;
 
    // Bring base class send method into scope
    using udp_transport::send;
 
    void set_simulate_success(bool success) {
        simulate_success_ = success;
    }
 
    common::VoidResult connect(const std::string& host, uint16_t port) override {
        if (!simulate_success_) {
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Simulated connection failure",
                "stub_udp_transport"
            ).to_common_error());
        }
        host_ = host;
        port_ = port;
        connected_ = true;
        return common::ok();
    }
 
    common::VoidResult send(std::span<const uint8_t> data) override {
        if (!connected_) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Not connected",
                "stub_udp_transport"
            ).to_common_error());
        }
 
        if (!simulate_success_) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Simulated send failure",
                "stub_udp_transport"
            ).to_common_error());
        }
 
        packets_sent_.fetch_add(1, std::memory_order_relaxed);
        bytes_sent_.fetch_add(data.size(), std::memory_order_relaxed);
        return common::ok();
    }
 
    bool is_connected() const override {
        return connected_;
    }
 
    void disconnect() override {
        connected_ = false;
        host_.clear();
        port_ = 0;
    }
 
    bool is_available() const override {
        return true;
    }
 
    std::string name() const override {
        return "stub";
    }
 
    udp_statistics get_statistics() const override {
        return {
            packets_sent_.load(std::memory_order_relaxed),
            bytes_sent_.load(std::memory_order_relaxed),
            send_failures_.load(std::memory_order_relaxed)
        };
    }
 
    void reset_statistics() override {
        packets_sent_.store(0, std::memory_order_relaxed);
        bytes_sent_.store(0, std::memory_order_relaxed);
        send_failures_.store(0, std::memory_order_relaxed);
    }
 
    // Test helpers
    std::string get_host() const { return host_; }
    uint16_t get_port() const { return port_; }
};
 
} } // namespace kcenon::monitoring
 
#ifdef MONITORING_HAS_COMMON_TRANSPORT_INTERFACES
#include <kcenon/common/interfaces/transport.h>
 
namespace kcenon { namespace monitoring {
 
class common_udp_transport : public udp_transport {
private:
    std::shared_ptr<::kcenon::common::interfaces::IUdpClient> client_;
    mutable std::atomic<std::size_t> packets_sent_{0};
    mutable std::atomic<std::size_t> bytes_sent_{0};
    mutable std::atomic<std::size_t> send_failures_{0};
 
public:
    explicit common_udp_transport(std::shared_ptr<::kcenon::common::interfaces::IUdpClient> client)
        : client_(std::move(client)) {}
 
    common::VoidResult connect(const std::string& host, uint16_t port) override {
        if (!client_) {
            return common::VoidResult::err(error_info(
                monitoring_error_code::dependency_missing,
                "UDP client not available",
                "common_udp_transport"
            ).to_common_error());
        }
 
        auto result = client_->connect(host, port);
        if (result.is_err()) {
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Connection failed: " + result.error().message,
                "common_udp_transport"
            ).to_common_error());
        }
        return common::ok();
    }
 
    common::VoidResult send(std::span<const uint8_t> data) override {
        if (!client_) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::dependency_missing,
                "UDP client not available",
                "common_udp_transport"
            ).to_common_error());
        }
 
        if (!client_->is_connected()) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Not connected",
                "common_udp_transport"
            ).to_common_error());
        }
 
        auto result = client_->send(data);
        if (result.is_err()) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Send failed: " + result.error().message,
                "common_udp_transport"
            ).to_common_error());
        }
 
        packets_sent_.fetch_add(1, std::memory_order_relaxed);
        bytes_sent_.fetch_add(data.size(), std::memory_order_relaxed);
        return common::ok();
    }
 
    bool is_connected() const override {
        return client_ && client_->is_connected();
    }
 
    void disconnect() override {
        if (client_) {
            client_->disconnect();
        }
    }
 
    bool is_available() const override {
        return client_ != nullptr;
    }
 
    std::string name() const override {
        if (client_) {
            return "common:" + client_->get_implementation_name();
        }
        return "common:unavailable";
    }
 
    udp_statistics get_statistics() const override {
        return {
            packets_sent_.load(std::memory_order_relaxed),
            bytes_sent_.load(std::memory_order_relaxed),
            send_failures_.load(std::memory_order_relaxed)
        };
    }
 
    void reset_statistics() override {
        packets_sent_.store(0, std::memory_order_relaxed);
        bytes_sent_.store(0, std::memory_order_relaxed);
        send_failures_.store(0, std::memory_order_relaxed);
    }
};
 
} } // namespace kcenon::monitoring
 
#endif // MONITORING_HAS_COMMON_TRANSPORT_INTERFACES
 
#ifdef MONITORING_HAS_NETWORK_SYSTEM
#include <kcenon/network/udp/udp_client.h>
 
namespace kcenon { namespace monitoring {
 
class network_udp_transport : public udp_transport {
private:
    std::unique_ptr<network_system::udp::udp_client> client_;
    std::string host_;
    uint16_t port_{0};
    bool connected_{false};
    mutable std::atomic<std::size_t> packets_sent_{0};
    mutable std::atomic<std::size_t> bytes_sent_{0};
    mutable std::atomic<std::size_t> send_failures_{0};
 
public:
    network_udp_transport() = default;
 
    common::VoidResult connect(const std::string& host, uint16_t port) override {
        try {
            client_ = std::make_unique<network_system::udp::udp_client>(host, port);
            host_ = host;
            port_ = port;
            connected_ = true;
            return common::ok();
        } catch (const std::exception& e) {
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                std::string("Connection failed: ") + e.what(),
                "network_udp_transport"
            ).to_common_error());
        }
    }
 
    common::VoidResult send(std::span<const uint8_t> data) override {
        if (!connected_ || !client_) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                "Not connected",
                "network_udp_transport"
            ).to_common_error());
        }
 
        try {
            auto result = client_->send(data);
            if (!result) {
                send_failures_.fetch_add(1, std::memory_order_relaxed);
                return common::VoidResult::err(error_info(
                    monitoring_error_code::network_error,
                    "Send failed: " + result.error().message,
                    "network_udp_transport"
                ).to_common_error());
            }
 
            packets_sent_.fetch_add(1, std::memory_order_relaxed);
            bytes_sent_.fetch_add(data.size(), std::memory_order_relaxed);
            return common::ok();
        } catch (const std::exception& e) {
            send_failures_.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(
                monitoring_error_code::network_error,
                std::string("Send failed: ") + e.what(),
                "network_udp_transport"
            ).to_common_error());
        }
    }
 
    bool is_connected() const override {
        return connected_ && client_ != nullptr;
    }
 
    void disconnect() override {
        client_.reset();
        connected_ = false;
        host_.clear();
        port_ = 0;
    }
 
    bool is_available() const override {
        return true;
    }
 
    std::string name() const override {
        return "network_system";
    }
 
    udp_statistics get_statistics() const override {
        return {
            packets_sent_.load(std::memory_order_relaxed),
            bytes_sent_.load(std::memory_order_relaxed),
            send_failures_.load(std::memory_order_relaxed)
        };
    }
 
    void reset_statistics() override {
        packets_sent_.store(0, std::memory_order_relaxed);
        bytes_sent_.store(0, std::memory_order_relaxed);
        send_failures_.store(0, std::memory_order_relaxed);
    }
};
 
} } // namespace kcenon::monitoring
 
#endif // MONITORING_HAS_NETWORK_SYSTEM
 
namespace kcenon { namespace monitoring {
 
inline std::unique_ptr<udp_transport> create_default_udp_transport() {
#ifdef MONITORING_HAS_NETWORK_SYSTEM
    return std::make_unique<network_udp_transport>();
#else
    return std::make_unique<stub_udp_transport>();
#endif
}
 
inline std::unique_ptr<stub_udp_transport> create_stub_udp_transport() {
    return std::make_unique<stub_udp_transport>();
}
 
#ifdef MONITORING_HAS_COMMON_TRANSPORT_INTERFACES
inline std::unique_ptr<common_udp_transport> create_common_udp_transport(
    std::shared_ptr<::kcenon::common::interfaces::IUdpClient> client) {
    return std::make_unique<common_udp_transport>(std::move(client));
}
#endif
 
#ifdef MONITORING_HAS_NETWORK_SYSTEM
inline std::unique_ptr<network_udp_transport> create_network_udp_transport() {
    return std::make_unique<network_udp_transport>();
}
#endif
 
} } // namespace kcenon::monitoring