[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..17cc25c35a6e448141f0fe76b0f580368c68c9ec
--- /dev/null
+++ b/net/filter/sdch_stream_source.cc
@@ -0,0 +1,239 @@
+// 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 "base/auto_reset.h"
+#include "base/bind.h"
+#include "net/filter/block_buffer.h"
+#include "net/filter/sdch_stream_source.h"
+#include "sdch/open-vcdiff/src/google/vcdecoder.h"
+
+namespace net {
+
+namespace {
+
+const size_t kServerIdLength = 9;
+
+}  // namespace
+
+SdchStreamSource::SdchStreamSource(scoped_ptr<StreamSource> previous,
+                                   SdchStreamSourceDelegate* delegate)
+    : StreamSource(StreamSource::SOURCE_SDCH, std::move(previous)),
+      decoder_(new open_vcdiff::VCDiffStreamingDecoder),
+      delegate_(delegate),
+      output_replaced_(false),
+      passthrough_(false),
+      dictionary_tried_load_(false) {}
+
+SdchStreamSource::~SdchStreamSource() {
+  // In unit tests, delegate_ may be null.
+  if (delegate_)
+    delegate_->NotifyStreamDone();
+}
+
+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_);
+  buffered_output_.assign(data, size);
+}
+
+Error SdchStreamSource::ReadInternal(IOBuffer* dest_buffer,
+                                     size_t buffer_size,
+                                     size_t* bytes_read) {
+  // If there is no loaded dictionary, try loading a dictionary ID from
+  // |buffer_| and then loading a dictionary. If anything goes wrong in
+  // LoadDictionary, the delegate's HandleDictionaryError() 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 HandleDictionaryError cannot recover, this function
+  // returns a read error.
+  if (!dictionary_tried_load_ && !LoadDictionary())
+    return ERR_FAILED;
+
+  // Run the synchronous decompressor.
+  SdchStreamState state = Decompress(dest_buffer, buffer_size, bytes_read);
+
+  // If decompression failed, call HandleDecodingError 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) {
+    bool handled = AskDelegateToHandleDecodingError();
+    if (!handled)
+      return ERR_FAILED;
+    state = Decompress(dest_buffer, buffer_size, bytes_read);
+  }
+
+  if (*bytes_read > 0) {
+    return OK;
+  }
+
+  DCHECK(state == SDCH_STREAM_MORE_INPUT);
+  DCHECK(!buffer_->HasMoreBytes());
+
+  // If someone called ReplaceOutput to replace this stream's output,
+  // Decompress will return all the buffered output; if it did not do so, this
+  // stream is at EOF.
+  if (output_replaced_)
+    return OK;
+
+  return OK;
+}
+
+// Synchronous decompressor core.
+SdchStreamSource::SdchStreamState SdchStreamSource::Decompress(
+    IOBuffer* dest_buffer,
+    size_t buffer_size,
+    size_t* bytes_read) {
+  // 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(dest_buffer, buffer_size, bytes_read);
+
+  // If this stream is no longer decoding, pass any input through.
+  if (passthrough_)
+    return Passthrough(dest_buffer, buffer_size, bytes_read);
+
+  // Can't do any decoding until a dictionary is loaded, which requires more
+  // input.
+  if (!dictionary_tried_load_)
+    return SDCH_STREAM_MORE_INPUT;
+
+  // If there's any data left in the output buffer, try flushing some of it.
+  if (Flush(dest_buffer, buffer_size, bytes_read) != SDCH_STREAM_MORE_INPUT)
+    return SDCH_STREAM_MORE_OUTPUT_SPACE;
+
+  // This is awkward. In principle, this function is called only when the input
+  // buffer has more bytes, since vcdiff always consumes input; unfortunately,
+  // the dictionary-loading code earlier in the Read() loop can consume all the
+  // currently buffered input. If that happened, Decompress returns early here.
+  if (!buffer_->HasMoreBytes())
+    return SDCH_STREAM_MORE_INPUT;
+
+  // Now there is no data left in the output buffer and there is data in the
+  // input buffer, so run the vcdiff decoder.
+  DCHECK(buffered_output_.empty());
+  DCHECK(buffer_->HasMoreBytes());
+
+  // Calls to DecodeChunk always consume all their input, so this always drains
+  // the entire buffer.
+  bool ok = decoder_->DecodeChunk(buffer_->bytes(), buffer_->bytes_left(),
+                                  &buffered_output_);
+  if (ok) {
+    buffer_->WasDrained(buffer_->bytes_left());
+  } else {
+    if (!AskDelegateToHandleDecodingError())
+      return SDCH_STREAM_ERROR;
+  }
+
+  return Flush(dest_buffer, buffer_size, bytes_read);
+}
+
+SdchStreamSource::SdchStreamState SdchStreamSource::Passthrough(
+    IOBuffer* dest_buffer,
+    size_t buffer_size,
+    size_t* bytes_read) {
+  SdchStreamState state = Flush(dest_buffer, 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 (!buffer_->HasMoreBytes())
+    return SDCH_STREAM_MORE_INPUT;
+
+  size_t to_copy = buffer_size;
+  if (to_copy > buffer_->bytes_left())
+    to_copy = buffer_->bytes_left();
+  memcpy(dest_buffer->data(), buffer_->bytes(), to_copy);
+  buffer_->WasDrained(to_copy);
+  if (buffer_->HasMoreBytes())
+    return SDCH_STREAM_MORE_OUTPUT_SPACE;
+  else
+    return SDCH_STREAM_MORE_INPUT;
+}
+
+SdchStreamSource::SdchStreamState SdchStreamSource::Flush(IOBuffer* dest_buffer,
+                                                          size_t buffer_size,
+                                                          size_t* bytes_read) {
+  size_t to_copy = buffer_size;
+  if (to_copy > buffered_output_.length())
+    to_copy = buffered_output_.length();
+  if (to_copy == 0)
+    return SDCH_STREAM_MORE_INPUT;
+  memcpy(dest_buffer->data(), buffered_output_.data(), to_copy);
+  buffered_output_.erase(0, to_copy);
+  *bytes_read = to_copy;
+  return SDCH_STREAM_MORE_OUTPUT_SPACE;
+}
+
+// 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() {
+  size_t to_copy = kServerIdLength - dictionary_id_.length();
+  const std::string* dictionary_text = nullptr;
+  if (to_copy > buffer_->bytes_left())
+    to_copy = buffer_->bytes_left();
+  dictionary_id_.append(buffer_->bytes(), to_copy);
+  buffer_->WasDrained(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 HandleDictionaryError 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(),
+  // |stem_id| here removes the last byte blindly, and this method only calls
+  // GetDictionary with it if CouldBeDictionaryId returns true.
+  std::string stem_id = dictionary_id_.substr(0, kServerIdLength - 1);
+  if (!CouldBeDictionaryId(dictionary_id_) ||
+      !delegate_->GetDictionary(stem_id, &dictionary_text)) {
+    buffered_output_.append(dictionary_id_);
+    return AskDelegateToHandleDictionaryError();
+  }
+
+  decoder_->StartDecoding(dictionary_text->data(), dictionary_text->length());
+  return true;
+}
+
+bool SdchStreamSource::CouldBeDictionaryId(const std::string& id) const {
+  std::string kBase64UrlChars =
+      "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-";
+  for (size_t i = 0; i < kServerIdLength - 1; i++) {
+    if (kBase64UrlChars.find(id[i]) == std::string::npos)
+      return false;
+  }
+  if (id[kServerIdLength - 1] != '\0')
+    return false;
+  return true;
+}
+
+bool SdchStreamSource::AskDelegateToHandleDictionaryError() {
+  base::AutoReset<bool> resetter(&in_delegate_handler_, true);
+  return delegate_->HandleDictionaryError(this);
+}
+
+bool SdchStreamSource::AskDelegateToHandleDecodingError() {
+  base::AutoReset<bool> resetter(&in_delegate_handler_, true);
+  return delegate_->HandleDecodingError(this);
+}
+
+}  // namespace net