text:
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc.  All rights reserved.
// https://developers.google.com/protocol-buffers/
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//     * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//     * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

// Author: kenton@google.com (Kenton Varda)
//  Based on original Protocol Buffers design by
//  Sanjay Ghemawat, Jeff Dean, and others.
//
// Testing strategy:  For each type of I/O (array, string, file, etc.) we
// create an output stream and write some data to it, then create a
// corresponding input stream to read the same data back and expect it to
// match.  When the data is written, it is written in several small chunks
// of varying sizes, with a BackUp() after each chunk.  It is read back
// similarly, but with chunks separated at different points.  The whole
// process is run with a variety of block sizes for both the input and
// the output.
//
// TODO(kenton):  Rewrite this test to bring it up to the standards of all
//   the other proto2 tests.  May want to wait for gTest to implement
//   "parametized tests" so that one set of tests can be used on all the
//   implementations.

#ifndef _MSC_VER
#include <unistd.h>
#endif
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <memory>
#ifndef _SHARED_PTR_H
#include <google/protobuf/stubs/shared_ptr.h>
#endif
#include <sstream>

#include <google/protobuf/testing/file.h>
#include <google/protobuf/io/coded_stream.h>
#include <google/protobuf/io/zero_copy_stream_impl.h>

#if HAVE_ZLIB
#include <google/protobuf/io/gzip_stream.h>
#endif

#include <google/protobuf/stubs/common.h>
#include <google/protobuf/stubs/logging.h>
#include <google/protobuf/testing/googletest.h>
#include <google/protobuf/testing/file.h>
#include <gtest/gtest.h>
#include <google/protobuf/stubs/io_win32.h>

namespace google {
namespace protobuf {
namespace io {
namespace {

#ifdef _WIN32
#define pipe(fds) _pipe(fds, 4096, O_BINARY)
// DO NOT include <io.h>, instead create functions in io_win32.{h,cc} and import
// them like we do below.
using google::protobuf::internal::win32::access;
using google::protobuf::internal::win32::mkdir;
using google::protobuf::internal::win32::open;
using google::protobuf::internal::win32::close;
#endif

#ifndef O_BINARY
#ifdef _O_BINARY
#define O_BINARY _O_BINARY
#else
#define O_BINARY 0     // If this isn't defined, the platform doesn't need it.
#endif
#endif

class IoTest : public testing::Test {
 protected:
  // Test helpers.

  // Helper to write an array of data to an output stream.
  bool WriteToOutput(ZeroCopyOutputStream* output, const void* data, int size);
  // Helper to read a fixed-length array of data from an input stream.
  int ReadFromInput(ZeroCopyInputStream* input, void* data, int size);
  // Write a string to the output stream.
  void WriteString(ZeroCopyOutputStream* output, const string& str);
  // Read a number of bytes equal to the size of the given string and checks
  // that it matches the string.
  void ReadString(ZeroCopyInputStream* input, const string& str);
  // Writes some text to the output stream in a particular order.  Returns
  // the number of bytes written, incase the caller needs that to set up an
  // input stream.
  int WriteStuff(ZeroCopyOutputStream* output);
  // Reads text from an input stream and expects it to match what
  // WriteStuff() writes.
  void ReadStuff(ZeroCopyInputStream* input);

  // Similar to WriteStuff, but performs more sophisticated testing.
  int WriteStuffLarge(ZeroCopyOutputStream* output);
  // Reads and tests a stream that should have been written to
  // via WriteStuffLarge().
  void ReadStuffLarge(ZeroCopyInputStream* input);

#if HAVE_ZLIB
  string Compress(const string& data, const GzipOutputStream::Options& options);
  string Uncompress(const string& data);
#endif

  static const int kBlockSizes[];
  static const int kBlockSizeCount;
};

const int IoTest::kBlockSizes[] = {-1, 1, 2, 5, 7, 10, 23, 64};
const int IoTest::kBlockSizeCount = GOOGLE_ARRAYSIZE(IoTest::kBlockSizes);

bool IoTest::WriteToOutput(ZeroCopyOutputStream* output,
                           const void* data, int size) {
  const uint8* in = reinterpret_cast<const uint8*>(data);
  int in_size = size;

  void* out;
  int out_size;

  while (true) {
    if (!output->Next(&out, &out_size)) {
      return false;
    }
    EXPECT_GT(out_size, 0);

    if (in_size <= out_size) {
      memcpy(out, in, in_size);
      output->BackUp(out_size - in_size);
      return true;
    }

    memcpy(out, in, out_size);
    in += out_size;
    in_size -= out_size;
  }
}

#define MAX_REPEATED_ZEROS 100

int IoTest::ReadFromInput(ZeroCopyInputStream* input, void* data, int size) {
  uint8* out = reinterpret_cast<uint8*>(data);
  int out_size = size;

  const void* in;
  int in_size = 0;

  int repeated_zeros = 0;

  while (true) {
    if (!input->Next(&in, &in_size)) {
      return size - out_size;
    }
    EXPECT_GT(in_size, -1);
    if (in_size == 0) {
      repeated_zeros++;
    } else {
      repeated_zeros = 0;
    }
    EXPECT_LT(repeated_zeros, MAX_REPEATED_ZEROS);

    if (out_size <= in_size) {
      memcpy(out, in, out_size);
      if (in_size > out_size) {
        input->BackUp(in_size - out_size);
      }
      return size;  // Copied all of it.
    }

    memcpy(out, in, in_size);
    out += in_size;
    out_size -= in_size;
  }
}

void IoTest::WriteString(ZeroCopyOutputStream* output, const string& str) {
  EXPECT_TRUE(WriteToOutput(output, str.c_str(), str.size()));
}

void IoTest::ReadString(ZeroCopyInputStream* input, const string& str) {
  google::protobuf::scoped_array<char> buffer(new char[str.size() + 1]);
  buffer[str.size()] = '\0';
  EXPECT_EQ(ReadFromInput(input, buffer.get(), str.size()), str.size());
  EXPECT_STREQ(str.c_str(), buffer.get());
}

int IoTest::WriteStuff(ZeroCopyOutputStream* output) {
  WriteString(output, "Hello world!\n");
  WriteString(output, "Some te");
  WriteString(output, "xt.  Blah blah.");
  WriteString(output, "abcdefg");
  WriteString(output, "01234567890123456789");
  WriteString(output, "foobar");

  EXPECT_EQ(output->ByteCount(), 68);

  int result = output->ByteCount();
  return result;
}

// Reads text from an input stream and expects it to match what WriteStuff()
// writes.
void IoTest::ReadStuff(ZeroCopyInputStream* input) {
  ReadString(input, "Hello world!\n");
  ReadString(input, "Some text.  ");
  ReadString(input, "Blah ");
  ReadString(input, "blah.");
  ReadString(input, "abcdefg");
  EXPECT_TRUE(input->Skip(20));
  ReadString(input, "foo");
  ReadString(input, "bar");

  EXPECT_EQ(input->ByteCount(), 68);

  uint8 byte;
  EXPECT_EQ(ReadFromInput(input, &byte, 1), 0);
}

int IoTest::WriteStuffLarge(ZeroCopyOutputStream* output) {
  WriteString(output, "Hello world!\n");
  WriteString(output, "Some te");
  WriteString(output, "xt.  Blah blah.");
  WriteString(output, string(100000, 'x'));  // A very long string
  WriteString(output, string(100000, 'y'));  // A very long string
  WriteString(output, "01234567890123456789");

  EXPECT_EQ(output->ByteCount(), 200055);

  int result = output->ByteCount();
  return result;
}

// Reads text from an input stream and expects it to match what WriteStuff()
// writes.
void IoTest::ReadStuffLarge(ZeroCopyInputStream* input) {
  ReadString(input, "Hello world!\nSome text.  ");
  EXPECT_TRUE(input->Skip(5));
  ReadString(input, "blah.");
  EXPECT_TRUE(input->Skip(100000 - 10));
  ReadString(input, string(10, 'x') + string(100000 - 20000, 'y'));
  EXPECT_TRUE(input->Skip(20000 - 10));
  ReadString(input, "yyyyyyyyyy01234567890123456789");

  EXPECT_EQ(input->ByteCount(), 200055);

  uint8 byte;
  EXPECT_EQ(ReadFromInput(input, &byte, 1), 0);
}

// ===================================================================

TEST_F(IoTest, ArrayIo) {
  const int kBufferSize = 256;
  uint8 buffer[kBufferSize];

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      int size;
      {
        ArrayOutputStream output(buffer, kBufferSize, kBlockSizes[i]);
        size = WriteStuff(&output);
      }
      {
        ArrayInputStream input(buffer, size, kBlockSizes[j]);
        ReadStuff(&input);
      }
    }
  }
}

TEST_F(IoTest, TwoSessionWrite) {
  // Test that two concatenated write sessions read correctly

  static const char* strA = "0123456789";
  static const char* strB = "WhirledPeas";
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  char* temp_buffer = new char[40];

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      ArrayOutputStream* output =
          new ArrayOutputStream(buffer, kBufferSize, kBlockSizes[i]);
      CodedOutputStream* coded_output = new CodedOutputStream(output);
      coded_output->WriteVarint32(strlen(strA));
      coded_output->WriteRaw(strA, strlen(strA));
      delete coded_output;  // flush
      int64 pos = output->ByteCount();
      delete output;
      output = new ArrayOutputStream(
          buffer + pos, kBufferSize - pos, kBlockSizes[i]);
      coded_output = new CodedOutputStream(output);
      coded_output->WriteVarint32(strlen(strB));
      coded_output->WriteRaw(strB, strlen(strB));
      delete coded_output;  // flush
      int64 size = pos + output->ByteCount();
      delete output;

      ArrayInputStream* input =
          new ArrayInputStream(buffer, size, kBlockSizes[j]);
      CodedInputStream* coded_input = new CodedInputStream(input);
      uint32 insize;
      EXPECT_TRUE(coded_input->ReadVarint32(&insize));
      EXPECT_EQ(strlen(strA), insize);
      EXPECT_TRUE(coded_input->ReadRaw(temp_buffer, insize));
      EXPECT_EQ(0, memcmp(temp_buffer, strA, insize));

      EXPECT_TRUE(coded_input->ReadVarint32(&insize));
      EXPECT_EQ(strlen(strB), insize);
      EXPECT_TRUE(coded_input->ReadRaw(temp_buffer, insize));
      EXPECT_EQ(0, memcmp(temp_buffer, strB, insize));

      delete coded_input;
      delete input;
    }
  }

