DatagramSocket.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "BufferView.hpp"
#include "common.hpp"
#include "DatagramPacket.hpp"
#include "detail/buffer_traits.hpp"
#include "SocketOptions.hpp"
 
#include <atomic>
#include <bit>
#include <optional>
#include <span>
#include <string>
 
namespace jsocketpp
{

enum class DatagramReceiveMode : std::uint8_t
{
    NoPreflight = 0,

    PreflightSize = 1,

    PreflightMax = 2
};

struct DatagramReadOptions
{
    DatagramReceiveMode mode = DatagramReceiveMode::NoPreflight;

    bool allowGrow = true;

    bool allowShrink = true;

    int recvFlags = 0;

    bool updateLastRemote = true;

    bool resolveNumeric = true;

    bool errorOnTruncate{true};
};

struct DatagramReadResult
{
    std::size_t bytes = 0;

    std::size_t datagramSize = 0;

    bool truncated = false;

    sockaddr_storage src{};

    socklen_t srcLen = 0;
};

struct ReadExactOptions
{
    DatagramReadOptions base{};

    bool requireExact = true;

    bool padIfSmaller = false;

    bool errorOnTruncate = true;

    bool autoResizeDynamic = true;
};

enum class Direction : std::uint8_t
{
    Read,     
    Write,    
    ReadWrite 
};

class DatagramSocket : public SocketOptions
{
  public:
    explicit DatagramSocket(Port localPort = 0, std::string_view localAddress = "",
                            std::optional<std::size_t> recvBufferSize = std::nullopt,
                            std::optional<std::size_t> sendBufferSize = std::nullopt,
                            std::optional<std::size_t> internalBufferSize = std::nullopt, bool reuseAddress = true,
                            int soRecvTimeoutMillis = -1, int soSendTimeoutMillis = -1, bool nonBlocking = false,
                            bool dualStack = true, bool autoBind = true, bool autoConnect = false,
                            std::string_view remoteAddress = "", Port remotePort = 0, int connectTimeoutMillis = -1);

    ~DatagramSocket() noexcept override;

    DatagramSocket(const DatagramSocket&) = delete;

    DatagramSocket& operator=(const DatagramSocket&) = delete;

    DatagramSocket(DatagramSocket&& rhs) noexcept
        : SocketOptions(rhs.getSocketFd()), _remoteAddr(rhs._remoteAddr), _remoteAddrLen(rhs._remoteAddrLen),
          _haveRemoteAddr(rhs._haveRemoteAddr.load(std::memory_order_relaxed)), _localAddr(rhs._localAddr),
          _localAddrLen(rhs._localAddrLen), _haveLocalAddr(rhs._haveLocalAddr.load(std::memory_order_relaxed)),
          _internalBuffer(std::move(rhs._internalBuffer)), _port(rhs._port), _isBound(rhs._isBound),
          _isConnected(rhs._isConnected)
    {
        rhs.cleanup();
    }

    DatagramSocket& operator=(DatagramSocket&& rhs) noexcept
    {
        if (this != &rhs)
        {
            // Clean up current socket
            try
            {
                close(); // Clean up existing resources
            }
            catch (...)
            {
                // CloseSocket error is ignored.
                // TODO: Consider adding an internal flag, nested exception, or user-configurable error handler
                //       to report errors in future versions.
            }
 
            // Transfer ownership
            setSocketFd(rhs.getSocketFd());
            _remoteAddr = rhs._remoteAddr;
            _remoteAddrLen = rhs._remoteAddrLen;
            _haveRemoteAddr.store(rhs._haveRemoteAddr.load(std::memory_order_relaxed));
            _localAddr = rhs._localAddr;
            _localAddrLen = rhs._localAddrLen;
            _haveLocalAddr.store(rhs._haveLocalAddr.load(std::memory_order_relaxed));
            _internalBuffer = std::move(rhs._internalBuffer);
            _port = rhs._port;
            _isBound = rhs._isBound;
            _isConnected = rhs._isConnected;
 
            // Reset source
            rhs.cleanup();
        }
        return *this;
    }

