filter_source_stream_unittest.cc 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565
  1. // Copyright 2016 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include <algorithm>
  5. #include <string>
  6. #include "base/bind.h"
  7. #include "base/callback.h"
  8. #include "base/numerics/safe_conversions.h"
  9. #include "net/base/io_buffer.h"
  10. #include "net/base/net_errors.h"
  11. #include "net/base/test_completion_callback.h"
  12. #include "net/filter/filter_source_stream.h"
  13. #include "net/filter/mock_source_stream.h"
  14. #include "testing/gtest/include/gtest/gtest.h"
  15. namespace net {
  16. namespace {
  17. const size_t kDefaultBufferSize = 4096;
  18. const size_t kSmallBufferSize = 1;
  19. class TestFilterSourceStreamBase : public FilterSourceStream {
  20. public:
  21. explicit TestFilterSourceStreamBase(std::unique_ptr<SourceStream> upstream)
  22. : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
  23. TestFilterSourceStreamBase(const TestFilterSourceStreamBase&) = delete;
  24. TestFilterSourceStreamBase& operator=(const TestFilterSourceStreamBase&) =
  25. delete;
  26. ~TestFilterSourceStreamBase() override { DCHECK(buffer_.empty()); }
  27. std::string GetTypeAsString() const override { return type_string_; }
  28. void set_type_string(const std::string& type_string) {
  29. type_string_ = type_string;
  30. }
  31. protected:
  32. // Writes contents of |buffer_| to |output_buffer| and returns the number of
  33. // bytes written or an error code. Additionally removes consumed data from
  34. // |buffer_|.
  35. size_t WriteBufferToOutput(IOBuffer* output_buffer,
  36. size_t output_buffer_size) {
  37. size_t bytes_to_filter = std::min(buffer_.length(), output_buffer_size);
  38. memcpy(output_buffer->data(), buffer_.data(), bytes_to_filter);
  39. buffer_.erase(0, bytes_to_filter);
  40. return bytes_to_filter;
  41. }
  42. // Buffer used by subclasses to hold data that is yet to be passed to the
  43. // caller.
  44. std::string buffer_;
  45. private:
  46. std::string type_string_;
  47. };
  48. // A FilterSourceStream that needs all input data before it can return non-zero
  49. // bytes read.
  50. class NeedsAllInputFilterSourceStream : public TestFilterSourceStreamBase {
  51. public:
  52. NeedsAllInputFilterSourceStream(std::unique_ptr<SourceStream> upstream,
  53. size_t expected_input_bytes)
  54. : TestFilterSourceStreamBase(std::move(upstream)),
  55. expected_input_bytes_(expected_input_bytes) {}
  56. NeedsAllInputFilterSourceStream(const NeedsAllInputFilterSourceStream&) =
  57. delete;
  58. NeedsAllInputFilterSourceStream& operator=(
  59. const NeedsAllInputFilterSourceStream&) = delete;
  60. base::expected<size_t, Error> FilterData(IOBuffer* output_buffer,
  61. size_t output_buffer_size,
  62. IOBuffer* input_buffer,
  63. size_t input_buffer_size,
  64. size_t* consumed_bytes,
  65. bool upstream_eof_reached) override {
  66. buffer_.append(input_buffer->data(), input_buffer_size);
  67. EXPECT_GE(expected_input_bytes_, input_buffer_size);
  68. expected_input_bytes_ -= input_buffer_size;
  69. *consumed_bytes = input_buffer_size;
  70. if (!upstream_eof_reached) {
  71. // Keep returning 0 bytes read until all input has been consumed.
  72. return 0;
  73. }
  74. EXPECT_EQ(0u, expected_input_bytes_);
  75. return WriteBufferToOutput(output_buffer, output_buffer_size);
  76. }
  77. private:
  78. // Expected remaining bytes to be received from |upstream|.
  79. size_t expected_input_bytes_;
  80. };
  81. // A FilterSourceStream that repeat every input byte by |multiplier| amount of
  82. // times.
  83. class MultiplySourceStream : public TestFilterSourceStreamBase {
  84. public:
  85. MultiplySourceStream(std::unique_ptr<SourceStream> upstream, int multiplier)
  86. : TestFilterSourceStreamBase(std::move(upstream)),
  87. multiplier_(multiplier) {}
  88. MultiplySourceStream(const MultiplySourceStream&) = delete;
  89. MultiplySourceStream& operator=(const MultiplySourceStream&) = delete;
  90. base::expected<size_t, Error> FilterData(
  91. IOBuffer* output_buffer,
  92. size_t output_buffer_size,
  93. IOBuffer* input_buffer,
  94. size_t input_buffer_size,
  95. size_t* consumed_bytes,
  96. bool /*upstream_eof_reached*/) override {
  97. for (size_t i = 0; i < input_buffer_size; i++) {
  98. for (int j = 0; j < multiplier_; j++)
  99. buffer_.append(input_buffer->data() + i, 1);
  100. }
  101. *consumed_bytes = input_buffer_size;
  102. return WriteBufferToOutput(output_buffer, output_buffer_size);
  103. }
  104. private:
  105. int multiplier_;
  106. };
  107. // A FilterSourceStream passes through data unchanged to consumer.
  108. class PassThroughFilterSourceStream : public TestFilterSourceStreamBase {
  109. public:
  110. explicit PassThroughFilterSourceStream(std::unique_ptr<SourceStream> upstream)
  111. : TestFilterSourceStreamBase(std::move(upstream)) {}
  112. PassThroughFilterSourceStream(const PassThroughFilterSourceStream&) = delete;
  113. PassThroughFilterSourceStream& operator=(
  114. const PassThroughFilterSourceStream&) = delete;
  115. base::expected<size_t, Error> FilterData(
  116. IOBuffer* output_buffer,
  117. size_t output_buffer_size,
  118. IOBuffer* input_buffer,
  119. size_t input_buffer_size,
  120. size_t* consumed_bytes,
  121. bool /*upstream_eof_reached*/) override {
  122. buffer_.append(input_buffer->data(), input_buffer_size);
  123. *consumed_bytes = input_buffer_size;
  124. return WriteBufferToOutput(output_buffer, output_buffer_size);
  125. }
  126. };
  127. // A FilterSourceStream passes throttle input data such that it returns them to
  128. // caller only one bytes at a time.
  129. class ThrottleSourceStream : public TestFilterSourceStreamBase {
  130. public:
  131. explicit ThrottleSourceStream(std::unique_ptr<SourceStream> upstream)
  132. : TestFilterSourceStreamBase(std::move(upstream)) {}
  133. ThrottleSourceStream(const ThrottleSourceStream&) = delete;
  134. ThrottleSourceStream& operator=(const ThrottleSourceStream&) = delete;
  135. base::expected<size_t, Error> FilterData(
  136. IOBuffer* output_buffer,
  137. size_t output_buffer_size,
  138. IOBuffer* input_buffer,
  139. size_t input_buffer_size,
  140. size_t* consumed_bytes,
  141. bool /*upstream_eof_reached*/) override {
  142. buffer_.append(input_buffer->data(), input_buffer_size);
  143. *consumed_bytes = input_buffer_size;
  144. size_t bytes_to_read = std::min(size_t{1}, buffer_.size());
  145. memcpy(output_buffer->data(), buffer_.data(), bytes_to_read);
  146. buffer_.erase(0, bytes_to_read);
  147. return bytes_to_read;
  148. }
  149. };
  150. // A FilterSourceStream that consumes all input data but return no output.
  151. class NoOutputSourceStream : public TestFilterSourceStreamBase {
  152. public:
  153. NoOutputSourceStream(std::unique_ptr<SourceStream> upstream,
  154. size_t expected_input_size)
  155. : TestFilterSourceStreamBase(std::move(upstream)),
  156. expected_input_size_(expected_input_size) {}
  157. NoOutputSourceStream(const NoOutputSourceStream&) = delete;
  158. NoOutputSourceStream& operator=(const NoOutputSourceStream&) = delete;
  159. base::expected<size_t, Error> FilterData(
  160. IOBuffer* output_buffer,
  161. size_t output_buffer_size,
  162. IOBuffer* input_buffer,
  163. size_t input_buffer_size,
  164. size_t* consumed_bytes,
  165. bool /*upstream_eof_reached*/) override {
  166. EXPECT_GE(expected_input_size_, input_buffer_size);
  167. expected_input_size_ -= input_buffer_size;
  168. *consumed_bytes = input_buffer_size;
  169. consumed_all_input_ = (expected_input_size_ == 0);
  170. return 0;
  171. }
  172. bool consumed_all_input() const { return consumed_all_input_; }
  173. private:
  174. // Expected remaining bytes to be received from |upstream|.
  175. size_t expected_input_size_;
  176. bool consumed_all_input_ = false;
  177. };
  178. // A FilterSourceStream return an error code in FilterData().
  179. class ErrorFilterSourceStream : public FilterSourceStream {
  180. public:
  181. explicit ErrorFilterSourceStream(std::unique_ptr<SourceStream> upstream)
  182. : FilterSourceStream(SourceStream::TYPE_NONE, std::move(upstream)) {}
  183. ErrorFilterSourceStream(const ErrorFilterSourceStream&) = delete;
  184. ErrorFilterSourceStream& operator=(const ErrorFilterSourceStream&) = delete;
  185. base::expected<size_t, Error> FilterData(
  186. IOBuffer* output_buffer,
  187. size_t output_buffer_size,
  188. IOBuffer* input_buffer,
  189. size_t input_buffer_size,
  190. size_t* consumed_bytes,
  191. bool /*upstream_eof_reached*/) override {
  192. return base::unexpected(ERR_CONTENT_DECODING_FAILED);
  193. }
  194. std::string GetTypeAsString() const override { return ""; }
  195. };
  196. } // namespace
  197. class FilterSourceStreamTest
  198. : public ::testing::TestWithParam<MockSourceStream::Mode> {
  199. protected:
  200. // If MockSourceStream::Mode is ASYNC, completes |num_reads| from
  201. // |mock_stream| and wait for |callback| to complete. If Mode is not ASYNC,
  202. // does nothing and returns |previous_result|.
  203. int CompleteReadIfAsync(int previous_result,
  204. TestCompletionCallback* callback,
  205. MockSourceStream* mock_stream,
  206. size_t num_reads) {
  207. if (GetParam() == MockSourceStream::ASYNC) {
  208. EXPECT_EQ(ERR_IO_PENDING, previous_result);
  209. while (num_reads > 0) {
  210. mock_stream->CompleteNextRead();
  211. num_reads--;
  212. }
  213. return callback->WaitForResult();
  214. }
  215. return previous_result;
  216. }
  217. };
  218. INSTANTIATE_TEST_SUITE_P(FilterSourceStreamTests,
  219. FilterSourceStreamTest,
  220. ::testing::Values(MockSourceStream::SYNC,
  221. MockSourceStream::ASYNC));
  222. // Tests that a FilterSourceStream subclass (NeedsAllInputFilterSourceStream)
  223. // can return 0 bytes for FilterData()s when it has not consumed EOF from the
  224. // upstream. In this case, FilterSourceStream should continue reading from
  225. // upstream to complete filtering.
  226. TEST_P(FilterSourceStreamTest, FilterDataReturnNoBytesExceptLast) {
  227. auto source = std::make_unique<MockSourceStream>();
  228. std::string input("hello, world!");
  229. size_t read_size = 2;
  230. size_t num_reads = 0;
  231. // Add a sequence of small reads.
  232. for (size_t offset = 0; offset < input.length(); offset += read_size) {
  233. source->AddReadResult(input.data() + offset,
  234. std::min(read_size, input.length() - offset), OK,
  235. GetParam());
  236. num_reads++;
  237. }
  238. source->AddReadResult(input.data(), 0, OK, GetParam()); // EOF
  239. num_reads++;
  240. MockSourceStream* mock_stream = source.get();
  241. NeedsAllInputFilterSourceStream stream(std::move(source), input.length());
  242. scoped_refptr<IOBufferWithSize> output_buffer =
  243. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  244. TestCompletionCallback callback;
  245. std::string actual_output;
  246. while (true) {
  247. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  248. callback.callback());
  249. if (rv == ERR_IO_PENDING)
  250. rv = CompleteReadIfAsync(rv, &callback, mock_stream, num_reads);
  251. if (rv == OK)
  252. break;
  253. ASSERT_GT(rv, OK);
  254. actual_output.append(output_buffer->data(), rv);
  255. }
  256. EXPECT_EQ(input, actual_output);
  257. }
  258. // Tests that FilterData() returns 0 byte read because the upstream gives an
  259. // EOF.
  260. TEST_P(FilterSourceStreamTest, FilterDataReturnNoByte) {
  261. auto source = std::make_unique<MockSourceStream>();
  262. std::string input;
  263. source->AddReadResult(input.data(), 0, OK, GetParam());
  264. MockSourceStream* mock_stream = source.get();
  265. PassThroughFilterSourceStream stream(std::move(source));
  266. scoped_refptr<IOBufferWithSize> output_buffer =
  267. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  268. TestCompletionCallback callback;
  269. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  270. callback.callback());
  271. rv = CompleteReadIfAsync(rv, &callback, mock_stream, 1);
  272. EXPECT_EQ(OK, rv);
  273. }
  274. // Tests that FilterData() returns 0 byte filtered even though the upstream
  275. // produces data.
  276. TEST_P(FilterSourceStreamTest, FilterDataOutputNoData) {
  277. auto source = std::make_unique<MockSourceStream>();
  278. std::string input = "hello, world!";
  279. size_t read_size = 2;
  280. size_t num_reads = 0;
  281. // Add a sequence of small reads.
  282. for (size_t offset = 0; offset < input.length(); offset += read_size) {
  283. source->AddReadResult(input.data() + offset,
  284. std::min(read_size, input.length() - offset), OK,
  285. GetParam());
  286. num_reads++;
  287. }
  288. // Add a 0 byte read to signal EOF.
  289. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  290. num_reads++;
  291. MockSourceStream* mock_stream = source.get();
  292. NoOutputSourceStream stream(std::move(source), input.length());
  293. scoped_refptr<IOBufferWithSize> output_buffer =
  294. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  295. TestCompletionCallback callback;
  296. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  297. callback.callback());
  298. rv = CompleteReadIfAsync(rv, &callback, mock_stream, num_reads);
  299. EXPECT_EQ(OK, rv);
  300. EXPECT_TRUE(stream.consumed_all_input());
  301. }
  302. // Tests that FilterData() returns non-zero bytes because the upstream
  303. // returns data.
  304. TEST_P(FilterSourceStreamTest, FilterDataReturnData) {
  305. auto source = std::make_unique<MockSourceStream>();
  306. std::string input = "hello, world!";
  307. size_t read_size = 2;
  308. // Add a sequence of small reads.
  309. for (size_t offset = 0; offset < input.length(); offset += read_size) {
  310. source->AddReadResult(input.data() + offset,
  311. std::min(read_size, input.length() - offset), OK,
  312. GetParam());
  313. }
  314. // Add a 0 byte read to signal EOF.
  315. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  316. MockSourceStream* mock_stream = source.get();
  317. PassThroughFilterSourceStream stream(std::move(source));
  318. scoped_refptr<IOBufferWithSize> output_buffer =
  319. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  320. TestCompletionCallback callback;
  321. std::string actual_output;
  322. while (true) {
  323. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  324. callback.callback());
  325. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  326. if (rv == OK)
  327. break;
  328. ASSERT_GE(static_cast<int>(read_size), rv);
  329. ASSERT_GT(rv, OK);
  330. actual_output.append(output_buffer->data(), rv);
  331. }
  332. EXPECT_EQ(input, actual_output);
  333. }
  334. // Tests that FilterData() returns more data than what it consumed.
  335. TEST_P(FilterSourceStreamTest, FilterDataReturnMoreData) {
  336. auto source = std::make_unique<MockSourceStream>();
  337. std::string input = "hello, world!";
  338. size_t read_size = 2;
  339. // Add a sequence of small reads.
  340. for (size_t offset = 0; offset < input.length(); offset += read_size) {
  341. source->AddReadResult(input.data() + offset,
  342. std::min(read_size, input.length() - offset), OK,
  343. GetParam());
  344. }
  345. // Add a 0 byte read to signal EOF.
  346. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  347. MockSourceStream* mock_stream = source.get();
  348. int multiplier = 2;
  349. MultiplySourceStream stream(std::move(source), multiplier);
  350. scoped_refptr<IOBufferWithSize> output_buffer =
  351. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  352. TestCompletionCallback callback;
  353. std::string actual_output;
  354. while (true) {
  355. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  356. callback.callback());
  357. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  358. if (rv == OK)
  359. break;
  360. ASSERT_GE(static_cast<int>(read_size) * multiplier, rv);
  361. ASSERT_GT(rv, OK);
  362. actual_output.append(output_buffer->data(), rv);
  363. }
  364. EXPECT_EQ("hheelllloo,, wwoorrlldd!!", actual_output);
  365. }
  366. // Tests that FilterData() returns non-zero bytes and output buffer size is
  367. // smaller than the number of bytes read from the upstream.
  368. TEST_P(FilterSourceStreamTest, FilterDataOutputSpace) {
  369. auto source = std::make_unique<MockSourceStream>();
  370. std::string input = "hello, world!";
  371. size_t read_size = 2;
  372. // Add a sequence of small reads.
  373. for (size_t offset = 0; offset < input.length(); offset += read_size) {
  374. source->AddReadResult(input.data() + offset,
  375. std::min(read_size, input.length() - offset), OK,
  376. GetParam());
  377. }
  378. // Add a 0 byte read to signal EOF.
  379. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  380. // Use an extremely small buffer size, so FilterData will need more output
  381. // space.
  382. scoped_refptr<IOBufferWithSize> output_buffer =
  383. base::MakeRefCounted<IOBufferWithSize>(kSmallBufferSize);
  384. MockSourceStream* mock_stream = source.get();
  385. PassThroughFilterSourceStream stream(std::move(source));
  386. TestCompletionCallback callback;
  387. std::string actual_output;
  388. while (true) {
  389. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  390. callback.callback());
  391. if (rv == ERR_IO_PENDING)
  392. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  393. if (rv == OK)
  394. break;
  395. ASSERT_GT(rv, OK);
  396. ASSERT_GE(kSmallBufferSize, static_cast<size_t>(rv));
  397. actual_output.append(output_buffer->data(), rv);
  398. }
  399. EXPECT_EQ(input, actual_output);
  400. }
  401. // Tests that FilterData() returns an error code, which is then surfaced as
  402. // the result of calling Read().
  403. TEST_P(FilterSourceStreamTest, FilterDataReturnError) {
  404. auto source = std::make_unique<MockSourceStream>();
  405. std::string input;
  406. source->AddReadResult(input.data(), 0, OK, GetParam());
  407. scoped_refptr<IOBufferWithSize> output_buffer =
  408. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  409. MockSourceStream* mock_stream = source.get();
  410. ErrorFilterSourceStream stream(std::move(source));
  411. TestCompletionCallback callback;
  412. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  413. callback.callback());
  414. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  415. EXPECT_EQ(ERR_CONTENT_DECODING_FAILED, rv);
  416. // Reading from |stream| again should return the same error.
  417. rv = stream.Read(output_buffer.get(), output_buffer->size(),
  418. callback.callback());
  419. EXPECT_EQ(ERR_CONTENT_DECODING_FAILED, rv);
  420. }
  421. TEST_P(FilterSourceStreamTest, FilterChaining) {
  422. auto source = std::make_unique<MockSourceStream>();
  423. std::string input = "hello, world!";
  424. source->AddReadResult(input.data(), input.length(), OK, GetParam());
  425. source->AddReadResult(input.data(), 0, OK, GetParam()); // EOF
  426. MockSourceStream* mock_stream = source.get();
  427. auto pass_through_source =
  428. std::make_unique<PassThroughFilterSourceStream>(std::move(source));
  429. pass_through_source->set_type_string("FIRST_PASS_THROUGH");
  430. auto needs_all_input_source =
  431. std::make_unique<NeedsAllInputFilterSourceStream>(
  432. std::move(pass_through_source), input.length());
  433. needs_all_input_source->set_type_string("NEEDS_ALL");
  434. auto second_pass_through_source =
  435. std::make_unique<PassThroughFilterSourceStream>(
  436. std::move(needs_all_input_source));
  437. second_pass_through_source->set_type_string("SECOND_PASS_THROUGH");
  438. scoped_refptr<IOBufferWithSize> output_buffer =
  439. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  440. TestCompletionCallback callback;
  441. std::string actual_output;
  442. while (true) {
  443. int rv = second_pass_through_source->Read(
  444. output_buffer.get(), output_buffer->size(), callback.callback());
  445. if (rv == ERR_IO_PENDING)
  446. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/2);
  447. if (rv == OK)
  448. break;
  449. ASSERT_GT(rv, OK);
  450. actual_output.append(output_buffer->data(), rv);
  451. }
  452. EXPECT_EQ(input, actual_output);
  453. // Type string (from left to right) should be the order of data flow.
  454. EXPECT_EQ("FIRST_PASS_THROUGH,NEEDS_ALL,SECOND_PASS_THROUGH",
  455. second_pass_through_source->Description());
  456. }
  457. // Tests that FilterData() returns multiple times for a single MockStream
  458. // read, because there is not enough output space.
  459. TEST_P(FilterSourceStreamTest, OutputSpaceForOneRead) {
  460. auto source = std::make_unique<MockSourceStream>();
  461. std::string input = "hello, world!";
  462. source->AddReadResult(input.data(), input.length(), OK, GetParam());
  463. // Add a 0 byte read to signal EOF.
  464. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  465. // Use an extremely small buffer size (1 byte), so FilterData will need more
  466. // output space.
  467. scoped_refptr<IOBufferWithSize> output_buffer =
  468. base::MakeRefCounted<IOBufferWithSize>(kSmallBufferSize);
  469. MockSourceStream* mock_stream = source.get();
  470. PassThroughFilterSourceStream stream(std::move(source));
  471. TestCompletionCallback callback;
  472. std::string actual_output;
  473. while (true) {
  474. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  475. callback.callback());
  476. if (rv == ERR_IO_PENDING)
  477. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  478. if (rv == OK)
  479. break;
  480. ASSERT_GT(rv, OK);
  481. ASSERT_GE(kSmallBufferSize, static_cast<size_t>(rv));
  482. actual_output.append(output_buffer->data(), rv);
  483. }
  484. EXPECT_EQ(input, actual_output);
  485. }
  486. // Tests that FilterData() returns multiple times for a single MockStream
  487. // read, because the filter returns one byte at a time.
  488. TEST_P(FilterSourceStreamTest, ThrottleSourceStream) {
  489. auto source = std::make_unique<MockSourceStream>();
  490. std::string input = "hello, world!";
  491. source->AddReadResult(input.data(), input.length(), OK, GetParam());
  492. // Add a 0 byte read to signal EOF.
  493. source->AddReadResult(input.data() + input.length(), 0, OK, GetParam());
  494. scoped_refptr<IOBufferWithSize> output_buffer =
  495. base::MakeRefCounted<IOBufferWithSize>(kDefaultBufferSize);
  496. MockSourceStream* mock_stream = source.get();
  497. ThrottleSourceStream stream(std::move(source));
  498. TestCompletionCallback callback;
  499. std::string actual_output;
  500. while (true) {
  501. int rv = stream.Read(output_buffer.get(), output_buffer->size(),
  502. callback.callback());
  503. if (rv == ERR_IO_PENDING)
  504. rv = CompleteReadIfAsync(rv, &callback, mock_stream, /*num_reads=*/1);
  505. if (rv == OK)
  506. break;
  507. ASSERT_GT(rv, OK);
  508. // ThrottleSourceStream returns 1 byte at a time.
  509. ASSERT_GE(1u, static_cast<size_t>(rv));
  510. actual_output.append(output_buffer->data(), rv);
  511. }
  512. EXPECT_EQ(input, actual_output);
  513. }
  514. } // namespace net