Refactor CastSocket code for the following:

chrome/browser/extensions/api/cast_channel/cast_socket.cc

Index: chrome/browser/extensions/api/cast_channel/cast_socket.cc
===================================================================
--- chrome/browser/extensions/api/cast_channel/cast_socket.cc	(revision 236353)
+++ chrome/browser/extensions/api/cast_channel/cast_socket.cc	(working copy)
@@ -74,14 +74,13 @@
     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();
@@ -102,15 +101,15 @@
 
 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.
@@ -130,7 +129,7 @@
   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_);
 
@@ -151,77 +150,51 @@
   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;
+bool CastSocket::VerifyChallengeReply() {
+  return AuthenticateChallengeReply(*challenge_reply_.get(), peer_cert_);
 }
 
-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);
-}
-
 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;
   }
+
+  ready_state_ = READY_STATE_CONNECTING;
   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);

[Ryan Sleevi, 2013/12/11 08:14:14]

DESIGN nit: the net/ idiom is as follows

This becomes

ready_state_ = READY_STATE_CONNECTING;
connect_state_ = CONN_STATE_TCP_CONNECT;
int rv = DoConnectLoop(net::OK);
if (rv == net::ERR_IO_PENDING) {
  connect_callback_ = callback;
  return;
}

// Note: |callback| may delete |this|
callback.Run(rv);

[Munjal (Google), 2013/12/11 23:39:41]

Yes. I changed it todo return void and do thecallback as part of the loop so
that multiple places don't have to remember to do the above.

This is also useful since connect loop can start write loop or read loop and can
expect a callback into it without the code calling it back doing the aboe
(because DoConnectLoop is now void).

 }
 
+void CastSocket::PostTaskToStartConnectLoop(int result) {

[Ryan Sleevi, 2013/12/11 08:14:14]

DESIGN NIT: Rather than have DoConnectLoop invoke the callback, this is
traditionally done by something as follows

void CastSocket::RestartConnectLoop(int result) {
  DCHECK(CalledOnValidThread());
  base::MessageLoop::current()->PostTask(FROM_HERE,
base::Bind(&CastSocket::OnConnectIOComplete, ...)))
}

void CastSocket::OnConnectIOComplete(int result) {
  int rv = DoConnectLoop(result);
  if (rv == net::ERR_IO_PENDING)
    return;

  // Note: |connect_callback_| may cause |this| to be deleted.
  base::ResetAndReturn(&connect_callback_).Run(result);
};


The goal with this paradigm is to make it absolutely clear where the exit points
are in terms of ownership (that is, when |callback| will be run, and when it may
delete |this|). The current implementation throws that ownership model
throughout the code.

[Munjal (Google), 2013/12/11 23:39:41]

The current code only has one place where the callback is performed:
DoConnectCallback. It also has only one point of entry - DoConnectLoop.

[Ryan Sleevi, 2013/12/12 00:48:14]

I apologize if my comment before was not clearer.

DoConnectLoop is not the only point of entry.

Any public API that causes DoConnectLoop to be called is the point of entry. My
goal in the above suggestion was to make sure that the callback invocation was
not buried in the details of the implementation, but as close to possible of the
actual public API entry points, or the start of a task.

I still think the way you've altered the code is dangerous and error-prone, even
if (probably) correct.

[Munjal (Google), 2013/12/12 17:53:47]

There is only one public method that starts the connect loop - Connect.

I understand your point - in the case where the caller of DoConnectLoop calls
the callback, DoConnectLoop won't be on teh stack when callback is called. But
the downside of that approach is that there are multiple places which have to
duplicate the code snippet you pasted above. If we create a helper to do that
code snippet, then we have the same issue of
code-calling-callback-buried-deeper. So I slightly prefer the approach of doing
the callback in the connect flow itself.

+  DCHECK(CalledOnValidThread());
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::DoConnectLoop, AsWeakPtr(), result));
+}
+
 // 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.

[Ryan Sleevi, 2013/12/11 08:14:14]

comment nit: This makes the comment describe implementation details, rather than
goals.

// Default to CONN_STATE_NONE, which break the processing loop if any
// handler fails to transition to another state to continue processing.

[Munjal (Google), 2013/12/11 23:39:41]

Done.