    void bind();

    void bind(Port localPort);

    void bind(std::string_view localAddress, Port localPort);

    [[nodiscard]] bool isBound() const noexcept { return _isBound; }

    void connect(std::string_view host, Port port, int timeoutMillis);

    void disconnect();

    [[nodiscard]] bool isConnected() const noexcept { return _isConnected; }

    [[nodiscard]] std::string getLocalIp(bool convertIPv4Mapped);

    [[nodiscard]] Port getLocalPort();

    [[nodiscard]] std::string getLocalSocketAddress(bool convertIPv4Mapped);

    [[nodiscard]] std::string getRemoteIp(bool convertIPv4Mapped) const;

    [[nodiscard]] Port getRemotePort() const;

    [[nodiscard]] std::string getRemoteSocketAddress(bool convertIPv4Mapped) const;

    void write(std::string_view message) const;

    template <typename T> void write(const T& value) const
    {
        static_assert(std::is_trivially_copyable_v<T>, "DatagramSocket::write<T>() requires trivially copyable type");
        static_assert(std::is_standard_layout_v<T>, "DatagramSocket::write<T>() requires standard layout type");
 
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::write<T>(): socket is not open.");
 
        if (!_isConnected)
            throw SocketException("DatagramSocket::write<T>(): socket is not connected. Use writeTo() instead.");
 
        enforceSendCapConnected(sizeof(T));
 
        const auto buffer = std::bit_cast<std::array<std::byte, sizeof(T)>>(value);
        internal::sendExact(getSocketFd(), buffer.data(), buffer.size());
    }

    void write(std::span<const std::byte> data) const;

    void writeAll(std::string_view message) const;

    template <typename T> void writePrefixed(const std::string_view payload) const
    {
        sendPrefixedConnected<T>(asBytes(payload));
    }

    template <typename T> void writePrefixed(const std::span<const std::byte> payload) const
    {
        sendPrefixedConnected<T>(payload);
    }

    template <typename T>
    void writePrefixedTo(const std::string_view host, const Port port, const std::string_view payload) const
    {
        sendPrefixedUnconnected<T>(host, port, asBytes(payload));
    }

    template <typename T>
    void writePrefixedTo(const std::string_view host, const Port port, const std::span<const std::byte> payload) const
    {
        sendPrefixedUnconnected<T>(host, port, payload);
    }

    void writev(std::span<const std::string_view> buffers) const;

    void writevAll(std::span<const std::string_view> buffers) const;

    void writeFrom(const void* data, std::size_t len) const;

    void writeWithTimeout(std::string_view data, int timeoutMillis) const;

    void write(const DatagramPacket& packet);

    void writeTo(std::string_view host, Port port, std::string_view message);

    void writeTo(std::string_view host, Port port, std::span<const std::byte> data);

    template <typename T> void writeTo(const std::string_view host, const Port port, const T& value) const
    {
        static_assert(std::is_trivially_copyable_v<T>, "DatagramSocket::writeTo<T>() requires trivially copyable type");
        static_assert(std::is_standard_layout_v<T>, "DatagramSocket::writeTo<T>() requires standard layout type");
 
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::writeTo<T>(): socket is not open.");
 
        if constexpr (sizeof(T) == 0)
            return;
 
        const auto buf = std::bit_cast<std::array<std::byte, sizeof(T)>>(value);
        sendUnconnectedTo(host, port, buf.data(), buf.size());
    }

    DatagramReadResult read(DatagramPacket& packet, const DatagramReadOptions& opts) const;

    [[nodiscard]] DatagramReadResult readInto(void* buffer, std::size_t len,
                                              const DatagramReadOptions& opts = {}) const;

    [[nodiscard]] DatagramReadResult readInto(std::span<char> out, const DatagramReadOptions& opts = {}) const;

