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

net/filter/sdch_stream_source.cc

Index: net/filter/sdch_stream_source.cc
diff --git a/net/filter/sdch_stream_source.cc b/net/filter/sdch_stream_source.cc
new file mode 100644
index 0000000000000000000000000000000000000000..0af5d6b24601cdcdf9d94a07a752d365d360735b
--- /dev/null
+++ b/net/filter/sdch_stream_source.cc
@@ -0,0 +1,243 @@
+// 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 "net/filter/sdch_stream_source.h"
+
+#include "base/auto_reset.h"
+#include "base/bind.h"
+#include "base/numerics/safe_conversions.h"
+#include "sdch/open-vcdiff/src/google/vcdecoder.h"
+
+namespace net {
+
+namespace {
+
+const size_t kServerIdLength = 9;
+const char kSDCH[] = "SDCH";
+
+}  // namespace
+
+SdchStreamSource::SdchStreamSource(std::unique_ptr<StreamSource> previous,
+                                   Delegate* delegate)
+    : FilterStreamSource(StreamSource::TYPE_SDCH, std::move(previous)),
+      delegate_(delegate),
+      output_replaced_(false),
+      passthrough_(false),
+      dictionary_tried_load_(false) {
+  CHECK(delegate);
+}
+
+SdchStreamSource::~SdchStreamSource() {}
+
+bool SdchStreamSource::Init() {
+  return true;
+}
+
+void SdchStreamSource::StopDecoding() {
+  DCHECK(in_delegate_handler_);
+  passthrough_ = true;
+}
+
+void SdchStreamSource::ReplaceOutput(const char* data, size_t size) {
+  DCHECK(in_delegate_handler_);
+  output_replaced_ = true;
+  buffered_output_.assign(data, size);
+}
+
+std::string SdchStreamSource::GetTypeAsString() const {
+  return kSDCH;
+}
+
+int SdchStreamSource::FilterData(IOBuffer* output_buffer,
+                                 size_t output_buffer_size,
+                                 DrainableIOBuffer* input_buffer) {
+  size_t bytes_read = 0;
+  // Run the synchronous decompressor.
+  SdchStreamState state =
+      Decompress(output_buffer, output_buffer_size, input_buffer, &bytes_read);
+
+  // If decompression failed, call OnDecodingError on the delegate, then
+  // rerun the decompressor. The decompressor shouldn't actually decompress
+  // this time, since the delegate should have called either PassThrough or
+  // ReplaceOutput.
+  if (state == SDCH_STREAM_ERROR) {
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    bool handled = AskDelegateToHandleDecodingError();
+    if (!handled)
+      return ERR_FAILED;
+    state = Decompress(output_buffer, output_buffer_size, input_buffer,
+                       &bytes_read);
+  }
+
+  DCHECK_EQ(0, input_buffer->BytesRemaining());
+  return bytes_read;
+}
+
+// Synchronous decompressor core.
+SdchStreamSource::SdchStreamState SdchStreamSource::Decompress(
+    IOBuffer* output_buffer,
+    size_t output_buffer_size,
+    DrainableIOBuffer* input_buffer,
+    size_t* bytes_read) {
+  // If there is no loaded dictionary, try loading a dictionary ID from
+  // |input_buffer| and then loading a dictionary. If anything goes wrong in
+  // LoadDictionary, the delegate's OnDictionaryError() call is expected
+  // to recover, either by calling StopDecoding() or ReplaceOutput(). Both of
+  // these are handled by Decompress (as is having no dictionary loaded). If
+  // the delegate's OnDictionaryError cannot recover, this function
+  // returns a read error.
+  if (!dictionary_tried_load_ && !LoadDictionary(input_buffer)) {
+    return SDCH_STREAM_ERROR;
+  }
+
+  // If there's any data left in |buffered_output_|, try flushing some of it.
+  if (Flush(output_buffer, output_buffer_size, bytes_read) !=
+      SDCH_STREAM_MORE_INPUT)
+    return SDCH_STREAM_MORE_OUTPUT_SPACE;
+
+  // If LoadDictionary() consume all |input_buffer|, return early here.
+  if (input_buffer->BytesRemaining() == 0)
+    return SDCH_STREAM_MORE_INPUT;
+
+  DCHECK(dictionary_tried_load_);
+
+  // If output has been replaced via ReplaceOutput, short-circuit any other
+  // processing, and emit as much of the buffered output as possible. If there's
+  // no buffered output left, return EOF.
+  if (output_replaced_)
+    return Flush(output_buffer, output_buffer_size, bytes_read);
+
+  // If this stream is no longer decoding, pass any input through.
+  if (passthrough_)
+    return PassThrough(output_buffer, output_buffer_size, input_buffer,
+                       bytes_read);
+
+  // There is no data left in |buffered_output_| and there is data in
+  // |input_buffer|, so run the vcdiff decoder.
+  DCHECK(buffered_output_.empty());
+  DCHECK_LT(0, input_buffer->BytesRemaining());
+
+  // Calls to DecodeChunk always consume all their input, so this always drains
+  // the entire buffer.
+  bool ok = decoder_->DecodeChunk(
+      input_buffer->data(), input_buffer->BytesRemaining(), &buffered_output_);
+
+  input_buffer->DidConsume(input_buffer->BytesRemaining());
+
+  if (!ok) {
+    // Don't call it again.
+    decoder_.reset();
+    return SDCH_STREAM_ERROR;
+  }
+  return Flush(output_buffer, output_buffer_size, bytes_read);
+}
+
+SdchStreamSource::SdchStreamState SdchStreamSource::PassThrough(
+    IOBuffer* output_buffer,
+    size_t output_buffer_size,
+    DrainableIOBuffer* input_buffer,
+    size_t* bytes_read) {
+  *bytes_read = 0;
+  SdchStreamState state = Flush(output_buffer, output_buffer_size, bytes_read);
+  if (state == SDCH_STREAM_MORE_OUTPUT_SPACE)
+    return state;
+
+  // No buffered data to flush. See if there's any input.
+  DCHECK(buffered_output_.empty());
+  if (input_buffer->BytesRemaining() == 0)
+    return SDCH_STREAM_MORE_INPUT;
+
+  size_t to_copy = output_buffer_size;
+  if (to_copy > base::checked_cast<size_t>(input_buffer->BytesRemaining()))
+    to_copy = input_buffer->BytesRemaining();
+  memcpy(output_buffer->data(), input_buffer->data(), to_copy);
+  input_buffer->DidConsume(to_copy);
+  *bytes_read = *bytes_read + to_copy;
+  if (input_buffer->BytesRemaining() > 0)
+    return SDCH_STREAM_MORE_OUTPUT_SPACE;
+  else
+    return SDCH_STREAM_MORE_INPUT;
+}
+
+SdchStreamSource::SdchStreamState SdchStreamSource::Flush(
+    IOBuffer* output_buffer,
+    size_t output_buffer_size,
+    size_t* bytes_read) {
+  size_t to_copy = buffered_output_.length();
+  if (to_copy > output_buffer_size)
+    to_copy = output_buffer_size;
+  memcpy(output_buffer->data(), buffered_output_.data(), to_copy);
+  buffered_output_.erase(0, to_copy);
+  *bytes_read = to_copy;
+  if (!buffered_output_.empty()) {
+    return SDCH_STREAM_MORE_OUTPUT_SPACE;
+  }
+  return SDCH_STREAM_MORE_INPUT;
+}
+
+// Attempts to find a dictionary ID at the start of the input stream and load
+// the dictionary with that identifier. Returns true if no error happened or an
+// error happened but the delegate repaired it, and false if an error happened
+// and the delegate did not repair it.
+bool SdchStreamSource::LoadDictionary(DrainableIOBuffer* input_buffer) {
+  // Copy at most |kServerIdLength| from |input_buffer|.
+  size_t to_copy = kServerIdLength - dictionary_id_.length();
+  if (to_copy > base::checked_cast<size_t>(input_buffer->BytesRemaining()))
+    to_copy = input_buffer->BytesRemaining();
+  dictionary_id_.append(input_buffer->data(), to_copy);
+  input_buffer->DidConsume(to_copy);
+
+  // Not enough bytes for a dictionary ID accumulated yet.
+  if (dictionary_id_.length() != kServerIdLength)
+    return true;
+
+  dictionary_tried_load_ = true;
+
+  // If the dictionary ID looks bogus, or the delegate can't find a dictionary
+  // with that identifier, then call OnDictionaryError and bail.
+  // Also, stick the dictionary ID into the output buffer here; LoadDictionary
+  // will consume it assuming it's a dictionary ID, so if it's really not, the
+  // "dictionary ID" should appear in the output stream.
+  //
+  // To avoid passing a std::string with a null terminator into GetDictionary(),
+  // |server_hash| here removes the last byte blindly, and this method only
+  // calls
+  // GetDictionary with it if CouldBeDictionaryId returns true.
+  std::string server_hash = dictionary_id_.substr(0, kServerIdLength - 1);
+  const std::string* dictionary_text = nullptr;
+  if (!CouldBeDictionaryId(dictionary_id_) ||
+      !delegate_->OnGetDictionary(server_hash, &dictionary_text)) {
+    input_buffer->DidConsume(input_buffer->BytesRemaining());
+    buffered_output_.append(dictionary_id_);
+    return AskDelegateToHandleDictionaryError();
+  }
+
+  decoder_.reset(new open_vcdiff::VCDiffStreamingDecoder);
+  decoder_->SetAllowVcdTarget(false);
+  decoder_->StartDecoding(dictionary_text->data(), dictionary_text->length());
+  return true;
+}
+
+bool SdchStreamSource::CouldBeDictionaryId(const std::string& id) const {
+  for (size_t i = 0; i < kServerIdLength - 1; i++) {
+    char base64_char = id[i];
+    if (!isalnum(base64_char) && '-' != base64_char && '_' != base64_char)
+      return false;
+  }
+  if (id[kServerIdLength - 1] != '\0')
+    return false;
+  return true;
+}
+
+bool SdchStreamSource::AskDelegateToHandleDictionaryError() {
+  base::AutoReset<bool> resetter(&in_delegate_handler_, true);
+  return delegate_->OnDictionaryError(this);
+}
+
+bool SdchStreamSource::AskDelegateToHandleDecodingError() {
+  base::AutoReset<bool> resetter(&in_delegate_handler_, true);
+  return delegate_->OnDecodingError(this);
+}
+
+}  // namespace net