// 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() { DCHECK_LT(tail_, buffer_.size()); 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_); DCHECK_LT(head_, buffer_.size()); 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_); DCHECK_LT(head_, buffer_.size()); 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