Create dedicated class for handling wire message formatting and parsing.

extensions/browser/api/cast_channel/cast_socket.cc

 // Copyright 2014 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 "extensions/browser/api/cast_channel/cast_socket.h"
 
 #include <stdlib.h>
 #include <string.h>
 
 #include "base/bind.h"
 #include "base/callback_helpers.h"
 #include "base/format_macros.h"
 #include "base/lazy_instance.h"
 #include "base/numerics/safe_conversions.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/strings/stringprintf.h"
 #include "base/sys_byteorder.h"
 #include "extensions/browser/api/cast_channel/cast_auth_util.h"
 #include "extensions/browser/api/cast_channel/cast_channel.pb.h"
+#include "extensions/browser/api/cast_channel/cast_framer.h"
 #include "extensions/browser/api/cast_channel/cast_message_util.h"
 #include "extensions/browser/api/cast_channel/logger.h"
 #include "extensions/browser/api/cast_channel/logger_util.h"
 #include "net/base/address_list.h"
 #include "net/base/host_port_pair.h"
 #include "net/base/net_errors.h"
 #include "net/base/net_util.h"
 #include "net/cert/cert_verifier.h"
 #include "net/cert/x509_certificate.h"
 #include "net/http/transport_security_state.h"
@@ skipping 10 matching lines @@
 #define VLOG_WITH_CONNECTION(level) VLOG(level) << "[" << \
     ip_endpoint_.ToString() << ", auth=" << channel_auth_ << "] "
 
 namespace {
 
 // The default keepalive delay.  On Linux, keepalives probes will be sent after
 // the socket is idle for this length of time, and the socket will be closed
 // after 9 failed probes.  So the total idle time before close is 10 *
 // kTcpKeepAliveDelaySecs.
 const int kTcpKeepAliveDelaySecs = 10;
-
 }  // namespace
 
 namespace extensions {
 
 static base::LazyInstance<BrowserContextKeyedAPIFactory<
     ApiResourceManager<core_api::cast_channel::CastSocket> > > g_factory =
     LAZY_INSTANCE_INITIALIZER;
 
 // static
 template <>
@@ skipping 121 matching lines @@
                        ChannelAuthType channel_auth,
                        CastSocket::Delegate* delegate,
                        net::NetLog* net_log,
                        const base::TimeDelta& timeout,
                        const scoped_refptr<Logger>& logger)
     : ApiResource(owner_extension_id),
       channel_id_(0),
       ip_endpoint_(ip_endpoint),
       channel_auth_(channel_auth),
       delegate_(delegate),
