Message.cpp

#include <xrpld/overlay/Message.h>
 
#include <xrpld/overlay/Compression.h>
#include <xrpld/overlay/detail/TrafficCount.h>
 
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/protocol/PublicKey.h>
 
#include <google/protobuf/message.h>
 
#include <xrpl.pb.h>
 
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <optional>
#include <vector>
 
namespace xrpl {
 
Message::Message(
    ::google::protobuf::Message const& message,
    protocol::MessageType type,
    std::optional<PublicKey> const& validator)
    : category_(static_cast<std::size_t>(TrafficCount::categorize(message, type, false)))
    , validatorKey_(validator)
{
    using namespace xrpl::compression;
 
    auto const messageBytes = messageSize(message);
 
    XRPL_ASSERT(messageBytes, "xrpl::Message::Message : non-empty message input");
 
    buffer_.resize(kHeaderBytes + messageBytes);
 
    setHeader(buffer_.data(), messageBytes, type, Algorithm::None, 0);
 
    if (messageBytes != 0)
        message.SerializeToArray(buffer_.data() + kHeaderBytes, messageBytes);
 
    XRPL_ASSERT(
        getBufferSize() == totalSize(message),
        "xrpl::Message::Message : message size matches the buffer");
}
 
// static
std::size_t
Message::messageSize(::google::protobuf::Message const& message)
{
#if defined(GOOGLE_PROTOBUF_VERSION) && (GOOGLE_PROTOBUF_VERSION >= 3011000)
    return message.ByteSizeLong();
#else
    return message.ByteSize();
#endif
}
 
// static
std::size_t
Message::totalSize(::google::protobuf::Message const& message)
{
    return messageSize(message) + compression::kHeaderBytes;
}
 
void
Message::compress()
{
    using namespace xrpl::compression;
    auto const messageBytes = buffer_.size() - kHeaderBytes;
 
    auto type = getType(buffer_.data());
 
    bool const compressible = [&] {
        if (messageBytes <= 70)
            return false;
 
        // NOLINTNEXTLINE(bugprone-switch-missing-default-case)
        switch (type)
        {
            case protocol::mtMANIFESTS:
            case protocol::mtENDPOINTS:
            case protocol::mtTRANSACTION:
            case protocol::mtGET_LEDGER:
            case protocol::mtLEDGER_DATA:
            case protocol::mtGET_OBJECTS:
            case protocol::mtVALIDATOR_LIST:
            case protocol::mtVALIDATOR_LIST_COLLECTION:
            case protocol::mtREPLAY_DELTA_RESPONSE:
            case protocol::mtTRANSACTIONS:
                return true;
            case protocol::mtPING:
            case protocol::mtCLUSTER:
            case protocol::mtPROPOSE_LEDGER:
            case protocol::mtSTATUS_CHANGE:
            case protocol::mtHAVE_SET:
            case protocol::mtVALIDATION:
            case protocol::mtPROOF_PATH_REQ:
            case protocol::mtPROOF_PATH_RESPONSE:
            case protocol::mtREPLAY_DELTA_REQ:
            case protocol::mtHAVE_TRANSACTIONS:
                break;
        }
        return false;
    }();
 
    if (compressible)
    {
        auto payload = static_cast<void const*>(buffer_.data() + kHeaderBytes);
 
        auto compressedSize = xrpl::compression::compress(
            payload,
            messageBytes,
            [&](std::size_t inSize) {  // size of required compressed buffer
                bufferCompressed_.resize(inSize + kHeaderBytesCompressed);
                return (bufferCompressed_.data() + kHeaderBytesCompressed);
            });
 
        if (compressedSize < (messageBytes - (kHeaderBytesCompressed - kHeaderBytes)))
        {
            bufferCompressed_.resize(kHeaderBytesCompressed + compressedSize);
            // NOLINTNEXTLINE(readability-suspicious-call-argument)
            setHeader(bufferCompressed_.data(), compressedSize, type, Algorithm::LZ4, messageBytes);
        }
        else
        {
            bufferCompressed_.resize(0);
        }
    }
}

void
Message::setHeader(
    std::uint8_t* in,
    std::uint32_t payloadBytes,
    int type,
    Algorithm compression,
    std::uint32_t uncompressedBytes)
{
    auto h = in;
 
    auto pack = [](std::uint8_t*& in, std::uint32_t size) {
        *in++ = static_cast<std::uint8_t>((size >> 24) & 0x0F);  // leftmost 4 are compression bits
        *in++ = static_cast<std::uint8_t>((size >> 16) & 0xFF);
        *in++ = static_cast<std::uint8_t>((size >> 8) & 0xFF);
        *in++ = static_cast<std::uint8_t>(size & 0xFF);
    };
 
    pack(in, payloadBytes);
 
    *in++ = static_cast<std::uint8_t>((type >> 8) & 0xFF);
    *in++ = static_cast<std::uint8_t>(type & 0xFF);
 
    if (compression != Algorithm::None)
    {
        pack(in, uncompressedBytes);
        *h |= static_cast<std::uint8_t>(compression);
    }
}
 
std::size_t
Message::getBufferSize()
{
    return buffer_.size();
}
 
std::vector<uint8_t> const&
Message::getBuffer(Compressed tryCompressed)
{
    if (tryCompressed == Compressed::Off)
        return buffer_;
 
    std::call_once(onceFlag_, &Message::compress, this);
 
    if (!bufferCompressed_.empty())
    {
        return bufferCompressed_;
    }
 
    return buffer_;
}
 
int
Message::getType(std::uint8_t const* in)
{
    int const type = (static_cast<int>(*(in + 4)) << 8) + *(in + 5);
    return type;
}
 
}  // namespace xrpl