-    next_state_ = CONN_STATE_NONE;
+    connect_state_ = CONN_STATE_NONE;
     switch (state) {
       case CONN_STATE_TCP_CONNECT:
         rv = DoTcpConnect();
@@ -245,27 +218,28 @@
       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.

[Ryan Sleevi, 2013/12/11 08:14:14]

comment nit: pronouns are considered harmful -
https://groups.google.com/a/chromium.org/d/msg/chromium-dev/NH-S6KCkr2M/yv0z3...

That said, this all seems like it shouldn't be necessary.

[Munjal (Google), 2013/12/11 23:39:41]

Done.

+  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) {
@@ -274,17 +248,17 @@
     // 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;
+    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) {
@@ -292,30 +266,47 @@
   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.

[Ryan Sleevi, 2013/12/11 08:14:14]

This is really unfortunate that you're forcing always-async. this has a very
real cost on Android (~5ms in worst case).

Having the code clearly structured on ownership lets you have a smooth
synchronous implementation. I still need to page all of this code into make a
solid recommendation, but given that we've accomplished it for SPDY and QUIC
(for example), I feel like we should be able to do the same here.

[Munjal (Google), 2013/12/11 23:39:41]

See my overall comment on the patch that addresses the various comments
regarding design.

+  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

[Ryan Sleevi, 2013/12/11 08:14:14]

comment nits re: pronouns apply throughout this file. See the thread for easy
and simple ways to reword comments without pronouns to be clearer and less
ambiguous.

[Munjal (Google), 2013/12/11 23:39:41]

Done.

+  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;
@@ -323,20 +314,13 @@
   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) {
@@ -353,102 +337,206 @@
 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_);

[Ryan Sleevi, 2013/12/11 08:14:14]

|this| can be deleted, can't it? Subtle.

+}
+
+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()));
+      base::Bind(&CastSocket::DoWriteLoop, AsWeakPtr()));
+}
 
-  if (result != net::ERR_IO_PENDING)
-    OnWriteData(result);
+int CastSocket::DoWriteComplete(int result) {
+  DCHECK(!write_queue_.empty());
+  if (result <= 0) {  // NOTE that 0 also indicates an error
+    error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
+    write_state_ = WRITE_STATE_ERROR;
+    return result == 0 ? net::ERR_FAILED : result;
+  }
 
-  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;
+
+  return net::OK;
 }
 
-void CastSocket::OnWriteData(int result) {
-  DCHECK(CalledOnValidThread());
-  VLOG(1) << "OnWriteComplete result = " << result;
-  DCHECK(write_callback_pending_);
+int CastSocket::DoWriteCallback() {
   DCHECK(!write_queue_.empty());
-  write_callback_pending_ = false;
   WriteRequest& request = write_queue_.front();
-  scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
+  int bytes_consumed = request.io_buffer->BytesConsumed();
 
-  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();
+  // If we are in connection flow, then we should have exaclty one item in
+  // the write queue.
+  if (ready_state_ == READY_STATE_CONNECTING) {
+    write_queue_.pop();
+    DCHECK(write_queue_.empty());
+    PostTaskToStartConnectLoop(bytes_consumed);
+  } else {
+    WriteRequest& request = write_queue_.front();
+    request.callback.Run(bytes_consumed);
+    write_queue_.pop();
   }
+  write_state_ = WRITE_STATE_WRITE;
+  return net::OK;
+}
 
-  request.callback.Run(result);
-  write_queue_.pop();
+int CastSocket::DoWriteError(int result) {
+  DCHECK(!write_queue_.empty());
+  DCHECK_LT(result, 0);
 
-  VLOG(1) << "OnWriteComplete size = " << io_buffer->size()
-          << " consumed " << io_buffer->BytesConsumed()
-          << " remaining " << io_buffer->BytesRemaining()
-          << " # requests " << write_queue_.size();
+  // If we are in connection flow, then we should have exactly one item in
+  // the write queue.
+  if (ready_state_ == READY_STATE_CONNECTING) {
+    write_queue_.pop();
+    DCHECK(write_queue_.empty());
+    PostTaskToStartConnectLoop(result);
+    return net::ERR_FAILED;
+  }
 
-  if (result < 0) {
-    CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
-    return;
+  while (!write_queue_.empty()) {
+    WriteRequest& request = write_queue_.front();
+    request.callback.Run(result);
+    write_queue_.pop();
   }
+  return net::ERR_FAILED;
+}
 
-  if (!write_queue_.empty())
-    WriteData();
+void CastSocket::PostTaskToStartReadLoop() {
+  DCHECK(CalledOnValidThread());
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
 }
 
-int CastSocket::ReadData() {
+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());
-  if (!socket_.get())
-    return net::ERR_FAILED;
-  DCHECK(!read_callback_pending_);
-  read_callback_pending_ = true;
+  // 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);
+      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) {
@@ -462,59 +550,88 @@
     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) {
-  DCHECK(CalledOnValidThread());
-  VLOG(1) << "OnReadData result = " << result
+int CastSocket::DoReadComplete(int 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 == 0 ? net::ERR_FAILED : 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 we received an auth message then check that we are in connect flow.
+    if (ready_state_ == READY_STATE_CONNECTING) {
+      challenge_reply_.reset(new CastMessage(current_message_));
+      PostTaskToStartConnectLoop(net::OK);
+    } else {
+      read_state_ = READ_STATE_ERROR;
+    }
+  } 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;
@@ -523,8 +640,9 @@
 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;
@@ -534,29 +652,6 @@
   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) {
@@ -621,7 +716,6 @@
 };
 
 void CastSocket::FillChannelInfo(ChannelInfo* channel_info) const {
-  DCHECK(CalledOnValidThread());
   channel_info->channel_id = channel_id_;
   channel_info->url = url_.spec();
   channel_info->ready_state = ready_state_;