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..3851dd5c253e197ec9f5281cea4b2dcc47eebb85
--- /dev/null
+++ b/net/filter/filter_source_stream_unittest.cc
@@ -0,0 +1,535 @@
+// 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;
+
+class TestFilterSourceStreamBase : public FilterSourceStream {
+ public:
+  TestFilterSourceStreamBase(std::unique_ptr<SourceStream> upstream)
+      : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
+  ~TestFilterSourceStreamBase() override { DCHECK(buffer_.empty()); }
+  std::string GetTypeAsString() const override { return type_string_; }
+
+  void set_type_string(const std::string& type_string) {
+    type_string_ = type_string;
+  }
+
+ protected:
+  // Buffer used by subclasses to hold data that is yet to be passed to the
+  // caller.
+  std::string buffer_;
+
+ private:
+  std::string type_string_;
+
+  DISALLOW_COPY_AND_ASSIGN(TestFilterSourceStreamBase);
+};
+
+// A FilterSourceStream that needs all input data before it can return non-zero
+// bytes read.
+class NeedsAllInputFilterSourceStream : public TestFilterSourceStreamBase {
+ public:
+  NeedsAllInputFilterSourceStream(std::unique_ptr<SourceStream> upstream,
+                                  size_t expected_input_bytes)
+      : TestFilterSourceStreamBase(std::move(upstream)),
+        expected_input_bytes_(expected_input_bytes) {}
+  int FilterData(IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    buffer_.append(input_buffer->data(), input_buffer->BytesRemaining());
+    EXPECT_GE(expected_input_bytes_, input_buffer->BytesRemaining());
+    expected_input_bytes_ -= input_buffer->BytesRemaining();
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    if (expected_input_bytes_ > 0) {
+      // Keep returning 0 byte read until all |expected_input_bytes| have

[Randy Smith (Not in Mondays), 2016/08/22 22:37:16]

nit: "0 bytes".

[xunjieli, 2016/08/29 16:25:29]

Done.

+      // been read from |upstream|.
+      return OK;

[Randy Smith (Not in Mondays), 2016/08/22 22:37:16]

nitty nit, suggestion: "return 0" (since we're returning a byte count, not an
error code).

[xunjieli, 2016/08/29 16:25:29]

Done.

+    }
+    size_t bytes_to_filter =
+        std::min(buffer_.length(), static_cast<size_t>(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);
+  }
+
+ private:
+  // Expected remaining bytes to be received from |upstream|.
+  int expected_input_bytes_;
+
+  DISALLOW_COPY_AND_ASSIGN(NeedsAllInputFilterSourceStream);
+};
+
+// A FilterSourceStream that repeat every input byte by |multiplier| amount of
+// times.
+class MultiplySourceStream : public TestFilterSourceStreamBase {
+ public:
+  MultiplySourceStream(std::unique_ptr<SourceStream> upstream, int multiplier)
+      : TestFilterSourceStreamBase(std::move(upstream)),
+        multiplier_(multiplier) {}
+  int FilterData(IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    for (int i = 0; i < input_buffer->BytesRemaining(); i++) {
+      for (int j = 0; j < multiplier_; j++) {
+        buffer_.append(input_buffer->data() + i, 1);
+      }

[Randy Smith (Not in Mondays), 2016/08/22 22:37:16]

nit, suggestion: You don't need the curly braces here.

[xunjieli, 2016/08/29 16:25:29]

Done.

+    }
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    size_t bytes_to_filter =
+        std::min(buffer_.length(), static_cast<size_t>(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);
+  }
+
+ private:
+  int multiplier_;
+
+  DISALLOW_COPY_AND_ASSIGN(MultiplySourceStream);
+};
+
+// A FilterSourceStream passes through data unchanged to consumer.
+class PassThroughFilterSourceStream : public TestFilterSourceStreamBase {
+ public:
+  explicit PassThroughFilterSourceStream(std::unique_ptr<SourceStream> upstream)
+      : TestFilterSourceStreamBase(std::move(upstream)) {}
+  int FilterData(IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    buffer_.append(input_buffer->data(), input_buffer->BytesRemaining());
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    int bytes_to_read =
+        std::min(static_cast<int>(buffer_.size()), output_buffer_size);
+    memcpy(output_buffer->data(), buffer_.data(), bytes_to_read);
+    buffer_.erase(0, bytes_to_read);
+    return bytes_to_read;

[Randy Smith (Not in Mondays), 2016/08/22 22:37:16]

I'm tempted to suggest you take these last four lines and move them into the
base class as a utility function--I think they're pretty much the same
everywhere (at least so far).  Up to you.

[xunjieli, 2016/08/29 16:25:29]

Done.

+  }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(PassThroughFilterSourceStream);
+};
+
+// A FilterSourceStream passes throttle input data such that it returns them to
+// caller only one bytes at a time.
+class ThrottleSourceStream : public TestFilterSourceStreamBase {
+ public:
+  explicit ThrottleSourceStream(std::unique_ptr<SourceStream> upstream)
+      : TestFilterSourceStreamBase(std::move(upstream)) {}
+  int FilterData(IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    buffer_.append(input_buffer->data(), input_buffer->BytesRemaining());
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    int bytes_to_read = std::min(1, static_cast<int>(buffer_.size()));
+    memcpy(output_buffer->data(), buffer_.data(), bytes_to_read);
+    buffer_.erase(0, bytes_to_read);
+    return bytes_to_read;
+  }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(ThrottleSourceStream);
+};
+
+// A FilterSourceStream that consumes all input data but return no output.
+class NoOutputSourceStream : public TestFilterSourceStreamBase {
+ public:
+  NoOutputSourceStream(std::unique_ptr<SourceStream> upstream,
+                       size_t expected_input_size)
+      : TestFilterSourceStreamBase(std::move(upstream)),
+        expected_input_size_(expected_input_size),
+        consumed_all_input_(false) {}
+  int FilterData(IOBuffer* output_buffer,
+                 int output_buffer_size,
+                 DrainableIOBuffer* input_buffer) override {
+    expected_input_size_ -= input_buffer->BytesRemaining();
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    EXPECT_LE(0, expected_input_size_);
+    consumed_all_input_ = (expected_input_size_ == 0);
+    return OK;
+  }
+
+  bool consumed_all_input() const { return consumed_all_input_; }
+
+ private:
+  // Expected remaining bytes to be received from |upstream|.
+  int expected_input_size_;
+  bool consumed_all_input_;
+
+  DISALLOW_COPY_AND_ASSIGN(NoOutputSourceStream);
+};
+
+// A FilterSourceStream return an error code in FilterData().
+class ErrorFilterSourceStream : public FilterSourceStream {
+ public:
+  explicit ErrorFilterSourceStream(std::unique_ptr<SourceStream> upstream)
+      : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
+
+  int FilterData(IOBuffer* output_buffer,
+                 int 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
+
+class FilterSourceStreamTest
+    : public ::testing::TestWithParam<MockSourceStream::Mode> {};
+
+INSTANTIATE_TEST_CASE_P(FilterSourceStreamTests,
+                        FilterSourceStreamTest,
+                        ::testing::Values(MockSourceStream::SYNC,
+                                          MockSourceStream::ASYNC));
+
+// Tests that a FilterSourceStream subclass (NeedsAllInputFilterSourceStream)
+// 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_P(FilterSourceStreamTest, FilterDataReturnNoBytesExceptLast) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input("hello, world!");
+  size_t read_size = 2;
+  size_t num_reads = 0;
+  // Add a sequence of small reads.
+  for (size_t offset = 0; offset < input.length(); offset += read_size) {
+    source->AddReadResult(input.data() + offset,
+                          std::min(read_size, input.length() - offset), OK,
+                          GetParam());
+    num_reads++;
+  }
+  MockSourceStream* mock_stream = source.get();
+  NeedsAllInputFilterSourceStream 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());
+  if (GetParam() == MockSourceStream::ASYNC) {
+    ASSERT_EQ(ERR_IO_PENDING, rv);
+    while (num_reads > 0) {
+      mock_stream->CompleteNextRead();
+      num_reads--;
+    }
+    rv = callback.WaitForResult();
+  }

[Randy Smith (Not in Mondays), 2016/08/22 22:37:16]

Suggestion: Utility function?

[xunjieli, 2016/08/29 16:25:29]

Done.

+  ASSERT_EQ(static_cast<int>(input.length()), rv);
+  EXPECT_EQ(input, std::string(output_buffer->data(), rv));
+}
+
+// Tests that FilterData() returns 0 byte read because the upstream gives an
+// EOF.
+TEST_P(FilterSourceStreamTest, FilterDataReturnNoByte) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input;
+  source->AddReadResult(input.data(), 0, OK, GetParam());
+  MockSourceStream* mock_stream = source.get();
+  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());
+  if (GetParam() == MockSourceStream::ASYNC) {
+    ASSERT_EQ(ERR_IO_PENDING, rv);
+    mock_stream->CompleteNextRead();
+    rv = callback.WaitForResult();
+  }
+  EXPECT_EQ(OK, rv);
+}
+
+// Tests that FilterData() returns 0 byte filtered even though the upstream
+// produces data.
+TEST_P(FilterSourceStreamTest, FilterDataOutputNoData) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  size_t read_size = 2;
+  size_t num_reads = 0;
+  // Add a sequence of small reads.
+  for (size_t offset = 0; offset < input.length(); offset += read_size) {
+    source->AddReadResult(input.data() + offset,
+                          std::min(read_size, input.length() - offset), OK,
+                          GetParam());
+    num_reads++;
+  }
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  num_reads++;
+  MockSourceStream* mock_stream = source.get();
+  NoOutputSourceStream 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());
+  if (GetParam() == MockSourceStream::ASYNC) {
+    ASSERT_EQ(ERR_IO_PENDING, rv);
+    while (num_reads > 0) {
+      mock_stream->CompleteNextRead();
+      num_reads--;
+    }
+    rv = callback.WaitForResult();
+  }
+  EXPECT_EQ(OK, rv);
+  EXPECT_TRUE(stream.consumed_all_input());
+}
+
+// Tests that FilterData() returns non-zero bytes because the upstream
+// returns data.
+TEST_P(FilterSourceStreamTest, FilterDataReturnData) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  size_t read_size = 2;
+  // Add a sequence of small reads.
+  for (size_t offset = 0; offset < input.length(); offset += read_size) {
+    source->AddReadResult(input.data() + offset,
+                          std::min(read_size, input.length() - offset), OK,
+                          GetParam());
+  }
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  MockSourceStream* mock_stream = source.get();
+  PassThroughFilterSourceStream stream(std::move(source));
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  std::string actual_output;
+  while (true) {
+    int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                         callback.callback());
+    if (GetParam() == MockSourceStream::ASYNC) {
+      mock_stream->CompleteNextRead();
+      rv = callback.WaitForResult();
+    }
+    if (rv == OK)
+      break;
+    ASSERT_GE(static_cast<int>(read_size), rv);
+    ASSERT_GT(rv, OK);
+    actual_output.append(output_buffer->data(), rv);
+  }
+  EXPECT_EQ(input, actual_output);
+}
+
+// Tests that FilterData() returns more data than what it consumed.
+TEST_P(FilterSourceStreamTest, FilterDataReturnMoreData) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  size_t read_size = 2;
+  // Add a sequence of small reads.
+  for (size_t offset = 0; offset < input.length(); offset += read_size) {
+    source->AddReadResult(input.data() + offset,
+                          std::min(read_size, input.length() - offset), OK,
+                          GetParam());
+  }
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  MockSourceStream* mock_stream = source.get();
+  int multiplier = 2;
+  MultiplySourceStream stream(std::move(source), multiplier);
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  TestCompletionCallback callback;
+  std::string actual_output;
+  while (true) {
+    int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                         callback.callback());
+    if (GetParam() == MockSourceStream::ASYNC) {
+      mock_stream->CompleteNextRead();
+      rv = callback.WaitForResult();
+    }
+    if (rv == OK)
+      break;
+    ASSERT_GE(static_cast<int>(read_size) * multiplier, rv);
+    ASSERT_GT(rv, OK);
+    actual_output.append(output_buffer->data(), rv);
+  }
+  EXPECT_EQ("hheelllloo,,  wwoorrlldd!!", actual_output);
+}
+
+// Tests that FilterData() returns non-zero bytes and output buffer size is
+// smaller than the number of bytes read from the upstream.
+TEST_P(FilterSourceStreamTest, FilterDataOutputSpace) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  size_t read_size = 2;
+  // Add a sequence of small reads.
+  for (size_t offset = 0; offset < input.length(); offset += read_size) {
+    source->AddReadResult(input.data() + offset,
+                          std::min(read_size, input.length() - offset), OK,
+                          GetParam());
+  }
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  // Use an extremely small buffer size, so FilterData will need more output
+  // space.
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kSmallBufferSize);
+  MockSourceStream* mock_stream = source.get();
+  PassThroughFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  std::string actual_output;
+  while (true) {
+    int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                         callback.callback());
+    if (rv == ERR_IO_PENDING) {
+      ASSERT_EQ(MockSourceStream::ASYNC, GetParam());
+      mock_stream->CompleteNextRead();
+      rv = callback.WaitForResult();
+    }
+    if (rv == OK)
+      break;
+    ASSERT_GT(rv, OK);
+    ASSERT_GE(kSmallBufferSize, static_cast<size_t>(rv));
+    actual_output.append(output_buffer->data(), rv);
+  }
+  EXPECT_EQ(input, actual_output);
+}
+
+// Tests that FilterData() returns an error code, which is then surfaced as the
+// result of calling Read().
+TEST_P(FilterSourceStreamTest, FilterDataReturnError) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input;
+  source->AddReadResult(input.data(), 0, OK, GetParam());
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  MockSourceStream* mock_stream = source.get();
+  ErrorFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                       callback.callback());
+  if (GetParam() == MockSourceStream::ASYNC) {
+    ASSERT_EQ(ERR_IO_PENDING, rv);
+    mock_stream->CompleteNextRead();
+    rv = callback.WaitForResult();
+  }
+  EXPECT_EQ(ERR_CONTENT_DECODING_FAILED, rv);
+  // Reading from |stream| again should return the same error.
+  rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                   callback.callback());
+  EXPECT_EQ(ERR_CONTENT_DECODING_FAILED, rv);
+}
+
+TEST_P(FilterSourceStreamTest, FilterChaining) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  source->AddReadResult(input.data(), input.length(), OK, GetParam());
+  MockSourceStream* mock_stream = source.get();
+  std::unique_ptr<PassThroughFilterSourceStream> pass_through_source(
+      new PassThroughFilterSourceStream(std::move(source)));
+  pass_through_source->set_type_string("FIRST_PASS_THROUGH");
+  std::unique_ptr<NeedsAllInputFilterSourceStream> needs_all_input_source(
+      new NeedsAllInputFilterSourceStream(std::move(pass_through_source),
+                                          input.length()));
+  needs_all_input_source->set_type_string("NEEDS_ALL");
+  std::unique_ptr<PassThroughFilterSourceStream> second_pass_through_source(
+      new PassThroughFilterSourceStream(std::move(needs_all_input_source)));
+  second_pass_through_source->set_type_string("SECOND_PASS_THROUGH");
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+
+  TestCompletionCallback callback;
+  int bytes_read = second_pass_through_source->Read(
+      output_buffer.get(), output_buffer->size(), callback.callback());
+
+  if (GetParam() == MockSourceStream::ASYNC) {
+    ASSERT_EQ(ERR_IO_PENDING, bytes_read);
+    mock_stream->CompleteNextRead();
+    bytes_read = callback.WaitForResult();
+  }
+  ASSERT_EQ(input.length(), static_cast<size_t>(bytes_read));
+  EXPECT_EQ(input, std::string(output_buffer->data(), bytes_read));
+  // Type string (from left to right) should be the order of data flow.
+  EXPECT_EQ("FIRST_PASS_THROUGH,NEEDS_ALL,SECOND_PASS_THROUGH",
+            second_pass_through_source->OrderedTypeStringList());
+}
+
+// Tests that FilterData() returns multiple times for a single MockStream read,
+// because there is not enough output space.
+TEST_P(FilterSourceStreamTest, OutputSpaceForOneRead) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  source->AddReadResult(input.data(), input.length(), OK, GetParam());
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  // Use an extremely small buffer size (1 byte), so FilterData will need more
+  // output space.
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kSmallBufferSize);
+  MockSourceStream* mock_stream = source.get();
+  PassThroughFilterSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  std::string actual_output;
+  while (true) {
+    int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                         callback.callback());
+    if (rv == ERR_IO_PENDING) {
+      ASSERT_EQ(MockSourceStream::ASYNC, GetParam());
+      mock_stream->CompleteNextRead();
+      rv = callback.WaitForResult();
+    }
+    if (rv == OK)
+      break;
+    ASSERT_GT(rv, OK);
+    ASSERT_GE(kSmallBufferSize, static_cast<size_t>(rv));
+    actual_output.append(output_buffer->data(), rv);
+  }
+  EXPECT_EQ(input, actual_output);
+}
+
+// Tests that FilterData() returns multiple times for a single MockStream read,
+// because the filter returns one byte at a time.
+TEST_P(FilterSourceStreamTest, ThrottleSourceStream) {
+  std::unique_ptr<MockSourceStream> source(new MockSourceStream);
+  std::string input = "hello, world!";
+  source->AddReadResult(input.data(), input.length(), OK, GetParam());
+  // Add a 0 byte read to signal EOF.
+  source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
+  scoped_refptr<IOBufferWithSize> output_buffer =
+      new IOBufferWithSize(kDefaultBufferSize);
+  MockSourceStream* mock_stream = source.get();
+  ThrottleSourceStream stream(std::move(source));
+  TestCompletionCallback callback;
+  std::string actual_output;
+  while (true) {
+    int rv = stream.Read(output_buffer.get(), output_buffer->size(),
+                         callback.callback());
+    if (rv == ERR_IO_PENDING) {
+      ASSERT_EQ(MockSourceStream::ASYNC, GetParam());
+      mock_stream->CompleteNextRead();
+      rv = callback.WaitForResult();
+    }
+    if (rv == OK)
+      break;
+    ASSERT_GT(rv, OK);
+    // ThrottleSourceStream returns 1 byte at a time.
+    ASSERT_GE(1u, static_cast<size_t>(rv));
+    actual_output.append(output_buffer->data(), rv);
+  }
+  EXPECT_EQ(input, actual_output);
+}
+
+}  // namespace net