Function CheckStreamingRoundtrip

cpp/src/arrow/util/compression_test.cc:227–308

// Check the streaming compressor and decompressor together

void CheckStreamingRoundtrip(std::shared_ptr<Compressor> compressor,
                             std::shared_ptr<Decompressor> decompressor,
                             const std::vector<uint8_t>& data) {
  std::default_random_engine engine(42);
  std::uniform_int_distribution<int> buf_size_distribution(10, 40);

  auto make_buf_size = [&]() -> int64_t { return buf_size_distribution(engine); };

  // Compress...

  std::vector<uint8_t> compressed(1);
  int64_t compressed_size = 0;
  {
    const uint8_t* input = data.data();
    int64_t remaining = data.size();

    while (remaining > 0) {
      // Feed a varying amount each time
      int64_t input_len = std::min(remaining, make_buf_size());
      int64_t output_len = compressed.size() - compressed_size;
      uint8_t* output = compressed.data() + compressed_size;
      ASSERT_OK_AND_ASSIGN(auto result,
                           compressor->Compress(input_len, input, output_len, output));
      ASSERT_LE(result.bytes_read, input_len);
      ASSERT_LE(result.bytes_written, output_len);
      compressed_size += result.bytes_written;
      input += result.bytes_read;
      remaining -= result.bytes_read;
      if (result.bytes_read == 0) {
        compressed.resize(compressed.capacity() * 2);
      }
    }
    // End the compressed stream
    Compressor::EndResult result;
    do {
      int64_t output_len = compressed.size() - compressed_size;
      uint8_t* output = compressed.data() + compressed_size;
      ASSERT_OK_AND_ASSIGN(result, compressor->End(output_len, output));
      ASSERT_LE(result.bytes_written, output_len);
      compressed_size += result.bytes_written;
      if (result.should_retry) {
        compressed.resize(compressed.capacity() * 2);
      }
    } while (result.should_retry);

    compressed.resize(compressed_size);
  }

  // Then decompress...

  std::vector<uint8_t> decompressed(2);
  int64_t decompressed_size = 0;
  {
    const uint8_t* input = compressed.data();
    int64_t remaining = compressed.size();

    while (!decompressor->IsFinished()) {
      // Feed a varying amount each time