  delete [] temp_buffer;
  delete [] buffer;
}

#if HAVE_ZLIB
TEST_F(IoTest, GzipIo) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      for (int z = 0; z < kBlockSizeCount; z++) {
        int gzip_buffer_size = kBlockSizes[z];
        int size;
        {
          ArrayOutputStream output(buffer, kBufferSize, kBlockSizes[i]);
          GzipOutputStream::Options options;
          options.format = GzipOutputStream::GZIP;
          if (gzip_buffer_size != -1) {
            options.buffer_size = gzip_buffer_size;
          }
          GzipOutputStream gzout(&output, options);
          WriteStuff(&gzout);
          gzout.Close();
          size = output.ByteCount();
        }
        {
          ArrayInputStream input(buffer, size, kBlockSizes[j]);
          GzipInputStream gzin(
              &input, GzipInputStream::GZIP, gzip_buffer_size);
          ReadStuff(&gzin);
        }
      }
    }
  }
  delete [] buffer;
}

TEST_F(IoTest, GzipIoWithFlush) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  // We start with i = 4 as we want a block size > 6. With block size <= 6
  // Flush() fills up the entire 2K buffer with flush markers and the test
  // fails. See documentation for Flush() for more detail.
  for (int i = 4; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      for (int z = 0; z < kBlockSizeCount; z++) {
        int gzip_buffer_size = kBlockSizes[z];
        int size;
        {
          ArrayOutputStream output(buffer, kBufferSize, kBlockSizes[i]);
          GzipOutputStream::Options options;
          options.format = GzipOutputStream::GZIP;
          if (gzip_buffer_size != -1) {
            options.buffer_size = gzip_buffer_size;
          }
          GzipOutputStream gzout(&output, options);
          WriteStuff(&gzout);
          EXPECT_TRUE(gzout.Flush());
          gzout.Close();
          size = output.ByteCount();
        }
        {
          ArrayInputStream input(buffer, size, kBlockSizes[j]);
          GzipInputStream gzin(
              &input, GzipInputStream::GZIP, gzip_buffer_size);
          ReadStuff(&gzin);
        }
      }
    }
  }
  delete [] buffer;
}

TEST_F(IoTest, GzipIoContiguousFlushes) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];

  int block_size = kBlockSizes[4];
  int gzip_buffer_size = block_size;
  int size;

  ArrayOutputStream output(buffer, kBufferSize, block_size);
  GzipOutputStream::Options options;
  options.format = GzipOutputStream::GZIP;
  if (gzip_buffer_size != -1) {
    options.buffer_size = gzip_buffer_size;
  }
  GzipOutputStream gzout(&output, options);
  WriteStuff(&gzout);
  EXPECT_TRUE(gzout.Flush());
  EXPECT_TRUE(gzout.Flush());
  gzout.Close();
  size = output.ByteCount();

  ArrayInputStream input(buffer, size, block_size);
  GzipInputStream gzin(
      &input, GzipInputStream::GZIP, gzip_buffer_size);
  ReadStuff(&gzin);

  delete [] buffer;
}

TEST_F(IoTest, GzipIoReadAfterFlush) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];

  int block_size = kBlockSizes[4];
  int gzip_buffer_size = block_size;
  int size;
  ArrayOutputStream output(buffer, kBufferSize, block_size);
  GzipOutputStream::Options options;
  options.format = GzipOutputStream::GZIP;
  if (gzip_buffer_size != -1) {
    options.buffer_size = gzip_buffer_size;
  }

  GzipOutputStream gzout(&output, options);
  WriteStuff(&gzout);
  EXPECT_TRUE(gzout.Flush());
  size = output.ByteCount();

  ArrayInputStream input(buffer, size, block_size);
  GzipInputStream gzin(
      &input, GzipInputStream::GZIP, gzip_buffer_size);
  ReadStuff(&gzin);

  gzout.Close();

  delete [] buffer;
}

TEST_F(IoTest, ZlibIo) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      for (int z = 0; z < kBlockSizeCount; z++) {
        int gzip_buffer_size = kBlockSizes[z];
        int size;
        {
          ArrayOutputStream output(buffer, kBufferSize, kBlockSizes[i]);
          GzipOutputStream::Options options;
          options.format = GzipOutputStream::ZLIB;
          if (gzip_buffer_size != -1) {
            options.buffer_size = gzip_buffer_size;
          }
          GzipOutputStream gzout(&output, options);
          WriteStuff(&gzout);
          gzout.Close();
          size = output.ByteCount();
        }
        {
          ArrayInputStream input(buffer, size, kBlockSizes[j]);
          GzipInputStream gzin(
              &input, GzipInputStream::ZLIB, gzip_buffer_size);
          ReadStuff(&gzin);
        }
      }
    }
  }
  delete [] buffer;
}

