blob: 38de5fa2ecab770882d3594a2ba95f7de5810e0c [file] [log] [blame]
// 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_deflate_stream.h"
#include <algorithm>
#include <string>
#include "base/bind.h"
#include "base/logging.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/memory/scoped_vector.h"
#include "net/base/completion_callback.h"
#include "net/base/io_buffer.h"
#include "net/base/net_errors.h"
#include "net/websockets/websocket_deflate_predictor.h"
#include "net/websockets/websocket_deflater.h"
#include "net/websockets/websocket_errors.h"
#include "net/websockets/websocket_frame.h"
#include "net/websockets/websocket_inflater.h"
#include "net/websockets/websocket_stream.h"
class GURL;
namespace net {
namespace {
const int kWindowBits = 15;
const size_t kChunkSize = 4 * 1024;
} // namespace
WebSocketDeflateStream::WebSocketDeflateStream(
scoped_ptr<WebSocketStream> stream,
WebSocketDeflater::ContextTakeOverMode mode,
int client_window_bits,
scoped_ptr<WebSocketDeflatePredictor> predictor)
: stream_(stream.Pass()),
deflater_(mode),
inflater_(kChunkSize, kChunkSize),
reading_state_(NOT_READING),
writing_state_(NOT_WRITING),
current_reading_opcode_(WebSocketFrameHeader::kOpCodeText),
current_writing_opcode_(WebSocketFrameHeader::kOpCodeText),
predictor_(predictor.Pass()) {
DCHECK(stream_);
DCHECK_GE(client_window_bits, 8);
DCHECK_LE(client_window_bits, 15);
deflater_.Initialize(client_window_bits);
inflater_.Initialize(kWindowBits);
}
WebSocketDeflateStream::~WebSocketDeflateStream() {}
int WebSocketDeflateStream::ReadFrames(ScopedVector<WebSocketFrame>* frames,
const CompletionCallback& callback) {
int result = stream_->ReadFrames(
frames,
base::Bind(&WebSocketDeflateStream::OnReadComplete,
base::Unretained(this),
base::Unretained(frames),
callback));
if (result < 0)
return result;
DCHECK_EQ(OK, result);
DCHECK(!frames->empty());
return InflateAndReadIfNecessary(frames, callback);
}
int WebSocketDeflateStream::WriteFrames(ScopedVector<WebSocketFrame>* frames,
const CompletionCallback& callback) {
int result = Deflate(frames);
if (result != OK)
return result;
if (frames->empty())
return OK;
return stream_->WriteFrames(frames, callback);
}
void WebSocketDeflateStream::Close() { stream_->Close(); }
std::string WebSocketDeflateStream::GetSubProtocol() const {
return stream_->GetSubProtocol();
}
std::string WebSocketDeflateStream::GetExtensions() const {
return stream_->GetExtensions();
}
void WebSocketDeflateStream::OnReadComplete(
ScopedVector<WebSocketFrame>* frames,
const CompletionCallback& callback,
int result) {
if (result != OK) {
frames->clear();
callback.Run(result);
return;
}
int r = InflateAndReadIfNecessary(frames, callback);
if (r != ERR_IO_PENDING)
callback.Run(r);
}
int WebSocketDeflateStream::Deflate(ScopedVector<WebSocketFrame>* frames) {
ScopedVector<WebSocketFrame> frames_to_write;
// Store frames of the currently processed message if writing_state_ equals to
// WRITING_POSSIBLY_COMPRESSED_MESSAGE.
ScopedVector<WebSocketFrame> frames_of_message;
for (size_t i = 0; i < frames->size(); ++i) {
DCHECK(!(*frames)[i]->header.reserved1);
if (!WebSocketFrameHeader::IsKnownDataOpCode((*frames)[i]->header.opcode)) {
frames_to_write.push_back((*frames)[i]);
(*frames)[i] = NULL;
continue;
}
if (writing_state_ == NOT_WRITING)
OnMessageStart(*frames, i);
scoped_ptr<WebSocketFrame> frame((*frames)[i]);
(*frames)[i] = NULL;
predictor_->RecordInputDataFrame(frame.get());
if (writing_state_ == WRITING_UNCOMPRESSED_MESSAGE) {
if (frame->header.final)
writing_state_ = NOT_WRITING;
predictor_->RecordWrittenDataFrame(frame.get());
frames_to_write.push_back(frame.release());
current_writing_opcode_ = WebSocketFrameHeader::kOpCodeContinuation;
} else {
if (frame->data && !deflater_.AddBytes(frame->data->data(),
frame->header.payload_length)) {
DVLOG(1) << "WebSocket protocol error. "
<< "deflater_.AddBytes() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
if (frame->header.final && !deflater_.Finish()) {
DVLOG(1) << "WebSocket protocol error. "
<< "deflater_.Finish() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
if (writing_state_ == WRITING_COMPRESSED_MESSAGE) {
if (deflater_.CurrentOutputSize() >= kChunkSize ||
frame->header.final) {
int result = AppendCompressedFrame(frame->header, &frames_to_write);
if (result != OK)
return result;
}
if (frame->header.final)
writing_state_ = NOT_WRITING;
} else {
DCHECK_EQ(WRITING_POSSIBLY_COMPRESSED_MESSAGE, writing_state_);
bool final = frame->header.final;
frames_of_message.push_back(frame.release());
if (final) {
int result = AppendPossiblyCompressedMessage(&frames_of_message,
&frames_to_write);
if (result != OK)
return result;
frames_of_message.clear();
writing_state_ = NOT_WRITING;
}
}
}
}
DCHECK_NE(WRITING_POSSIBLY_COMPRESSED_MESSAGE, writing_state_);
frames->swap(frames_to_write);
return OK;
}
void WebSocketDeflateStream::OnMessageStart(
const ScopedVector<WebSocketFrame>& frames, size_t index) {
WebSocketFrame* frame = frames[index];
current_writing_opcode_ = frame->header.opcode;
DCHECK(current_writing_opcode_ == WebSocketFrameHeader::kOpCodeText ||
current_writing_opcode_ == WebSocketFrameHeader::kOpCodeBinary);
WebSocketDeflatePredictor::Result prediction =
predictor_->Predict(frames, index);
switch (prediction) {
case WebSocketDeflatePredictor::DEFLATE:
writing_state_ = WRITING_COMPRESSED_MESSAGE;
return;
case WebSocketDeflatePredictor::DO_NOT_DEFLATE:
writing_state_ = WRITING_UNCOMPRESSED_MESSAGE;
return;
case WebSocketDeflatePredictor::TRY_DEFLATE:
writing_state_ = WRITING_POSSIBLY_COMPRESSED_MESSAGE;
return;
}
NOTREACHED();
}
int WebSocketDeflateStream::AppendCompressedFrame(
const WebSocketFrameHeader& header,
ScopedVector<WebSocketFrame>* frames_to_write) {
const WebSocketFrameHeader::OpCode opcode = current_writing_opcode_;
scoped_refptr<IOBufferWithSize> compressed_payload =
deflater_.GetOutput(deflater_.CurrentOutputSize());
if (!compressed_payload) {
DVLOG(1) << "WebSocket protocol error. "
<< "deflater_.GetOutput() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
scoped_ptr<WebSocketFrame> compressed(new WebSocketFrame(opcode));
compressed->header.CopyFrom(header);
compressed->header.opcode = opcode;
compressed->header.final = header.final;
compressed->header.reserved1 =
(opcode != WebSocketFrameHeader::kOpCodeContinuation);
compressed->data = compressed_payload;
compressed->header.payload_length = compressed_payload->size();
current_writing_opcode_ = WebSocketFrameHeader::kOpCodeContinuation;
predictor_->RecordWrittenDataFrame(compressed.get());
frames_to_write->push_back(compressed.release());
return OK;
}
int WebSocketDeflateStream::AppendPossiblyCompressedMessage(
ScopedVector<WebSocketFrame>* frames,
ScopedVector<WebSocketFrame>* frames_to_write) {
DCHECK(!frames->empty());
const WebSocketFrameHeader::OpCode opcode = current_writing_opcode_;
scoped_refptr<IOBufferWithSize> compressed_payload =
deflater_.GetOutput(deflater_.CurrentOutputSize());
if (!compressed_payload) {
DVLOG(1) << "WebSocket protocol error. "
<< "deflater_.GetOutput() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
uint64 original_payload_length = 0;
for (size_t i = 0; i < frames->size(); ++i) {
WebSocketFrame* frame = (*frames)[i];
// Asserts checking that frames represent one whole data message.
DCHECK(WebSocketFrameHeader::IsKnownDataOpCode(frame->header.opcode));
DCHECK_EQ(i == 0,
WebSocketFrameHeader::kOpCodeContinuation !=
frame->header.opcode);
DCHECK_EQ(i == frames->size() - 1, frame->header.final);
original_payload_length += frame->header.payload_length;
}
if (original_payload_length <=
static_cast<uint64>(compressed_payload->size())) {
// Compression is not effective. Use the original frames.
for (size_t i = 0; i < frames->size(); ++i) {
WebSocketFrame* frame = (*frames)[i];
frames_to_write->push_back(frame);
predictor_->RecordWrittenDataFrame(frame);
(*frames)[i] = NULL;
}
frames->weak_clear();
return OK;
}
scoped_ptr<WebSocketFrame> compressed(new WebSocketFrame(opcode));
compressed->header.CopyFrom((*frames)[0]->header);
compressed->header.opcode = opcode;
compressed->header.final = true;
compressed->header.reserved1 = true;
compressed->data = compressed_payload;
compressed->header.payload_length = compressed_payload->size();
predictor_->RecordWrittenDataFrame(compressed.get());
frames_to_write->push_back(compressed.release());
return OK;
}
int WebSocketDeflateStream::Inflate(ScopedVector<WebSocketFrame>* frames) {
ScopedVector<WebSocketFrame> frames_to_output;
ScopedVector<WebSocketFrame> frames_passed;
frames->swap(frames_passed);
for (size_t i = 0; i < frames_passed.size(); ++i) {
scoped_ptr<WebSocketFrame> frame(frames_passed[i]);
frames_passed[i] = NULL;
DVLOG(3) << "Input frame: opcode=" << frame->header.opcode
<< " final=" << frame->header.final
<< " reserved1=" << frame->header.reserved1
<< " payload_length=" << frame->header.payload_length;
if (!WebSocketFrameHeader::IsKnownDataOpCode(frame->header.opcode)) {
frames_to_output.push_back(frame.release());
continue;
}
if (reading_state_ == NOT_READING) {
if (frame->header.reserved1)
reading_state_ = READING_COMPRESSED_MESSAGE;
else
reading_state_ = READING_UNCOMPRESSED_MESSAGE;
current_reading_opcode_ = frame->header.opcode;
} else {
if (frame->header.reserved1) {
DVLOG(1) << "WebSocket protocol error. "
<< "Receiving a non-first frame with RSV1 flag set.";
return ERR_WS_PROTOCOL_ERROR;
}
}
if (reading_state_ == READING_UNCOMPRESSED_MESSAGE) {
if (frame->header.final)
reading_state_ = NOT_READING;
current_reading_opcode_ = WebSocketFrameHeader::kOpCodeContinuation;
frames_to_output.push_back(frame.release());
} else {
DCHECK_EQ(reading_state_, READING_COMPRESSED_MESSAGE);
if (frame->data && !inflater_.AddBytes(frame->data->data(),
frame->header.payload_length)) {
DVLOG(1) << "WebSocket protocol error. "
<< "inflater_.AddBytes() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
if (frame->header.final) {
if (!inflater_.Finish()) {
DVLOG(1) << "WebSocket protocol error. "
<< "inflater_.Finish() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
}
// TODO(yhirano): Many frames can be generated by the inflater and
// memory consumption can grow.
// We could avoid it, but avoiding it makes this class much more
// complicated.
while (inflater_.CurrentOutputSize() >= kChunkSize ||
frame->header.final) {
size_t size = std::min(kChunkSize, inflater_.CurrentOutputSize());
scoped_ptr<WebSocketFrame> inflated(
new WebSocketFrame(WebSocketFrameHeader::kOpCodeText));
scoped_refptr<IOBufferWithSize> data = inflater_.GetOutput(size);
bool is_final = !inflater_.CurrentOutputSize() && frame->header.final;
if (!data) {
DVLOG(1) << "WebSocket protocol error. "
<< "inflater_.GetOutput() returns an error.";
return ERR_WS_PROTOCOL_ERROR;
}
inflated->header.CopyFrom(frame->header);
inflated->header.opcode = current_reading_opcode_;
inflated->header.final = is_final;
inflated->header.reserved1 = false;
inflated->data = data;
inflated->header.payload_length = data->size();
DVLOG(3) << "Inflated frame: opcode=" << inflated->header.opcode
<< " final=" << inflated->header.final
<< " reserved1=" << inflated->header.reserved1
<< " payload_length=" << inflated->header.payload_length;
frames_to_output.push_back(inflated.release());
current_reading_opcode_ = WebSocketFrameHeader::kOpCodeContinuation;
if (is_final)
break;
}
if (frame->header.final)
reading_state_ = NOT_READING;
}
}
frames->swap(frames_to_output);
return frames->empty() ? ERR_IO_PENDING : OK;
}
int WebSocketDeflateStream::InflateAndReadIfNecessary(
ScopedVector<WebSocketFrame>* frames,
const CompletionCallback& callback) {
int result = Inflate(frames);
while (result == ERR_IO_PENDING) {
DCHECK(frames->empty());
result = stream_->ReadFrames(
frames,
base::Bind(&WebSocketDeflateStream::OnReadComplete,
base::Unretained(this),
base::Unretained(frames),
callback));
if (result < 0)
break;
DCHECK_EQ(OK, result);
DCHECK(!frames->empty());
result = Inflate(frames);
}
if (result < 0)
frames->clear();
return result;
}
} // namespace net