Introduce WebSocketInflater.

Implement WebSocketInflater, a utility class for the permessage-deflate WebSocket extension.

BUG=280910
R=ricea, tyoshino

Review URL: https://chromiumcodereview.appspot.com/23480049

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@224271 0039d316-1c4b-4281-b951-d872f2087c98

net/net.gyp

@@ -1122,6 +1122,8 @@
         'websockets/websocket_handshake_constants.h',
         'websockets/websocket_handshake_handler.cc',
         'websockets/websocket_handshake_handler.h',
+        'websockets/websocket_inflater.cc',
+        'websockets/websocket_inflater.h',
         'websockets/websocket_job.cc',
         'websockets/websocket_job.h',
         'websockets/websocket_mux.h',
@@ -1893,6 +1895,7 @@
         'websockets/websocket_frame_test.cc',
         'websockets/websocket_handshake_handler_test.cc',
         'websockets/websocket_handshake_handler_spdy_test.cc',
+        'websockets/websocket_inflater_test.cc',
         'websockets/websocket_job_test.cc',
         'websockets/websocket_net_log_params_test.cc',
         'websockets/websocket_throttle_test.cc',

net/websockets/README

@@ -53,6 +53,9 @@ websocket_frame_parser.cc
 websocket_frame_parser.h
 websocket_frame_parser_test.cc
 websocket_frame_test.cc
+websocket_inflater.cc
+websocket_inflater.h
+websocket_inflater_test.cc
 websocket_mux.h
 websocket_stream_base.h
 websocket_stream.cc

net/websockets/websocket_inflater.cc

