google_apis/gcm/engine/connection_handler_impl.cc

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "google_apis/gcm/engine/connection_handler_impl.h"
   6
   7 #include "base/message_loop/message_loop.h"
   8 #include "google/protobuf/io/coded_stream.h"
   9 #include "google_apis/gcm/base/mcs_util.h"
  10 #include "google_apis/gcm/base/socket_stream.h"
  11 #include "google_apis/gcm/protocol/mcs.pb.h"
  12 #include "net/base/net_errors.h"
  13 #include "net/socket/stream_socket.h"
  14
  15 using namespace google::protobuf::io;
  16
  17 namespace gcm {
  18
  19 namespace {
  20
  21 // # of bytes a MCS version packet consumes.
  22 const int kVersionPacketLen = 1;
  23 // # of bytes a tag packet consumes.
  24 const int kTagPacketLen = 1;
  25 // Max # of bytes a length packet consumes.
  26 const int kSizePacketLenMin = 1;
  27 const int kSizePacketLenMax = 2;
  28
  29 // The current MCS protocol version.
  30 const int kMCSVersion = 41;
  31
  32 }  // namespace
  33
  34 ConnectionHandlerImpl::ConnectionHandlerImpl(
  35     base::TimeDelta read_timeout,
  36     const ProtoReceivedCallback& read_callback,
  37     const ProtoSentCallback& write_callback,
  38     const ConnectionChangedCallback& connection_callback)
  39     : read_timeout_(read_timeout),
  40       socket_(NULL),
  41       handshake_complete_(false),
  42       message_tag_(0),
  43       message_size_(0),
  44       read_callback_(read_callback),
  45       write_callback_(write_callback),
  46       connection_callback_(connection_callback),
  47       weak_ptr_factory_(this) {
  48 }
  49
  50 ConnectionHandlerImpl::~ConnectionHandlerImpl() {
  51 }
  52
  53 void ConnectionHandlerImpl::Init(
  54     const mcs_proto::LoginRequest& login_request,
  55     net::StreamSocket* socket) {
  56   DCHECK(!read_callback_.is_null());
  57   DCHECK(!write_callback_.is_null());
  58   DCHECK(!connection_callback_.is_null());
  59
  60   // Invalidate any previously outstanding reads.
  61   weak_ptr_factory_.InvalidateWeakPtrs();
  62
  63   handshake_complete_ = false;
  64   message_tag_ = 0;
  65   message_size_ = 0;
  66   socket_ = socket;
  67   input_stream_.reset(new SocketInputStream(socket_));
  68   output_stream_.reset(new SocketOutputStream(socket_));
  69
  70   Login(login_request);
  71 }
  72
  73 void ConnectionHandlerImpl::Reset() {
  74   CloseConnection();
  75 }
  76
  77 bool ConnectionHandlerImpl::CanSendMessage() const {
  78   return handshake_complete_ && output_stream_.get() &&
  79       output_stream_->GetState() == SocketOutputStream::EMPTY;
  80 }
  81
  82 void ConnectionHandlerImpl::SendMessage(
  83     const google::protobuf::MessageLite& message) {
  84   DCHECK_EQ(output_stream_->GetState(), SocketOutputStream::EMPTY);
  85   DCHECK(handshake_complete_);
  86
  87   {
  88     CodedOutputStream coded_output_stream(output_stream_.get());
  89     DVLOG(1) << "Writing proto of size " << message.ByteSize();
  90     int tag = GetMCSProtoTag(message);
  91     DCHECK_NE(tag, -1);
  92     coded_output_stream.WriteRaw(&tag, 1);
  93     coded_output_stream.WriteVarint32(message.ByteSize());
  94     message.SerializeToCodedStream(&coded_output_stream);
  95   }
  96
  97   if (output_stream_->Flush(
  98           base::Bind(&ConnectionHandlerImpl::OnMessageSent,
  99                      weak_ptr_factory_.GetWeakPtr())) != net::ERR_IO_PENDING) {
 100     OnMessageSent();
 101   }
 102 }
 103
 104 void ConnectionHandlerImpl::Login(
 105     const google::protobuf::MessageLite& login_request) {
 106   DCHECK_EQ(output_stream_->GetState(), SocketOutputStream::EMPTY);
 107
 108   const char version_byte[1] = {kMCSVersion};
 109   const char login_request_tag[1] = {kLoginRequestTag};
 110   {
 111     CodedOutputStream coded_output_stream(output_stream_.get());
 112     coded_output_stream.WriteRaw(version_byte, 1);
 113     coded_output_stream.WriteRaw(login_request_tag, 1);
 114     coded_output_stream.WriteVarint32(login_request.ByteSize());
 115     login_request.SerializeToCodedStream(&coded_output_stream);
 116   }
 117
 118   if (output_stream_->Flush(
 119           base::Bind(&ConnectionHandlerImpl::OnMessageSent,
 120                      weak_ptr_factory_.GetWeakPtr())) != net::ERR_IO_PENDING) {
 121     base::MessageLoop::current()->PostTask(
 122         FROM_HERE,
 123         base::Bind(&ConnectionHandlerImpl::OnMessageSent,
 124                    weak_ptr_factory_.GetWeakPtr()));
 125   }
 126
 127   read_timeout_timer_.Start(FROM_HERE,
 128                             read_timeout_,
 129                             base::Bind(&ConnectionHandlerImpl::OnTimeout,
 130                                        weak_ptr_factory_.GetWeakPtr()));
 131   WaitForData(MCS_VERSION_TAG_AND_SIZE);
 132 }
 133
 134 void ConnectionHandlerImpl::OnMessageSent() {
 135   if (!output_stream_.get()) {
 136     // The connection has already been closed. Just return.
 137     DCHECK(!input_stream_.get());
 138     DCHECK(!read_timeout_timer_.IsRunning());
 139     return;
 140   }
 141
 142   if (output_stream_->GetState() != SocketOutputStream::EMPTY) {
 143     int last_error = output_stream_->last_error();
 144     CloseConnection();
 145     // If the socket stream had an error, plumb it up, else plumb up FAILED.
 146     if (last_error == net::OK)
 147       last_error = net::ERR_FAILED;
 148     connection_callback_.Run(last_error);
 149     return;
 150   }
 151
 152   write_callback_.Run();
 153 }
 154
 155 void ConnectionHandlerImpl::GetNextMessage() {
 156   DCHECK(SocketInputStream::EMPTY == input_stream_->GetState() ||
 157          SocketInputStream::READY == input_stream_->GetState());
 158   message_tag_ = 0;
 159   message_size_ = 0;
 160
 161   WaitForData(MCS_TAG_AND_SIZE);
 162 }
 163
 164 void ConnectionHandlerImpl::WaitForData(ProcessingState state) {
 165   DVLOG(1) << "Waiting for MCS data: state == " << state;
 166
 167   if (!input_stream_) {
 168     // The connection has already been closed. Just return.
 169     DCHECK(!output_stream_.get());
 170     DCHECK(!read_timeout_timer_.IsRunning());
 171     return;
 172   }
 173
 174   if (input_stream_->GetState() != SocketInputStream::EMPTY &&
 175       input_stream_->GetState() != SocketInputStream::READY) {
 176     // An error occurred.
 177     int last_error = output_stream_->last_error();
 178     CloseConnection();
 179     // If the socket stream had an error, plumb it up, else plumb up FAILED.
 180     if (last_error == net::OK)
 181       last_error = net::ERR_FAILED;
 182     connection_callback_.Run(last_error);
 183     return;
 184   }
 185
 186   // Used to determine whether a Socket::Read is necessary.
 187   size_t min_bytes_needed = 0;
 188   // Used to limit the size of the Socket::Read.
 189   size_t max_bytes_needed = 0;
 190
 191   switch(state) {
 192     case MCS_VERSION_TAG_AND_SIZE:
 193       min_bytes_needed = kVersionPacketLen + kTagPacketLen + kSizePacketLenMin;
 194       max_bytes_needed = kVersionPacketLen + kTagPacketLen + kSizePacketLenMax;
 195       break;
 196     case MCS_TAG_AND_SIZE:
 197       min_bytes_needed = kTagPacketLen + kSizePacketLenMin;
 198       max_bytes_needed = kTagPacketLen + kSizePacketLenMax;
 199       break;
 200     case MCS_FULL_SIZE:
 201       // If in this state, the minimum size packet length must already have been
 202       // insufficient, so set both to the max length.
 203       min_bytes_needed = kSizePacketLenMax;
 204       max_bytes_needed = kSizePacketLenMax;
 205       break;
 206     case MCS_PROTO_BYTES:
 207       read_timeout_timer_.Reset();
 208       // No variability in the message size, set both to the same.
 209       min_bytes_needed = message_size_;
 210       max_bytes_needed = message_size_;
 211       break;
 212     default:
 213       NOTREACHED();
 214   }
 215   DCHECK_GE(max_bytes_needed, min_bytes_needed);
 216
 217   size_t unread_byte_count = input_stream_->UnreadByteCount();
 218   if (min_bytes_needed > unread_byte_count &&
 219       input_stream_->Refresh(
 220           base::Bind(&ConnectionHandlerImpl::WaitForData,
 221                      weak_ptr_factory_.GetWeakPtr(),
 222                      state),
 223           max_bytes_needed - unread_byte_count) == net::ERR_IO_PENDING) {
 224     return;
 225   }
 226
 227   // Check for refresh errors.
 228   if (input_stream_->GetState() != SocketInputStream::READY) {
 229     // An error occurred.
 230     int last_error = input_stream_->last_error();
 231     CloseConnection();
 232     // If the socket stream had an error, plumb it up, else plumb up FAILED.
 233     if (last_error == net::OK)
 234       last_error = net::ERR_FAILED;
 235     connection_callback_.Run(last_error);
 236     return;
 237   }
 238
 239   // Check whether read is complete, or needs to be continued (
 240   // SocketInputStream::Refresh can finish without reading all the data).
 241   if (input_stream_->UnreadByteCount() < min_bytes_needed) {
 242     DVLOG(1) << "Socket read finished prematurely. Waiting for "
 243              << min_bytes_needed - input_stream_->UnreadByteCount()
 244              << " more bytes.";
 245     base::MessageLoop::current()->PostTask(
 246         FROM_HERE,
 247         base::Bind(&ConnectionHandlerImpl::WaitForData,
 248                    weak_ptr_factory_.GetWeakPtr(),
 249                    MCS_PROTO_BYTES));
 250     return;
 251   }
 252
 253   // Received enough bytes, process them.
 254   DVLOG(1) << "Processing MCS data: state == " << state;
 255   switch(state) {
 256     case MCS_VERSION_TAG_AND_SIZE:
 257       OnGotVersion();
 258       break;
 259     case MCS_TAG_AND_SIZE:
 260       OnGotMessageTag();
 261       break;
 262     case MCS_FULL_SIZE:
 263       OnGotMessageSize();
 264       break;
 265     case MCS_PROTO_BYTES:
 266       OnGotMessageBytes();
 267       break;
 268     default:
 269       NOTREACHED();
 270   }
 271 }
 272
 273 void ConnectionHandlerImpl::OnGotVersion() {
 274   uint8 version = 0;
 275   {
 276     CodedInputStream coded_input_stream(input_stream_.get());
 277     coded_input_stream.ReadRaw(&version, 1);
 278   }
 279   // TODO(zea): remove this when the server is ready.
 280   if (version < kMCSVersion && version != 38) {
 281     LOG(ERROR) << "Invalid GCM version response: " << static_cast<int>(version);
 282     connection_callback_.Run(net::ERR_FAILED);
 283     return;
 284   }
 285
 286   input_stream_->RebuildBuffer();
 287
 288   // Process the LoginResponse message tag.
 289   OnGotMessageTag();
 290 }
 291
 292 void ConnectionHandlerImpl::OnGotMessageTag() {
 293   if (input_stream_->GetState() != SocketInputStream::READY) {
 294     LOG(ERROR) << "Failed to receive protobuf tag.";
 295     read_callback_.Run(scoped_ptr<google::protobuf::MessageLite>());
 296     return;
 297   }
 298
 299   {
 300     CodedInputStream coded_input_stream(input_stream_.get());
 301     coded_input_stream.ReadRaw(&message_tag_, 1);
 302   }
 303
 304   DVLOG(1) << "Received proto of type "
 305            << static_cast<unsigned int>(message_tag_);
 306
 307   if (!read_timeout_timer_.IsRunning()) {
 308     read_timeout_timer_.Start(FROM_HERE,
 309                               read_timeout_,
 310                               base::Bind(&ConnectionHandlerImpl::OnTimeout,
 311                                          weak_ptr_factory_.GetWeakPtr()));
 312   }
 313   OnGotMessageSize();
 314 }
 315
 316 void ConnectionHandlerImpl::OnGotMessageSize() {
 317   if (input_stream_->GetState() != SocketInputStream::READY) {
 318     LOG(ERROR) << "Failed to receive message size.";
 319     read_callback_.Run(scoped_ptr<google::protobuf::MessageLite>());
 320     return;
 321   }
 322
 323   bool need_another_byte = false;
 324   int prev_byte_count = input_stream_->ByteCount();
 325   {
 326     CodedInputStream coded_input_stream(input_stream_.get());
 327     if (!coded_input_stream.ReadVarint32(&message_size_))
 328       need_another_byte = true;
 329   }
 330
 331   if (need_another_byte) {
 332     DVLOG(1) << "Expecting another message size byte.";
 333     if (prev_byte_count >= kSizePacketLenMax) {
 334       // Already had enough bytes, something else went wrong.
 335       LOG(ERROR) << "Failed to process message size.";
 336       read_callback_.Run(scoped_ptr<google::protobuf::MessageLite>());
 337       return;
 338     }
 339     // Back up by the amount read (should always be 1 byte).
 340     int bytes_read = prev_byte_count - input_stream_->ByteCount();
 341     DCHECK_EQ(bytes_read, 1);
 342     input_stream_->BackUp(bytes_read);
 343     WaitForData(MCS_FULL_SIZE);
 344     return;
 345   }
 346
 347   DVLOG(1) << "Proto size: " << message_size_;
 348
 349   if (message_size_ > 0)
 350     WaitForData(MCS_PROTO_BYTES);
 351   else
 352     OnGotMessageBytes();
 353 }
 354
 355 void ConnectionHandlerImpl::OnGotMessageBytes() {
 356   read_timeout_timer_.Stop();
 357   scoped_ptr<google::protobuf::MessageLite> protobuf(
 358       BuildProtobufFromTag(message_tag_));
 359   // Messages with no content are valid; just use the default protobuf for
 360   // that tag.
 361   if (protobuf.get() && message_size_ == 0) {
 362     base::MessageLoop::current()->PostTask(
 363         FROM_HERE,
 364         base::Bind(&ConnectionHandlerImpl::GetNextMessage,
 365                    weak_ptr_factory_.GetWeakPtr()));
 366     read_callback_.Run(protobuf.Pass());
 367     return;
 368   }
 369
 370   if (!protobuf.get() ||
 371       input_stream_->GetState() != SocketInputStream::READY) {
 372     LOG(ERROR) << "Failed to extract protobuf bytes of type "
 373                << static_cast<unsigned int>(message_tag_);
 374     // Reset the connection.
 375     connection_callback_.Run(net::ERR_FAILED);
 376     return;
 377   }
 378
 379   {
 380     CodedInputStream coded_input_stream(input_stream_.get());
 381     if (!protobuf->ParsePartialFromCodedStream(&coded_input_stream)) {
 382       LOG(ERROR) << "Unable to parse GCM message of type "
 383                  << static_cast<unsigned int>(message_tag_);
 384       // Reset the connection.
 385       connection_callback_.Run(net::ERR_FAILED);
 386       return;
 387     }
 388   }
 389
 390   input_stream_->RebuildBuffer();
 391   base::MessageLoop::current()->PostTask(
 392       FROM_HERE,
 393       base::Bind(&ConnectionHandlerImpl::GetNextMessage,
 394                  weak_ptr_factory_.GetWeakPtr()));
 395   if (message_tag_ == kLoginResponseTag) {
 396     if (handshake_complete_) {
 397       LOG(ERROR) << "Unexpected login response.";
 398     } else {
 399       handshake_complete_ = true;
 400       DVLOG(1) << "GCM Handshake complete.";
 401       connection_callback_.Run(net::OK);
 402     }
 403   }
 404   read_callback_.Run(protobuf.Pass());
 405 }
 406
 407 void ConnectionHandlerImpl::OnTimeout() {
 408   LOG(ERROR) << "Timed out waiting for GCM Protocol buffer.";
 409   CloseConnection();
 410   connection_callback_.Run(net::ERR_TIMED_OUT);
 411 }
 412
 413 void ConnectionHandlerImpl::CloseConnection() {
 414   DVLOG(1) << "Closing connection.";
 415   read_timeout_timer_.Stop();
 416   if (socket_)
 417     socket_->Disconnect();
 418   socket_ = NULL;
 419   handshake_complete_ = false;
 420   message_tag_ = 0;
 421   message_size_ = 0;
 422   input_stream_.reset();
 423   output_stream_.reset();
 424   weak_ptr_factory_.InvalidateWeakPtrs();
 425 }
 426
 427 }  // namespace gcm