TEST_F(IoTest, ZlibIoInputAutodetect) {
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  int size;
  {
    ArrayOutputStream output(buffer, kBufferSize);
    GzipOutputStream::Options options;
    options.format = GzipOutputStream::ZLIB;
    GzipOutputStream gzout(&output, options);
    WriteStuff(&gzout);
    gzout.Close();
    size = output.ByteCount();
  }
  {
    ArrayInputStream input(buffer, size);
    GzipInputStream gzin(&input, GzipInputStream::AUTO);
    ReadStuff(&gzin);
  }
  {
    ArrayOutputStream output(buffer, kBufferSize);
    GzipOutputStream::Options options;
    options.format = GzipOutputStream::GZIP;
    GzipOutputStream gzout(&output, options);
    WriteStuff(&gzout);
    gzout.Close();
    size = output.ByteCount();
  }
  {
    ArrayInputStream input(buffer, size);
    GzipInputStream gzin(&input, GzipInputStream::AUTO);
    ReadStuff(&gzin);
  }
  delete [] buffer;
}

string IoTest::Compress(const string& data,
                        const GzipOutputStream::Options& options) {
  string result;
  {
    StringOutputStream output(&result);
    GzipOutputStream gzout(&output, options);
    WriteToOutput(&gzout, data.data(), data.size());
  }
  return result;
}

string IoTest::Uncompress(const string& data) {
  string result;
  {
    ArrayInputStream input(data.data(), data.size());
    GzipInputStream gzin(&input);
    const void* buffer;
    int size;
    while (gzin.Next(&buffer, &size)) {
      result.append(reinterpret_cast<const char*>(buffer), size);
    }
  }
  return result;
}

TEST_F(IoTest, CompressionOptions) {
  // Some ad-hoc testing of compression options.

  string golden;
  GOOGLE_CHECK_OK(File::GetContents(
      TestSourceDir() +
          "/google/protobuf/testdata/golden_message",
      &golden, true));

  GzipOutputStream::Options options;
  string gzip_compressed = Compress(golden, options);

  options.compression_level = 0;
  string not_compressed = Compress(golden, options);

  // Try zlib compression for fun.
  options = GzipOutputStream::Options();
  options.format = GzipOutputStream::ZLIB;
  string zlib_compressed = Compress(golden, options);

  // Uncompressed should be bigger than the original since it should have some
  // sort of header.
  EXPECT_GT(not_compressed.size(), golden.size());

  // Higher compression levels should result in smaller sizes.
  EXPECT_LT(zlib_compressed.size(), not_compressed.size());

  // ZLIB format should differ from GZIP format.
  EXPECT_TRUE(zlib_compressed != gzip_compressed);

  // Everything should decompress correctly.
  EXPECT_TRUE(Uncompress(not_compressed) == golden);
  EXPECT_TRUE(Uncompress(gzip_compressed) == golden);
  EXPECT_TRUE(Uncompress(zlib_compressed) == golden);
}

TEST_F(IoTest, TwoSessionWriteGzip) {
  // Test that two concatenated gzip streams can be read correctly

  static const char* strA = "0123456789";
  static const char* strB = "QuickBrownFox";
  const int kBufferSize = 2*1024;
  uint8* buffer = new uint8[kBufferSize];
  char* temp_buffer = new char[40];

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      ArrayOutputStream* output =
          new ArrayOutputStream(buffer, kBufferSize, kBlockSizes[i]);
      GzipOutputStream* gzout = new GzipOutputStream(output);
      CodedOutputStream* coded_output = new CodedOutputStream(gzout);
      int32 outlen = strlen(strA) + 1;
      coded_output->WriteVarint32(outlen);
      coded_output->WriteRaw(strA, outlen);
      delete coded_output;  // flush
      delete gzout;  // flush
      int64 pos = output->ByteCount();
      delete output;
      output = new ArrayOutputStream(
          buffer + pos, kBufferSize - pos, kBlockSizes[i]);
      gzout = new GzipOutputStream(output);
      coded_output = new CodedOutputStream(gzout);
      outlen = strlen(strB) + 1;
      coded_output->WriteVarint32(outlen);
      coded_output->WriteRaw(strB, outlen);
      delete coded_output;  // flush
      delete gzout;  // flush
      int64 size = pos + output->ByteCount();
      delete output;

      ArrayInputStream* input =
          new ArrayInputStream(buffer, size, kBlockSizes[j]);
      GzipInputStream* gzin = new GzipInputStream(input);
      CodedInputStream* coded_input = new CodedInputStream(gzin);
      uint32 insize;
      EXPECT_TRUE(coded_input->ReadVarint32(&insize));
      EXPECT_EQ(strlen(strA) + 1, insize);
      EXPECT_TRUE(coded_input->ReadRaw(temp_buffer, insize));
      EXPECT_EQ(0, memcmp(temp_buffer, strA, insize))
          << "strA=" << strA << " in=" << temp_buffer;

      EXPECT_TRUE(coded_input->ReadVarint32(&insize));
      EXPECT_EQ(strlen(strB) + 1, insize);
      EXPECT_TRUE(coded_input->ReadRaw(temp_buffer, insize));
      EXPECT_EQ(0, memcmp(temp_buffer, strB, insize))
          << " out_block_size=" << kBlockSizes[i]
          << " in_block_size=" << kBlockSizes[j]
          << " pos=" << pos
          << " size=" << size
          << " strB=" << strB << " in=" << temp_buffer;

      delete coded_input;
      delete gzin;
      delete input;
    }
  }

  delete [] temp_buffer;
  delete [] buffer;
}

TEST_F(IoTest, GzipInputByteCountAfterClosed) {
  string golden = "abcdefghijklmnopqrstuvwxyz";
  string compressed = Compress(golden, GzipOutputStream::Options());

  for (int i = 0; i < kBlockSizeCount; i++) {
    ArrayInputStream arr_input(compressed.data(), compressed.size(),
                               kBlockSizes[i]);
    GzipInputStream gz_input(&arr_input);
    const void* buffer;
    int size;
    while (gz_input.Next(&buffer, &size)) {
      EXPECT_LE(gz_input.ByteCount(), golden.size());
    }
    EXPECT_EQ(golden.size(), gz_input.ByteCount());
  }
}

TEST_F(IoTest, GzipInputByteCountAfterClosedConcatenatedStreams) {
  string golden1 = "abcdefghijklmnopqrstuvwxyz";
  string golden2 = "the quick brown fox jumps over the lazy dog";
  const size_t total_size = golden1.size() + golden2.size();
  string compressed = Compress(golden1, GzipOutputStream::Options()) +
                      Compress(golden2, GzipOutputStream::Options());

  for (int i = 0; i < kBlockSizeCount; i++) {
    ArrayInputStream arr_input(compressed.data(), compressed.size(),
                               kBlockSizes[i]);
    GzipInputStream gz_input(&arr_input);
    const void* buffer;
    int size;
    while (gz_input.Next(&buffer, &size)) {
      EXPECT_LE(gz_input.ByteCount(), total_size);
    }
    EXPECT_EQ(total_size, gz_input.ByteCount());
  }
}
#endif

// There is no string input, only string output.  Also, it doesn't support
// explicit block sizes.  So, we'll only run one test and we'll use
// ArrayInput to read back the results.
TEST_F(IoTest, StringIo) {
  string str;
  {
    StringOutputStream output(&str);
    WriteStuff(&output);
  }
  {
    ArrayInputStream input(str.data(), str.size());
    ReadStuff(&input);
  }
}

// To test files, we create a temporary file, write, read, truncate, repeat.
TEST_F(IoTest, FileIo) {
  string filename = TestTempDir() + "/zero_copy_stream_test_file";

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      // Make a temporary file.
      int file =
        open(filename.c_str(), O_RDWR | O_CREAT | O_TRUNC | O_BINARY, 0777);
      ASSERT_GE(file, 0);

      {
        FileOutputStream output(file, kBlockSizes[i]);
        WriteStuff(&output);
        EXPECT_EQ(0, output.GetErrno());
      }

      // Rewind.
      ASSERT_NE(lseek(file, 0, SEEK_SET), (off_t)-1);

      {
        FileInputStream input(file, kBlockSizes[j]);
        ReadStuff(&input);
        EXPECT_EQ(0, input.GetErrno());
      }

      close(file);
    }
  }
}

