[GCM] Add TTL support

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 "base/time/clock.h"
+#include "base/time/time.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 {
@@ skipping 17 matching lines @@
 // 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.
 const size_t kMaxSendQueueSize = 10 * 1024;
 
 // The maximum message size that can be sent to the server.
 const int kMaxMessageBytes = 4 * 1024;  // 4KB, like the server.
 
+// Maximum amount of time to save an unsent outgoing message for.
+const int kMaxTTLSeconds = 4 * 7 * 24 * 60 * 60;  // 4 weeks.
+
 // Helper for converting a proto persistent id list to a vector of strings.
 bool BuildPersistentIdListFromProto(const google::protobuf::string& bytes,
                                     std::vector<std::string>* id_list) {
   mcs_proto::SelectiveAck selective_ack;
   if (!selective_ack.ParseFromString(bytes))
     return false;
   std::vector<std::string> new_list;
   for (int i = 0; i < selective_ack.id_size(); ++i) {
     DCHECK(!selective_ack.id(i).empty());
     new_list.push_back(selective_ack.id(i));
@@ skipping 19 matching lines @@
 
   // The protobuf of the message itself.
   MCSProto protobuf;
 };
 
 ReliablePacketInfo::ReliablePacketInfo()
   : stream_id(0), tag(0) {
 }
 ReliablePacketInfo::~ReliablePacketInfo() {}
 
-MCSClient::MCSClient(ConnectionFactory* connection_factory, RMQStore* rmq_store)
-    : state_(UNINITIALIZED),
+MCSClient::MCSClient(base::Clock* clock,
+                     ConnectionFactory* connection_factory,
+                     RMQStore* rmq_store)
+    : clock_(clock),
+      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_store),
       weak_ptr_factory_(this) {
@@ skipping 44 matching lines @@
 
   DVLOG(1) << "RMQ Load finished with " << load_result.incoming_messages.size()
            << " incoming acks pending and "
            << load_result.outgoing_messages.size()
            << " outgoing messages pending.";
 
   restored_unackeds_server_ids_ = load_result.incoming_messages;
 
   // First go through and order the outgoing messages by recency.
   std::map<uint64, google::protobuf::MessageLite*> ordered_messages;
+  std::vector<PersistentId> dropped_ids;
   for (std::map<PersistentId, google::protobuf::MessageLite*>::const_iterator
            iter = load_result.outgoing_messages.begin();
        iter != load_result.outgoing_messages.end(); ++iter) {
     uint64 timestamp = 0;
     if (!base::StringToUint64(iter->first, &timestamp)) {
       LOG(ERROR) << "Invalid restored message.";
       return;
     }
+
+    // Check if the TTL has expired for this message.
+    if (HasTTLExpired(*iter->second, clock_)) {
+      dropped_ids.push_back(iter->first);
+      message_sent_callback_.Run("TTL expired for " + iter->first);
+      delete iter->second;
+      continue;
+    }
+
     ordered_messages[timestamp] = iter->second;
   }
 
+  if (!dropped_ids.empty()) {
+    rmq_store_->RemoveOutgoingMessages(
+        dropped_ids,
+        base::Bind(&MCSClient::OnRMQUpdateFinished,
+                   weak_ptr_factory_.GetWeakPtr()));
+  }
+
   // Now go through and add the outgoing messages to the send queue in their
   // appropriate order (oldest at front, most recent at back).
   for (std::map<uint64, google::protobuf::MessageLite*>::const_iterator
            iter = ordered_messages.begin();
        iter != ordered_messages.end(); ++iter) {
     ReliablePacketInfo* packet_info = new ReliablePacketInfo();
     packet_info->protobuf.reset(iter->second);
     packet_info->persistent_id = base::Uint64ToString(iter->first);
     to_send_.push_back(make_linked_ptr(packet_info));
   }
@@ skipping 14 matching lines @@
         android_id_,
         security_token_,
         base::Bind(&MCSClient::OnRMQUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
   }
 
   state_ = CONNECTING;
   connection_factory_->Connect();
 }
 