    template <typename T, std::enable_if_t<detail::is_dynamic_buffer_v<T>, int> = 0>
    [[nodiscard]] T read(const DatagramReadOptions& opts = {},
                         std::size_t minCapacity = DefaultDatagramReceiveSize) const
    {
        using Ptr = decltype(std::declval<T&>().data());
        using Elem = std::remove_pointer_t<Ptr>;
        static_assert(std::is_same_v<Elem, char> || std::is_same_v<Elem, unsigned char> ||
                          std::is_same_v<Elem, std::byte>,
                      "T must be a contiguous byte-like container of char, unsigned char, or std::byte");
 
        if (minCapacity == 0)
            minCapacity = 1; // allow clean reception of zero-length datagrams
 
        std::size_t capacity = (std::min) (minCapacity, MaxDatagramPayloadSafe);
 
        // Prefer an exact allocation when we can learn the size (either explicitly requested,
        // or because strict no-truncation is desired).
        std::size_t probed = 0;
        const bool wantPreflight = (opts.mode != DatagramReceiveMode::NoPreflight) || opts.errorOnTruncate;
 
        if (wantPreflight)
        {
            try
            {
                if (const std::size_t exact = internal::nextDatagramSize(getSocketFd()); exact > 0)
                {
                    probed = (std::min) (exact, MaxDatagramPayloadSafe);
                    capacity = (std::max) (capacity, probed); // grow to fit the datagram exactly
                }
            }
            catch (const SocketException&)
            {
                // graceful degradation: keep current capacity
            }
        }
 
        T out;
        out.resize(capacity);
 
        DatagramReadOptions local = opts;
        if (probed > 0)
            local.mode = DatagramReceiveMode::NoPreflight; // avoid redundant preflight on the actual recv
 
        // Use the policy-aware backbone; it will enforce errorOnTruncate as documented.
        if (auto res = readInto(reinterpret_cast<void*>(out.data()), out.size(), local); res.bytes < out.size())
            out.resize(res.bytes); // shrink to actual bytes received (no null terminator implied)
 
        return out;
    }

    template <typename T, std::enable_if_t<detail::is_fixed_buffer_v<T>, int> = 0>
    [[nodiscard]] T read(const DatagramReadOptions& opts = {}) const
    {
        using Ptr = decltype(std::declval<T&>().data());
        using Elem = std::remove_pointer_t<Ptr>;
        static_assert(std::is_same_v<Elem, char> || std::is_same_v<Elem, unsigned char> ||
                          std::is_same_v<Elem, std::byte>,
                      "T must be a contiguous byte-like container of char, unsigned char, or std::byte");
        static_assert(sizeof(Elem) == 1, "T element type must be 1 byte wide");
 
        T out{}; // value-init so any unused tail is predictable (typically zero)
        if (out.size() == 0)
            return out;
 
        // Zero-allocation, policy-aware read
        (void) readInto(std::span<char>(reinterpret_cast<char*>(out.data()), out.size()), opts);
        return out;
    }