#if HAVE_ZLIB
TEST_F(IoTest, GzipFileIo) {
  string filename = TestTempDir() + "/zero_copy_stream_test_file";

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      // Make a temporary file.
      int file =
        open(filename.c_str(), O_RDWR | O_CREAT | O_TRUNC | O_BINARY, 0777);
      ASSERT_GE(file, 0);
      {
        FileOutputStream output(file, kBlockSizes[i]);
        GzipOutputStream gzout(&output);
        WriteStuffLarge(&gzout);
        gzout.Close();
        output.Flush();
        EXPECT_EQ(0, output.GetErrno());
      }

      // Rewind.
      ASSERT_NE(lseek(file, 0, SEEK_SET), (off_t)-1);

      {
        FileInputStream input(file, kBlockSizes[j]);
        GzipInputStream gzin(&input);
        ReadStuffLarge(&gzin);
        EXPECT_EQ(0, input.GetErrno());
      }

      close(file);
    }
  }
}
#endif

// MSVC raises various debugging exceptions if we try to use a file
// descriptor of -1, defeating our tests below.  This class will disable
// these debug assertions while in scope.
class MsvcDebugDisabler {
 public:
#if defined(_MSC_VER) && _MSC_VER >= 1400
  MsvcDebugDisabler() {
    old_handler_ = _set_invalid_parameter_handler(MyHandler);
    old_mode_ = _CrtSetReportMode(_CRT_ASSERT, 0);
  }
  ~MsvcDebugDisabler() {
    old_handler_ = _set_invalid_parameter_handler(old_handler_);
    old_mode_ = _CrtSetReportMode(_CRT_ASSERT, old_mode_);
  }

  static void MyHandler(const wchar_t *expr,
                        const wchar_t *func,
                        const wchar_t *file,
                        unsigned int line,
                        uintptr_t pReserved) {
    // do nothing
  }

  _invalid_parameter_handler old_handler_;
  int old_mode_;
#else
  // Dummy constructor and destructor to ensure that GCC doesn't complain
  // that debug_disabler is an unused variable.
  MsvcDebugDisabler() {}
  ~MsvcDebugDisabler() {}
#endif
};

// Test that FileInputStreams report errors correctly.
TEST_F(IoTest, FileReadError) {
  MsvcDebugDisabler debug_disabler;

  // -1 = invalid file descriptor.
  FileInputStream input(-1);

  const void* buffer;
  int size;
  EXPECT_FALSE(input.Next(&buffer, &size));
  EXPECT_EQ(EBADF, input.GetErrno());
}

// Test that FileOutputStreams report errors correctly.
TEST_F(IoTest, FileWriteError) {
  MsvcDebugDisabler debug_disabler;

  // -1 = invalid file descriptor.
  FileOutputStream input(-1);

  void* buffer;
  int size;

  // The first call to Next() succeeds because it doesn't have anything to
  // write yet.
  EXPECT_TRUE(input.Next(&buffer, &size));

  // Second call fails.
  EXPECT_FALSE(input.Next(&buffer, &size));

  EXPECT_EQ(EBADF, input.GetErrno());
}

// Pipes are not seekable, so File{Input,Output}Stream ends up doing some
// different things to handle them.  We'll test by writing to a pipe and
// reading back from it.
TEST_F(IoTest, PipeIo) {
  int files[2];

  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      // Need to create a new pipe each time because ReadStuff() expects
      // to see EOF at the end.
      ASSERT_EQ(pipe(files), 0);

      {
        FileOutputStream output(files[1], kBlockSizes[i]);
        WriteStuff(&output);
        EXPECT_EQ(0, output.GetErrno());
      }
      close(files[1]);  // Send EOF.

      {
        FileInputStream input(files[0], kBlockSizes[j]);
        ReadStuff(&input);
        EXPECT_EQ(0, input.GetErrno());
      }
      close(files[0]);
    }
  }
}

// Test using C++ iostreams.
TEST_F(IoTest, IostreamIo) {
  for (int i = 0; i < kBlockSizeCount; i++) {
    for (int j = 0; j < kBlockSizeCount; j++) {
      {
        std::stringstream stream;

        {
          OstreamOutputStream output(&stream, kBlockSizes[i]);
          WriteStuff(&output);
          EXPECT_FALSE(stream.fail());
        }

        {
          IstreamInputStream input(&stream, kBlockSizes[j]);
          ReadStuff(&input);
          EXPECT_TRUE(stream.eof());
        }
      }

      {
        std::stringstream stream;

        {
          OstreamOutputStream output(&stream, kBlockSizes[i]);
          WriteStuffLarge(&output);
          EXPECT_FALSE(stream.fail());
        }

        {
          IstreamInputStream input(&stream, kBlockSizes[j]);
          ReadStuffLarge(&input);
          EXPECT_TRUE(stream.eof());
        }
      }
    }
  }
}

// To test ConcatenatingInputStream, we create several ArrayInputStreams
// covering a buffer and then concatenate them.
TEST_F(IoTest, ConcatenatingInputStream) {
  const int kBufferSize = 256;
  uint8 buffer[kBufferSize];

  // Fill the buffer.
  ArrayOutputStream output(buffer, kBufferSize);
  WriteStuff(&output);

  // Now split it up into multiple streams of varying sizes.
  ASSERT_EQ(68, output.ByteCount());  // Test depends on this.
  ArrayInputStream input1(buffer     , 12);
  ArrayInputStream input2(buffer + 12,  7);
  ArrayInputStream input3(buffer + 19,  6);
  ArrayInputStream input4(buffer + 25, 15);
  ArrayInputStream input5(buffer + 40,  0);
  // Note:  We want to make sure we have a stream boundary somewhere between
  // bytes 42 and 62, which is the range that it Skip()ed by ReadStuff().  This
  // tests that a bug that existed in the original code for Skip() is fixed.
  ArrayInputStream input6(buffer + 40, 10);
  ArrayInputStream input7(buffer + 50, 18);  // Total = 68 bytes.

  ZeroCopyInputStream* streams[] =
    {&input1, &input2, &input3, &input4, &input5, &input6, &input7};

  // Create the concatenating stream and read.
  ConcatenatingInputStream input(streams, GOOGLE_ARRAYSIZE(streams));
  ReadStuff(&input);
}

// To test LimitingInputStream, we write our golden text to a buffer, then
// create an ArrayInputStream that contains the whole buffer (not just the
// bytes written), then use a LimitingInputStream to limit it just to the
// bytes written.
TEST_F(IoTest, LimitingInputStream) {
  const int kBufferSize = 256;
  uint8 buffer[kBufferSize];

  // Fill the buffer.
  ArrayOutputStream output(buffer, kBufferSize);
  WriteStuff(&output);

  // Set up input.
  ArrayInputStream array_input(buffer, kBufferSize);
  LimitingInputStream input(&array_input, output.ByteCount());

  ReadStuff(&input);
}

// Checks that ByteCount works correctly for LimitingInputStreams where the
// underlying stream has already been read.
TEST_F(IoTest, LimitingInputStreamByteCount) {
  const int kHalfBufferSize = 128;
  const int kBufferSize = kHalfBufferSize * 2;
  uint8 buffer[kBufferSize];

  // Set up input. Only allow half to be read at once.
  ArrayInputStream array_input(buffer, kBufferSize, kHalfBufferSize);
  const void* data;
  int size;
  EXPECT_TRUE(array_input.Next(&data, &size));
  EXPECT_EQ(kHalfBufferSize, array_input.ByteCount());
  // kHalfBufferSize - 1 to test limiting logic as well.
  LimitingInputStream input(&array_input, kHalfBufferSize - 1);
  EXPECT_EQ(0, input.ByteCount());
  EXPECT_TRUE(input.Next(&data, &size));
  EXPECT_EQ(kHalfBufferSize - 1 , input.ByteCount());
}