-      current_message_size_(0),
-      current_message_(new CastMessage()),
       net_log_(net_log),
       logger_(logger),
       connect_timeout_(timeout),
       connect_timeout_timer_(new base::OneShotTimer<CastSocket>),
       is_canceled_(false),
       connect_state_(CONN_STATE_NONE),
       write_state_(WRITE_STATE_NONE),
       read_state_(READ_STATE_NONE),
       error_state_(CHANNEL_ERROR_NONE),
       ready_state_(READY_STATE_NONE) {
   DCHECK(net_log_);
   DCHECK(channel_auth_ == CHANNEL_AUTH_TYPE_SSL ||
          channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED);
   net_log_source_.type = net::NetLog::SOURCE_SOCKET;
   net_log_source_.id = net_log_->NextID();
 
-  // Reuse these buffers for each message.
-  header_read_buffer_ = new net::GrowableIOBuffer();
-  header_read_buffer_->SetCapacity(MessageHeader::header_size());
-  body_read_buffer_ = new net::GrowableIOBuffer();
-  body_read_buffer_->SetCapacity(MessageHeader::max_message_size());
-  current_read_buffer_ = header_read_buffer_;
+  // Buffer is reused across messages.
+  read_buffer_ = new net::GrowableIOBuffer();
+  read_buffer_->SetCapacity(MessageFramer::MessageHeader::max_message_size());
+  framer_.reset(new MessageFramer(read_buffer_));
 }
 
 CastSocket::~CastSocket() {
   // Ensure that resources are freed but do not run pending callbacks to avoid
   // any re-entrancy.
   CloseInternal();
 }
 
 ReadyState CastSocket::ready_state() const {
   return ready_state_;
@@ skipping 39 matching lines @@
       ip_endpoint_);
 
   return net::ClientSocketFactory::GetDefaultFactory()->CreateSSLClientSocket(
       connection.Pass(), host_and_port, ssl_config, context);
 }
 
 bool CastSocket::ExtractPeerCert(std::string* cert) {
   DCHECK(cert);
   DCHECK(peer_cert_.empty());
   net::SSLInfo ssl_info;
-  if (!socket_->GetSSLInfo(&ssl_info) || !ssl_info.cert.get())
+  if (!socket_->GetSSLInfo(&ssl_info) || !ssl_info.cert.get()) {
     return false;
+  }
 
   logger_->LogSocketEvent(channel_id_, proto::SSL_INFO_OBTAINED);
 
   bool result = net::X509Certificate::GetDEREncoded(
      ssl_info.cert->os_cert_handle(), cert);
-  if (result)
+  if (result) {
     VLOG_WITH_CONNECTION(1) << "Successfully extracted peer certificate: "
                             << *cert;
+  }
 
   logger_->LogSocketEventWithRv(
       channel_id_, proto::DER_ENCODED_CERT_OBTAIN, result ? 1 : 0);
   return result;
 }
 
 bool CastSocket::VerifyChallengeReply() {
   AuthResult result = AuthenticateChallengeReply(*challenge_reply_, peer_cert_);
   logger_->LogSocketChallengeReplyEvent(channel_id_, result);
   return result.success();
@@ skipping 186 matching lines @@
 void CastSocket::DoAuthChallengeSendWriteComplete(int result) {
   send_auth_challenge_callback_.Cancel();
   VLOG_WITH_CONNECTION(2) << "DoAuthChallengeSendWriteComplete: " << result;
   DCHECK_GT(result, 0);
   DCHECK_EQ(write_queue_.size(), 1UL);
   PostTaskToStartConnectLoop(result);
 }
 
 int CastSocket::DoAuthChallengeSendComplete(int result) {
   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
-  if (result < 0)
+  if (result < 0) {
     return result;
+  }
   SetConnectState(CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE);
 
   // Post a task to start read loop so that DoReadLoop is not nested inside
   // DoConnectLoop. This is not strictly necessary but keeps the read loop
   // code decoupled from connect loop code.
   PostTaskToStartReadLoop();
   // Always return IO_PENDING since the result is always asynchronous.
   return net::ERR_IO_PENDING;
 }
 
 int CastSocket::DoAuthChallengeReplyComplete(int result) {
   VLOG_WITH_CONNECTION(1) << "DoAuthChallengeReplyComplete: " << result;
-  if (result < 0)
+  if (result < 0) {
     return result;
-  if (!VerifyChallengeReply())
+  }
+  if (!VerifyChallengeReply()) {
     return net::ERR_FAILED;
+  }
   VLOG_WITH_CONNECTION(1) << "Auth challenge verification succeeded";
   return net::OK;
 }
 
 void CastSocket::DoConnectCallback(int result) {
   SetReadyState((result == net::OK) ? READY_STATE_OPEN : READY_STATE_CLOSED);
   if (result == net::OK) {
     SetErrorState(CHANNEL_ERROR_NONE);
     PostTaskToStartReadLoop();
     VLOG_WITH_CONNECTION(1) << "Calling Connect_Callback";
@@ skipping 140 matching lines @@
            write_state_ != WRITE_STATE_NONE);
 
   // No state change occurred in do-while loop above. This means state has
   // transitioned to NONE.
   if (write_state_ == WRITE_STATE_NONE) {
     logger_->LogSocketWriteState(channel_id_, WriteStateToProto(write_state_));
   }
 
   // If write loop is done because the queue is empty then set write
   // state to NONE
-  if (write_queue_.empty())
+  if (write_queue_.empty()) {
     SetWriteState(WRITE_STATE_NONE);
+  }
 
   // Write loop is done - if the result is ERR_FAILED then close with error.
-  if (rv == net::ERR_FAILED)
+  if (rv == net::ERR_FAILED) {
     CloseWithError();
+  }
 }
 
 int CastSocket::DoWrite() {
   DCHECK(!write_queue_.empty());
   WriteRequest& request = write_queue_.front();
 
   VLOG_WITH_CONNECTION(2) << "WriteData byte_count = "
                           << request.io_buffer->size() << " bytes_written "
                           << request.io_buffer->BytesConsumed();
 
@@ skipping 13 matching lines @@
   if (result <= 0) {  // NOTE that 0 also indicates an error
     SetErrorState(CHANNEL_ERROR_SOCKET_ERROR);
     SetWriteState(WRITE_STATE_ERROR);
     return result == 0 ? net::ERR_FAILED : result;
   }
 
   // Some bytes were successfully written
   WriteRequest& request = write_queue_.front();
   scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
   io_buffer->DidConsume(result);
-  if (io_buffer->BytesRemaining() == 0)  // Message fully sent
+  if (io_buffer->BytesRemaining() == 0) {  // Message fully sent
     SetWriteState(WRITE_STATE_DO_CALLBACK);
-  else
+  } else {
     SetWriteState(WRITE_STATE_WRITE);
+  }
 
   return net::OK;
 }
 
 int CastSocket::DoWriteCallback() {
   DCHECK(!write_queue_.empty());
 
   SetWriteState(WRITE_STATE_WRITE);
 
   WriteRequest& request = write_queue_.front();
@@ skipping 95 matching lines @@
     } else {
       // Connection is already established.  Close and send error status via the
       // OnError delegate.
       CloseWithError();
     }
   }
 }
 
 int CastSocket::DoRead() {
   SetReadState(READ_STATE_READ_COMPLETE);
-  // Figure out whether to read header or body, and the remaining bytes.
-  uint32 num_bytes_to_read = 0;
-  if (header_read_buffer_->RemainingCapacity() > 0) {
-    current_read_buffer_ = header_read_buffer_;
-    num_bytes_to_read = header_read_buffer_->RemainingCapacity();
-    CHECK_LE(num_bytes_to_read, MessageHeader::header_size());
-  } else {
-    DCHECK_GT(current_message_size_, 0U);
-    num_bytes_to_read = current_message_size_ - body_read_buffer_->offset();
-    current_read_buffer_ = body_read_buffer_;
-    CHECK_LE(num_bytes_to_read, MessageHeader::max_message_size());
-  }
-  CHECK_GT(num_bytes_to_read, 0U);
+
+  // Determine how many bytes need to be read.
+  size_t num_bytes_to_read = framer_->BytesRequested();
 
   // Read up to num_bytes_to_read into |current_read_buffer_|.
   int rv = socket_->Read(
-      current_read_buffer_.get(),
-      num_bytes_to_read,
+      read_buffer_.get(),
+      base::checked_cast<uint32>(num_bytes_to_read),
       base::Bind(&CastSocket::DoReadLoop, base::Unretained(this)));
   logger_->LogSocketEventWithRv(channel_id_, proto::SOCKET_READ, rv);
 
   return rv;
 }
 
 int CastSocket::DoReadComplete(int result) {
-  VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result
-                          << " header offset = "
-                          << header_read_buffer_->offset()
-                          << " body offset = " << body_read_buffer_->offset();
+  VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result;
+
   if (result <= 0) {  // 0 means EOF: the peer closed the socket
     VLOG_WITH_CONNECTION(1) << "Read error, peer closed the socket";
     SetErrorState(CHANNEL_ERROR_SOCKET_ERROR);
     SetReadState(READ_STATE_ERROR);
     return result == 0 ? net::ERR_FAILED : result;
   }
 
-  // Some data was read.  Move the offset in the current buffer forward.
-  CHECK_LE(current_read_buffer_->offset() + result,
-           current_read_buffer_->capacity());
-  current_read_buffer_->set_offset(current_read_buffer_->offset() + result);
-
-  if (current_read_buffer_.get() == header_read_buffer_.get() &&
-      current_read_buffer_->RemainingCapacity() == 0) {
-    // A full header is read, process the contents.
-    if (!ProcessHeader()) {
-      SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE);
-      SetReadState(READ_STATE_ERROR);
-    } else {
-      // Processed header, now read the body.
-      SetReadState(READ_STATE_READ);
-    }
-  } else if (current_read_buffer_.get() == body_read_buffer_.get() &&
-             static_cast<uint32>(current_read_buffer_->offset()) ==
-             current_message_size_) {
-    // Store a copy of current_message_size_ since it will be reset by
-    // ProcessBody().
-    uint32 message_size = current_message_size_;
-    // Full body is read, process the contents.
-    if (ProcessBody()) {
-      logger_->LogSocketEventForMessage(
-          channel_id_,
-          proto::MESSAGE_READ,
-          current_message_->namespace_(),
-          base::StringPrintf("Message size: %u", message_size));
-      SetReadState(READ_STATE_DO_CALLBACK);
-    } else {
-      SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE);
-      SetReadState(READ_STATE_ERROR);
-    }
+  size_t message_size;
+  DCHECK(current_message_.get() == NULL);
+  current_message_ = framer_->Ingest(result, &message_size, &error_state_);
+  if (current_message_.get()) {
+    DCHECK_EQ(error_state_, CHANNEL_ERROR_NONE);
+    DCHECK_GT(message_size, static_cast<size_t>(0));
+    logger_->LogSocketEventForMessage(
+        channel_id_,
+        proto::MESSAGE_READ,
+        current_message_->namespace_(),
+        base::StringPrintf("Message size: %u",
+                           static_cast<uint32>(message_size)));
+    SetReadState(READ_STATE_DO_CALLBACK);
+  } else if (error_state_ != CHANNEL_ERROR_NONE) {
+    DCHECK(current_message_.get() == NULL);
+    SetReadState(READ_STATE_ERROR);
   } else {
-    // Have not received full header or full body yet; keep reading.
+    DCHECK(current_message_.get() == NULL);
     SetReadState(READ_STATE_READ);
   }