-void MCSClient::SendMessage(const MCSMessage& message, bool use_rmq) {
-  DCHECK_EQ(state_, CONNECTED);
+void MCSClient::SendMessage(const MCSMessage& message) {
+  int ttl = GetTTL(message.GetProtobuf());
+  DCHECK_GE(ttl, 0);
+  DCHECK_LE(ttl, kMaxTTLSeconds);
   if (to_send_.size() > kMaxSendQueueSize) {
-    base::MessageLoop::current()->PostTask(
-        FROM_HERE,
-        base::Bind(message_sent_callback_, "Message queue full."));
+    message_sent_callback_.Run("Message queue full.");
     return;
   }
   if (message.size() > kMaxMessageBytes) {
-    base::MessageLoop::current()->PostTask(
-        FROM_HERE,
-        base::Bind(message_sent_callback_, "Message too large."));
+    message_sent_callback_.Run("Message too large.");
     return;
   }
 
   ReliablePacketInfo* packet_info = new ReliablePacketInfo();
   packet_info->protobuf = message.CloneProtobuf();
 
-  if (use_rmq) {
+  if (ttl > 0) {
     PersistentId persistent_id = GetNextPersistentId();
     DVLOG(1) << "Setting persistent id to " << persistent_id;
     packet_info->persistent_id = persistent_id;
     SetPersistentId(persistent_id,
                     packet_info->protobuf.get());
     rmq_store_->AddOutgoingMessage(persistent_id,
                                    MCSMessage(message.tag(),
                                               *(packet_info->protobuf)),
                                    base::Bind(&MCSClient::OnRMQUpdateFinished,
                                               weak_ptr_factory_.GetWeakPtr()));
-  } else {
-    // Check that there is an active connection to the endpoint.
-    if (!connection_handler_->CanSendMessage()) {
-      base::MessageLoop::current()->PostTask(
-          FROM_HERE,
-          base::Bind(message_sent_callback_, "Unable to reach endpoint"));
-      return;
-    }
+  } else if (!connection_factory_->IsEndpointReachable()) {
+    DVLOG(1) << "No active connection, dropping message.";
+    message_sent_callback_.Run("TTL expired");
+    return;
   }
   to_send_.push_back(make_linked_ptr(packet_info));
   MaybeSendMessage();
 }
 
 void MCSClient::Destroy() {
   rmq_store_->Destroy(base::Bind(&MCSClient::OnRMQUpdateFinished,
                                  weak_ptr_factory_.GetWeakPtr()));
 }
 
 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();
 
   // Any acknowledged server ids which have not been confirmed by the server
   // are treated like unacknowledged ids.
@@ skipping 16 matching lines @@
   acked_server_ids_[stream_id_out_] = restored_unackeds_server_ids_;
   restored_unackeds_server_ids_.clear();
 
   // Push all unacknowledged messages to front of send queue. No need to save
   // 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();
   }
+
+  // Drop all TTL == 0 or expired TTL messages from the queue.
+  std::deque<MCSPacketInternal> new_to_send;
+  std::vector<PersistentId> dropped_ids;
+  while (!to_send_.empty()) {
+    MCSPacketInternal packet = to_send_.front();
+    to_send_.pop_front();
+    if (GetTTL(*packet->protobuf) > 0 &&
+        !HasTTLExpired(*packet->protobuf, clock_)) {
+      new_to_send.push_back(packet);
+    } else {
+      if (!packet->persistent_id.empty())
+        dropped_ids.push_back(packet->persistent_id);

[jianli, 2014/01/02 18:52:58]

If TTL is 0, we do not give a persisten ID as in SendMessage. It seems that
dropped_ids will not be filled with TTL==0 message.

[Nicolas Zea, 2014/01/02 21:39:08]

Right, but those ids weren't stored in the RMQStore anyways, so it's okay to
skip them. dropped_ids is just for removing messages from the rmq store.

[jianli, 2014/01/02 21:51:31]

But the comment at line 311 says: Drop all TTL == 0 or ...

Also you have the check for "GetTTL(*packet->protobuf) > 0". All theses make the
reader think we're handling TTL == 0 case here.

[Nicolas Zea, 2014/01/02 21:56:13]

We are handling the case by removing those messages from the send queue. But,
they don't need to be removed from the persistent storage because they were
never added there. Do you think the name dropped_ids should be different to make
that distinction clearer?

+      message_sent_callback_.Run("TTL expired");
+    }
+  }
+
+  if (!dropped_ids.empty()) {
+    DVLOG(1) << "Connection reset, " << dropped_ids.size()
+             << " messages expired.";
+    rmq_store_->RemoveOutgoingMessages(
+        dropped_ids,
+        base::Bind(&MCSClient::OnRMQUpdateFinished,
+                   weak_ptr_factory_.GetWeakPtr()));
+  }
+
+  to_send_.swap(new_to_send);
+
   DVLOG(1) << "Resetting state, with " << request->received_persistent_id_size()
            << " incoming acks pending, and " << to_send_.size()
            << " pending outgoing messages.";
 
   state_ = CONNECTING;
 }
 
 void MCSClient::SendHeartbeat() {
-  SendMessage(MCSMessage(kHeartbeatPingTag, mcs_proto::HeartbeatPing()),
-              false);
+  SendMessage(MCSMessage(kHeartbeatPingTag, mcs_proto::HeartbeatPing()));
 }
 
 void MCSClient::OnRMQUpdateFinished(bool success) {
   LOG_IF(ERROR, !success) << "RMQ Update failed!";
   // TODO(zea): Rebuild the store from scratch in case of persistence failure?
 }
 
 void MCSClient::MaybeSendMessage() {
   if (to_send_.empty())
     return;
 
-  if (!connection_handler_->CanSendMessage())
+  // If the connection has been reset, do nothing. On reconnection
+  // MaybeSendMessage will be automatically invoked again.
+  // TODO(zea): consider doing TTL expiration at connection reset time, rather
+  // than reconnect time.
+  if (!connection_factory_->IsEndpointReachable())
     return;
 
-  // TODO(zea): drop messages older than their TTL.
-
-  DVLOG(1) << "Pending output message found, sending.";
   MCSPacketInternal packet = to_send_.front();
   to_send_.pop_front();
+  if (HasTTLExpired(*packet->protobuf, clock_)) {
+    DCHECK(packet->persistent_id.empty());
+    DVLOG(1) << "Dropping expired message " << packet->persistent_id << ".";
+    message_sent_callback_.Run("TTL expired for " + packet->persistent_id);
+    rmq_store_->RemoveOutgoingMessage(
+        packet->persistent_id,
+        base::Bind(&MCSClient::OnRMQUpdateFinished,
+                   weak_ptr_factory_.GetWeakPtr()));
+    base::MessageLoop::current()->PostTask(
+            FROM_HERE,
+            base::Bind(&MCSClient::MaybeSendMessage,
+                       weak_ptr_factory_.GetWeakPtr()));
+    return;
+  }
+  DVLOG(1) << "Pending output message found, sending.";
   if (!packet->persistent_id.empty())
     to_resend_.push_back(packet);
   SendPacketToWire(packet.get());
 }
 
 void MCSClient::SendPacketToWire(ReliablePacketInfo* packet_info) {
   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
@@ skipping 40 matching lines @@
       data.set_key(kIdleNotification);
       data.set_value("false");
       response->add_app_data()->CopyFrom(data);
       response->set_category(kMCSCategory);
     }
   }
 
   if (send) {
     SendMessage(
         MCSMessage(kDataMessageStanzaTag,
-                   response.PassAs<const google::protobuf::MessageLite>()),
-        false);
+                   response.PassAs<const google::protobuf::MessageLite>()));
   }
 }
 
 void MCSClient::HandlePacketFromWire(
     scoped_ptr<google::protobuf::MessageLite> protobuf) {
   if (!protobuf.get())
     return;
   uint8 tag = GetMCSProtoTag(*protobuf);
   PersistentId persistent_id = GetPersistentId(*protobuf);
   StreamId last_stream_id_received = GetLastStreamIdReceived(*protobuf);
@@ skipping 33 matching lines @@
   DVLOG(1) << "Received message of type " << protobuf->GetTypeName()
            << " with persistent id "
            << (persistent_id.empty() ? "NULL" : persistent_id)
            << ", stream id " << stream_id_in_ << " and last stream id received "
            << 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);
