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_;
@@ skipping 49 matching lines @@
   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) {

[Ryan Sleevi, 2013/11/21 22:16:15]

This is an area where it seems like you could further adopt the net/ pattern of
sync-or-async return (much as we do for Write and Read)

Namely

int CastSocket::Connect(const net::CompletionCallback& callback) {
  int rv = ConnectInternal();
  if (rv != net::ERR_IO_PENDING)
    return rv;
  connect_callback_ = callback;
  return net::ERR_IO_PENDING;
}

Then the caller can handle synchronous completion. This avoids a mandatory task
hop, which can introduce performance issues on mobile (where, admittedly, this
is less critical)

[Munjal (Google), 2013/11/21 22:44:11]

I was trying to avoid invoking the callback synchronously from Connect.

[Munjal (Google), 2013/11/21 22:53:36]

Actually I did not explain my reasons very well. It is fine to invoke the
callback synchronously as long as I make sure I don't access member veriables in
the unwind path (like you said) since the caller may delete this object in the
callback.

BUT Wez suggested that may be it is cleaner to have DoConnectLoop as the
callback for various async operations (instead of OnConnectComplete,
OnChallengeEvent, etc.). I also observed that in the previous version of the
code there were a few places where I was doing:

int rv = DoConnectLoop(result);
if (rv != net::ERR_IO_PENDING)
  DoConnectCallback(rv);

So I agreed with Wez's suggestion.

So I thought if DoConnectLoop also takes car of calling the connect callback as
part of its flow, then I can address the above points. In doing that, I needed
to make DoConnectLoop return type void instead of int. And that seemed OK.

I can make DoConnectLoop return int again and then invoke the callback
synchronously if needed, but that will introduce another method, say
OnConnectCallback. Then there will be two places - Connect and OnConnectCallback
- which need to do the code snippet above.

[Munjal (Google), 2013/11/21 22:56:48]

Here is an approach that might work better given all of the above:
- Add a method, say void PumpWriteLoop, that all places call
- Make PumpWriteLoop the callback method for all async operations DoConnectLoop
calls
- Remove the PostTask from Connect

What do you think?

   DCHECK(CalledOnValidThread());
-  int result = net::ERR_CONNECTION_FAILED;
+  connect_callback_ = callback;
+  // Post a task so that we can safely run the callback
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::ConnectInternal, AsWeakPtr()));
+}
+
+void CastSocket::ConnectInternal() {
+  DCHECK(CalledOnValidThread());
   VLOG(1) << "Connect readyState = " << ready_state_;
   if (ready_state_ != READY_STATE_NONE) {
-    callback.Run(result);
+    DoConnectCallback(net::ERR_CONNECTION_FAILED);
     return;
   }
   if (!ParseChannelUrl(url_)) {
     CloseWithError(cast_channel::CHANNEL_ERROR_CONNECT_ERROR);
-    callback.Run(result);
+    DoConnectCallback(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. ConnectInternal 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);
+    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;
+  connect_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
   socket_ = CreateSslSocket();
   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);
+  PostTaskToSendCastMessage(
+      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;
+  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;
+  // Start the ReadData loop if not already started.
+  if (result == net::OK)
+    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;
+  // Seems safe to invoke the callback synchronously for validation errors.

[Ryan Sleevi, 2013/11/21 22:16:15]

This comment doesn't really help readers of this code; If you're not sure about
it, then it's unsafe, and if you are sure about it, say so :) It creates
confusion as to what the safe API guarantees are - and this is your API, you get
to make the rules.

   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);
+
+  // Post a task so that we safely run the callback
+  PostTaskToSendCastMessage(message_proto, callback);

[Ryan Sleevi, 2013/11/21 22:16:15]

I don't understand why you need to Post a task to do this. Properly designed,
you shouldn't need to do a context switch here - you should be able to
synchronously read (and possibly fail) and invoke the callback safely.

There are other ways to do this that also avoid the PostTask overhead (both
mental and computational), IF it's absolutely necessary (and reading this code
in isolation doesn't seem to suggest it is), such as using a RefCounted or
WeakPtr'd inner class that holds onto whatever data members you need after the
callback.

The way net/ sockets are used, it's the higher layers responsibility to only
write one message at a time; that is, if there is an outstanding write that
returned ERR_IO_PENDING, the caller should not Write more data until the Write
completion callback is returned. This moves the queuing logic up a layer.

I realize you've adopted that queuing logic here, but I'm wondering whether it
makes sense to push it up even one layer further (I'm not sure what that layer
looks like, mind you), to handle the message queue. Just $.02

[Munjal (Google), 2013/11/21 22:44:11]

Actually, we do not need to post a task here. When SendMessage is called, it is
safe to invoke the callback synchronously since the caller should not delete
this object in the write callback. If a write loop is already running we will
return asynchronously anyway. I called PostTask* here by mistake. Changed now.

Pushing the queuing logic outside does not make sense since the consumer of this
class is the extension API implementation and this class is supposed to be the
abstraction. We could introduce another class in the middle, but that would just
move the queuing logic from here to there.

 }
 