// Check that a zero-size array doesn't confuse the code.
TEST(ZeroSizeArray, Input) {
  ArrayInputStream input(NULL, 0);
  const void* data;
  int size;
  EXPECT_FALSE(input.Next(&data, &size));
}

TEST(ZeroSizeArray, Output) {
  ArrayOutputStream output(NULL, 0);
  void* data;
  int size;
  EXPECT_FALSE(output.Next(&data, &size));
}

}  // namespace
}  // namespace io
}  // namespace protobuf
}  // namespace google


text:
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc.  All rights reserved.
// https://developers.google.com/protocol-buffers/
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//     * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//     * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

// Author: kenton@google.com (Kenton Varda)
//  Based on original Protocol Buffers design by
//  Sanjay Ghemawat, Jeff Dean, and others.

#ifndef _MSC_VER
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#endif
#include <errno.h>
#include <iostream>
#include <algorithm>
#include <google/protobuf/io/zero_copy_stream_impl.h>
#include <google/protobuf/stubs/common.h>
#include <google/protobuf/stubs/logging.h>
#include <google/protobuf/stubs/stl_util.h>
#include <google/protobuf/stubs/io_win32.h>

namespace google {
namespace protobuf {
namespace io {

#ifdef _WIN32
// Win32 lseek is broken:  If invoked on a non-seekable file descriptor, its
// return value is undefined.  We re-define it to always produce an error.
#define lseek(fd, offset, origin) ((off_t)-1)
// DO NOT include <io.h>, instead create functions in io_win32.{h,cc} and import
// them like we do below.
using google::protobuf::internal::win32::access;
using google::protobuf::internal::win32::close;
using google::protobuf::internal::win32::open;
using google::protobuf::internal::win32::read;
using google::protobuf::internal::win32::write;
#endif

namespace {

// EINTR sucks.
int close_no_eintr(int fd) {
  int result;
  do {
    result = close(fd);
  } while (result < 0 && errno == EINTR);
  return result;
}

}  // namespace

// ===================================================================

FileInputStream::FileInputStream(int file_descriptor, int block_size)
  : copying_input_(file_descriptor),
    impl_(&copying_input_, block_size) {
}

bool FileInputStream::Close() {
  return copying_input_.Close();
}

bool FileInputStream::Next(const void** data, int* size) {
  return impl_.Next(data, size);
}

void FileInputStream::BackUp(int count) {
  impl_.BackUp(count);
}

bool FileInputStream::Skip(int count) {
  return impl_.Skip(count);
}

int64 FileInputStream::ByteCount() const {
  return impl_.ByteCount();
}

FileInputStream::CopyingFileInputStream::CopyingFileInputStream(
    int file_descriptor)
  : file_(file_descriptor),
    close_on_delete_(false),
    is_closed_(false),
    errno_(0),
    previous_seek_failed_(false) {
}

FileInputStream::CopyingFileInputStream::~CopyingFileInputStream() {
  if (close_on_delete_) {
    if (!Close()) {
      GOOGLE_LOG(ERROR) << "close() failed: " << strerror(errno_);
    }
  }
}

bool FileInputStream::CopyingFileInputStream::Close() {
  GOOGLE_CHECK(!is_closed_);

  is_closed_ = true;
  if (close_no_eintr(file_) != 0) {
    // The docs on close() do not specify whether a file descriptor is still
    // open after close() fails with EIO.  However, the glibc source code
    // seems to indicate that it is not.
    errno_ = errno;
    return false;
  }

  return true;
}

int FileInputStream::CopyingFileInputStream::Read(void* buffer, int size) {
  GOOGLE_CHECK(!is_closed_);

  int result;
  do {
    result = read(file_, buffer, size);
  } while (result < 0 && errno == EINTR);

  if (result < 0) {
    // Read error (not EOF).
    errno_ = errno;
  }

  return result;
}

int FileInputStream::CopyingFileInputStream::Skip(int count) {
  GOOGLE_CHECK(!is_closed_);

  if (!previous_seek_failed_ &&
      lseek(file_, count, SEEK_CUR) != (off_t)-1) {
    // Seek succeeded.
    return count;
  } else {
    // Failed to seek.

    // Note to self:  Don't seek again.  This file descriptor doesn't
    // support it.
    previous_seek_failed_ = true;

    // Use the default implementation.
    return CopyingInputStream::Skip(count);
  }
}

// ===================================================================

FileOutputStream::FileOutputStream(int file_descriptor, int block_size)
  : copying_output_(file_descriptor),
    impl_(&copying_output_, block_size) {
}

FileOutputStream::~FileOutputStream() {
  impl_.Flush();
}

bool FileOutputStream::Close() {
  bool flush_succeeded = impl_.Flush();
  return copying_output_.Close() && flush_succeeded;
}

bool FileOutputStream::Flush() {
  return impl_.Flush();
}

bool FileOutputStream::Next(void** data, int* size) {
  return impl_.Next(data, size);
}

void FileOutputStream::BackUp(int count) {
  impl_.BackUp(count);
}

int64 FileOutputStream::ByteCount() const {
  return impl_.ByteCount();
}

FileOutputStream::CopyingFileOutputStream::CopyingFileOutputStream(
    int file_descriptor)
  : file_(file_descriptor),
    close_on_delete_(false),
    is_closed_(false),
    errno_(0) {
}

FileOutputStream::CopyingFileOutputStream::~CopyingFileOutputStream() {
  if (close_on_delete_) {
    if (!Close()) {
      GOOGLE_LOG(ERROR) << "close() failed: " << strerror(errno_);
    }
  }
}

bool FileOutputStream::CopyingFileOutputStream::Close() {
  GOOGLE_CHECK(!is_closed_);

  is_closed_ = true;
  if (close_no_eintr(file_) != 0) {
    // The docs on close() do not specify whether a file descriptor is still
    // open after close() fails with EIO.  However, the glibc source code
    // seems to indicate that it is not.
    errno_ = errno;
    return false;
  }

  return true;
}

bool FileOutputStream::CopyingFileOutputStream::Write(
    const void* buffer, int size) {
  GOOGLE_CHECK(!is_closed_);
  int total_written = 0;

  const uint8* buffer_base = reinterpret_cast<const uint8*>(buffer);

  while (total_written < size) {
    int bytes;
    do {
      bytes = write(file_, buffer_base + total_written, size - total_written);
    } while (bytes < 0 && errno == EINTR);

    if (bytes <= 0) {
      // Write error.

      // FIXME(kenton):  According to the man page, if write() returns zero,
      //   there was no error; write() simply did not write anything.  It's
      //   unclear under what circumstances this might happen, but presumably
      //   errno won't be set in this case.  I am confused as to how such an
      //   event should be handled.  For now I'm treating it as an error, since
      //   retrying seems like it could lead to an infinite loop.  I suspect
      //   this never actually happens anyway.

      if (bytes < 0) {
        errno_ = errno;
      }
      return false;
    }
    total_written += bytes;
  }

  return true;
}

// ===================================================================

IstreamInputStream::IstreamInputStream(std::istream* input, int block_size)
    : copying_input_(input), impl_(&copying_input_, block_size) {}

bool IstreamInputStream::Next(const void** data, int* size) {
  return impl_.Next(data, size);
}

void IstreamInputStream::BackUp(int count) {
  impl_.BackUp(count);
}

bool IstreamInputStream::Skip(int count) {
  return impl_.Skip(count);
}

int64 IstreamInputStream::ByteCount() const {
  return impl_.ByteCount();
}

IstreamInputStream::CopyingIstreamInputStream::CopyingIstreamInputStream(
    std::istream* input)
    : input_(input) {}

IstreamInputStream::CopyingIstreamInputStream::~CopyingIstreamInputStream() {}

int IstreamInputStream::CopyingIstreamInputStream::Read(
    void* buffer, int size) {
  input_->read(reinterpret_cast<char*>(buffer), size);
  int result = input_->gcount();
  if (result == 0 && input_->fail() && !input_->eof()) {
    return -1;
  }
  return result;
}

// ===================================================================

OstreamOutputStream::OstreamOutputStream(std::ostream* output, int block_size)
    : copying_output_(output), impl_(&copying_output_, block_size) {}

OstreamOutputStream::~OstreamOutputStream() {
  impl_.Flush();
}

bool OstreamOutputStream::Next(void** data, int* size) {
  return impl_.Next(data, size);
}

void OstreamOutputStream::BackUp(int count) {
  impl_.BackUp(count);
}

int64 OstreamOutputStream::ByteCount() const {
  return impl_.ByteCount();
}

OstreamOutputStream::CopyingOstreamOutputStream::CopyingOstreamOutputStream(
    std::ostream* output)
    : output_(output) {}

OstreamOutputStream::CopyingOstreamOutputStream::~CopyingOstreamOutputStream() {
}

bool OstreamOutputStream::CopyingOstreamOutputStream::Write(
    const void* buffer, int size) {
  output_->write(reinterpret_cast<const char*>(buffer), size);
  return output_->good();
}

// ===================================================================

ConcatenatingInputStream::ConcatenatingInputStream(
    ZeroCopyInputStream* const streams[], int count)
  : streams_(streams), stream_count_(count), bytes_retired_(0) {
}

bool ConcatenatingInputStream::Next(const void** data, int* size) {
  while (stream_count_ > 0) {
    if (streams_[0]->Next(data, size)) return true;

    // That stream is done.  Advance to the next one.
    bytes_retired_ += streams_[0]->ByteCount();
    ++streams_;
    --stream_count_;
  }

  // No more streams.
  return false;
}

void ConcatenatingInputStream::BackUp(int count) {
  if (stream_count_ > 0) {
    streams_[0]->BackUp(count);
  } else {
    GOOGLE_LOG(DFATAL) << "Can't BackUp() after failed Next().";
  }
}

bool ConcatenatingInputStream::Skip(int count) {
  while (stream_count_ > 0) {
    // Assume that ByteCount() can be used to find out how much we actually
    // skipped when Skip() fails.
    int64 target_byte_count = streams_[0]->ByteCount() + count;
    if (streams_[0]->Skip(count)) return true;

    // Hit the end of the stream.  Figure out how many more bytes we still have
    // to skip.
    int64 final_byte_count = streams_[0]->ByteCount();
    GOOGLE_DCHECK_LT(final_byte_count, target_byte_count);
    count = target_byte_count - final_byte_count;

    // That stream is done.  Advance to the next one.
    bytes_retired_ += final_byte_count;
    ++streams_;
    --stream_count_;
  }

  return false;
}

int64 ConcatenatingInputStream::ByteCount() const {
  if (stream_count_ == 0) {
    return bytes_retired_;
  } else {
    return bytes_retired_ + streams_[0]->ByteCount();
  }
}

// ===================================================================

LimitingInputStream::LimitingInputStream(ZeroCopyInputStream* input,
                                         int64 limit)
  : input_(input), limit_(limit) {
  prior_bytes_read_ = input_->ByteCount();
}

LimitingInputStream::~LimitingInputStream() {
  // If we overshot the limit, back up.
  if (limit_ < 0) input_->BackUp(-limit_);
}

bool LimitingInputStream::Next(const void** data, int* size) {
  if (limit_ <= 0) return false;
  if (!input_->Next(data, size)) return false;

  limit_ -= *size;
  if (limit_ < 0) {
    // We overshot the limit.  Reduce *size to hide the rest of the buffer.
    *size += limit_;
  }
  return true;
}

void LimitingInputStream::BackUp(int count) {
  if (limit_ < 0) {
    input_->BackUp(count - limit_);
    limit_ = count;
  } else {
    input_->BackUp(count);
    limit_ += count;
  }
}

bool LimitingInputStream::Skip(int count) {
  if (count > limit_) {
    if (limit_ < 0) return false;
    input_->Skip(limit_);
    limit_ = 0;
    return false;
  } else {
    if (!input_->Skip(count)) return false;
    limit_ -= count;
    return true;
  }
}

int64 LimitingInputStream::ByteCount() const {
  if (limit_ < 0) {
    return input_->ByteCount() + limit_ - prior_bytes_read_;
  } else {
    return input_->ByteCount() - prior_bytes_read_;
  }
}

// ===================================================================

}  // namespace io
}  // namespace protobuf
}  // namespace google

