Add net::SourceStream and net::FilterSourceStream

net/filter/filter_source_stream_unittest.cc

Index: net/filter/filter_source_stream_unittest.cc
diff --git a/net/filter/filter_source_stream_unittest.cc b/net/filter/filter_source_stream_unittest.cc
new file mode 100644
index 0000000000000000000000000000000000000000..3650de2c7daae6f99db2259d258516c61955567f
--- /dev/null
+++ b/net/filter/filter_source_stream_unittest.cc
@@ -0,0 +1,296 @@
+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <algorithm>
+#include <string>
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/macros.h"
+#include "base/numerics/safe_conversions.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/base/test_completion_callback.h"
+#include "net/filter/filter_source_stream.h"
+#include "net/filter/mock_source_stream.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace net {
+
+namespace {
+
+const size_t kDefaultBufferSize = 4096;
+const size_t kSmallBufferSize = 1;
+
+// A FilterSourceStream that needs all input data before it can return non-zero
+// bytes read.
+class NeedAllInputFilterSourceStream : public FilterSourceStream {
+ public:
+  NeedAllInputFilterSourceStream(std::unique_ptr<SourceStream> upstream,
+                                 size_t expected_remaining_bytes)
+      : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)),
+        expected_remaining_bytes_(expected_remaining_bytes) {}
+
+  int FilterData(IOBuffer* output_buffer,
+                 size_t output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    buffer_.append(input_buffer->data(), input_buffer->BytesRemaining());
+    expected_remaining_bytes_ -= input_buffer->BytesRemaining();
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    if (expected_remaining_bytes_ > 0) {
+      // Keep returning 0 byte read until all |expected_remaining_bytes| have
+      // been read from |upstream|.
+      return OK;
+    }
+    EXPECT_EQ(0u, expected_remaining_bytes_);
+    size_t bytes_to_filter = std::min(buffer_.length(), output_buffer_size);
+    memcpy(output_buffer->data(), buffer_.data(), bytes_to_filter);
+    buffer_.erase(0, bytes_to_filter);
+    return base::checked_cast<int>(bytes_to_filter);
+  }
+
+  std::string GetTypeAsString() const override { return ""; }
+
+ private:
+  // Buffer data that is yet to be passed through to the caller.
+  std::string buffer_;
+  // Expected remaining bytes to be received from |upstream|.
+  size_t expected_remaining_bytes_;
+
+  DISALLOW_COPY_AND_ASSIGN(NeedAllInputFilterSourceStream);
+};
+
+// A FilterSourceStream passes through data unchanged to consumer.
+class PassThroughFilterSourceStream : public FilterSourceStream {
+ public:
+  PassThroughFilterSourceStream(std::unique_ptr<SourceStream> upstream)
+      : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
+
+  int FilterData(IOBuffer* output_buffer,
+                 size_t output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    int bytes_flushed = Flush(output_buffer, output_buffer_size);
+    output_buffer_size -= bytes_flushed;
+    size_t bytes_to_filter =
+        base::checked_cast<size_t>(input_buffer->BytesRemaining());
+    if (output_buffer_size < bytes_to_filter) {
+      buffer_.append(input_buffer->data() + output_buffer_size,
+                     bytes_to_filter - output_buffer_size);
+      bytes_to_filter = output_buffer_size;
+    }
+    memcpy(output_buffer->data() + bytes_flushed, input_buffer->data(),
+           bytes_to_filter);
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    return bytes_flushed + base::checked_cast<int>(bytes_to_filter);
+  }
+
+  std::string GetTypeAsString() const override { return ""; }
+
+ private:
+  int Flush(IOBuffer* output_buffer, size_t output_buffer_size) {
+    int to_flush = std::min(buffer_.length(), output_buffer_size);
+    memcpy(output_buffer->data(), buffer_.data(), to_flush);
+    buffer_.clear();
+    return to_flush;
+  }
+
+  // Buffer data that is yet to be passed through to the caller.
+  std::string buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(PassThroughFilterSourceStream);
+};
+
+// A FilterSourceStream return an error code in FilterData().
+class ErrorFilterSourceStream : public FilterSourceStream {
+ public:
+  ErrorFilterSourceStream(std::unique_ptr<SourceStream> upstream)
+      : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
+
+  int FilterData(IOBuffer* output_buffer,
+                 size_t output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    return ERR_CONTENT_DECODING_FAILED;
+  }
+
+  std::string GetTypeAsString() const override { return ""; }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(ErrorFilterSourceStream);
+};
+
+}  // namespace
+
+// Tests that a FilterSourceStream subclass (NeedAllInputFilterSourceStream)
+// can return 0 bytes for FilterData()s when it has not consumed EOF from the
+// upstream. In this case, FilterSourceStream should continue reading from
+// upstream to complete filtering.
+TEST(FilterSourceStreamTest, FilterDataReturnNoBytesExceptLast) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input("hello, world!");
+  int read_size = 2;
+  int num_bytes_remaining = input.length();
+  // Add a sequence of small reads.
+  while (num_bytes_remaining > 0) {
+    int bytes_added = std::min(read_size, num_bytes_remaining);
+    source->AddReadResult(input.data() + input.length() - num_bytes_remaining,
+                          bytes_added, OK, /*synchronous=*/true);
+    num_bytes_remaining -= bytes_added;
+  }
+  NeedAllInputFilterSourceStream stream(std::move(source), input.length());
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                       callback.callback());
+  EXPECT_EQ(static_cast<int>(input.length()), rv);
+  EXPECT_EQ(0, memcmp(input.data(), output_buffer->data(), rv));
+}
+
+// Tests that FilterData() returns 0 because the upstream gives an EOF
+// synchronously.
+TEST(FilterSourceStreamTest, FilterDataReturnNoBytesSync) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input;
+  source->AddReadResult(input.data(), 0, OK, /*synchronous=*/true);
+  PassThroughFilterSourceStream stream(std::move(source));
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                       callback.callback());
+  EXPECT_EQ(OK, rv);
+}
+
+// Tests that FilterData() returns 0 because the upstream gives an EOF
+// asynchronously.
+TEST(FilterSourceStreamTest, FilterDataReturnNoBytesSyncAsync) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  MockSourceStream* source_ptr = source.get();
+  std::string input;
+  source->AddReadResult(input.data(), 0, OK, /*synchronous=*/false);
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  PassThroughFilterSourceStream stream(std::move(source));
+  int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                       callback.callback());
+  EXPECT_EQ(ERR_IO_PENDING, rv);
+  source_ptr->CompleteNextRead();
+  EXPECT_EQ(OK, callback.WaitForResult());
+}
+
+// Tests that FilterData() returns non-zero bytes because the upstream
+// returns data synchronously.
+TEST(FilterSourceStreamTest, FilterDataReturnDataSync) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  int read_size = 2;
+  int num_bytes_remaining = input.length();
+  // Add a sequence of small reads.
+  while (num_bytes_remaining > 0) {
+    int bytes_added = std::min(read_size, num_bytes_remaining);
+    source->AddReadResult(input.data() + input.length() - num_bytes_remaining,
+                          bytes_added, OK, /*synchronous=*/true);
+    num_bytes_remaining -= bytes_added;
+  }
+  PassThroughFilterSourceStream stream(std::move(source));
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  std::string actual_output;
+  int rv = OK;
+  do {
+    rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                     callback.callback());
+    EXPECT_GE(read_size, rv);
+    EXPECT_GE(rv, OK);
+    actual_output.append(output_buffer->data(), rv);
+  } while (rv != OK);
+  EXPECT_EQ(actual_output, input);
+}
+
+// Tests that FilterData() returns non-zero bytes because the upstream
+// returns data asynchronously.
+TEST(FilterSourceStreamTest, FilterDataReturnDataAsync) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  MockSourceStream* source_ptr = source.get();
+  std::string input = "hello, world!";
+  int read_size = 2;
+  int num_bytes_remaining = input.length();
+  // Add a sequence of small reads.
+  while (num_bytes_remaining > 0) {
+    int bytes_added = std::min(read_size, num_bytes_remaining);
+    source->AddReadResult(input.data() + input.length() - num_bytes_remaining,
+                          bytes_added, OK, /*synchronous=*/false);
+    num_bytes_remaining -= bytes_added;
+  }
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  PassThroughFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  std::string actual_output;
+  int rv = OK;
+  while (true) {
+    rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                     callback.callback());
+    if (rv == OK)
+      break;
+    EXPECT_EQ(ERR_IO_PENDING, rv);
+    source_ptr->CompleteNextRead();
+    rv = callback.WaitForResult();
+    EXPECT_GE(read_size, rv);
+    EXPECT_GE(rv, OK);
+    actual_output.append(output_buffer->data(), rv);
+  };
+  EXPECT_EQ(actual_output, input);
+}
+
+// Tests that FilterData() returns non-zero bytes and output buffer size is
+// smaller than the number of bytes read from the upstream.
+TEST(FilterSourceStreamTest, FilterDataNeedsOutputSpace) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  int read_size = 2;
+  int num_bytes_remaining = input.length();
+  // Add a sequence of small reads.
+  while (num_bytes_remaining > 0) {
+    int bytes_added = std::min(read_size, num_bytes_remaining);
+    source->AddReadResult(input.data() + input.length() - num_bytes_remaining,
+                          bytes_added, OK, /*synchronous=*/true);
+    num_bytes_remaining -= bytes_added;
+  }
+  // Use an extremely small buffer size, so FilterData will need more output
+  // space.
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kSmallBufferSize);
+  PassThroughFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  std::string actual_output;
+  int rv = OK;
+  do {
+    rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                     callback.callback());
+    EXPECT_GE(rv, OK);
+    EXPECT_GE(kSmallBufferSize, static_cast<size_t>(rv));
+    actual_output.append(output_buffer->data(), rv);
+  } while (rv != OK);
+  EXPECT_EQ(actual_output, input);
+}
+
+// Tests that FilterData() returns an error code, which is then surfaced as the
+// result of calling Read().
+TEST(FilterSourceStreamTest, FilterDataReturnError) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input;
+  source->AddReadResult(input.data(), 0, OK, /*synchronous=*/true);
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  ErrorFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                       callback.callback());
+  EXPECT_EQ(ERR_CONTENT_DECODING_FAILED, rv);
+}
+
+}  // namespace net