#include "chrome/browser/extensions/api/cast_channel/cast_socket.h"
+#include <stdlib.h>
#include <string.h>
#include "base/bind.h"
#include "base/callback_helpers.h"
#include "base/lazy_instance.h"
+#include "base/numerics/safe_conversions.h"
#include "base/strings/string_number_conversions.h"
#include "base/sys_byteorder.h"
#include "chrome/browser/extensions/api/cast_channel/cast_auth_util.h"
#include "net/ssl/ssl_config_service.h"
#include "net/ssl/ssl_info.h"
-namespace {
-
-// Allowed schemes for Cast device URLs.
-const char kCastInsecureScheme[] = "cast";
-const char kCastSecureScheme[] = "casts";
+// Assumes |ip_endpoint_| of type net::IPEndPoint and |channel_auth_| of enum
+// type ChannelAuthType are available in the current scope.
+#define VLOG_WITH_CONNECTION(level) VLOG(level) << "[" << \
+ ip_endpoint_.ToString() << ", auth=" << channel_auth_ << "] "
-// Size of the message header, in bytes. Don't use sizeof(MessageHeader)
-// because of alignment; instead, sum the sizeof() for the fields.
-const uint32 kMessageHeaderSize = sizeof(uint32);
+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
namespace extensions {
-static base::LazyInstance<
- ProfileKeyedAPIFactory<ApiResourceManager<api::cast_channel::CastSocket> > >
- g_factory = LAZY_INSTANCE_INITIALIZER;
+static base::LazyInstance<BrowserContextKeyedAPIFactory<
+ ApiResourceManager<api::cast_channel::CastSocket> > > g_factory =
+ LAZY_INSTANCE_INITIALIZER;
// static
template <>
-ProfileKeyedAPIFactory<ApiResourceManager<api::cast_channel::CastSocket> >*
+BrowserContextKeyedAPIFactory<
+ ApiResourceManager<api::cast_channel::CastSocket> >*
ApiResourceManager<api::cast_channel::CastSocket>::GetFactoryInstance() {
- return &g_factory.Get();
+ return g_factory.Pointer();
}
namespace api {
namespace cast_channel {
-const uint32 kMaxMessageSize = 65536;
-
CastSocket::CastSocket(const std::string& owner_extension_id,
- const GURL& url,
+ const net::IPEndPoint& ip_endpoint,
+ ChannelAuthType channel_auth,
CastSocket::Delegate* delegate,
net::NetLog* net_log) :
ApiResource(owner_extension_id),
channel_id_(0),
- url_(url),
+ ip_endpoint_(ip_endpoint),
+ channel_auth_(channel_auth),
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),
+ current_message_(new CastMessage()),
net_log_(net_log),
- next_state_(CONN_STATE_NONE) {
+ 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();
- // We reuse these buffers for each message.
+ // Reuse these buffers for each message.
header_read_buffer_ = new net::GrowableIOBuffer();
- header_read_buffer_->SetCapacity(kMessageHeaderSize);
+ header_read_buffer_->SetCapacity(MessageHeader::header_size());
body_read_buffer_ = new net::GrowableIOBuffer();
- body_read_buffer_->SetCapacity(kMaxMessageSize);
+ body_read_buffer_->SetCapacity(MessageHeader::max_message_size());
current_read_buffer_ = header_read_buffer_;
}
CastSocket::~CastSocket() { }
-const GURL& CastSocket::url() const {
- return url_;
+ReadyState CastSocket::ready_state() const {
+ return ready_state_;
+}
+
+ChannelError CastSocket::error_state() const {
+ return error_state_;
}
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_));
- // Enable keepalive
- tcp_socket->SetKeepAlive(true, kTcpKeepAliveDelaySecs);
- return tcp_socket.Pass();
+ // Options cannot be set on the TCPClientSocket yet, because the
+ // underlying platform socket will not be created until Bind()
+ // or Connect() is called.
}
-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.
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_);
bool result = net::X509Certificate::GetDEREncoded(
ssl_info.cert->os_cert_handle(), cert);
if (result)
- DVLOG(1) << "Successfully extracted peer certificate: " << *cert;
+ VLOG_WITH_CONNECTION(1) << "Successfully extracted peer certificate: "
+ << *cert;
return result;
}
-int CastSocket::SendAuthChallenge() {
- CastMessage challenge_message;
- CreateAuthChallengeMessage(&challenge_message);
- DVLOG(1) << "Sending challenge: " << CastMessageToString(challenge_message);
- return SendMessageInternal(
- challenge_message,
- base::Bind(&CastSocket::OnChallengeEvent, AsWeakPtr()));
-}
-
-int CastSocket::ReadAuthChallengeReply() {
- return ReadData();
-}
-
-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_, peer_cert_);
}
void CastSocket::Connect(const net::CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
- int result = net::ERR_CONNECTION_FAILED;
- DVLOG(1) << "Connect readyState = " << ready_state_;
+ VLOG_WITH_CONNECTION(1) << "Connect readyState = " << ready_state_;
if (ready_state_ != READY_STATE_NONE) {
- callback.Run(result);
- 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);
+}
+
+void CastSocket::PostTaskToStartConnectLoop(int result) {
+ 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) {
- // Network operations can either finish sycnronously or asynchronously.
+// 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*
- // method changed state or not.
- next_state_ = CONN_STATE_NONE;
+ ConnectionState state = connect_state_;
+ // Default to CONN_STATE_NONE, which breaks the processing loop if any
+ // handler fails to transition to another state to continue processing.
+ connect_state_ = CONN_STATE_NONE;
switch (state) {
case CONN_STATE_TCP_CONNECT:
rv = DoTcpConnect();
case CONN_STATE_AUTH_CHALLENGE_REPLY_COMPLETE:
rv = DoAuthChallengeReplyComplete(rv);
break;
-
default:
- NOTREACHED() << "BUG in CastSocket state machine code";
+ NOTREACHED() << "BUG in connect flow. Unknown state: " << state;
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
- return rv;
+ // Connect loop is finished: if there is no pending IO invoke the callback.
+ if (rv != net::ERR_IO_PENDING)
+ DoConnectCallback(rv);
}
int CastSocket::DoTcpConnect() {
- DVLOG(1) << "DoTcpConnect";
- next_state_ = CONN_STATE_TCP_CONNECT_COMPLETE;
+ VLOG_WITH_CONNECTION(1) << "DoTcpConnect";
+ 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) {
- DVLOG(1) << "DoTcpConnectComplete: " << result;
- if (result == net::OK)
- next_state_ = CONN_STATE_SSL_CONNECT;
+ VLOG_WITH_CONNECTION(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.";
+ connect_state_ = CONN_STATE_SSL_CONNECT;
+ }
return result;
}
int CastSocket::DoSslConnect() {
- DVLOG(1) << "DoSslConnect";
- next_state_ = CONN_STATE_SSL_CONNECT_COMPLETE;
- socket_ = CreateSslSocket();
+ VLOG_WITH_CONNECTION(1) << "DoSslConnect";
+ 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) {
- DVLOG(1) << "DoSslConnectComplete: " << result;
+ VLOG_WITH_CONNECTION(1) << "DoSslConnectComplete: " << result;
if (result == net::ERR_CERT_AUTHORITY_INVALID &&
- peer_cert_.empty() &&
- ExtractPeerCert(&peer_cert_)) {
- next_state_ = CONN_STATE_TCP_CONNECT;
- } else if (result == net::OK && auth_required_) {
- next_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
+ peer_cert_.empty() && ExtractPeerCert(&peer_cert_)) {
+ connect_state_ = CONN_STATE_TCP_CONNECT;
+ } else if (result == net::OK &&
+ channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) {
+ connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND;
}
return result;
}
int CastSocket::DoAuthChallengeSend() {
- DVLOG(1) << "DoAuthChallengeSend";
- next_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
- return SendAuthChallenge();
+ VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSend";
+ connect_state_ = CONN_STATE_AUTH_CHALLENGE_SEND_COMPLETE;
+ CastMessage challenge_message;
+ CreateAuthChallengeMessage(&challenge_message);
+ VLOG_WITH_CONNECTION(1) << "Sending challenge: "
+ << CastMessageToString(challenge_message);
+ // Post a task to send auth challenge 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 the result is always asynchronous.
+ return net::ERR_IO_PENDING;
}
int CastSocket::DoAuthChallengeSendComplete(int result) {
- DVLOG(1) << "DoAuthChallengeSendComplete: " << result;
- if (result != net::OK)
+ VLOG_WITH_CONNECTION(1) << "DoAuthChallengeSendComplete: " << result;
+ 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 the result is always asynchronous.
+ return net::ERR_IO_PENDING;
}
int CastSocket::DoAuthChallengeReplyComplete(int result) {
- DVLOG(1) << "DoAuthChallengeReplyComplete: " << result;
- if (result != net::OK)
+ VLOG_WITH_CONNECTION(1) << "DoAuthChallengeReplyComplete: " << result;
+ if (result < 0)
return result;
if (!VerifyChallengeReply())
return net::ERR_FAILED;
- DVLOG(1) << "Auth challenge verification succeeded";
+ VLOG_WITH_CONNECTION(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());
- DVLOG(1) << "Close ReadyState = " << ready_state_;
- tcp_socket_.reset(NULL);
- socket_.reset(NULL);
- cert_verifier_.reset(NULL);
- transport_security_state_.reset(NULL);
+ VLOG_WITH_CONNECTION(1) << "Close ReadyState = " << ready_state_;
+ tcp_socket_.reset();
+ socket_.reset();
+ cert_verifier_.reset();
+ transport_security_state_.reset();
ready_state_ = READY_STATE_CLOSED;
callback.Run(net::OK);
+ // |callback| can delete |this|
}
void CastSocket::SendMessage(const MessageInfo& message,
const net::CompletionCallback& callback) {
DCHECK(CalledOnValidThread());
- DVLOG(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());
- DVLOG(1) << "WriteData q = " << write_queue_.size();
- if (write_queue_.empty() || write_callback_pending_)
- return net::ERR_FAILED;
+ VLOG_WITH_CONNECTION(1) << "DoWriteLoop queue size: " << write_queue_.size();
+
+ if (write_queue_.empty()) {
+ write_state_ = WRITE_STATE_NONE;
+ 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 write flow. Unknown state: " << state;
+ break;
+ }
+ } while (!write_queue_.empty() &&
+ rv != net::ERR_IO_PENDING &&
+ write_state_ != WRITE_STATE_NONE);
+
+ // If write loop is done because the queue is empty then set write
+ // state to NONE
+ if (write_queue_.empty())
+ write_state_ = WRITE_STATE_NONE;
+
+ // Write loop is done - if 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();
- DVLOG(1) << "WriteData byte_count = " << request.io_buffer->size() <<
- " bytes_written " << request.io_buffer->BytesConsumed();
+ VLOG_WITH_CONNECTION(2) << "WriteData byte_count = "
+ << request.io_buffer->size() << " bytes_written "
+ << request.io_buffer->BytesConsumed();
+
+ write_state_ = WRITE_STATE_WRITE_COMPLETE;
- write_callback_pending_ = true;
- int result = socket_->Write(
+ 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());
- DVLOG(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 inside connection flow, then there should be 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;
+}
- if (result >= 0) {
- io_buffer->DidConsume(result);
- if (io_buffer->BytesRemaining() > 0) {
- DVLOG(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();
+int CastSocket::DoWriteError(int result) {
+ DCHECK(!write_queue_.empty());
+ DCHECK_LT(result, 0);
+
+ // If inside connection flow, then there should be exactly one item in
+ // the write queue.
+ if (ready_state_ == READY_STATE_CONNECTING) {
+ write_queue_.pop();
+ DCHECK(write_queue_.empty());
+ PostTaskToStartConnectLoop(result);
+ // Connect loop will handle the error. Return net::OK so that write flow
+ // does not try to report error also.
+ return net::OK;
}
- request.callback.Run(result);
- write_queue_.pop();
+ while (!write_queue_.empty()) {
+ WriteRequest& request = write_queue_.front();
+ request.callback.Run(result);
+ write_queue_.pop();
+ }
+ return net::ERR_FAILED;
+}
- DVLOG(1) << "OnWriteComplete size = " << io_buffer->size() <<
- " consumed " << io_buffer->BytesConsumed() <<
- " remaining " << io_buffer->BytesRemaining() <<
- " # requests " << write_queue_.size();
+void CastSocket::PostTaskToStartReadLoop() {
+ DCHECK(CalledOnValidThread());
+ base::MessageLoop::current()->PostTask(
+ FROM_HERE,
+ base::Bind(&CastSocket::StartReadLoop, AsWeakPtr()));
+}
- if (result < 0) {
- CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
- return;
+void CastSocket::StartReadLoop() {
+ // Read loop would have already been started if read state is not NONE
+ if (read_state_ == READ_STATE_NONE) {
+ read_state_ = READ_STATE_READ;
+ DoReadLoop(net::OK);
}
-
- if (!write_queue_.empty())
- WriteData();
}
-int CastSocket::ReadData() {
+void CastSocket::DoReadLoop(int result) {
DCHECK(CalledOnValidThread());
- if (!socket_.get())
- return net::ERR_FAILED;
- DCHECK(!read_callback_pending_);
- read_callback_pending_ = true;
- // Figure out if we are reading the header or body, and the remaining bytes.
+ // 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);
+ DCHECK_EQ(read_state_, READ_STATE_NONE);
+ break;
+ default:
+ NOTREACHED() << "BUG in read flow. Unknown state: " << state;
+ break;
+ }
+ } while (rv != net::ERR_IO_PENDING && read_state_ != READ_STATE_NONE);
+
+ if (rv == net::ERR_FAILED) {
+ if (ready_state_ == READY_STATE_CONNECTING) {
+ // Read errors during the handshake should notify the caller via
+ // the connect callback, rather than the message event delegate.
+ PostTaskToStartConnectLoop(net::ERR_FAILED);
+ } else {
+ // Connection is already established.
+ // Close and send error status via the message event delegate.
+ CloseWithError(error_state_);
+ }
+ }
+}
+
+int CastSocket::DoRead() {
+ read_state_ = 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();
- DCHECK_LE(num_bytes_to_read, kMessageHeaderSize);
+ 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_;
- DCHECK_LE(num_bytes_to_read, kMaxMessageSize);
+ CHECK_LE(num_bytes_to_read, MessageHeader::max_message_size());
}
- DCHECK_GT(num_bytes_to_read, 0U);
- // We read up to num_bytes_to_read into |current_read_buffer_|.
- int result = socket_->Read(
+ CHECK_GT(num_bytes_to_read, 0U);
+
+ // Read up to num_bytes_to_read into |current_read_buffer_|.
+ return socket_->Read(
current_read_buffer_.get(),
num_bytes_to_read,
- base::Bind(&CastSocket::OnReadData, AsWeakPtr()));
- DVLOG(1) << "ReadData result = " << result;
- if (result > 0) {
- OnReadData(result);
- } else if (result != net::ERR_IO_PENDING) {
- CloseWithError(CHANNEL_ERROR_SOCKET_ERROR);
+ base::Bind(&CastSocket::DoReadLoop, AsWeakPtr()));
+}
+
+int CastSocket::DoReadComplete(int result) {
+ VLOG_WITH_CONNECTION(2) << "DoReadComplete result = " << result
+ << " header offset = "
+ << header_read_buffer_->offset()
+ << " body offset = " << body_read_buffer_->offset();
+ if (result <= 0) { // 0 means EOF: the peer closed the socket
+ VLOG_WITH_CONNECTION(1) << "Read error, peer closed the socket";
+ error_state_ = CHANNEL_ERROR_SOCKET_ERROR;
+ read_state_ = READ_STATE_ERROR;
+ return result == 0 ? net::ERR_FAILED : result;
}
- return result;
-}
-void CastSocket::OnReadData(int result) {
- DCHECK(CalledOnValidThread());
- DVLOG(1) << "OnReadData 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;
- }
- // We read some data. Move the offset in the current buffer forward.
- DCHECK_LE(current_read_buffer_->offset() + result,
- current_read_buffer_->capacity());
+ // 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);
+ 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();
+ // A full header is read, 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();
+ // Full body is read, process the contents.
+ if (ProcessBody()) {
+ read_state_ = READ_STATE_DO_CALLBACK;
+ } else {
+ error_state_ = cast_channel::CHANNEL_ERROR_INVALID_MESSAGE;
+ read_state_ = READ_STATE_ERROR;
+ }
+ }
+
+ return net::OK;
+}
+
+int CastSocket::DoReadCallback() {
+ read_state_ = READ_STATE_READ;
+ const CastMessage& message = *current_message_;
+ if (ready_state_ == READY_STATE_CONNECTING) {
+ if (IsAuthMessage(message)) {
+ challenge_reply_.reset(new CastMessage(message));
+ PostTaskToStartConnectLoop(net::OK);
+ return net::OK;
+ } else {
+ // Expected an auth message, got something else instead. Handle as error.
+ read_state_ = READ_STATE_ERROR;
+ return net::ERR_INVALID_RESPONSE;
+ }
}
- if (should_continue)
- ReadData();
+
+ MessageInfo message_info;
+ if (!CastMessageToMessageInfo(message, &message_info)) {
+ current_message_->Clear();
+ read_state_ = READ_STATE_ERROR;
+ return net::ERR_INVALID_RESPONSE;
+ }
+ delegate_->OnMessage(this, message_info);
+ current_message_->Clear();
+ return net::OK;
+}
+
+int CastSocket::DoReadError(int result) {
+ DCHECK_LE(result, 0);
+ return net::ERR_FAILED;
}
bool CastSocket::ProcessHeader() {
- DCHECK_EQ(static_cast<uint32>(header_read_buffer_->offset()),
- kMessageHeaderSize);
+ 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 > kMaxMessageSize) {
- CloseWithError(cast_channel::CHANNEL_ERROR_INVALID_MESSAGE);
+ if (header.message_size > MessageHeader::max_message_size())
return false;
- }
- DVLOG(1) << "Parsed header { message_size: " << header.message_size << " }";
+
+ VLOG_WITH_CONNECTION(2) << "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);
+ 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;
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;
- DVLOG(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) {
+ if (message_size > MessageHeader::max_message_size()) {
message_data->clear();
return false;
}
delegate_->OnError(this, error);
}
-bool CastSocket::ParseChannelUrl(const GURL& url) {
- DVLOG(1) << "url = " + url.spec();
- if (url.SchemeIs(kCastInsecureScheme)) {
- auth_required_ = false;
- } else if (url.SchemeIs(kCastSecureScheme)) {
- auth_required_ = true;
- } else {
- return false;
- }
- // TODO(mfoltz): Manual parsing, yech. Register cast[s] as standard schemes?
- // TODO(mfoltz): Test for IPv6 addresses. Brackets or no brackets?
- // TODO(mfoltz): Maybe enforce restriction to IPv4 private and IPv6 link-local
- // networks
- const std::string& path = url.path();
- // Shortest possible: //A:B
- if (path.size() < 5) {
- return false;
- }
- if (path.find("//") != 0) {
- return false;
- }
- size_t colon = path.find_last_of(':');
- if (colon == std::string::npos || colon < 3 || colon > path.size() - 2) {
- return false;
- }
- const std::string& ip_address_str = path.substr(2, colon - 2);
- const std::string& port_str = path.substr(colon + 1);
- DVLOG(1) << "addr " << ip_address_str << " port " << port_str;
- int port;
- if (!base::StringToInt(port_str, &port))
- return false;
- net::IPAddressNumber ip_address;
- if (!net::ParseIPLiteralToNumber(ip_address_str, &ip_address))
- return false;
- ip_endpoint_ = net::IPEndPoint(ip_address, port);
- return true;
-};
-
-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_;
- channel_info->error_state = error_state_;
+std::string CastSocket::CastUrl() const {
+ return ((channel_auth_ == CHANNEL_AUTH_TYPE_SSL_VERIFIED) ?
+ "casts://" : "cast://") + ip_endpoint_.ToString();
}
bool CastSocket::CalledOnValidThread() const {
void CastSocket::MessageHeader::SetMessageSize(size_t size) {
DCHECK(size < static_cast<size_t>(kuint32max));
DCHECK(size > 0);
- message_size = static_cast<size_t>(size);
+ 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);
- char char_array[kMessageHeaderSize];
- memcpy(&char_array, &output, arraysize(char_array));
- str->insert(0, char_array, arraysize(char_array));
+ 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;
- memcpy(&message_size, buffer->StartOfBuffer(), kMessageHeaderSize);
+ 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);
}
} // namespace cast_channel
} // namespace api
} // namespace extensions
+
+#undef VLOG_WITH_CONNECTION
projects / platform / framework / web / crosswalk.git / blobdiff