<commit_before>//
// detail/impl/response_read.h
// ---------------------------------
//
// Copyright (c) 2015 <NAME>
//
// Distributed under MIT license. (See file LICENSE)
//

#ifndef REQUEST_READ_H_2C4DEC07_68A7_48A3_A19E_4ECDE2AF19D9
#define REQUEST_READ_H_2C4DEC07_68A7_48A3_A19E_4ECDE2AF19D9

#include <boost/asio.hpp>
#include <libkafka_asio/detail/endian.h>
#include <libkafka_asio/detail/compression.h>

namespace libkafka_asio
{
namespace detail
{

inline Int8 ReadInt8(std::istream& is)
{
  Int8 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int8));
  return result;
}

inline Int16 ReadInt16(std::istream& is)
{
  Int16 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int16));
  result = ntohs(result);
  return result;
}

inline Int32 ReadInt32(std::istream& is)
{
  Int32 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int32));
  result = ntohl(result);
  return result;
}

inline Int64 ReadInt64(std::istream& is)
{
  Int64 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int64));
  result = be_to_host_64(result);
  return result;
}

inline String ReadString(std::istream& is)
{
  Int16 length = ReadInt16(is);
  if (length > 0)
  {
    String result(length, '\0');
    is.read(&result[0], length);
    return result;
  }
  return "";
}

inline void ReadBytes(std::istream& is, Bytes& bytes)
{
  Int32 length = ReadInt32(is);
  if (length > 0)
  {
    bytes.reset(new Bytes::element_type(length, 0));
    is.read(reinterpret_cast<char *>(&(*bytes)[0]), length);
  }
}

inline void ReadMessage(std::istream& is, Message& message)
{
  Int32 crc = ReadInt32(is);
  Int8 magic_byte = ReadInt8(is);
  Int8 attributes = ReadInt8(is);
  ReadBytes(is, message.mutable_key());
  ReadBytes(is, message.mutable_value());
  if (magic_byte == 0x01)
  {
    message.SetCompression(attributes);
    boost::system::error_code ec;
    Bytes data = Decompress(message.value(), message.compression(), ec);
    if (!ec && data && !data->empty())
    {
      std::stringbuf intermediate_buffer;
      intermediate_buffer.pubsetbuf(reinterpret_cast<char*>(&(*data)[0]),
                                    data->size());
      std::istream intermediate_is(&intermediate_buffer);
      ReadMessageSet(is, message.mutable_nested_message_set(), data->size());
    }
  }
}

inline void ReadMessageSet(std::istream& is,
                           MessageSet& message_set,
                           size_t size)
{
  size_t read_count = 0;
  while (read_count < size)
  {
    MessageAndOffset message;
    message.set_offset(ReadInt64(is));
    Int32 message_size = ReadInt32(is);
    if ((Int32)(size - read_count) < message_size)
    {
      // Ignore partial messages
      is.seekg(size - read_count, std::ios::cur);
      read_count = size;
    }
    else
    {
      ReadMessage(is, message);<commit_msg>Fixed recursive read of compressed message sets<commit_after>//
// detail/impl/response_read.h
// ---------------------------------
//
// Copyright (c) 2015 <NAME>
//
// Distributed under MIT license. (See file LICENSE)
//

#ifndef REQUEST_READ_H_2C4DEC07_68A7_48A3_A19E_4ECDE2AF19D9
#define REQUEST_READ_H_2C4DEC07_68A7_48A3_A19E_4ECDE2AF19D9

#include <boost/asio.hpp>
#include <libkafka_asio/detail/endian.h>
#include <libkafka_asio/detail/compression.h>

namespace libkafka_asio
{
namespace detail
{

inline Int8 ReadInt8(std::istream& is)
{
  Int8 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int8));
  return result;
}

inline Int16 ReadInt16(std::istream& is)
{
  Int16 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int16));
  result = ntohs(result);
  return result;
}

inline Int32 ReadInt32(std::istream& is)
{
  Int32 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int32));
  result = ntohl(result);
  return result;
}

inline Int64 ReadInt64(std::istream& is)
{
  Int64 result = 0;
  is.read(reinterpret_cast<char *>(&result), sizeof(Int64));
  result = be_to_host_64(result);
  return result;
}

inline String ReadString(std::istream& is)
{
  Int16 length = ReadInt16(is);
  if (length > 0)
  {
    String result(length, '\0');
    is.read(&result[0], length);
    return result;
  }
  return "";
}

inline void ReadBytes(std::istream& is, Bytes& bytes)
{
  Int32 length = ReadInt32(is);
  if (length > 0)
  {
    bytes.reset(new Bytes::element_type(length, 0));
    is.read(reinterpret_cast<char *>(&(*bytes)[0]), length);
  }
}

inline void ReadMessage(std::istream& is, Message& message)
{
  Int32 crc = ReadInt32(is);
  Int8 magic_byte = ReadInt8(is);
  Int8 attributes = ReadInt8(is);
  ReadBytes(is, message.mutable_key());
  ReadBytes(is, message.mutable_value());

  message.SetCompression(attributes);
  boost::system::error_code ec;
  Bytes data = Decompress(message.value(), message.compression(), ec);
  if (!ec && data && !data->empty())
  {
    std::stringbuf intermediate_buffer;
    intermediate_buffer.pubsetbuf(reinterpret_cast<char *>(&(*data)[0]),
                                  data->size());
    std::istream intermediate_is(&intermediate_buffer);
    ReadMessageSet(intermediate_is, message.mutable_nested_message_set(),
                   data->size());
  }

}

inline void ReadMessageSet(std::istream& is,
                           MessageSet& message_set,
                           size_t size)
{
  size_t read_count = 0;
  while (read_count < size)
  {
    MessageAndOffset message;
    message.set_offset(ReadInt64(is));
    Int32 message_size = ReadInt32(is);
    if ((Int32) (size - read_count) < message_size)
    {
      // Ignore partial messages
      is.seekg(size - read_count, std::ios::cur);
      read_count = size;
    }
    else
    {
      ReadMessage(is, message);
{'fraction_non_alphanumeric': 0.1511142061281337, 'fraction_numerical': 0.03673398328690808, 'mean_word_length': 3.516509433962264, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7240272', 'n_tokens_mistral': 2132, 'n_tokens_neox': 2024, 'n_words': 394}