-int CastSocket::SendMessageInternal(const CastMessage& message_proto,
-                                    const net::CompletionCallback& callback) {
-  WriteRequest write_request(callback);
-  if (!write_request.SetContent(message_proto))
-    return net::ERR_FAILED;
-  write_queue_.push(write_request);
-  return WriteData();
+void CastSocket::PostTaskToSendCastMessage(
+    const CastMessage& message, const net::CompletionCallback& callback) {
+  base::MessageLoop::current()->PostTask(
+      FROM_HERE,
+      base::Bind(&CastSocket::SendCastMessageInternal, AsWeakPtr(),
+      message, callback));
 }
 
-int CastSocket::WriteData() {
+void CastSocket::SendCastMessageInternal(
+    const CastMessage& message,
+    const net::CompletionCallback& callback) {
+  WriteRequest write_request(callback);
+  if (!write_request.SetContent(message)) {
+    callback.Run(net::ERR_FAILED);
+    return;
+  }
+
+  write_queue_.push(write_request);
+  if (write_state_ == WRITE_STATE_NONE) {
+    write_state_ = WRITE_STATE_WRITE;
+    DoWriteLoop(net::OK);
+  }
+}
+
+void CastSocket::ClearWriteQueue() {
+  while (!write_queue_.empty())
+    write_queue_.pop();
+}
+
+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;
+
+  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_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);
+}
+
+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;
   WriteRequest& request = write_queue_.front();
-  scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
 
-  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 (result < 0) {
+    // ERROR: do the write completion callback and get into error state
+    // to close the session
+    request.callback.Run(result);

[Ryan Sleevi, 2013/11/21 22:16:15]

If the callback deletes this object, transitively, then you'll have a memory
error on line 430.

Seems like you need to shuffle some of this error handling (and make sure
DoWriteCompelte doesn't access member variables in its unwind path)

[Munjal (Google), 2013/11/21 22:44:11]

The caller should not delete this object in the write callback. It should delete
this object in the OnError callback of the delegate.

+    write_queue_.pop();
+    write_state_ = WRITE_STATE_ERROR;
+    return 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 (result > 0) {
+    scoped_refptr<net::DrainableIOBuffer> io_buffer = request.io_buffer;
+    // Some bytes were successfully written
+    io_buffer->DidConsume(result);
+    if (io_buffer->BytesRemaining() == 0) {
+      // A message is sent fully, do the write completion callback
+      request.callback.Run(io_buffer->BytesConsumed());
+      write_queue_.pop();
+    }
   }
 
-  if (!write_queue_.empty())
-    WriteData();
+  write_state_ = WRITE_STATE_WRITE;

[Ryan Sleevi, 2013/11/21 22:16:15]

BUG: You don't handle == 0 (aka EOF).

[Munjal (Google), 2013/11/21 22:44:11]

I looked at the code in spdy_session.cc and (wrongly?) assumed that 0 is never
returned by write. I will find out and add the appropriate handling here.

+  return net::OK;
 }
 
-int CastSocket::ReadData() {
+int CastSocket::DoWriteError(int result) {
+  DCHECK(!write_queue_.empty());
+  DCHECK_LT(result, 0);
+  ClearWriteQueue();
+  CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
+  return net::OK;
+}
+
+void CastSocket::PostTaskToStartReadLoop() {
   DCHECK(CalledOnValidThread());
-  if (!socket_.get())
-    return net::ERR_FAILED;
-  DCHECK(!read_callback_pending_);
-  read_callback_pending_ = true;
+  // Post a task to avoid re-entrancy into DoConnectLoop
+  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());
+  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_ERROR:
+        rv = DoReadError(rv);
+        break;
+      default:
+        NOTREACHED() << "BUG in read state machine";
+        break;
+    }
+  } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE);
+}
+
+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;
+    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);
 
-  bool should_continue = true;
+  bool success = 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.
+    success = ProcessHeader();
   } 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();
+    success = ProcessBody();
   }
-  if (should_continue)
-    ReadData();
+
+  if (success)
+    read_state_ = READ_STATE_READ;
+  return net::OK;
+}
+
+int CastSocket::DoReadError(int result) {
+  DCHECK_LT(result, 0);
+  CloseWithError(error_state_);
+  return net::OK;
 }
 
 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);
+    error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
     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);
+    error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
     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_))
+      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);
+    DoConnectLoop(net::OK);
   } else if (delegate_) {
     MessageInfo message;
     if (!CastMessageToMessageInfo(message_proto, &message))
       return false;
     delegate_->OnMessage(this, message);
   }
   return true;
 }
 
 // static
@@ skipping 109 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