-
   return net::OK;
 }
 
 int CastSocket::DoReadCallback() {
   SetReadState(READ_STATE_READ);
   const CastMessage& message = *current_message_;
   if (ready_state_ == READY_STATE_CONNECTING) {
     if (IsAuthMessage(message)) {
       challenge_reply_.reset(new CastMessage(message));
       logger_->LogSocketEvent(channel_id_, proto::RECEIVED_CHALLENGE_REPLY);
       PostTaskToStartConnectLoop(net::OK);
+      current_message_.reset();
       return net::OK;
     } else {
       SetReadState(READ_STATE_ERROR);
       SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE);
+      current_message_.reset();
       return net::ERR_INVALID_RESPONSE;
     }
   }
 
   MessageInfo message_info;
   if (!CastMessageToMessageInfo(message, &message_info)) {
-    current_message_->Clear();
+    current_message_.reset();
     SetReadState(READ_STATE_ERROR);
     SetErrorState(CHANNEL_ERROR_INVALID_MESSAGE);
     return net::ERR_INVALID_RESPONSE;
   }
 
   logger_->LogSocketEventForMessage(channel_id_,
                                     proto::NOTIFY_ON_MESSAGE,
                                     message.namespace_(),
                                     std::string());
   delegate_->OnMessage(this, message_info);