mime_type: text/plain


text:
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc.  All rights reserved.
// https://developers.google.com/protocol-buffers/
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//     * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//     * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

// Author: kenton@google.com (Kenton Varda)
//  Based on original Protocol Buffers design by
//  Sanjay Ghemawat, Jeff Dean, and others.
//
// This file contains the ZeroCopyInputStream and ZeroCopyOutputStream
// interfaces, which represent abstract I/O streams to and from which
// protocol buffers can be read and written.  For a few simple
// implementations of these interfaces, see zero_copy_stream_impl.h.
//
// These interfaces are different from classic I/O streams in that they
// try to minimize the amount of data copying that needs to be done.
// To accomplish this, responsibility for allocating buffers is moved to
// the stream object, rather than being the responsibility of the caller.
// So, the stream can return a buffer which actually points directly into
// the final data structure where the bytes are to be stored, and the caller
// can interact directly with that buffer, eliminating an intermediate copy
// operation.
//
// As an example, consider the common case in which you are reading bytes
// from an array that is already in memory (or perhaps an mmap()ed file).
// With classic I/O streams, you would do something like:
//   char buffer[BUFFER_SIZE];
//   input->Read(buffer, BUFFER_SIZE);
//   DoSomething(buffer, BUFFER_SIZE);
// Then, the stream basically just calls memcpy() to copy the data from
// the array into your buffer.  With a ZeroCopyInputStream, you would do
// this instead:
//   const void* buffer;
//   int size;
//   input->Next(&buffer, &size);
//   DoSomething(buffer, size);
// Here, no copy is performed.  The input stream returns a pointer directly
// into the backing array, and the caller ends up reading directly from it.
//
// If you want to be able to read the old-fashion way, you can create
// a CodedInputStream or CodedOutputStream wrapping these objects and use
// their ReadRaw()/WriteRaw() methods.  These will, of course, add a copy
// step, but Coded*Stream will handle buffering so at least it will be
// reasonably efficient.
//
// ZeroCopyInputStream example:
//   // Read in a file and print its contents to stdout.
//   int fd = open("myfile", O_RDONLY);
//   ZeroCopyInputStream* input = new FileInputStream(fd);
//
//   const void* buffer;
//   int size;
//   while (input->Next(&buffer, &size)) {
//     cout.write(buffer, size);
//   }
//
//   delete input;
//   close(fd);
//
// ZeroCopyOutputStream example:
//   // Copy the contents of "infile" to "outfile", using plain read() for
//   // "infile" but a ZeroCopyOutputStream for "outfile".
//   int infd = open("infile", O_RDONLY);
//   int outfd = open("outfile", O_WRONLY);
//   ZeroCopyOutputStream* output = new FileOutputStream(outfd);
//
//   void* buffer;
//   int size;
//   while (output->Next(&buffer, &size)) {
//     int bytes = read(infd, buffer, size);
//     if (bytes < size) {
//       // Reached EOF.
//       output->BackUp(size - bytes);
//       break;
//     }
//   }
//
//   delete output;
//   close(infd);
//   close(outfd);

#ifndef GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_H__
#define GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_H__

#include <string>
#include <google/protobuf/stubs/common.h>

