[ON HOLD] Implement pull-based design for content decoding

net/tools/content_decoder_tool/content_decoder_tool.cc

Index: net/tools/content_decoder_tool/content_decoder_tool.cc
diff --git a/net/tools/content_decoder_tool/content_decoder_tool.cc b/net/tools/content_decoder_tool/content_decoder_tool.cc
index 7b295e0a6259198ca7237e129a4d055601a1ca91..596c30c835def9b5a435f999b9c3a25432fd477f 100644
--- a/net/tools/content_decoder_tool/content_decoder_tool.cc
+++ b/net/tools/content_decoder_tool/content_decoder_tool.cc
@@ -5,14 +5,28 @@
 #include <iostream>
 
 #include "base/command_line.h"
+#include "base/macros.h"
+#include "base/memory/ptr_util.h"
+#include "base/strings/string_util.h"
 #include "net/base/io_buffer.h"
-#include "net/filter/filter.h"
-#include "net/filter/mock_filter_context.h"
-
-using net::Filter;
+#include "net/base/test_completion_callback.h"
+#include "net/filter/brotli_stream_source.h"
+#include "net/filter/filter_stream_source.h"
+#include "net/filter/gzip_stream_source.h"
+#include "net/filter/sdch_stream_source.h"
+#include "net/filter/stream_source.h"
 
 namespace {
 
+const int kBufferLen = 4096;
+
+// TODO(xunjielI): refactor to share logic with url_request_http_job.cc
+const char kDeflate[] = "deflate";
+const char kGZip[] = "gzip";
+const char kSdch[] = "sdch";
+const char kXGZip[] = "x-gzip";
+const char kBrotli[] = "br";
+
 // Print the command line help.
 void PrintHelp(const char* command_line_name) {
   std::cout << command_line_name << " content_encoding [content_encoding]..."
@@ -24,6 +38,34 @@ void PrintHelp(const char* command_line_name) {
             << std::endl;
 }
 
+class StdinStreamSource : public net::StreamSource {
+ public:
+  StdinStreamSource()
+      : net::StreamSource(net::StreamSource::TYPE_NONE),
+        raw_read_buffer_(new net::IOBufferWithSize(kBufferLen)) {}
+  ~StdinStreamSource() override {}
+
+  // StreamSource implementation
+  int Read(net::IOBuffer* dest_buffer,
+           size_t buffer_size,
+           const net::CompletionCallback& callback) override {
+    scoped_refptr<net::IOBuffer> pre_filter_buf =
+        new net::IOBufferWithSize(kBufferLen);
+    if (std::cin.eof()) {
+      return net::OK;
+    }
+    if (std::cin) {
+      std::cin.read(dest_buffer->data(), buffer_size);
+      return std::cin.gcount();
+    }
+    return net::ERR_FAILED;
+  }
+
+ private:
+  scoped_refptr<net::IOBuffer> raw_read_buffer_;
+  DISALLOW_COPY_AND_ASSIGN(StdinStreamSource);
+};
+
 }  // namespace
 
 int main(int argc, char* argv[]) {
@@ -37,45 +79,85 @@ int main(int argc, char* argv[]) {
     return 1;
   }
 
-  std::vector<Filter::FilterType> filter_types;
+  std::vector<net::StreamSource::SourceType> types;
   for (const auto& content_encoding : content_encodings) {
-    Filter::FilterType filter_type =
-        Filter::ConvertEncodingToType(content_encoding);
-    if (filter_type == Filter::FILTER_TYPE_UNSUPPORTED) {
-      std::cerr << "Unsupported decoder '" << content_encoding << "'."
-                << std::endl;
+    if (base::LowerCaseEqualsASCII(content_encoding, kBrotli)) {
+      types.push_back(net::StreamSource::TYPE_BROTLI);
+    } else if (base::LowerCaseEqualsASCII(content_encoding, kDeflate)) {
+      types.push_back(net::StreamSource::TYPE_DEFLATE);
+    } else if (base::LowerCaseEqualsASCII(content_encoding, kGZip) ||
+               base::LowerCaseEqualsASCII(content_encoding, kXGZip)) {
+      types.push_back(net::StreamSource::TYPE_GZIP);
+    } else if (base::LowerCaseEqualsASCII(content_encoding, kSdch)) {
+      types.push_back(net::StreamSource::TYPE_SDCH);
+    }
+  }
+
+  // Keep a raw ptr to the front of the filter chain.
+  StdinStreamSource* stdin_stream_source = new StdinStreamSource();
+  std::unique_ptr<net::StreamSource> previous(
+      base::WrapUnique(stdin_stream_source));
+  for (std::vector<net::StreamSource::SourceType>::reverse_iterator riter =
+           types.rbegin();
+       riter != types.rend(); ++riter) {
+    std::unique_ptr<net::FilterStreamSource> next = nullptr;
+    net::StreamSource::SourceType type = *riter;
+    switch (type) {
+      case net::StreamSource::TYPE_BROTLI:
+        next = CreateBrotliStreamSource(std::move(previous));
+        break;
+      case net::StreamSource::TYPE_SDCH: {
+        next.reset(new net::SdchStreamSource(std::move(previous), nullptr));
+        break;
+      }
+      case net::StreamSource::TYPE_GZIP:
+        next.reset(new net::GzipStreamSource(
+            std::move(previous),
+            net::GzipStreamSource::GZIP_STREAM_SOURCE_GZIP));
+        break;
+      case net::StreamSource::TYPE_DEFLATE:
+        next.reset(new net::GzipStreamSource(
+            std::move(previous),
+            net::GzipStreamSource::GZIP_STREAM_SOURCE_DEFLATE));
+        break;
+      case net::StreamSource::TYPE_GZIP_FALLBACK:
+        next.reset(new net::GzipStreamSource(
+            std::move(previous),
+            net::GzipStreamSource::GZIP_STREAM_SOURCE_GZIP_WITH_FALLBACK));
+        break;
+      default:
+        std::cerr << "Unsupported decoder '" << type << "'." << std::endl;
+        NOTREACHED();
+        return 1;
+    }
+    if (next == nullptr || !next->Init()) {
+      std::cerr << "Couldn't create the decoder." << std::endl;
       return 1;
     }
-    filter_types.push_back(filter_type);
+    previous = std::move(next);
   }
 
-  net::MockFilterContext filter_context;
-  std::unique_ptr<Filter> filter(Filter::Factory(filter_types, filter_context));
-  if (!filter) {
+  if (!previous) {
     std::cerr << "Couldn't create the decoder." << std::endl;
     return 1;
   }
 
-  net::IOBuffer* pre_filter_buf = filter->stream_buffer();
-  int pre_filter_buf_len = filter->stream_buffer_size();
-  while (std::cin) {
-    std::cin.read(pre_filter_buf->data(), pre_filter_buf_len);
-    int pre_filter_data_len = std::cin.gcount();
-    filter->FlushStreamBuffer(pre_filter_data_len);
-
-    while (true) {
-      const int kPostFilterBufLen = 4096;
-      char post_filter_buf[kPostFilterBufLen];
-      int post_filter_data_len = kPostFilterBufLen;
-      Filter::FilterStatus filter_status =
-          filter->ReadData(post_filter_buf, &post_filter_data_len);
-      std::cout.write(post_filter_buf, post_filter_data_len);
-      if (filter_status == Filter::FILTER_ERROR) {
-        std::cerr << "Couldn't decode stdin." << std::endl;
-        return 1;
-      } else if (filter_status != Filter::FILTER_OK) {
-        break;
-      }
+  while (true) {
+    scoped_refptr<net::IOBuffer> read_buffer =
+        new net::IOBufferWithSize(kBufferLen);
+    net::TestCompletionCallback callback;
+    int bytes_read =
+        previous->Read(read_buffer.get(), kBufferLen, callback.callback());
+    if (bytes_read == net::ERR_IO_PENDING)
+      bytes_read = callback.WaitForResult();
+
+    if (bytes_read > net::OK) {
+      std::cout.write(read_buffer->data(), bytes_read);
+    } else if (bytes_read < net::OK) {
+      std::cerr << "Couldn't decode stdin." << std::endl;
+      return 1;
+    } else if (bytes_read == net::OK) {
+      return 0;
     }
   }