    template <typename T>
    [[nodiscard]] DatagramReadResult readFrom(T& buffer, std::string* senderAddr, Port* senderPort,
                                              const DatagramReadOptions& opts) const
    {
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::readFrom(): socket is not open.");
 
        DatagramReadResult result{};
 
        // Determine current capacity and whether we can grow.
        constexpr bool isDynamic = detail::is_dynamic_buffer_v<T>;
        constexpr bool isFixed = detail::is_fixed_buffer_v<T>;
        static_assert(isDynamic || isFixed, "T must be a supported dynamic or fixed-size contiguous byte container");
 
        auto currSize = static_cast<std::size_t>(buffer.size());
 
        // For dynamic containers, ensure at least 1 byte to properly consume zero-length datagrams.
        if constexpr (isDynamic)
        {
            if (currSize == 0)
                currSize = (std::min) (DefaultDatagramReceiveSize, MaxDatagramPayloadSafe);
        }
 
        // Optional preflight: if we can learn the size, either grow (dynamic) or enforce policy (fixed).
        std::size_t probed = 0;
        const bool wantPreflight = (opts.mode != DatagramReceiveMode::NoPreflight) || opts.errorOnTruncate;
 
        if (wantPreflight)
        {
            try
            {
                const std::size_t exact = internal::nextDatagramSize(getSocketFd());
                if (exact > 0)
                {
                    probed = (std::min) (exact, MaxDatagramPayloadSafe);
 
                    if constexpr (isDynamic)
                    {
                        if (probed > currSize)
                            currSize = probed; // grow to fit the datagram exactly
                    }
                    else
                    { // fixed
                        if (opts.errorOnTruncate && probed > currSize)
                        {
                            // Early, non-destructive failure: datagram remains queued.
                            throw SocketException("DatagramSocket::readFrom(): fixed buffer too small for incoming "
                                                  "datagram (preflight).");
                        }
                    }
                }
            }
            catch (const SocketException&)
            {
                // graceful degradation; fall back to single-recv path
                probed = 0;
            }
        }
 
        // Clamp to safety cap.
        currSize = (std::min) (currSize, MaxDatagramPayloadSafe);
 
        // Resize dynamic container as needed.
        if constexpr (isDynamic)
            buffer.resize(currSize);
 
        // Perform exactly one receive using the low-level backbone to obtain sender info.
        sockaddr_storage src{};
        auto srcLen = static_cast<socklen_t>(sizeof(src));
        std::size_t datagramSize = 0;
        bool truncated = false;
 
        const DatagramReceiveMode effectiveMode = (probed > 0 ? DatagramReceiveMode::NoPreflight : opts.mode);
 
        const std::size_t n = readIntoBuffer(reinterpret_cast<char*>(buffer.data()), currSize, effectiveMode,
                                             opts.recvFlags, &src, &srcLen, &datagramSize, &truncated);
 
        // Enforce strict no-truncation after the read if we couldn't fail early.
        if (opts.errorOnTruncate && (truncated || (datagramSize > 0 && datagramSize > currSize)))
        {
            throw SocketException("DatagramSocket::readFrom(): datagram truncated into destination buffer.");
        }
 
        // Shrink dynamic container to the actual bytes received.
        if constexpr (isDynamic)
        {
            if (n < buffer.size())
                buffer.resize(n);
        }
 
        // Fill the result structure.
        result.bytes = n;
        result.datagramSize = (datagramSize > 0 ? datagramSize : n);
        result.truncated = truncated;
        result.src = src;
        result.srcLen = srcLen;
 
        // Optional: update last-remote if requested by options.
        if (opts.updateLastRemote)
            rememberRemote(src, srcLen);
 
        // Resolve numeric sender info if requested.
        if ((senderAddr || senderPort) && srcLen > 0)
        {
            std::string addrStr;
            Port portNum{};
            internal::resolveNumericHostPort(reinterpret_cast<const sockaddr*>(&src), srcLen, addrStr, portNum);
            if (senderAddr)
                *senderAddr = std::move(addrStr);
            if (senderPort)
                *senderPort = portNum;
        }
 
        return result;
    }

    [[nodiscard]] DatagramReadResult readExact(void* buffer, std::size_t exactLen,
                                               const ReadExactOptions& opts = {}) const;

    [[nodiscard]] DatagramReadResult readExact(std::span<char> out, const ReadExactOptions& opts = {}) const;