namespace google {

namespace protobuf {
namespace io {

// Defined in this file.
class ZeroCopyInputStream;
class ZeroCopyOutputStream;

// Abstract interface similar to an input stream but designed to minimize
// copying.
class LIBPROTOBUF_EXPORT ZeroCopyInputStream {
 public:
  ZeroCopyInputStream() {}
  virtual ~ZeroCopyInputStream() {}

  // Obtains a chunk of data from the stream.
  //
  // Preconditions:
  // * "size" and "data" are not NULL.
  //
  // Postconditions:
  // * If the returned value is false, there is no more data to return or
  //   an error occurred.  All errors are permanent.
  // * Otherwise, "size" points to the actual number of bytes read and "data"
  //   points to a pointer to a buffer containing these bytes.
  // * Ownership of this buffer remains with the stream, and the buffer
  //   remains valid only until some other method of the stream is called
  //   or the stream is destroyed.
  // * It is legal for the returned buffer to have zero size, as long
  //   as repeatedly calling Next() eventually yields a buffer with non-zero
  //   size.
  virtual bool Next(const void** data, int* size) = 0;

  // Backs up a number of bytes, so that the next call to Next() returns
  // data again that was already returned by the last call to Next().  This
  // is useful when writing procedures that are only supposed to read up
  // to a certain point in the input, then return.  If Next() returns a
  // buffer that goes beyond what you wanted to read, you can use BackUp()
  // to return to the point where you intended to finish.
  //
  // Preconditions:
  // * The last method called must have been Next().
  // * count must be less than or equal to the size of the last buffer
  //   returned by Next().
  //
  // Postconditions:
  // * The last "count" bytes of the last buffer returned by Next() will be
  //   pushed back into the stream.  Subsequent calls to Next() will return
  //   the same data again before producing new data.
  virtual void BackUp(int count) = 0;

  // Skips a number of bytes.  Returns false if the end of the stream is
  // reached or some input error occurred.  In the end-of-stream case, the
  // stream is advanced to the end of the stream (so ByteCount() will return
  // the total size of the stream).
  virtual bool Skip(int count) = 0;

  // Returns the total number of bytes read since this object was created.
  virtual int64 ByteCount() const = 0;

 private:
  GOOGLE_DISALLOW_EVIL_CONSTRUCTORS(ZeroCopyInputStream);
};

// Abstract interface similar to an output stream but designed to minimize
// copying.
class LIBPROTOBUF_EXPORT ZeroCopyOutputStream {
 public:
  ZeroCopyOutputStream() {}
  virtual ~ZeroCopyOutputStream() {}

  // Obtains a buffer into which data can be written.  Any data written
  // into this buffer will eventually (maybe instantly, maybe later on)
  // be written to the output.
  //
  // Preconditions:
  // * "size" and "data" are not NULL.
  //
  // Postconditions:
  // * If the returned value is false, an error occurred.  All errors are
  //   permanent.
  // * Otherwise, "size" points to the actual number of bytes in the buffer
  //   and "data" points to the buffer.
  // * Ownership of this buffer remains with the stream, and the buffer
  //   remains valid only until some other method of the stream is called
  //   or the stream is destroyed.
  // * Any data which the caller stores in this buffer will eventually be
  //   written to the output (unless BackUp() is called).
  // * It is legal for the returned buffer to have zero size, as long
  //   as repeatedly calling Next() eventually yields a buffer with non-zero
  //   size.
  virtual bool Next(void** data, int* size) = 0;

  // Backs up a number of bytes, so that the end of the last buffer returned
  // by Next() is not actually written.  This is needed when you finish
  // writing all the data you want to write, but the last buffer was bigger
  // than you needed.  You don't want to write a bunch of garbage after the
  // end of your data, so you use BackUp() to back up.
  //
  // Preconditions:
  // * The last method called must have been Next().
  // * count must be less than or equal to the size of the last buffer
  //   returned by Next().
  // * The caller must not have written anything to the last "count" bytes
  //   of that buffer.
  //
  // Postconditions:
  // * The last "count" bytes of the last buffer returned by Next() will be
  //   ignored.
  virtual void BackUp(int count) = 0;

  // Returns the total number of bytes written since this object was created.
  virtual int64 ByteCount() const = 0;

  // Write a given chunk of data to the output.  Some output streams may
  // implement this in a way that avoids copying. Check AllowsAliasing() before
  // calling WriteAliasedRaw(). It will GOOGLE_CHECK fail if WriteAliasedRaw() is
  // called on a stream that does not allow aliasing.
  //
  // NOTE: It is caller's responsibility to ensure that the chunk of memory
  // remains live until all of the data has been consumed from the stream.
  virtual bool WriteAliasedRaw(const void* data, int size);
  virtual bool AllowsAliasing() const { return false; }

 private:
  GOOGLE_DISALLOW_EVIL_CONSTRUCTORS(ZeroCopyOutputStream);
};

}  // namespace io
}  // namespace protobuf

}  // namespace google
#endif  // GOOGLE_PROTOBUF_IO_ZERO_COPY_STREAM_H__


text:
// Protocol Buffers - Google's data interchange format
// Copyright 2008 Google Inc.  All rights reserved.
// https://developers.google.com/protocol-buffers/
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//     * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//     * Neither the name of Google Inc. nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

// Author: brianolson@google.com (Brian Olson)
//
// This file contains the implementation of classes GzipInputStream and
// GzipOutputStream.

#if HAVE_ZLIB
#include <google/protobuf/io/gzip_stream.h>

#include <google/protobuf/stubs/common.h>
#include <google/protobuf/stubs/logging.h>

