WebSocketChannel implementation

net/websockets/websocket_channel.cc

+// 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_channel.h"
+
+#include <algorithm>
+
+#include "base/basictypes.h"  // for size_t
+#include "base/bind.h"
+#include "base/string_util.h"
+#include "net/base/big_endian.h"
+#include "net/base/io_buffer.h"
+#include "net/http/http_request_info.h"
+#include "net/http/http_stream_factory.h"
+#include "net/ssl/ssl_config_service.h"
+#include "net/websockets/websocket_errors.h"
+#include "net/websockets/websocket_event_interface.h"
+#include "net/websockets/websocket_frame.h"
+#include "net/websockets/websocket_mux.h"
+#include "net/websockets/websocket_stream.h"
+
+namespace net {
+
+namespace {
+
+const int kDefaultSendQuotaLowWaterMark = 1 << 16;
+const int kDefaultSendQuotaHighWaterMark = 1 << 17;
+const size_t kWebSocketCloseCodeLength = 2;
+
+// IOBuffer is too hardcore to offer a const accessor. Make our own.
+const char* GetConstData(const IOBuffer* iobuffer) {
+  return const_cast<IOBuffer*>(iobuffer)->data();
+}
+
+// Concatenate the data from two IOBufferWithSize objects into a single one.
+IOBufferWithSize* ConcatenateIOBuffers(const IOBufferWithSize* part1,
+                                       const IOBufferWithSize* part2) {
+  int newsize = part1->size() + part2->size();
+  IOBufferWithSize* newbuffer = new IOBufferWithSize(newsize);
+  std::copy(GetConstData(part1),
+            GetConstData(part1) + part1->size(),
+            newbuffer->data());
+  std::copy(GetConstData(part2),
+            GetConstData(part2) + part2->size(),
+            newbuffer->data() + part1->size());
+  return newbuffer;
+}
+
+}  // namespace
+
+struct WebSocketChannel::SendBuffer {
+  SendBuffer() : frames(), total_bytes(0) {}

[yhirano, 2013/06/28 06:49:17]

nit: You don't need frames().

[Adam Rice, 2013/06/28 07:59:29]

I have a habit of initialising all the member varables in the constructor, since
for built-in types it makes a difference.

For example I have seen cases where there is a scoped_ptr<C> member then later
someone changes it to be a C* but forgets to update the constructor and the code
becomes unstable.

However, this habit seems to worry code reviewers a lot so I think I will stop
doing it.

+  ScopedVector<WebSocketFrameChunk> frames;
+  size_t total_bytes;
+};
+
+// Implementation of WebSocketStream::ConnectDelegate that simply forwards the
+// calls on to the WebSocketChannel that created it.
+class WebSocketChannel::ConnectDelegate
+    : public WebSocketStream::ConnectDelegate {
+ public:
+  ConnectDelegate(WebSocketChannel* creator) : creator_(creator) {}
+
+  virtual void OnSuccess(scoped_ptr<WebSocketStream> stream) OVERRIDE {
+    creator_->OnConnectSuccess(stream.Pass());
+  }
+
+  virtual void OnFailure(unsigned short websocket_error) OVERRIDE {
+    creator_->OnConnectFailure(websocket_error);
+  }
+
+ private:
+  // A pointer to the WebSocketChannel that created us. We do not need to worry
+  // about this pointer being stale, because deleting WebSocketChannel cancels
+  // the connect process, deleting this object and preventing its callbacks from
+  //  being called.
+  WebSocketChannel* const creator_;
+
+  DISALLOW_COPY_AND_ASSIGN(ConnectDelegate);
+};
+
+WebSocketChannel::WebSocketChannel(
+    const GURL& socket_url,
+    scoped_ptr<WebSocketEventInterface> event_interface)
+    : socket_url_(socket_url),
+      event_interface_(event_interface.Pass()),
+      currently_sending_(),
+      send_next_(),
+      read_frame_chunks_(),
+      current_frame_header_(),
+      incomplete_control_frame_body_(),

[yhirano, 2013/06/28 06:49:17]

nit: You don't need call a constructor if it doesn't take arguments.

[Adam Rice, 2013/06/28 07:59:29]

I removed the redundant initialisers. See my comment above.

+      send_quota_low_water_mark_(kDefaultSendQuotaLowWaterMark),
+      send_quota_high_water_mark_(kDefaultSendQuotaHighWaterMark),
+      current_send_quota_(0),
+      state_(FRESHLY_CONSTRUCTED),
+      weak_factory_(this) {}
+
+WebSocketChannel::~WebSocketChannel() {
+  // The stream may hold a pointer to read_frame_chunks_, and so it needs to be
+  // destroyed first.
+  stream_.reset();
+}
+
+void WebSocketChannel::SendAddChannelRequest(
+    const std::vector<std::string>& requested_subprotocols,
+    const GURL& origin,
+    URLRequestContext* url_request_context) {
+  // Delegate to the tested version.
+  SendAddChannelRequestWithFactory(
+      requested_subprotocols,
+      origin,
+      url_request_context,
+      base::Bind(&WebSocketStream::CreateAndConnectStream));
+}
+
+void WebSocketChannel::SendAddChannelRequestWithFactory(
+    const std::vector<std::string>& requested_subprotocols,
+    const GURL& origin,
+    URLRequestContext* url_request_context,
+    base::Callback<scoped_ptr<WebSocketStreamRequest>(
+        const GURL&,
+        const std::vector<std::string>&,
+        const GURL&,
+        URLRequestContext*,
+        const BoundNetLog&,
+        scoped_ptr<WebSocketStream::ConnectDelegate>)> factory) {
+  DCHECK_EQ(FRESHLY_CONSTRUCTED, state_);
+  scoped_ptr<WebSocketStream::ConnectDelegate> connect_delegate(
+      new WebSocketChannel::ConnectDelegate(this));
+  stream_request_ = factory.Run(socket_url_,
+                                requested_subprotocols,
+                                origin,
+                                url_request_context,
+                                BoundNetLog(),
+                                connect_delegate.Pass());
+  state_ = CONNECTING;
+}
+
+void WebSocketChannel::OnConnectSuccess(scoped_ptr<WebSocketStream> stream) {
+  DCHECK(stream);
+  DCHECK_EQ(CONNECTING, state_);
+  stream_ = stream.Pass();
+  event_interface_->OnAddChannelResponse(false, stream_->GetSubProtocol());
+  // TODO(ricea): Get flow control information from the WebSocketStream once we
+  // have a multiplexing WebSocketStream.
+  event_interface_->OnFlowControl(send_quota_high_water_mark_);
+  current_send_quota_ = send_quota_high_water_mark_;
+  state_ = CONNECTED;
+  // We don't need this any more.
+  stream_request_.reset();
+  ReadFrames();
+}
+
+void WebSocketChannel::OnConnectFailure(unsigned short web_socket_error) {
+  DCHECK_EQ(CONNECTING, state_);
+  event_interface_->OnAddChannelResponse(true, "");
+  stream_request_.reset();
+  state_ = CLOSED;
+}
+
+void WebSocketChannel::SendFrame(bool fin,
+                                 WebSocketFrameHeader::OpCode op_code,
+                                 const std::vector<char>& data) {
+  if (data.size() > INT_MAX) {

[tyoshino (SeeGerritForStatus), 2013/06/28 07:55:27]

 > -> >= ?

what is this check for?

[Adam Rice, 2013/06/28 08:28:06]

Mostly just paranoia. The networking code uses int for buffer sizes in lots of
places and I wanted to avoid any risk of overflow right from the start.

[tyoshino (SeeGerritForStatus), 2013/06/28 08:30:51]

ok. it's fine for now. please write a comment that this method accepts only data
not bigger than INT_MAX and some TODO to support big message by doing
fragmentation at higher layer.

[Adam Rice, 2013/06/28 14:20:48]

I updated the comment.

+    DCHECK_LE(data.size(), static_cast<size_t>(INT_MAX));
+    return;
+  }
+  if (stream_ == NULL) {
+    LOG(DFATAL) << "Got SendFrame without a connection established; "
+                << "misbehaving renderer? fin=" << fin << " op_code=" << op_code
+                << " data.size()=" << data.size();
+    return;
+  }
+  if (state_ == SEND_CLOSED || state_ == RECV_CLOSED || state_ == CLOSED) {
+    VLOG(1) << "SendFrame called in state " << state_
+            << ". This may be a bug, or a harmless race.";
+    return;
+  }
+  DCHECK_EQ(CONNECTED, state_);
+  CHECK_GE(current_send_quota_, 0);  // Security-critical invariant
+  typedef std::vector<char>::size_type size_type;

[tyoshino (SeeGerritForStatus), 2013/06/28 07:55:27]

don't worry about size_type being different from size_t. please just use size_t.

[Adam Rice, 2013/06/28 08:28:06]

Done.

+  if (data.size() > static_cast<size_type>(current_send_quota_)) {
+    FailChannel(SEND_INTERNAL_ERROR,
+                kWebSocketMuxErrorSendQuotaViolation,
+                "Send quota exceeded");
+    return;
+  }
+  if (!WebSocketFrameHeader::IsKnownDataOpCode(op_code)) {
+    LOG(DFATAL) << "Got SendFrame with bogus op_code " << op_code
+                << "; misbehaving renderer? fin=" << fin
+                << " data.size()=" << data.size();
+    return;
+  }
+  current_send_quota_ -= data.size();
+  // TODO(ricea): If current_send_quota_ has dropped below
+  // send_quota_low_water_mark_, we may want to consider increasing the "low
+  // water mark" and "high water mark", but only if we think we are not
+  // saturating the link to the WebSocket server.
+  // TODO(ricea): For kOpCodeText, do UTF-8 validation?
+  Send(fin, op_code, data);
+}
+
+void WebSocketChannel::Send(bool fin,
+                            WebSocketFrameHeader::OpCode op_code,
+                            const std::vector<char>& data) {
+  scoped_refptr<IOBufferWithSize> buffer(new IOBufferWithSize(data.size()));
+  std::copy(data.begin(), data.end(), buffer->data());
+  SendIOBufferWithSize(fin, op_code, buffer);
+}
+
+void WebSocketChannel::SendIOBufferWithSize(
+    bool fin,
+    WebSocketFrameHeader::OpCode op_code,
+    const scoped_refptr<IOBufferWithSize>& buffer) {
+  DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
+  DCHECK(stream_);
+  scoped_ptr<WebSocketFrameHeader> header(new WebSocketFrameHeader(op_code));
+  header->final = fin;
+  header->masked = true;
+  header->payload_length = buffer->size();
+  scoped_ptr<WebSocketFrameChunk> chunk(new WebSocketFrameChunk());
+  chunk->header = header.Pass();
+  chunk->final_chunk = true;
+  chunk->data = buffer;
+  if (currently_sending_) {
+    // Either the link to the WebSocket server is saturated, or we are simply
+    // processing a batch of messages.
+    // TODO(ricea): We need to keep some statistics to work out which situation
+    // we are in and adjust quota appropriately.
+    if (!send_next_) {
+      send_next_.reset(new SendBuffer);
+    }
+    send_next_->frames.push_back(chunk.release());
+    send_next_->total_bytes += buffer->size();
+  } else {
+    currently_sending_.reset(new SendBuffer);
+    currently_sending_->frames.push_back(chunk.release());
+    currently_sending_->total_bytes += buffer->size();
+    WriteFrames();
+  }
+}
+
+void WebSocketChannel::WriteFrames() {
+  // This is safe because we own the WebSocketStream and destroying it cancels
+  // all callbacks.
+  int result = stream_->WriteFrames(
+      &(currently_sending_->frames),
+      base::Bind(&WebSocketChannel::OnWriteDone, base::Unretained(this)));
+  if (result != ERR_IO_PENDING) {
+    OnWriteDone(result);
+  }
+}
+
+void WebSocketChannel::OnWriteDone(int result) {
+  DCHECK(state_ != FRESHLY_CONSTRUCTED && state_ != CONNECTING);
+  DCHECK_NE(ERR_IO_PENDING, result);
+  DCHECK(currently_sending_);
+  switch (result) {
+    case OK:
+      if (send_next_) {
+        currently_sending_ = send_next_.Pass();
+        WriteFrames();
+      } else {
+        currently_sending_.reset();
+        if (current_send_quota_ < send_quota_low_water_mark_) {
+          // TODO(ricea): Increase low_water_mark and high_water_mark if
+          // throughput is high, reduce them if throughput is low.  Low water
+          // mark needs to be >= the bandwidth delay product *of the IPC
+          // channel*. Because factors like context-switch time, thread wake-up
+          // time, and bus speed come into play it is complex and probably needs
+          // to be determined empirically.
+          DCHECK_LE(send_quota_low_water_mark_, send_quota_high_water_mark_);
+          // TODO(ricea): Truncate quota by the quota specified by the remote
+          // server, if the protocol in use supports quota.
+          int fresh_quota = send_quota_high_water_mark_ - current_send_quota_;
+          event_interface_->OnFlowControl(fresh_quota);
+          current_send_quota_ += fresh_quota;
+        }
+      }
+      break;
+
+    // If a recoverable error condition existed, it would go here.
+
+    default:
+      DCHECK_LT(result, 0)
+          << "WriteFrames() should only return OK or ERR_ codes";
+      stream_->Close();
+      state_ = CLOSED;
+      event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                      "Abnormal Closure");
+      break;
+  }
+}
+
+void WebSocketChannel::ReadFrames() {
+  // This use if base::Unretained is safe because we own the WebSocketStream,
+  // and any pending reads will be cancelled when it is destroyed.
+  int result = stream_->ReadFrames(
+      &read_frame_chunks_,
+      base::Bind(&WebSocketChannel::OnReadDone, base::Unretained(this)));
+  if (result != ERR_IO_PENDING) {
+    OnReadDone(result);
+  }
+}
+
+void WebSocketChannel::OnReadDone(int result) {
+  DCHECK(state_ != FRESHLY_CONSTRUCTED && state_ != CONNECTING);
+  DCHECK_NE(ERR_IO_PENDING, result);
+  switch (result) {
+    case OK:
+      // ReadFrames() must use ERR_CONNECTION_CLOSED for a closed connection
+      // with no data read, not an empty response.
+      DCHECK(!read_frame_chunks_.empty())
+          << "ReadFrames() returned OK, but nothing was read.";
+      for (size_t i = 0; i < read_frame_chunks_.size(); ++i) {
+        scoped_ptr<WebSocketFrameChunk> chunk(read_frame_chunks_[i]);

[yhirano, 2013/06/28 06:49:17]

What happens when ProcessFrameChunk fails (i.e. FailChannel is called) and there
are more chunks to be processed?

Assume the channel is brand-new and read_frame_chunks_ is [chunk1(masked, with
header), chunk2 (without header)].
ProcessFrameChunk(chunk1) will fail and ProcessFrameChunk(chunk2) will be called
after that. Is it OK?
At least, current_frame_header_ will be null after ProcessFrameChunk(chunk1) and
it will cause SEGV in ProcessFrameChunk(chunk2), I think.

[Adam Rice, 2013/06/28 07:59:29]

It is important that is doesn't make any difference whether two chunks arrive in
one read or in two reads. So ProcessFrameChunk() needs to be able to handle
frames in every possible case anyway.

I tried to make ProcessFrameChunk() robust in all cases but I missed the one
that you mentioned. I have added a check for it now, and I will add a test in a
separate CL.

I looked for more cases that ProcessFrameChunk() isn't handling but I wasn't
able to find any. Please let me know if you spot one.

+        read_frame_chunks_[i] = NULL;
+        ProcessFrameChunk(chunk.Pass());
+      }
+      read_frame_chunks_.clear();
+      // We need to always keep a call to ReadFrames pending.
+      ReadFrames();
+      return;
+
+    case ERR_CONNECTION_CLOSED: {
+      State old_state = state_;
+      state_ = CLOSED;
+      if (old_state != RECV_CLOSED && old_state != CLOSED) {
+        // We need to inform the render process of the unexpected closure.
+        event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                        "Abnormal Closure");
+      }
+      return;
+    }
+
+    default: {
+      DCHECK_LT(result, 0)
+          << "ReadFrames() should only return OK or ERR_ codes";
+      stream_->Close();
+      State old_state = state_;
+      state_ = CLOSED;
+      if (old_state != RECV_CLOSED && old_state != CLOSED) {
+        event_interface_->OnDropChannel(kWebSocketErrorAbnormalClosure,
+                                        "Abnormal Closure");
+      }
+      return;
+    }
+  }
+}
+
+// TODO(ricea): This method is too long. Break it up.
+void WebSocketChannel::ProcessFrameChunk(
+    scoped_ptr<WebSocketFrameChunk> chunk) {
+  // frame_header stores the header for this frame, either saved from a previous
+  // chunk or from this chunk if it includes a header.
+  scoped_ptr<WebSocketFrameHeader> frame_header;
+  // Borrow the value of current_frame_header_. At the end of the function we
+  // will put it back if it is still valid, or replace it with the header from
+  // the new chunk.
+  frame_header.swap(current_frame_header_);
+  bool first_chunk = false;
+  if (chunk->header) {
+    first_chunk = true;
+    frame_header.swap(chunk->header);
+    if (frame_header->masked) {
+      // RFC6455 Section 5.1 "A client MUST close a connection if it detects a
+      // masked frame."
+      FailChannel(SEND_REAL_ERROR,
+                  kWebSocketErrorProtocolError,
+                  "Masked frame from server");
+      return;
+    }
+  }

[yhirano, 2013/06/28 06:49:17]

optional: DCHECK(frame_header) here?

[Adam Rice, 2013/06/28 07:59:29]

I made it return instead, since this can legitimately happen if the previous
frame was invalid.

+  scoped_refptr<IOBufferWithSize> data_buffer;
+  data_buffer.swap(chunk->data);
+  if (WebSocketFrameHeader::IsKnownControlOpCode(frame_header->opcode)) {
+    if (chunk->final_chunk) {
+      if (incomplete_control_frame_body_) {
+        VLOG(2) << "Rejoining a split control frame, opcode "
+                << frame_header->opcode;
+        scoped_refptr<IOBufferWithSize> old_data_buffer;
+        old_data_buffer.swap(incomplete_control_frame_body_);
+        scoped_refptr<IOBufferWithSize> new_data_buffer;
+        new_data_buffer.swap(data_buffer);
+        data_buffer =
+            ConcatenateIOBuffers(old_data_buffer.get(), new_data_buffer.get());
+      }
+    } else {
+      // TODO(ricea): Enforce a maximum size of 125 bytes on the control frames
+      // we accept.
+      VLOG(2) << "Encountered a split control frame, opcode "
+              << frame_header->opcode;
+      if (incomplete_control_frame_body_) {
+        // The really horrid case. We need to create a new IOBufferWithSize
+        // combining the new one and the old one. This should virtually never
+        // happen.
+        // TODO(ricea): This algorithm is O(N^2). Use a fixed 127-byte byffer
+        // instead.
+        VLOG(3) << "Hit the really horrid case";
+        scoped_refptr<IOBufferWithSize> old_body;
+        old_body.swap(incomplete_control_frame_body_);
+        incomplete_control_frame_body_ =
+            ConcatenateIOBuffers(old_body.get(), data_buffer.get());
+      } else {
+        // The merely horrid case. Store the IOBufferWithSize to use when the
+        // rest of the control frame arrives.
+        incomplete_control_frame_body_.swap(data_buffer);
+      }
+      current_frame_header_.swap(frame_header);
+      return;
+    }
+  }
+
+  WebSocketFrameHeader::OpCode opcode = frame_header->opcode;
+  switch (frame_header->opcode) {
+    case WebSocketFrameHeader::kOpCodeText:  // fall-thru
+    case WebSocketFrameHeader::kOpCodeBinary:
+      if (!first_chunk) {
+        opcode = WebSocketFrameHeader::kOpCodeContinuation;
+      }
+    // fall-thru
+    case WebSocketFrameHeader::kOpCodeContinuation:
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Data packet received after close");
+        return;
+      } else if (state_ == CONNECTED) {
+        const bool final = chunk->final_chunk && frame_header->final;
+        // TODO(ricea): Can this copy be eliminated?
+        const char* const data_begin = data_buffer->data();
+        const char* const data_end = data_begin + data_buffer->size();
+        const std::vector<char> data(data_begin, data_end);
+        // TODO(ricea): Handle the (improbable) case when ReadFrames returns far
+        // more data at once than we want to send in a single IPC (in which case
+        // we need to buffer the data and return to the event loop with a
+        // callback to send the rest in 32K chunks).
+
+        // Send the received frame to the renderer process.
+        event_interface_->OnSendFrame(final, opcode, data);
+      }
+      break;
+
+    case WebSocketFrameHeader::kOpCodePing:
+      VLOG(1) << "Got Ping of size " << data_buffer->size();
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Ping received after Close");
+        return;
+      } else if (state_ == CONNECTED) {
+        SendIOBufferWithSize(
+            true, WebSocketFrameHeader::kOpCodePong, data_buffer);
+      }
+      break;
+
+    case WebSocketFrameHeader::kOpCodePong:
+      VLOG(1) << "Got Pong of size " << data_buffer->size();
+      if (state_ == RECV_CLOSED) {
+        FailChannel(SEND_REAL_ERROR,
+                    kWebSocketErrorProtocolError,
+                    "Pong received after Close");
+        return;
+      }
+      // We do not need to do anything with pong messages.
+      break;
+
+    case WebSocketFrameHeader::kOpCodeClose: {
+      unsigned short reason = kWebSocketNormalClosure;
+      std::string reason_text;
+      ParseClose(*data_buffer, &reason, &reason_text);
+      // TODO(ricea): Find a way to safely log the message from the close
+      // message (escape control codes and so on).
+      VLOG(1) << "Got Close with reason " << reason;
+      switch (state_) {
+        case CONNECTED:
+          state_ = RECV_CLOSED;
+          SendClose(reason, reason_text);
+          event_interface_->OnDropChannel(reason, reason_text);
+          break;
+
+        case RECV_CLOSED:
+          FailChannel(SEND_REAL_ERROR,
+                      kWebSocketErrorProtocolError,
+                      "Close received after Close");
+          break;
+
+        case SEND_CLOSED:
+          state_ = CLOSED;
+          event_interface_->OnDropChannel(reason, reason_text);
+          break;
+
+        default:
+          LOG(DFATAL) << "Got Close in unexpected state " << state_;
+          break;
+      }
+      break;
+    }
+
+    default:
+      FailChannel(SEND_REAL_ERROR,
+                  kWebSocketErrorProtocolError,
+                  "Unknown opcode");
+
+      break;
+  }
+  if (!chunk->final_chunk) {
+    // Preserve the frame header for the next call.
+    current_frame_header_.swap(frame_header);
+  }
+}
+
+void WebSocketChannel::SendFlowControl(int64 quota) {
+  DCHECK_EQ(CONNECTED, state_);
+  // TODO(ricea): Add interface to WebSocketStream and implement.
+  // stream_->SendFlowControl(quota);
+}
+
+void WebSocketChannel::SendDropChannel(unsigned short reason,
+                                       const std::string& reason_text) {
+  if (state_ == SEND_CLOSED || state_ == CLOSED) {
+    VLOG(1) << "SendDropChannel called in state " << state_
+            << ". This may be a bug, or a harmless race.";
+    return;
+  }
+  DCHECK_EQ(CONNECTED, state_);
+  // TODO(ricea): Validate reason? Check that reason_text is valid UTF-8?
+  // TODO(ricea): There should probably be a timeout for the closing handshake.
+  SendClose(reason, reason_text);
+}
+
+void WebSocketChannel::FailChannel(ExposeError expose,
+                                   unsigned short code,
+                                   const std::string& reason) {
+  // TODO(ricea): Logging.
+  State old_state = state_;
+  if (state_ == CONNECTED) {
+    unsigned short send_code = kWebSocketErrorGoingAway;
+    std::string send_reason = "Internal Error";
+    if (expose == SEND_REAL_ERROR) {
+      send_code = code;
+      send_reason = reason;
+    }
+    SendClose(send_code, send_reason);
+  }
+  if (old_state != RECV_CLOSED && old_state != CLOSED) {
+    event_interface_->OnDropChannel(code, reason);
+  }
+}
+
+void WebSocketChannel::SendClose(unsigned short code,
+                                 const std::string& reason) {
+  DCHECK(state_ == CONNECTED || state_ == RECV_CLOSED);
+  uint64 payload_length = kWebSocketCloseCodeLength + reason.length();
+  std::vector<char> data(payload_length);
+  DCHECK(payload_length > 0);
+  WriteBigEndian(&data[0], code);
+  COMPILE_ASSERT(sizeof(code) == kWebSocketCloseCodeLength,
+                 they_should_both_be_two);
+  std::copy(reason.begin(),
+            reason.end(),
+            data.begin() + kWebSocketCloseCodeLength);
+  Send(true, WebSocketFrameHeader::kOpCodeClose, data);
+  state_ = state_ == CONNECTED ? SEND_CLOSED : CLOSED;
+}
+
+void WebSocketChannel::ParseClose(const IOBufferWithSize& buffer,
+                                  unsigned short* reason,
+                                  std::string* reason_text) {
+  const char* data = const_cast<IOBufferWithSize&>(buffer).data();
+  CHECK_GE(buffer.size(), 0);  // Possibly security-critical invariant.
+  size_t size = static_cast<size_t>(buffer.size());
+  reason_text->clear();
+  if (size < kWebSocketCloseCodeLength) {
+    *reason = kWebSocketErrorNoStatusReceived;
+    if (size != 0) {
+      VLOG(1) << "Close frame with payload size " << size << " received "
+              << "(the first byte is " << std::hex << static_cast<int>(data[0])
+              << ")";
+      return;
+    }
+    return;
+  }
+  unsigned short unchecked_reason = 0;
+  ReadBigEndian(data, &unchecked_reason);
+  COMPILE_ASSERT(sizeof(unchecked_reason) == kWebSocketCloseCodeLength,
+                 they_should_both_be_two_bytes);
+  if (unchecked_reason >=
+          static_cast<unsigned short>(kWebSocketNormalClosure) &&
+      unchecked_reason <=
+          static_cast<unsigned short>(kWebSocketErrorPrivateReservedMax)) {
+    *reason = unchecked_reason;
+  } else {
+    VLOG(1) << "Close frame contained reason code outside of the valid range: "
+            << unchecked_reason;
+    *reason = kWebSocketErrorProtocolError;
+  }
+  std::string text(data + kWebSocketCloseCodeLength, data + size);
+  // TODO(ricea): Is this check strict enough? In particular, check the
+  // "Security Considerations" from RFC3629.
+  if (IsStringUTF8(text)) {
+    using std::swap;
+    swap(*reason_text, text);
+  }
+}
+
+}  // namespace net