    template <typename T>
    [[nodiscard]] DatagramReadResult readExact(T& buffer, const std::size_t exactLen,
                                               const ReadExactOptions& opts = {}) const
    {
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::readExact(T&,size_t): socket is not open.");
        if (exactLen == 0)
            throw SocketException("DatagramSocket::readExact(T&,size_t): exactLen must be > 0.");
 
        // Validate container element type and contiguity at compile time
        using Ptr = decltype(std::declval<T&>().data());
        using Elem = std::remove_pointer_t<Ptr>;
        static_assert(sizeof(Elem) == 1, "T element type must be 1 byte wide");
        static_assert(std::is_same_v<Elem, char> || std::is_same_v<Elem, unsigned char> ||
                          std::is_same_v<Elem, std::byte>,
                      "T must be a contiguous byte-like container of char, unsigned char, or std::byte");
 
        DatagramReadResult res{};
 
        if constexpr (detail::is_dynamic_buffer_v<T>)
        {
            // Ensure capacity matches policy
            if (opts.autoResizeDynamic)
            {
                if (exactLen > MaxDatagramPayloadSafe)
                    throw SocketException(
                        "DatagramSocket::readExact(T&,size_t): exactLen exceeds MaxDatagramPayloadSafe.");
                buffer.resize(exactLen); // value-initializes any new bytes
            }
            else
            {
                if (buffer.size() < exactLen)
                    throw SocketException(
                        "DatagramSocket::readExact(T&,size_t): buffer too small and autoResizeDynamic=false.");
            }
 
            // Perform policy-aware exact read into the first exactLen bytes
            res = readExact(std::span(reinterpret_cast<char*>(buffer.data()), exactLen), opts);
 
            // Normalize container size to the logical exact length (keeps API guarantees)
            if (buffer.size() != exactLen)
                buffer.resize(exactLen); // any extension is value-initialized (zeros)
        }
        else if constexpr (detail::is_fixed_buffer_v<T>)
        {
            if (buffer.size() < exactLen)
                throw SocketException("DatagramSocket::readExact(T&,size_t): fixed buffer smaller than exactLen.");
 
            // Exact read directly into the fixed-size storage (only the first exactLen bytes are touched)
            res = readExact(std::span(reinterpret_cast<char*>(buffer.data()), exactLen), opts);
            // Size is fixed; if padding was requested, the span path already zero-filled the tail.
        }
        else
        {
            static_assert(detail::is_dynamic_buffer_v<T> || detail::is_fixed_buffer_v<T>,
                          "T must be a supported dynamic or fixed-size contiguous byte container");
        }
 
        return res;
    }

    std::size_t readAtMost(std::span<char> out, const DatagramReadOptions& opts = {}) const;

    [[nodiscard]] std::string readAtMost(std::size_t n) const;

    [[nodiscard]] std::string readAvailable() const;

    std::size_t readAvailable(std::span<char> out, const DatagramReadOptions& opts = {}) const;

    std::size_t readIntoExact(void* buffer, std::size_t len) const;

    [[nodiscard]] std::string readAtMostWithTimeout(std::size_t n, int timeoutMillis) const;

    template <
        typename T,
        std::enable_if_t<std::is_integral_v<T> && std::is_unsigned_v<T> && std::is_trivially_copyable_v<T>, int> = 0>
    [[nodiscard]] std::string readPrefixed(std::size_t maxPayloadLen = MaxDatagramPayloadSafe,
                                           const std::endian prefixEndian = std::endian::big) const
    {
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::readPrefixed<T>(): socket is not open.");
 
        // Clamp guard values
        if (maxPayloadLen == 0)
        {
            // Only valid if the incoming prefix is exactly zero; we’ll validate after the read.
            // Keep as 0 here; we still read the datagram and then enforce.
        }
        constexpr std::size_t safeMax = MaxDatagramPayloadSafe;
        if (maxPayloadLen > safeMax)
            maxPayloadLen = safeMax;
 
        // Prefer allocating exactly the datagram size when we can probe.
        std::size_t probed = 0;
        try
        {
            if (const std::size_t sz = internal::nextDatagramSize(getSocketFd()); sz > 0)
                probed = (std::min) (sz, safeMax);
        }
        catch (const SocketException&)
        {
            // Fallback to safe maximum below.
            probed = 0;
        }
 
        const std::size_t cap = (probed > 0) ? probed : safeMax;
        if (cap < sizeof(T))
        {
            // We’ll still read and then fail with a precise error if needed.
        }
 
        std::string frame(cap, '\0');
 
        // Read the entire datagram (no truncation); a single recv is used internally.
        DatagramReadOptions ro{};
        ro.errorOnTruncate = true; // enforce full-datagram semantics
        const std::size_t n = readAvailable(std::span<char>(frame.data(), frame.size()), ro);
        frame.resize(n);
 
        // Validate minimal size (must contain the prefix)
        if (frame.size() < sizeof(T))
            throw SocketException("DatagramSocket::readPrefixed<T>(): datagram smaller than length prefix.");
 
        // Decode prefix in requested endian
        const auto* p = reinterpret_cast<const unsigned char*>(frame.data());
        std::uint64_t decl = 0;
 
        if (prefixEndian == std::endian::big)
        {
            for (std::size_t i = 0; i < sizeof(T); ++i)
                decl = (decl << 8) | static_cast<std::uint64_t>(p[i]);
        }
        else
        { // little-endian
            for (std::size_t i = 0; i < sizeof(T); ++i)
                decl |= (static_cast<std::uint64_t>(p[i]) << (8 * i));
        }
 
        const std::size_t declared = decl;
 
        // Enforce declared size == actual payload length
        if (const std::size_t actual = frame.size() - sizeof(T); declared != actual)
            throw SocketException("DatagramSocket::readPrefixed<T>(): prefix/payload size mismatch.");
 
        // Enforce application-level max payload guard
        if (declared > maxPayloadLen)
            throw SocketException("DatagramSocket::readPrefixed<T>(): payload exceeds maxPayloadLen.");
 
        // Return payload slice
        return {frame.data() + sizeof(T), frame.data() + sizeof(T) + declared};
    }