@@ -0,0 +1,280 @@
+// Copyright 2013 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/websockets/websocket_inflater.h"
+
+#include <algorithm>
+#include <deque>
+#include <vector>
+
+#include "base/logging.h"
+#include "net/base/io_buffer.h"
+#include "third_party/zlib/zlib.h"
+
+namespace net {
+
+namespace {
+
+class ShrinkableIOBufferWithSize : public IOBufferWithSize {
+ public:
+  explicit ShrinkableIOBufferWithSize(int size)
+      : IOBufferWithSize(size) {}
+
+  void Shrink(int new_size) {
+    DCHECK_LE(new_size, size_);
+    size_ = new_size;
+  }
+
+ private:
+  virtual ~ShrinkableIOBufferWithSize() {}
+};
+
+}  // namespace
+
+WebSocketInflater::WebSocketInflater()
+    : input_queue_(kDefaultInputIOBufferCapacity),
+      output_buffer_(kDefaultBufferCapacity) {}
+
+WebSocketInflater::WebSocketInflater(size_t input_queue_capacity,
+                                     size_t output_buffer_capacity)
+    : input_queue_(input_queue_capacity),
+      output_buffer_(output_buffer_capacity) {
+  DCHECK_GT(input_queue_capacity, 0u);
+  DCHECK_GT(output_buffer_capacity, 0u);
+}
+
+bool WebSocketInflater::Initialize(int window_bits) {
+  DCHECK_LE(8, window_bits);
+  DCHECK_GE(15, window_bits);
+  stream_.reset(new z_stream);
+  memset(stream_.get(), 0, sizeof(*stream_));
+  int result = inflateInit2(stream_.get(), -window_bits);
+  if (result != Z_OK) {
+    inflateEnd(stream_.get());
+    stream_.reset();
+    return false;
+  }
+  return true;
+}
+
+WebSocketInflater::~WebSocketInflater() {
+  if (stream_) {
+    inflateEnd(stream_.get());
+    stream_.reset();
+  }
+}
+
+bool WebSocketInflater::AddBytes(const char* data, size_t size) {
+  if (!size)
+    return true;
+
+  if (!input_queue_.IsEmpty()) {
+    // choked
+    input_queue_.Push(data, size);
+    return true;
+  }
+
+  int result = InflateWithFlush(data, size);
+  if (stream_->avail_in > 0)
+    input_queue_.Push(&data[size - stream_->avail_in], stream_->avail_in);
+
+  return result == Z_OK || result == Z_BUF_ERROR;
+}
+
+bool WebSocketInflater::Finish() {
+  return AddBytes("\x00\x00\xff\xff", 4);
+}
+
+scoped_refptr<IOBufferWithSize> WebSocketInflater::GetOutput(size_t size) {
+  scoped_refptr<ShrinkableIOBufferWithSize> buffer =
+      new ShrinkableIOBufferWithSize(size);
+  size_t num_bytes_copied = 0;
+
+  while (num_bytes_copied < size && output_buffer_.Size() > 0) {
+    size_t num_bytes_to_copy =
+        std::min(output_buffer_.Size(), size - num_bytes_copied);
+    output_buffer_.Read(&buffer->data()[num_bytes_copied], num_bytes_to_copy);
+    num_bytes_copied += num_bytes_to_copy;
+    int result = InflateChokedInput();
+    if (result != Z_OK && result != Z_BUF_ERROR)
+      return NULL;
+  }
+  buffer->Shrink(num_bytes_copied);
+  return buffer;
+}
+
+int WebSocketInflater::InflateWithFlush(const char* next_in, size_t avail_in) {
+  int result = Inflate(next_in, avail_in, Z_NO_FLUSH);
+  if (result != Z_OK && result != Z_BUF_ERROR)
+    return result;
+
+  if (CurrentOutputSize() > 0)
+    return result;
+  // CurrentOutputSize() == 0 means there is no data to be output,
+  // so we should make sure it by using Z_SYNC_FLUSH.
+  return Inflate(reinterpret_cast<const char*>(stream_->next_in),
+                 stream_->avail_in,
+                 Z_SYNC_FLUSH);
+}
+
+int WebSocketInflater::Inflate(const char* next_in,
+                               size_t avail_in,
+                               int flush) {
+  stream_->next_in = reinterpret_cast<Bytef*>(const_cast<char*>(next_in));
+  stream_->avail_in = avail_in;
+
+  int result = Z_BUF_ERROR;
+  do {
+    std::pair<char*, size_t> tail = output_buffer_.GetTail();
+    if (!tail.second)
+      break;
+
+    stream_->next_out = reinterpret_cast<Bytef*>(tail.first);
+    stream_->avail_out = tail.second;
+    result = inflate(stream_.get(), flush);
+    output_buffer_.AdvanceTail(tail.second - stream_->avail_out);
+    if (result == Z_STREAM_END) {
+      // Received a block with BFINAL set to 1. Reset the decompression state.
+      result = inflateReset(stream_.get());
+    } else if (tail.second == stream_->avail_out) {
+      break;
+    }
+  } while (result == Z_OK || result == Z_BUF_ERROR);
+  return result;
+}
+
+int WebSocketInflater::InflateChokedInput() {
+  if (input_queue_.IsEmpty())
+    return InflateWithFlush(NULL, 0);
+
+  int result = Z_BUF_ERROR;
+  while (!input_queue_.IsEmpty()) {
+    std::pair<char*, size_t> top = input_queue_.Top();
+
+    result = InflateWithFlush(top.first, top.second);
+    input_queue_.Consume(top.second - stream_->avail_in);
+
+    if (result != Z_OK && result != Z_BUF_ERROR)
+      return result;
+
+    if (stream_->avail_in > 0) {
+      // There are some data which are not consumed.
+      break;
+    }
+  }
+  return result;
+}
+
+WebSocketInflater::OutputBuffer::OutputBuffer(size_t capacity)
+    : capacity_(capacity),
+      buffer_(capacity_ + 1),  // 1 for sentinel
+      head_(0),
+      tail_(0) {}
+
+WebSocketInflater::OutputBuffer::~OutputBuffer() {}
+
+size_t WebSocketInflater::OutputBuffer::Size() const {
+  return (tail_ + buffer_.size() - head_) % buffer_.size();
+}
+
+std::pair<char*, size_t> WebSocketInflater::OutputBuffer::GetTail() {
+  return std::make_pair(&buffer_[tail_],
+                        std::min(capacity_ - Size(), buffer_.size() - tail_));
+}
+
+void WebSocketInflater::OutputBuffer::Read(char* dest, size_t size) {
+  DCHECK_LE(size, Size());
+
+  size_t num_bytes_copied = 0;
+  if (tail_ < head_) {
+    size_t num_bytes_to_copy = std::min(size, buffer_.size() - head_);
+    memcpy(&dest[num_bytes_copied], &buffer_[head_], num_bytes_to_copy);
+    AdvanceHead(num_bytes_to_copy);
+    num_bytes_copied += num_bytes_to_copy;
+  }
+
+  if (num_bytes_copied == size)
+    return;
+  DCHECK_LE(head_, tail_);
+  size_t num_bytes_to_copy = size - num_bytes_copied;
+  DCHECK_LE(num_bytes_to_copy, tail_ - head_);
+  memcpy(&dest[num_bytes_copied], &buffer_[head_], num_bytes_to_copy);
+  AdvanceHead(num_bytes_to_copy);
+  num_bytes_copied += num_bytes_to_copy;
+  DCHECK_EQ(size, num_bytes_copied);
+  return;
+}
+
+void WebSocketInflater::OutputBuffer::AdvanceHead(size_t advance) {
+  DCHECK_LE(advance, Size());
+  head_ = (head_ + advance) % buffer_.size();
+}
+
+void WebSocketInflater::OutputBuffer::AdvanceTail(size_t advance) {
+  DCHECK_LE(advance + Size(), capacity_);
+  tail_ = (tail_ + advance) % buffer_.size();
+}
+
+WebSocketInflater::InputQueue::InputQueue(size_t capacity)
+    : capacity_(capacity), head_of_first_buffer_(0), tail_of_last_buffer_(0) {}
+
+WebSocketInflater::InputQueue::~InputQueue() {}
+
+std::pair<char*, size_t> WebSocketInflater::InputQueue::Top() {
+  DCHECK(!IsEmpty());
+  if (buffers_.size() == 1) {
+    return std::make_pair(&buffers_.front()->data()[head_of_first_buffer_],
+                          tail_of_last_buffer_ - head_of_first_buffer_);
+  }
+  return std::make_pair(&buffers_.front()->data()[head_of_first_buffer_],
+                        capacity_ - head_of_first_buffer_);
+}
+
+void WebSocketInflater::InputQueue::Push(const char* data, size_t size) {
+  if (!size)
+    return;
+
+  size_t num_copied_bytes = 0;
+  if (!IsEmpty())
+    num_copied_bytes += PushToLastBuffer(data, size);
+
+  while (num_copied_bytes < size) {
+    DCHECK(IsEmpty() || tail_of_last_buffer_ == capacity_);
+
+    buffers_.push_back(new IOBufferWithSize(capacity_));
+    tail_of_last_buffer_ = 0;
+    num_copied_bytes +=
+        PushToLastBuffer(&data[num_copied_bytes], size - num_copied_bytes);
+  }
+}
+
+void WebSocketInflater::InputQueue::Consume(size_t size) {
+  DCHECK(!IsEmpty());
+  DCHECK_LE(size + head_of_first_buffer_, capacity_);
+
+  head_of_first_buffer_ += size;
+  if (head_of_first_buffer_ == capacity_) {
+    buffers_.pop_front();
+    head_of_first_buffer_ = 0;
+  }
+  if (buffers_.size() == 1 && head_of_first_buffer_ == tail_of_last_buffer_) {
+    buffers_.pop_front();
+    head_of_first_buffer_ = 0;
+    tail_of_last_buffer_ = 0;
+  }
+}
+
+size_t WebSocketInflater::InputQueue::PushToLastBuffer(const char* data,
+                                                       size_t size) {
+  DCHECK(!IsEmpty());
+  size_t num_bytes_to_copy = std::min(size, capacity_ - tail_of_last_buffer_);
+  if (!num_bytes_to_copy)
+    return 0;
+  IOBufferWithSize* buffer = buffers_.back().get();
+  memcpy(&buffer->data()[tail_of_last_buffer_], data, num_bytes_to_copy);
+  tail_of_last_buffer_ += num_bytes_to_copy;
+  return num_bytes_to_copy;
+}
+
+}  // namespace net