Refactor CastSocket code for the following:

chrome/browser/extensions/api/cast_channel/cast_socket.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 "chrome/browser/extensions/api/cast_channel/cast_socket.h"
 
 #include <string.h>
 
 #include "base/bind.h"
 #include "base/callback_helpers.h"
@@ skipping 56 matching lines @@
 
 CastSocket::CastSocket(const std::string& owner_extension_id,
                        const GURL& url,
                        CastSocket::Delegate* delegate,
                        net::NetLog* net_log) :
     ApiResource(owner_extension_id),
     channel_id_(0),
     url_(url),
     delegate_(delegate),
     auth_required_(false),
-    error_state_(CHANNEL_ERROR_NONE),
-    ready_state_(READY_STATE_NONE),
-    write_callback_pending_(false),
-    read_callback_pending_(false),
     current_message_size_(0),
     net_log_(net_log),
-    next_state_(CONN_STATE_NONE),
-    in_connect_loop_(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_);
   net_log_source_.type = net::NetLog::SOURCE_SOCKET;
   net_log_source_.id = net_log_->NextID();
 
   // We reuse these buffers for each message.
   header_read_buffer_ = new net::GrowableIOBuffer();
   header_read_buffer_->SetCapacity(kMessageHeaderSize);
   body_read_buffer_ = new net::GrowableIOBuffer();
   body_read_buffer_->SetCapacity(kMaxMessageSize);
   current_read_buffer_ = header_read_buffer_;
 }
 
 CastSocket::~CastSocket() { }
 
 const GURL& CastSocket::url() const {
   return url_;
 }
 
 scoped_ptr<net::TCPClientSocket> CastSocket::CreateTcpSocket() {
   net::AddressList addresses(ip_endpoint_);
-  scoped_ptr<net::TCPClientSocket> tcp_socket(
+  return scoped_ptr<net::TCPClientSocket>(
       new net::TCPClientSocket(addresses, net_log_, net_log_source_));
   // Options cannot be set on the TCPClientSocket yet, because the
   // underlying platform socket will not be created until we Bind()
   // or Connect() it.
-  return tcp_socket.Pass();
 }
 
-scoped_ptr<net::SSLClientSocket> CastSocket::CreateSslSocket() {
+scoped_ptr<net::SSLClientSocket> CastSocket::CreateSslSocket(
+    scoped_ptr<net::StreamSocket> socket) {
   net::SSLConfig ssl_config;
   // If a peer cert was extracted in a previous attempt to connect, then
   // whitelist that cert.
   if (!peer_cert_.empty()) {
     net::SSLConfig::CertAndStatus cert_and_status;
     cert_and_status.cert_status = net::CERT_STATUS_AUTHORITY_INVALID;
     cert_and_status.der_cert = peer_cert_;
     ssl_config.allowed_bad_certs.push_back(cert_and_status);
   }
 
   cert_verifier_.reset(net::CertVerifier::CreateDefault());
   transport_security_state_.reset(new net::TransportSecurityState);
   net::SSLClientSocketContext context;
   // CertVerifier and TransportSecurityState are owned by us, not the
   // context object.
   context.cert_verifier = cert_verifier_.get();
   context.transport_security_state = transport_security_state_.get();
 
   scoped_ptr<net::ClientSocketHandle> connection(new net::ClientSocketHandle);
-  connection->SetSocket(tcp_socket_.PassAs<net::StreamSocket>());
+  connection->SetSocket(socket.Pass());
   net::HostPortPair host_and_port = net::HostPortPair::FromIPEndPoint(
       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())
     return false;
   bool result = net::X509Certificate::GetDEREncoded(
      ssl_info.cert->os_cert_handle(), cert);
   if (result)
     VLOG(1) << "Successfully extracted peer certificate: " << *cert;
   return result;
 }
 
-int CastSocket::SendAuthChallenge() {
-  CastMessage challenge_message;
-  CreateAuthChallengeMessage(&challenge_message);
-  VLOG(1) << "Sending challenge: " << CastMessageToString(challenge_message);
-  int result = SendMessageInternal(
-      challenge_message,
-      base::Bind(&CastSocket::OnChallengeEvent, AsWeakPtr()));
-  return (result < 0) ? result : net::OK;
-}
-
-int CastSocket::ReadAuthChallengeReply() {
-  int result = ReadData();
-  return (result < 0) ? result : net::OK;
-}
-
-void CastSocket::OnConnectComplete(int result) {
-  int rv = DoConnectLoop(result);
-  if (rv != net::ERR_IO_PENDING)
-    DoConnectCallback(rv);
-}
-
-void CastSocket::OnChallengeEvent(int result) {
-  // result >= 0 means read or write succeeded synchronously.
-  int rv = DoConnectLoop(result >= 0 ? net::OK : result);
-  if (rv != net::ERR_IO_PENDING)
-    DoConnectCallback(rv);
+bool CastSocket::VerifyChallengeReply() {
+  return AuthenticateChallengeReply(*challenge_reply_.get(), peer_cert_);
 }
 
 void CastSocket::Connect(const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
-  int result = net::ERR_CONNECTION_FAILED;
   VLOG(1) << "Connect readyState = " << ready_state_;
   if (ready_state_ != READY_STATE_NONE) {
-    callback.Run(result);
+    callback.Run(net::ERR_CONNECTION_FAILED);
     return;
   }
   if (!ParseChannelUrl(url_)) {
-    CloseWithError(cast_channel::CHANNEL_ERROR_CONNECT_ERROR);
-    callback.Run(result);
+    callback.Run(net::ERR_CONNECTION_FAILED);
     return;
   }
+
   connect_callback_ = callback;
-  next_state_ = CONN_STATE_TCP_CONNECT;
-  int rv = DoConnectLoop(net::OK);
-  if (rv != net::ERR_IO_PENDING)
-    DoConnectCallback(rv);
+  connect_state_ = CONN_STATE_TCP_CONNECT;
+  DoConnectLoop(net::OK);
 }
 
 // This method performs the state machine transitions for connection flow.
 // There are two entry points to this method:
-// 1. public Connect method: this starts the flow
-// 2. OnConnectComplete: callback method called when an async operation
-//    is done. OnConnectComplete calls this method to continue the state
-//    machine transitions.
-int CastSocket::DoConnectLoop(int result) {
-  // Avoid re-entrancy as a result of synchronous completion.
-  if (in_connect_loop_)
-    return net::ERR_IO_PENDING;
-  in_connect_loop_ = true;
-
+// 1. Connect method: this starts the flow
+// 2. Callback from network operations that finish asynchronously
+void CastSocket::DoConnectLoop(int result) {
   // Network operations can either finish synchronously or asynchronously.
   // This method executes the state machine transitions in a loop so that
   // correct state transitions happen even when network operations finish
   // synchronously.
   int rv = result;
   do {
-    ConnectionState state = next_state_;
-    // All the Do* methods do not set next_state_ in case of an
-    // error. So set next_state_ to NONE to figure out if the Do*
+    ConnectionState state = connect_state_;
+    // All the Do* methods do not set connect_state_ in case of an
+    // error. So set connect_state_ to NONE to figure out if the Do*
     // method changed state or not.
-    next_state_ = CONN_STATE_NONE;
+    connect_state_ = CONN_STATE_NONE;
     switch (state) {
       case CONN_STATE_TCP_CONNECT:
         rv = DoTcpConnect();
         break;
       case CONN_STATE_TCP_CONNECT_COMPLETE:
         rv = DoTcpConnectComplete(rv);
         break;
       case CONN_STATE_SSL_CONNECT:
         DCHECK_EQ(net::OK, rv);
         rv = DoSslConnect();
         break;
       case CONN_STATE_SSL_CONNECT_COMPLETE:
         rv = DoSslConnectComplete(rv);
         break;
       case CONN_STATE_AUTH_CHALLENGE_SEND:
         rv = DoAuthChallengeSend();
         break;
       case CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE:
         rv = DoAuthChallengeSendComplete(rv);
         break;
       case CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE:
         rv = DoAuthChallengeReplyComplete(rv);
         break;
-
       default:
-        NOTREACHED() << "BUG in CastSocket state machine code";
+        NOTREACHED() << "BUG in CastSocket connection state machine code";
         break;
     }
-  } while (rv != net::ERR_IO_PENDING && next_state_ != CONN_STATE_NONE);
+  } while (rv != net::ERR_IO_PENDING && connect_state_ != CONN_STATE_NONE);
   // Get out of the loop either when:
   // a. A network operation is pending, OR
   // b. The Do* method called did not change state
 
-  in_connect_loop_ = false;
-
-  return rv;
+  // If there is no pending IO and if we still got out of the loop then
+  // either we are done successfully or there was an error; invoke the
+  // callback in either case.
+  if (rv != net::ERR_IO_PENDING)
+    DoConnectCallback(rv);
 }
 
 int CastSocket::DoTcpConnect() {
   VLOG(1) << "DoTcpConnect";
-  next_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
+  connect_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
   tcp_socket_ = CreateTcpSocket();
   return tcp_socket_->Connect(
-      base::Bind(&CastSocket::OnConnectComplete, AsWeakPtr()));
+      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
 }
 
 int CastSocket::DoTcpConnectComplete(int result) {
   VLOG(1) << "DoTcpConnectComplete: " << result;
   if (result == net::OK) {
     // Enable TCP protocol-level keep-alive.
-    bool result = tcp_socket_->SetKeepAlive(true, kTcpKeepAliveDelaySecs);
-    LOG_IF(WARNING, !result) << "Failed to SetKeepAlive.";
-    next_state_ = CONN_STATE_SSL_CONNECT;
+    //// bool success = tcp_socket_->SetKeepAlive(true, kTcpKeepAliveDelaySecs);

[mark a. foltz, 2013/12/05 23:42:49]

Should these lines be commented out?

[Munjal (Google), 2013/12/06 19:20:21]

Done.

+    //// LOG_IF(WARNING, !success) << "Failed to SetKeepAlive.";
+    connect_state_ = CONN_STATE_SSL_CONNECT;
   }
   return result;
 }
 
 int CastSocket::DoSslConnect() {
   VLOG(1) << "DoSslConnect";
-  next_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
-  socket_ = CreateSslSocket();
+  connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
+  socket_ = CreateSslSocket(tcp_socket_.PassAs<net::StreamSocket>());
   return socket_->Connect(
-      base::Bind(&CastSocket::OnConnectComplete, AsWeakPtr()));
+      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr()));
 }
 
 int CastSocket::DoSslConnectComplete(int result) {
   VLOG(1) << "DoSslConnectComplete: " << result;
   if (result == net::ERR_CERT_AUTHORITY_INVALID &&
              peer_cert_.empty() &&
              ExtractPeerCert(&peer_cert_)) {
-    next_state_ = CONN_STATE_TCP_CONNECT;
+    connect_state_ = CONN_STATE_TCP_CONNECT;
   } else if (result == net::OK && auth_required_) {
-    next_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
+    connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
   }
   return result;
 }
 
 int CastSocket::DoAuthChallengeSend() {
   VLOG(1) << "DoAuthChallengeSend";
-  next_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
-  return SendAuthChallenge();
+  connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
+  CastMessage challenge_message;
+  CreateAuthChallengeMessage(&challenge_message);
+  VLOG(1) << "Sending challenge: " << CastMessageToString(challenge_message);
+  // Post a task to start write loop so that DoWriteLoop is not nested inside
+  // DoConnectLoop. This is not strictly necessary but keeps the write loop
+  // code decoupled from connect loop code.
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::SendCastMessageInternal, AsWeakPtr(),
+                 challenge_message,
+                 base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr())));
+  // Always return IO_PENDING since we always get the result asynchronously
+  return net::ERR_IO_PENDING;
 }
 
 int CastSocket::DoAuthChallengeSendComplete(int result) {
   VLOG(1) << "DoAuthChallengeSendComplete: " << result;
-  if (result != net::OK)
+  if (result < 0)
     return result;
-  next_state_ = CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE;
-  return ReadAuthChallengeReply();
+  connect_state_ = 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 we always get the result asynchronously
+  return net::ERR_IO_PENDING;
 }
 
 int CastSocket::DoAuthChallengeReplyComplete(int result) {
   VLOG(1) << "DoAuthChallengeReplyComplete: " << result;
-  if (result != net::OK)
+  if (result < 0)
     return result;
   if (!VerifyChallengeReply())
     return net::ERR_FAILED;
   VLOG(1) << "Auth challenge verification succeeded";
   return net::OK;
 }
 