    void discard(const DatagramReadOptions& opts = {}) const;

    void discardExact(std::size_t n, const DatagramReadOptions& opts = {}) const;

    [[nodiscard]] DatagramReadResult readv(std::span<BufferView> buffers, const DatagramReadOptions& opts = {}) const;

    [[nodiscard]] DatagramReadResult readvAll(std::span<BufferView> buffers,
                                              const DatagramReadOptions& opts = {}) const;

    std::size_t readvAllBytes(std::span<BufferView> buffers, const DatagramReadOptions& opts = {}) const
    {
        return readvAll(buffers, opts).bytes;
    }

    [[nodiscard]] DatagramReadResult readvAtMostWithTimeout(std::span<BufferView> buffers, int timeoutMillis,
                                                            const DatagramReadOptions& opts = {}) const;

    std::size_t readvAtMostWithTimeout(std::span<BufferView> buffers, int timeoutMillis) const
    {
        return readvAtMostWithTimeout(buffers, timeoutMillis, DatagramReadOptions{}).bytes;
    }

    [[nodiscard]] DatagramReadResult readvAllWithTotalTimeout(std::span<BufferView> buffers, int totalTimeoutMillis,
                                                              const DatagramReadOptions& opts = {}) const;

    std::size_t readvAllWithTotalTimeoutBytes(std::span<BufferView> buffers, int totalTimeoutMillis,
                                              const DatagramReadOptions& opts = {}) const
    {
        return readvAllWithTotalTimeout(buffers, totalTimeoutMillis, opts).bytes;
    }

    [[nodiscard]] DatagramReadResult peek(DatagramPacket& packet, bool allowResize = true,
                                          const DatagramReadOptions& opts = {}) const;

    bool hasPendingData(int timeoutMillis) const;

    [[nodiscard]] std::optional<int> getMTU() const;

    void waitReady(Direction dir, int timeoutMillis) const;

    void setInternalBufferSize(std::size_t newLen);

    void close();

    [[nodiscard]] bool isValid() const noexcept { return getSocketFd() != INVALID_SOCKET; }

    [[nodiscard]] bool isClosed() const noexcept { return getSocketFd() == INVALID_SOCKET; }

    [[nodiscard]] std::optional<std::pair<sockaddr_storage, socklen_t>> getLastPeerSockAddr() const
    {
        if (!_haveRemoteAddr)
        {
            // No communication has occurred; no peer info available
            return std::nullopt;
        }
 
        return std::make_pair(_remoteAddr, _remoteAddrLen);
    }
 
  protected:
    std::size_t readIntoBuffer(char* buf, std::size_t len, DatagramReceiveMode mode, int recvFlags,
                               sockaddr_storage* outSrc, socklen_t* outSrcLen, std::size_t* outDatagramSz,
                               bool* outTruncated) const;