namespace google {
namespace protobuf {
namespace io {

static const int kDefaultBufferSize = 65536;

GzipInputStream::GzipInputStream(
    ZeroCopyInputStream* sub_stream, Format format, int buffer_size)
    : format_(format), sub_stream_(sub_stream), zerror_(Z_OK), byte_count_(0) {
  zcontext_.state = Z_NULL;
  zcontext_.zalloc = Z_NULL;
  zcontext_.zfree = Z_NULL;
  zcontext_.opaque = Z_NULL;
  zcontext_.total_out = 0;
  zcontext_.next_in = NULL;
  zcontext_.avail_in = 0;
  zcontext_.total_in = 0;
  zcontext_.msg = NULL;
  if (buffer_size == -1) {
    output_buffer_length_ = kDefaultBufferSize;
  } else {
    output_buffer_length_ = buffer_size;
  }
  output_buffer_ = operator new(output_buffer_length_);
  GOOGLE_CHECK(output_buffer_ != NULL);
  zcontext_.next_out = static_cast<Bytef*>(output_buffer_);
  zcontext_.avail_out = output_buffer_length_;
  output_position_ = output_buffer_;
}
GzipInputStream::~GzipInputStream() {
  operator delete(output_buffer_);
  zerror_ = inflateEnd(&zcontext_);
}

static inline int internalInflateInit2(
    z_stream* zcontext, GzipInputStream::Format format) {
  int windowBitsFormat = 0;
  switch (format) {
    case GzipInputStream::GZIP: windowBitsFormat = 16; break;
    case GzipInputStream::AUTO: windowBitsFormat = 32; break;
    case GzipInputStream::ZLIB: windowBitsFormat = 0; break;
  }
  return inflateInit2(zcontext, /* windowBits */15 | windowBitsFormat);
}

int GzipInputStream::Inflate(int flush) {
  if ((zerror_ == Z_OK) && (zcontext_.avail_out == 0)) {
    // previous inflate filled output buffer. don't change input params yet.
  } else if (zcontext_.avail_in == 0) {
    const void* in;
    int in_size;
    bool first = zcontext_.next_in == NULL;
    bool ok = sub_stream_->Next(&in, &in_size);
    if (!ok) {
      zcontext_.next_out = NULL;
      zcontext_.avail_out = 0;
      return Z_STREAM_END;
    }
    zcontext_.next_in = static_cast<Bytef*>(const_cast<void*>(in));
    zcontext_.avail_in = in_size;
    if (first) {
      int error = internalInflateInit2(&zcontext_, format_);
      if (error != Z_OK) {
        return error;
      }
    }
  }
  zcontext_.next_out = static_cast<Bytef*>(output_buffer_);
  zcontext_.avail_out = output_buffer_length_;
  output_position_ = output_buffer_;
  int error = inflate(&zcontext_, flush);
  return error;
}

void GzipInputStream::DoNextOutput(const void** data, int* size) {
  *data = output_position_;
  *size = ((uintptr_t)zcontext_.next_out) - ((uintptr_t)output_position_);
  output_position_ = zcontext_.next_out;
}

// implements ZeroCopyInputStream ----------------------------------
bool GzipInputStream::Next(const void** data, int* size) {
  bool ok = (zerror_ == Z_OK) || (zerror_ == Z_STREAM_END)
      || (zerror_ == Z_BUF_ERROR);
  if ((!ok) || (zcontext_.next_out == NULL)) {
    return false;
  }
  if (zcontext_.next_out != output_position_) {
    DoNextOutput(data, size);
    return true;
  }
  if (zerror_ == Z_STREAM_END) {
    if (zcontext_.next_out != NULL) {
      // sub_stream_ may have concatenated streams to follow
      zerror_ = inflateEnd(&zcontext_);
      byte_count_ += zcontext_.total_out;
      if (zerror_ != Z_OK) {
        return false;
      }
      zerror_ = internalInflateInit2(&zcontext_, format_);
      if (zerror_ != Z_OK) {
        return false;
      }
    } else {
      *data = NULL;
      *size = 0;
      return false;
    }
  }
  zerror_ = Inflate(Z_NO_FLUSH);
  if ((zerror_ == Z_STREAM_END) && (zcontext_.next_out == NULL)) {
    // The underlying stream's Next returned false inside Inflate.
    return false;
  }
  ok = (zerror_ == Z_OK) || (zerror_ == Z_STREAM_END)
      || (zerror_ == Z_BUF_ERROR);
  if (!ok) {
    return false;
  }
  DoNextOutput(data, size);
  return true;
}
void GzipInputStream::BackUp(int count) {
  output_position_ = reinterpret_cast<void*>(
      reinterpret_cast<uintptr_t>(output_position_) - count);
}
bool GzipInputStream::Skip(int count) {
  const void* data;
  int size = 0;
  bool ok = Next(&data, &size);
  while (ok && (size < count)) {
    count -= size;
    ok = Next(&data, &size);
  }
  if (size > count) {
    BackUp(size - count);
  }
  return ok;
}
int64 GzipInputStream::ByteCount() const {
  int64 ret = byte_count_ + zcontext_.total_out;
  if (zcontext_.next_out != NULL && output_position_ != NULL) {
    ret += reinterpret_cast<uintptr_t>(zcontext_.next_out) -
           reinterpret_cast<uintptr_t>(output_position_);
  }
  return ret;
}

// =========================================================================

GzipOutputStream::Options::Options()
    : format(GZIP),
      buffer_size(kDefaultBufferSize),
      compression_level(Z_DEFAULT_COMPRESSION),
      compression_strategy(Z_DEFAULT_STRATEGY) {}

GzipOutputStream::GzipOutputStream(ZeroCopyOutputStream* sub_stream) {
  Init(sub_stream, Options());
}

GzipOutputStream::GzipOutputStream(ZeroCopyOutputStream* sub_stream,
                                   const Options& options) {
  Init(sub_stream, options);
}

void GzipOutputStream::Init(ZeroCopyOutputStream* sub_stream,
                            const Options& options) {
  sub_stream_ = sub_stream;
  sub_data_ = NULL;
  sub_data_size_ = 0;

  input_buffer_length_ = options.buffer_size;
  input_buffer_ = operator new(input_buffer_length_);
  GOOGLE_CHECK(input_buffer_ != NULL);

  zcontext_.zalloc = Z_NULL;
  zcontext_.zfree = Z_NULL;
  zcontext_.opaque = Z_NULL;
  zcontext_.next_out = NULL;
  zcontext_.avail_out = 0;
  zcontext_.total_out = 0;
  zcontext_.next_in = NULL;
  zcontext_.avail_in = 0;
  zcontext_.total_in = 0;
  zcontext_.msg = NULL;
  // default to GZIP format
  int windowBitsFormat = 16;
  if (options.format == ZLIB) {
    windowBitsFormat = 0;
  }
  zerror_ = deflateInit2(
      &zcontext_,
      options.compression_level,
      Z_DEFLATED,
      /* windowBits */15 | windowBitsFormat,
      /* memLevel (default) */8,
      options.compression_strategy);
}

GzipOutputStream::~GzipOutputStream() {
  Close();
  operator delete(input_buffer_);
}

// private
int GzipOutputStream::Deflate(int flush) {
  int error = Z_OK;
  do {
    if ((sub_data_ == NULL) || (zcontext_.avail_out == 0)) {
      bool ok = sub_stream_->Next(&sub_data_, &sub_data_size_);
      if (!ok) {
        sub_data_ = NULL;
        sub_data_size_ = 0;
        return Z_BUF_ERROR;
      }
      GOOGLE_CHECK_GT(sub_data_size_, 0);
      zcontext_.next_out = static_cast<Bytef*>(sub_data_);
      zcontext_.avail_out = sub_data_size_;
    }
    error = deflate(&zcontext_, flush);
  } while (error == Z_OK && zcontext_.avail_out == 0);
  if ((flush == Z_FULL_FLUSH) || (flush == Z_FINISH)) {
    // Notify lower layer of data.
    sub_stream_->BackUp(zcontext_.avail_out);
    // We don't own the buffer anymore.
    sub_data_ = NULL;
    sub_data_size_ = 0;
  }
  return error;
}

// implements ZeroCopyOutputStream ---------------------------------
bool GzipOutputStream::Next(void** data, int* size) {
  if ((zerror_ != Z_OK) && (zerror_ != Z_BUF_ERROR)) {
    return false;
  }
  if (zcontext_.avail_in != 0) {
    zerror_ = Deflate(Z_NO_FLUSH);
    if (zerror_ != Z_OK) {
      return false;
    }
  }
  if (zcontext_.avail_in == 0) {
    // all input was consumed. reset the buffer.
    zcontext_.next_in = static_cast<Bytef*>(input_buffer_);
    zcontext_.avail_in = input_buffer_length_;
    *data = input_buffer_;
    *size = input_buffer_length_;
  } else {
    // The loop in Deflate should consume all avail_in
    GOOGLE_LOG(DFATAL) << "Deflate left bytes unconsumed";
  }
  return true;
}
void GzipOutputStream::BackUp(int count) {
  GOOGLE_CHECK_GE(zcontext_.avail_in, count);
  zcontext_.avail_in -= count;
}
int64 GzipOutputStream::ByteCount() const {
  return zcontext_.total_in + zcontext_.avail_in;
}

bool GzipOutputStream::Flush() {
  zerror_ = Deflate(Z_FULL_FLUSH);
  // Return true if the flush succeeded or if it was a no-op.
  return  (zerror_ == Z_OK) ||
      (zerror_ == Z_BUF_ERROR && zcontext_.avail_in == 0 &&
       zcontext_.avail_out != 0);
}

bool GzipOutputStream::Close() {
  if ((zerror_ != Z_OK) && (zerror_ != Z_BUF_ERROR)) {
    return false;
  }
  do {
    zerror_ = Deflate(Z_FINISH);
  } while (zerror_ == Z_OK);
  zerror_ = deflateEnd(&zcontext_);
  bool ok = zerror_ == Z_OK;
  zerror_ = Z_STREAM_END;
  return ok;
}

}  // namespace io
}  // namespace protobuf
}  // namespace google

#endif  // HAVE_ZLIB