-bool CastSocket::VerifyChallengeReply() {
-  return AuthenticateChallengeReply(*challenge_reply_.get(), peer_cert_);
-}
-
 void CastSocket::DoConnectCallback(int result) {
   ready_state_ = (result == net::OK) ? READY_STATE_OPEN : READY_STATE_CLOSED;
   error_state_ = (result == net::OK) ?
       CHANNEL_ERROR_NONE : CHANNEL_ERROR_CONNECT_ERROR;
+  if (result == net::OK)  // Start the read loop
+    PostTaskToStartReadLoop();
   base::ResetAndReturn(&connect_callback_).Run(result);
-  // Start the ReadData loop if not already started.
-  // If auth_required_ is true we would've started a ReadData loop already.
-  // TODO(munjal): This is a bit ugly. Refactor read and write code.
-  if (result == net::OK && !auth_required_)
-    ReadData();
 }
 
 void CastSocket::Close(const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
   VLOG(1) << "Close ReadyState = " << ready_state_;
   tcp_socket_.reset(NULL);
   socket_.reset(NULL);
   cert_verifier_.reset(NULL);
   transport_security_state_.reset(NULL);
   ready_state_ = READY_STATE_CLOSED;
   callback.Run(net::OK);
 }
 
 void CastSocket::SendMessage(const MessageInfo& message,
                              const net::CompletionCallback& callback) {
   DCHECK(CalledOnValidThread());
-  VLOG(1) << "Send ReadyState " << ready_state_;
-  int result = net::ERR_FAILED;
   if (ready_state_ != READY_STATE_OPEN) {
-    callback.Run(result);
+    callback.Run(net::ERR_FAILED);
     return;
   }
   CastMessage message_proto;
   if (!MessageInfoToCastMessage(message, &message_proto)) {
-    CloseWithError(cast_channel::CHANNEL_ERROR_INVALID_MESSAGE);
-    // TODO(mfoltz): Do a better job of signaling cast_channel errors to the
-    // caller.
-    callback.Run(net::OK);
+    callback.Run(net::ERR_FAILED);
     return;
   }
-  SendMessageInternal(message_proto, callback);
+
+  SendCastMessageInternal(message_proto, callback);
 }
 