    void cleanup();

    void cleanupAndThrow(int errorCode);

    void cleanupAndRethrow();

    std::size_t chooseReceiveSize() const
    {
        if (const std::size_t exact = internal::nextDatagramSize(getSocketFd()); exact > 0)
            return (exact > MaxDatagramPayloadSafe) ? MaxDatagramPayloadSafe : exact;
 
        const std::size_t fallback = _internalBuffer.empty() ? DefaultDatagramReceiveSize : _internalBuffer.size();
 
        return (fallback > MaxDatagramPayloadSafe) ? MaxDatagramPayloadSafe : fallback;
    }

    void rememberRemote(const sockaddr_storage& src, const socklen_t len) const noexcept
    {
        _remoteAddr = src;
        _remoteAddrLen = len;
        _haveRemoteAddr.store(true, std::memory_order_relaxed);
    }

    static void throwSizeMismatch(const std::size_t expected, const std::size_t actual, const bool isProbedKnown)
    {
        // Two-arg pattern with wrapped message for consistency across the project
        constexpr int err = 0; // logical (not a system error)
        const std::string msg =
            "UDP datagram size mismatch: expected " + std::to_string(expected) +
            (isProbedKnown ? (", probed " + std::to_string(actual)) : (", received " + std::to_string(actual)));
        throw SocketException(err, msg);
    }

    void enforceSendCapConnected(const std::size_t payloadSize) const
    {
        // Zero-length datagrams are valid (often used as keep-alives); nothing to enforce.
        if (payloadSize == 0)
            return;
 
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException(0, "DatagramSocket::enforceSendCapConnected(): socket is not open.");
 
        if (!_isConnected)
            throw SocketException(0, "DatagramSocket::enforceSendCapConnected(): socket is not connected.");
 
        // Determine remote family: prefer cached last-peer, else query getpeername().
        int family = AF_UNSPEC;
        if (_remoteAddrLen > 0)
        {
            family = reinterpret_cast<const sockaddr*>(&_remoteAddr)->sa_family;
        }
        else
        {
            sockaddr_storage peer{};
            auto len = static_cast<socklen_t>(sizeof(peer));
            if (::getpeername(getSocketFd(), reinterpret_cast<sockaddr*>(&peer), &len) == SOCKET_ERROR)
            {
                const int err = GetSocketError();
                throw SocketException(err, SocketErrorMessage(err));
            }
            family = reinterpret_cast<const sockaddr*>(&peer)->sa_family;
        }
 
        // Always allow the conservative, cross-stack "safe" ceiling.
        if (payloadSize <= MaxDatagramPayloadSafe)
            return;
 
        // If the connected peer is IPv6, allow the IPv6 theoretical headroom.
        if (family == AF_INET6 && payloadSize <= MaxUdpPayloadIPv6)
            return;
 
        // Compose a precise, helpful error message and throw a logical (non-OS) error.
        std::string msg = "UDP datagram payload (" + std::to_string(payloadSize) + " bytes) ";
 
        if (payloadSize > MaxUdpPayloadIPv6)
        {
            msg += "exceeds IPv6 theoretical maximum (" + std::to_string(MaxUdpPayloadIPv6) + "). ";
        }
        else if (family == AF_INET)
        {
            msg += "exceeds IPv4 maximum (" + std::to_string(MaxUdpPayloadIPv4) + "). ";
        }
        else
        {
            // AF_UNSPEC or other: fell through the safe ceiling and not eligible for IPv6 headroom.
            msg += "exceeds MaxDatagramPayloadSafe (" + std::to_string(MaxDatagramPayloadSafe) + "). ";
        }
 
        msg += "Split the payload across multiple datagrams or switch to a stream protocol.";
 
        // Two-arg form for consistency with other logical throws in the project.
        throw SocketException(0, msg);
    }

    void sendUnconnectedTo(std::string_view host, Port port, const void* data, std::size_t len);

