[GCM] Add heartbeat manager and reconnection logic due to heartbeat failure

google_apis/gcm/engine/mcs_client.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 "google_apis/gcm/engine/mcs_client.h"
 
 #include "base/basictypes.h"
 #include "base/message_loop/message_loop.h"
 #include "base/strings/string_number_conversions.h"
 #include "google_apis/gcm/base/mcs_util.h"
 #include "google_apis/gcm/base/socket_stream.h"
 #include "google_apis/gcm/engine/connection_factory.h"
 #include "google_apis/gcm/engine/rmq_store.h"
 
 using namespace google::protobuf::io;
 
 namespace gcm {
 
 namespace {
 
 typedef scoped_ptr<google::protobuf::MessageLite> MCSProto;
 
-// TODO(zea): get these values from MCS settings.
-const int64 kHeartbeatDefaultSeconds = 60 * 15;  // 15 minutes.
-
 // The category of messages intended for the GCM client itself from MCS.
 const char kMCSCategory[] = "com.google.android.gsf.gtalkservice";
 
 // The from field for messages originating in the GCM client.
 const char kGCMFromField[] = "gcm@android.com";
 
 // MCS status message types.
+// TODO(zea): handle these at the GCMClient layer.

[fgorski, 2013/12/19 23:55:13]

What exactly do you mean?

[Nicolas Zea, 2013/12/20 22:52:56]

System status settings like these should be handled by the GCMClient, rather
than the MCSClient.

 const char kIdleNotification[] = "IdleNotification";
-// TODO(zea): consume the following message types:
 // const char kAlwaysShowOnIdle[] = "ShowAwayOnIdle";
 // const char kPowerNotification[] = "PowerNotification";
 // const char kDataActiveNotification[] = "DataActiveNotification";
 
 // The number of unacked messages to allow before sending a stream ack.
 // Applies to both incoming and outgoing messages.
 // TODO(zea): make this server configurable.
 const int kUnackedMessageBeforeStreamAck = 10;
 
 // The global maximum number of pending messages to have in the send queue.
@@ skipping 48 matching lines @@
     : state_(UNINITIALIZED),
       android_id_(0),
       security_token_(0),
       connection_factory_(connection_factory),
       connection_handler_(NULL),
       last_device_to_server_stream_id_received_(0),
       last_server_to_device_stream_id_received_(0),
       stream_id_out_(0),
       stream_id_in_(0),
       rmq_store_(rmq_path, blocking_task_runner),
-      heartbeat_interval_(
-          base::TimeDelta::FromSeconds(kHeartbeatDefaultSeconds)),
-      heartbeat_timer_(true, true),
       blocking_task_runner_(blocking_task_runner),
       weak_ptr_factory_(this) {
 }
 
 MCSClient::~MCSClient() {
 }
 
 void MCSClient::Initialize(
     const InitializationCompleteCallback& initialization_callback,
     const OnMessageReceivedCallback& message_received_callback,
@@ skipping 11 matching lines @@
       base::Bind(&MCSClient::ResetStateAndBuildLoginRequest,
                  weak_ptr_factory_.GetWeakPtr()),
       base::Bind(&MCSClient::HandlePacketFromWire,
                  weak_ptr_factory_.GetWeakPtr()),
       base::Bind(&MCSClient::MaybeSendMessage,
                  weak_ptr_factory_.GetWeakPtr()));
   connection_handler_ = connection_factory_->GetConnectionHandler();
 }
 
 void MCSClient::Login(uint64 android_id, uint64 security_token) {
-  DCHECK_EQ(state_, LOADED);
   if (android_id != android_id_ && security_token != security_token_) {
     DCHECK(android_id);
     DCHECK(security_token);
     DCHECK(restored_unackeds_server_ids_.empty());
     android_id_ = android_id;
     security_token_ = security_token;
     rmq_store_.SetDeviceCredentials(android_id_,
                                     security_token_,
                                     base::Bind(&MCSClient::OnRMQUpdateFinished,
                                                weak_ptr_factory_.GetWeakPtr()));
@@ skipping 52 matching lines @@
 
 void MCSClient::ResetStateAndBuildLoginRequest(
     mcs_proto::LoginRequest* request) {
   DCHECK(android_id_);
   DCHECK(security_token_);
   stream_id_in_ = 0;
   stream_id_out_ = 1;
   last_device_to_server_stream_id_received_ = 0;
   last_server_to_device_stream_id_received_ = 0;
 
+  heartbeat_manager_.Stop();
+
   // TODO(zea): expire all messages older than their TTL.
 
   // Add any pending acknowledgments to the list of ids.
   for (StreamIdToPersistentIdMap::const_iterator iter =
            unacked_server_ids_.begin();
        iter != unacked_server_ids_.end(); ++iter) {
     restored_unackeds_server_ids_.push_back(iter->second);
   }
   unacked_server_ids_.clear();
 
@@ skipping 22 matching lines @@
   // to RMQ, as all messages that reach this point should already have been
   // saved as necessary.
   while (!to_resend_.empty()) {
     to_send_.push_front(to_resend_.back());
     to_resend_.pop_back();
   }
   DVLOG(1) << "Resetting state, with " << request->received_persistent_id_size()
            << " incoming acks pending, and " << to_send_.size()
            << " pending outgoing messages.";
 
-  heartbeat_timer_.Stop();
-
   state_ = CONNECTING;
 }
 
 void MCSClient::SendHeartbeat() {
   SendMessage(MCSMessage(kHeartbeatPingTag, mcs_proto::HeartbeatPing()),
               false);
 }
 
 void MCSClient::OnRMQLoadFinished(const RMQStore::LoadResult& result) {
   if (!result.success) {
@@ skipping 63 matching lines @@
 
   DVLOG(1) << "Pending output message found, sending.";
   MCSPacketInternal packet = to_send_.front();
   to_send_.pop_front();
   if (!packet->persistent_id.empty())
     to_resend_.push_back(packet);
   SendPacketToWire(packet.get());
 }
 
 void MCSClient::SendPacketToWire(ReliablePacketInfo* packet_info) {
-  // Reset the heartbeat interval.
-  heartbeat_timer_.Reset();
   packet_info->stream_id = ++stream_id_out_;
   DVLOG(1) << "Sending packet of type " << packet_info->protobuf->GetTypeName();
 
   // Set the proper last received stream id to acknowledge received server
   // packets.
   DVLOG(1) << "Setting last stream id received to "
            << stream_id_in_;
   SetLastStreamIdReceived(stream_id_in_,
                           packet_info->protobuf.get());
   if (stream_id_in_ != last_server_to_device_stream_id_received_) {
@@ skipping 93 matching lines @@
            << last_stream_id_received;
 
   if (unacked_server_ids_.size() > 0 &&
       unacked_server_ids_.size() % kUnackedMessageBeforeStreamAck == 0) {
     SendMessage(MCSMessage(kIqStanzaTag,
                            BuildStreamAck().
                                PassAs<const google::protobuf::MessageLite>()),
                 false);
   }
 
+  // The connection is alive, treat this message as a heartbeat ack.
+  heartbeat_manager_.OnHeartbeatAcked();

[fgorski, 2013/12/19 23:55:13]

Will that work if the manager hasn't been started yet? as in, if it is a login
response.

[Nicolas Zea, 2013/12/20 22:52:56]

Yep, OnHeartbeatAcked does nothing if not running.

+
   switch (tag) {
     case kLoginResponseTag: {
       DCHECK_EQ(CONNECTING, state_);
       mcs_proto::LoginResponse* login_response =
           reinterpret_cast<mcs_proto::LoginResponse*>(protobuf.get());
       DVLOG(1) << "Received login response:";
       DVLOG(1) << "  Id: " << login_response->id();
       DVLOG(1) << "  Timestamp: " << login_response->server_timestamp();
-      if (login_response->has_error()) {
+      if (login_response->has_error() && login_response->error().code() != 0) {
         state_ = UNINITIALIZED;
         DVLOG(1) << "  Error code: " << login_response->error().code();
         DVLOG(1) << "  Error message: " << login_response->error().message();
         initialization_callback_.Run(false, 0, 0);
         return;
       }
 
+      if (login_response->has_heartbeat_config()) {
+        heartbeat_manager_.UpdateHeartbeatConfig(
+            login_response->heartbeat_config());
+      }
+
       state_ = CONNECTED;
       stream_id_in_ = 1;  // To account for the login response.
       DCHECK_EQ(1U, stream_id_out_);
 
       // Pass the login response on up.
       base::MessageLoop::current()->PostTask(
           FROM_HERE,
           base::Bind(message_received_callback_,
                      MCSMessage(tag,
                                 protobuf.PassAs<
                                     const google::protobuf::MessageLite>())));
 
       // If there are pending messages, attempt to send one.
       if (!to_send_.empty()) {
         base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(&MCSClient::MaybeSendMessage,
                        weak_ptr_factory_.GetWeakPtr()));
       }
 
-      heartbeat_timer_.Start(FROM_HERE,
-                             heartbeat_interval_,
-                             base::Bind(&MCSClient::SendHeartbeat,
-                                        weak_ptr_factory_.GetWeakPtr()));
+      heartbeat_manager_.Start(
+          base::Bind(&MCSClient::SendHeartbeat,
+                     weak_ptr_factory_.GetWeakPtr()),
+          base::Bind(&MCSClient::OnConnectionResetByHeartbeat,
+                     weak_ptr_factory_.GetWeakPtr()));
       return;
     }
     case kHeartbeatPingTag:
       DCHECK_GE(stream_id_in_, 1U);
       DVLOG(1) << "Received heartbeat ping, sending ack.";
       SendMessage(
           MCSMessage(kHeartbeatAckTag, mcs_proto::HeartbeatAck()), false);
       return;
     case kHeartbeatAckTag:
       DCHECK_GE(stream_id_in_, 1U);
       DVLOG(1) << "Received heartbeat ack.";
-      // TODO(zea): add logic to reconnect if no ack received within a certain
-      // timeout (with backoff).
+      // Do nothing else, all messages act as heartbeat acks.
       return;
     case kCloseTag:
       LOG(ERROR) << "Received close command, resetting connection.";
       state_ = LOADED;
       connection_factory_->SignalConnectionReset();
       return;
     case kIqStanzaTag: {
       DCHECK_GE(stream_id_in_, 1U);
       mcs_proto::IqStanza* iq_stanza =
           reinterpret_cast<mcs_proto::IqStanza*>(protobuf.get());
@@ skipping 134 matching lines @@
            << " acknowledged server messages.";
   rmq_store_.RemoveIncomingMessages(acked_incoming_ids,
                                     base::Bind(&MCSClient::OnRMQUpdateFinished,
                                                weak_ptr_factory_.GetWeakPtr()));
 }
 
 MCSClient::PersistentId MCSClient::GetNextPersistentId() {
   return base::Uint64ToString(base::TimeTicks::Now().ToInternalValue());
 }
 
+void MCSClient::OnConnectionResetByHeartbeat() {
+  connection_factory_->SignalConnectionReset();
+}
+
 } // namespace gcm