-int CastSocket::SendMessageInternal(const CastMessage& message_proto,
-                                    const net::CompletionCallback& callback) {
+void CastSocket::SendCastMessageInternal(
+    const CastMessage& message,
+    const net::CompletionCallback& callback) {
   WriteRequest write_request(callback);
-  if (!write_request.SetContent(message_proto))
-    return net::ERR_FAILED;
+  if (!write_request.SetContent(message)) {
+    callback.Run(net::ERR_FAILED);
+    return;
+  }
+
   write_queue_.push(write_request);
-  return WriteData();
+  if (write_state_ == WRITE_STATE_NONE) {
+    write_state_ = WRITE_STATE_WRITE;
+    DoWriteLoop(net::OK);
+  }
 }
 
-int CastSocket::WriteData() {
+void CastSocket::DoWriteLoop(int result) {
   DCHECK(CalledOnValidThread());
   VLOG(1) << "WriteData q = " << write_queue_.size();
-  if (write_queue_.empty() || write_callback_pending_)
-    return net::ERR_FAILED;
 
+  if (write_queue_.empty())
+    return;
+
+  // Network operations can either finish synchronously or asynchronously.
+  // This method executes the state machine transitions in a loop so that
+  // write state transitions happen even when network operations finish
+  // synchronously.
+  int rv = result;
+  do {
+    WriteState state = write_state_;
+    write_state_ = WRITE_STATE_NONE;
+    switch (state) {
+      case WRITE_STATE_WRITE:
+        rv = DoWrite();
+        break;
+      case WRITE_STATE_WRITE_COMPLETE:
+        rv = DoWriteComplete(rv);
+        break;
+      case WRITE_STATE_DO_CALLBACK:
+        rv = DoWriteCallback();
+        break;
+      case WRITE_STATE_ERROR:
+        rv = DoWriteError(rv);
+        break;
+      default:
+        NOTREACHED() << "BUG in CastSocket write state machine code";
+        break;
+    }
+  } while (!write_queue_.empty() &&
+           rv != net::ERR_IO_PENDING &&
+           write_state_ != WRITE_STATE_NONE);
+
+  // If we came out of the loop and the result is ERR_FAILED then close
+  // with error.
+  if (rv == net::ERR_FAILED)
+    CloseWithError(error_state_);
+}
+
+int CastSocket::DoWrite() {
+  DCHECK(!write_queue_.empty());
   WriteRequest& request = write_queue_.front();
 
   VLOG(1) << "WriteData byte_count = " << request.io_buffer->size()
-          << " bytes_written " << request.io_buffer->BytesConsumed();
+           << " bytes_written " << request.io_buffer->BytesConsumed();
 
-  write_callback_pending_ = true;
-  int result = socket_->Write(
+  write_state_ = WRITE_STATE_WRITE_COMPLETE;
+
+  return socket_->Write(
       request.io_buffer.get(),
       request.io_buffer->BytesRemaining(),
-      base::Bind(&CastSocket::OnWriteData, AsWeakPtr()));
-
-  if (result != net::ERR_IO_PENDING)
-    OnWriteData(result);
-
-  return result;
+      base::Bind(&CastSocket::DoWriteLoop, AsWeakPtr()));
 }
 
-void CastSocket::OnWriteData(int result) {
-  DCHECK(CalledOnValidThread());
-  VLOG(1) << "OnWriteComplete result = " << result;
-  DCHECK(write_callback_pending_);
+int CastSocket::DoWriteComplete(int result) {
   DCHECK(!write_queue_.empty());
-  write_callback_pending_ = false;
+  if (result <= 0) {  // NOTE that 0 also indicates an error
+    error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
+    write_state_ = WRITE_STATE_ERROR;
+    return 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
+    write_state_ = WRITE_STATE_DO_CALLBACK;
+  else
+    write_state_ = WRITE_STATE_WRITE;
 
-  if (result >= 0) {
-    io_buffer->DidConsume(result);
-    if (io_buffer->BytesRemaining() > 0) {
-      VLOG(1) << "OnWriteComplete size = " << io_buffer->size()
-              << " consumed " << io_buffer->BytesConsumed()
-              << " remaining " << io_buffer->BytesRemaining()
-              << " # requests " << write_queue_.size();
-      WriteData();
-      return;
-    }
-    DCHECK_EQ(io_buffer->BytesConsumed(), io_buffer->size());
-    DCHECK_EQ(io_buffer->BytesRemaining(), 0);
-    result = io_buffer->BytesConsumed();
+  return net::OK;
+}
+
+int CastSocket::DoWriteCallback() {
+  WriteRequest& request = write_queue_.front();
+  request.callback.Run(request.io_buffer->BytesConsumed());
+  write_queue_.pop();
+  write_state_ = WRITE_STATE_WRITE;
+  return net::OK;
+}
+
+int CastSocket::DoWriteError(int result) {
+  DCHECK(!write_queue_.empty());
+  DCHECK_LT(result, 0);
+
+  // Invoke completion callback for all pending writes
+  while (!write_queue_.empty()) {
+    WriteRequest& request = write_queue_.front();
+    write_queue_.pop();
+    request.callback.Run(result);
   }
 
-  request.callback.Run(result);
-  write_queue_.pop();
-
-  VLOG(1) << "OnWriteComplete size = " << io_buffer->size()
-          << " consumed " << io_buffer->BytesConsumed()
-          << " remaining " << io_buffer->BytesRemaining()
-          << " # requests " << write_queue_.size();
-
-  if (result < 0) {
-    CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
-    return;
-  }
-
-  if (!write_queue_.empty())
-    WriteData();
+  return net::ERR_FAILED;
 }
 
-int CastSocket::ReadData() {
+void CastSocket::PostTaskToStartReadLoop() {
   DCHECK(CalledOnValidThread());
-  if (!socket_.get())
-    return net::ERR_FAILED;
-  DCHECK(!read_callback_pending_);
-  read_callback_pending_ = true;
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
+}
+
+void CastSocket::StartReadLoop() {
+  // If we are in READ_STATE_NONE then get into appropriate
+  // starting state and start the read loop
+  if (read_state_ == READ_STATE_NONE) {
+    read_state_ = READ_STATE_READ;
+    DoReadLoop(net::OK);
+  }
+}
+
+void CastSocket::DoReadLoop(int result) {
+  DCHECK(CalledOnValidThread());
+  // Network operations can either finish synchronously or asynchronously.
+  // This method executes the state machine transitions in a loop so that
+  // write state transitions happen even when network operations finish
+  // synchronously.
+  int rv = result;
+  do {
+    ReadState state = read_state_;
+    read_state_ = READ_STATE_NONE;
+
+    switch (state) {
+      case READ_STATE_READ:
+        rv = DoRead();
+        break;
+      case READ_STATE_READ_COMPLETE:
+        rv = DoReadComplete(rv);
+        break;
+      case READ_STATE_DO_CALLBACK:
+        rv = DoReadCallback();
+        break;
+      case READ_STATE_ERROR:
+        rv = DoReadError(rv);
+        break;
+      default:
+        NOTREACHED() << "BUG in read state machine";
+        break;
+    }
+  } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE);
+
+  // If we came out of the loop and the result is ERR_FAILED then close
+  // with error.
+  if (rv == net::ERR_FAILED)
+    CloseWithError(error_state_);
+}
+
+int CastSocket::DoRead() {
+  read_state_ = READ_STATE_READ_COMPLETE;
   // Figure out if we are reading the 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();
     DCHECK_LE(num_bytes_to_read, kMessageHeaderSize);
   } else {
     DCHECK_GT(current_message_size_, 0U);
     num_bytes_to_read = current_message_size_ - body_read_buffer_->offset();
     current_read_buffer_ = body_read_buffer_;
     DCHECK_LE(num_bytes_to_read, kMaxMessageSize);
   }
   DCHECK_GT(num_bytes_to_read, 0U);
+
   // We read up to num_bytes_to_read into |current_read_buffer_|.
-  int result = socket_->Read(
+  return socket_->Read(
       current_read_buffer_.get(),
       num_bytes_to_read,
-      base::Bind(&CastSocket::OnReadData, AsWeakPtr()));
-  VLOG(1) << "ReadData result = " << result;
-  if (result > 0) {
-    OnReadData(result);
-  } else if (result != net::ERR_IO_PENDING) {
-    CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
-  }
-  return result;
+      base::Bind(&CastSocket::DoReadLoop, AsWeakPtr()));
 }
 
-void CastSocket::OnReadData(int result) {
+int CastSocket::DoReadComplete(int result) {
   DCHECK(CalledOnValidThread());
-  VLOG(1) << "OnReadData result = " << result
+  VLOG(1) << "DoReadDataComplete result = " << result
           << " header offset = " << header_read_buffer_->offset()
           << " body offset = " << body_read_buffer_->offset();
-  read_callback_pending_ = false;
-  if (result <= 0) {
-    CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
-    return;
+  if (result <= 0) {  // 0 means EOF: the peer closed the socket
+    error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
+    read_state_ = READ_STATE_ERROR;
+    return result;
   }
+
   // We read some data.  Move the offset in the current buffer forward.
   DCHECK_LE(current_read_buffer_->offset() + result,
             current_read_buffer_->capacity());
   current_read_buffer_->set_offset(current_read_buffer_->offset() + result);
+  read_state_ = READ_STATE_READ;
 
-  bool should_continue = true;
   if (current_read_buffer_.get() == header_read_buffer_.get() &&
       current_read_buffer_->RemainingCapacity() == 0) {
-  // If we have read a full header, process the contents.
-    should_continue = ProcessHeader();
+    // If we have read a full header, process the contents.
+    if (!ProcessHeader()) {
+      error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
+      read_state_ = READ_STATE_ERROR;
+    }
   } else if (current_read_buffer_.get() == body_read_buffer_.get() &&
              static_cast<uint32>(current_read_buffer_->offset()) ==
              current_message_size_) {
     // If we have read a full body, process the contents.
-    should_continue = ProcessBody();
+    if (ProcessBody()) {
+      read_state_ = READ_STATE_DO_CALLBACK;
+    } else {
+      error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
+      read_state_ = READ_STATE_ERROR;
+    }
   }
-  if (should_continue)
-    ReadData();
+
+  return net::OK;
+}
+
+int CastSocket::DoReadCallback() {
+  read_state_ = READ_STATE_READ;
+  if (IsAuthMessage(current_message_)) {
+    // If the message is an auth message then we handle it internally.

[mark a. foltz, 2013/12/05 23:42:49]

I feel that we should check connect_state_ here to ensure we are actually inside
the connect loop.  If the other side sends an auth reply message after we've
finished the connect loop (for whatever reason) we'll be in a weird flow where
we attempt to revalidate the message and fire the connect callback again.

If we're not in the connect loop I would CloseWithError() with a socket error.

[Munjal (Google), 2013/12/06 19:20:21]

Done. Good call.

+    challenge_reply_.reset(new CastMessage(current_message_));
+    DoConnectLoop(net::OK);
+  } else if (delegate_) {
+    MessageInfo message;
+    if (CastMessageToMessageInfo(current_message_, &message))
+      delegate_->OnMessage(this, message);
+    else
+      read_state_ = READ_STATE_ERROR;
+  }
+  current_message_.Clear();
+  return net::OK;
+}
+
+int CastSocket::DoReadError(int result) {
+  // This method is not strictly necessary but provides consistency with
+  // the write flow
+  DCHECK_LT(result, 0);
+  return net::ERR_FAILED;
 }
 
 bool CastSocket::ProcessHeader() {
   DCHECK_EQ(static_cast<uint32>(header_read_buffer_->offset()),
             kMessageHeaderSize);
   MessageHeader header;
   MessageHeader::ReadFromIOBuffer(header_read_buffer_.get(), &header);
-  if (header.message_size > kMaxMessageSize) {
-    CloseWithError(cast_channel::CHANNEL_ERROR_INVALID_MESSAGE);
+  if (header.message_size > kMaxMessageSize)
     return false;
-  }
+
   VLOG(1) << "Parsed header { message_size: " << header.message_size << " }";
   current_message_size_ = header.message_size;
   return true;
 }
 
 bool CastSocket::ProcessBody() {
   DCHECK_EQ(static_cast<uint32>(body_read_buffer_->offset()),
             current_message_size_);
-  if (!ParseMessageFromBody()) {
-    CloseWithError(cast_channel::CHANNEL_ERROR_INVALID_MESSAGE);
+  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;
 }
 
-bool CastSocket::ParseMessageFromBody() {
-  DCHECK(CalledOnValidThread());
-  DCHECK_EQ(static_cast<uint32>(body_read_buffer_->offset()),
-            current_message_size_);
-  CastMessage message_proto;
-  if (!message_proto.ParseFromArray(
-      body_read_buffer_->StartOfBuffer(),
-      current_message_size_))
-    return false;
-  VLOG(1) << "Parsed message " << CastMessageToString(message_proto);
-  // If the message is an auth message then we handle it internally.
-  if (IsAuthMessage(message_proto)) {
-    challenge_reply_.reset(new CastMessage(message_proto));
-    OnChallengeEvent(net::OK);
-  } else if (delegate_) {
-    MessageInfo message;
-    if (!CastMessageToMessageInfo(message_proto, &message))
-      return false;
-    delegate_->OnMessage(this, message);
-  }
-  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 > kMaxMessageSize) {
     message_data->clear();
     return false;
   }
@@ skipping 100 matching lines @@
   io_buffer = new net::DrainableIOBuffer(new net::StringIOBuffer(message_data),
                                          message_data.size());
   return true;
 }
 
 CastSocket::WriteRequest::~WriteRequest() { }
 
 }  // namespace cast_channel
 }  // namespace api
 }  // namespace extensions