    static std::span<const std::byte> asBytes(const std::string_view sv) noexcept
    {
        return {reinterpret_cast<const std::byte*>(sv.data()), sv.size()};
    }

    template <typename T> static std::array<std::byte, sizeof(T)> encodeLengthPrefixBE(std::size_t n)
    {
        static_assert(std::is_integral_v<T> && std::is_unsigned_v<T>,
                      "T must be an unsigned integral type for the length prefix.");
        if (n > static_cast<std::size_t>((std::numeric_limits<T>::max)()))
            throw SocketException("writePrefixed<T>(): payload too large for prefix type T.");
        std::array<std::byte, sizeof(T)> out{};
        T v = static_cast<T>(n);
        for (std::size_t i = 0; i < sizeof(T); ++i)
        {
            out[sizeof(T) - 1 - i] = static_cast<std::byte>(v & static_cast<T>(0xFF));
            v = static_cast<T>(v >> 8);
        }
        return out;
    }

    template <typename T> void sendPrefixedConnected(const std::span<const std::byte> payload) const
    {
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::sendPrefixedConnected<T>(): socket is not open.");
        if (!isConnected())
            throw SocketException(
                "DatagramSocket::sendPrefixedConnected<T>(): socket is not connected. Use write*To().");
 
        const std::size_t n = payload.size();
        const auto prefix = encodeLengthPrefixBE<T>(n);
        const std::size_t total = sizeof(T) + n;
 
        enforceSendCapConnected(total);
 
        // Coalesce [prefix | payload] into one datagram and send once.
        std::vector<std::byte> datagram(total);
        std::memcpy(datagram.data(), prefix.data(), sizeof(T));
        if (n != 0)
        {
            std::memcpy(datagram.data() + sizeof(T), payload.data(), n);
        }
        internal::sendExact(getSocketFd(), datagram.data(), datagram.size());
    }

    template <typename T>
    void sendPrefixedUnconnected(const std::string_view host, const Port port, const std::span<const std::byte> payload)
    {
        if (getSocketFd() == INVALID_SOCKET)
            throw SocketException("DatagramSocket::sendPrefixedUnconnected<T>(): socket is not open.");
 
        const std::size_t n = payload.size();
        const auto prefix = encodeLengthPrefixBE<T>(n);
        const std::size_t total = sizeof(T) + n;
 
        // Build once; family skipping & send are handled inside sendUnconnectedTo().
        std::vector<std::byte> datagram(total);
        std::memcpy(datagram.data(), prefix.data(), sizeof(T));
        if (n != 0)
        {
            std::memcpy(datagram.data() + sizeof(T), payload.data(), n);
        }
        sendUnconnectedTo(host, port, datagram.data(), datagram.size());
    }

    void cacheLocalEndpoint() noexcept
    {
        sockaddr_storage ss{};
        auto len = static_cast<socklen_t>(sizeof(ss));
#if defined(_WIN32)
        if (::getsockname(getSocketFd(), reinterpret_cast<sockaddr*>(&ss), &len) == SOCKET_ERROR)
            return;
#else
        if (::getsockname(getSocketFd(), reinterpret_cast<sockaddr*>(&ss), &len) == -1)
            return;
#endif
        _localAddr = ss;
        _localAddrLen = len;
        _haveLocalAddr.store(true, std::memory_order_relaxed); // or set your existing flag
    }

    [[nodiscard]] bool tryGetRemoteSockaddr(sockaddr_storage& out, socklen_t& outLen) const;
 
  private:
    mutable sockaddr_storage
        _remoteAddr{}; 
    mutable socklen_t _remoteAddrLen =
        0; 
    mutable std::atomic_bool _haveRemoteAddr = false;
    sockaddr_storage _localAddr{}; 
    socklen_t _localAddrLen = 0;   
    std::atomic_bool _haveLocalAddr =
        false;                         
    std::vector<char> _internalBuffer; 
    Port _port;                        
    bool _isBound = false;             
    bool _isConnected = false;         
};
 
} // namespace jsocketpp