+                               PassAs<const google::protobuf::MessageLite>()));
   }
 
   // The connection is alive, treat this message as a heartbeat ack.
   heartbeat_manager_.OnHeartbeatAcked();
 
   switch (tag) {
     case kLoginResponseTag: {
       DCHECK_EQ(CONNECTING, state_);
       mcs_proto::LoginResponse* login_response =
           reinterpret_cast<mcs_proto::LoginResponse*>(protobuf.get());
@@ skipping 37 matching lines @@
           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);
+          MCSMessage(kHeartbeatAckTag, mcs_proto::HeartbeatAck()));
       return;
     case kHeartbeatAckTag:
       DCHECK_GE(stream_id_in_, 1U);
       DVLOG(1) << "Received heartbeat ack.";
       // Do nothing else, all messages act as heartbeat acks.
       return;
     case kCloseTag:
       LOG(ERROR) << "Received close command, resetting connection.";
       state_ = LOADED;
       connection_factory_->SignalConnectionReset();
@@ skipping 147 matching lines @@
 
 MCSClient::PersistentId MCSClient::GetNextPersistentId() {
   return base::Uint64ToString(base::TimeTicks::Now().ToInternalValue());
 }
 
 void MCSClient::OnConnectionResetByHeartbeat() {
   connection_factory_->SignalConnectionReset();
 }
 
 } // namespace gcm