-  current_message_->Clear();
+  current_message_.reset();
 
   return net::OK;
 }
 
 int CastSocket::DoReadError(int result) {
   DCHECK_LE(result, 0);
   return net::ERR_FAILED;
 }
 
-bool CastSocket::ProcessHeader() {
-  CHECK_EQ(static_cast<uint32>(header_read_buffer_->offset()),
-           MessageHeader::header_size());
-  MessageHeader header;
-  MessageHeader::ReadFromIOBuffer(header_read_buffer_.get(), &header);
-  if (header.message_size > MessageHeader::max_message_size())
-    return false;
-
-  VLOG_WITH_CONNECTION(2) << "Parsed header { message_size: "
-                          << header.message_size << " }";
-  current_message_size_ = header.message_size;
-  return true;
-}
-
-bool CastSocket::ProcessBody() {
-  CHECK_EQ(static_cast<uint32>(body_read_buffer_->offset()),
-           current_message_size_);
-  if (!current_message_->ParseFromArray(
-      body_read_buffer_->StartOfBuffer(), current_message_size_)) {
-    return false;
-  }
-  current_message_size_ = 0;
-  header_read_buffer_->set_offset(0);
-  body_read_buffer_->set_offset(0);
-  current_read_buffer_ = header_read_buffer_;
-  return true;
-}
-
-// static
-bool CastSocket::Serialize(const CastMessage& message_proto,
-                           std::string* message_data) {
-  DCHECK(message_data);
-  message_proto.SerializeToString(message_data);
-  size_t message_size = message_data->size();
-  if (message_size > MessageHeader::max_message_size()) {
-    message_data->clear();
-    return false;
-  }
-  CastSocket::MessageHeader header;
-  header.SetMessageSize(message_size);
-  header.PrependToString(message_data);
-  return true;
-}
-
 void CastSocket::CloseWithError() {
   DCHECK(CalledOnValidThread());
   CloseInternal();
   RunPendingCallbacksOnClose();
   if (delegate_) {
     logger_->LogSocketEvent(channel_id_, proto::NOTIFY_ON_ERROR);
     delegate_->OnError(this, error_state_, logger_->GetLastErrors(channel_id_));
   }
 }
 
