[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 56 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> expired_ttl_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_)) {
+      expired_ttl_ids.push_back(iter->first);
+      message_sent_callback_.Run("TTL expired for " + iter->first);
+      delete iter->second;
+      continue;
+    }
+
     ordered_messages[timestamp] = iter->second;
   }
 
+  if (!expired_ttl_ids.empty()) {
+    rmq_store_->RemoveOutgoingMessages(
+        expired_ttl_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);
   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> expired_ttl_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 the TTL was 0 there is no persistent id, so no need to remove the
+      // message from the persistent store.
+      if (!packet->persistent_id.empty())
+        expired_ttl_ids.push_back(packet->persistent_id);
+      message_sent_callback_.Run("TTL expired");
+    }
+  }
+
+  if (!expired_ttl_ids.empty()) {
+    DVLOG(1) << "Connection reset, " << expired_ttl_ids.size()
+             << " messages expired.";
+    rmq_store_->RemoveOutgoingMessages(
+        expired_ttl_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 23 matching lines @@
 
 void MCSClient::HandleMCSDataMesssage(
     scoped_ptr<google::protobuf::MessageLite> protobuf) {
   mcs_proto::DataMessageStanza* data_message =
       reinterpret_cast<mcs_proto::DataMessageStanza*>(protobuf.get());
   // TODO(zea): implement a proper status manager rather than hardcoding these
   // values.
   scoped_ptr<mcs_proto::DataMessageStanza> response(
       new mcs_proto::DataMessageStanza());
   response->set_from(kGCMFromField);
+  response->set_sent(clock_->Now().ToInternalValue() /
+                         base::Time::kMicrosecondsPerSecond);
+  response->set_ttl(0);
   bool send = false;
   for (int i = 0; i < data_message->app_data_size(); ++i) {
     const mcs_proto::AppData& app_data = data_message->app_data(i);
     if (app_data.key() == kIdleNotification) {
       // Tell the MCS server the client is not idle.
       send = true;
       mcs_proto::AppData data;
       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