@@ skipping 39 matching lines @@
   }
 }
 
 void CastSocket::SetWriteState(WriteState write_state) {
   if (write_state_ != write_state) {
     write_state_ = write_state;
     logger_->LogSocketWriteState(channel_id_, WriteStateToProto(write_state_));
   }
 }
 
-CastSocket::MessageHeader::MessageHeader() : message_size(0) { }
-
-void CastSocket::MessageHeader::SetMessageSize(size_t size) {
-  DCHECK_LT(size, static_cast<size_t>(kuint32max));
-  DCHECK_GT(size, 0U);
-  message_size = size;
-}
-
-// TODO(mfoltz): Investigate replacing header serialization with base::Pickle,
-// if bit-for-bit compatible.
-void CastSocket::MessageHeader::PrependToString(std::string* str) {
-  MessageHeader output = *this;
-  output.message_size = base::HostToNet32(message_size);
-  size_t header_size = base::checked_cast<size_t, uint32>(
-      MessageHeader::header_size());
-  scoped_ptr<char, base::FreeDeleter> char_array(
-      static_cast<char*>(malloc(header_size)));
-  memcpy(char_array.get(), &output, header_size);
-  str->insert(0, char_array.get(), header_size);
-}
-
-// TODO(mfoltz): Investigate replacing header deserialization with base::Pickle,
-// if bit-for-bit compatible.
-void CastSocket::MessageHeader::ReadFromIOBuffer(
-    net::GrowableIOBuffer* buffer, MessageHeader* header) {
-  uint32 message_size;
-  size_t header_size = base::checked_cast<size_t, uint32>(
-      MessageHeader::header_size());
-  memcpy(&message_size, buffer->StartOfBuffer(), header_size);
-  header->message_size = base::NetToHost32(message_size);
-}
-
-std::string CastSocket::MessageHeader::ToString() {
-  return "{message_size: " + base::UintToString(message_size) + "}";
-}
-
 CastSocket::WriteRequest::WriteRequest(const net::CompletionCallback& callback)
   : callback(callback) { }
 
 bool CastSocket::WriteRequest::SetContent(const CastMessage& message_proto) {
   DCHECK(!io_buffer.get());
   std::string message_data;
-  if (!Serialize(message_proto, &message_data))
+  if (!MessageFramer::Serialize(message_proto, &message_data)) {
     return false;
+  }
   message_namespace = message_proto.namespace_();
   io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(message_data),
                                          message_data.size());
   return true;
 }
 
 CastSocket::WriteRequest::~WriteRequest() { }
-
 }  // namespace cast_channel
 }  // namespace core_api
 }  // namespace extensions
 
 #undef VLOG_WITH_CONNECTION