[GCM] Add basic collapse key support for upstream

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/metrics/histogram.h"
 #include "base/strings/string_number_conversions.h"
@@ skipping 45 matching lines @@
   for (int i = 0; i < selective_ack.id_size(); ++i) {
     DCHECK(!selective_ack.id(i).empty());
     new_list.push_back(selective_ack.id(i));
   }
   id_list->swap(new_list);
   return true;
 }
 
 }  // namespace
 
+class CollapseKey {
+ public:
+  explicit CollapseKey(const mcs_proto::DataMessageStanza& message);
+  ~CollapseKey();
+
+  // Comparison operator for use in maps.
+  bool operator<(const CollapseKey& right) const;
+
+  // Whether the message had a valid collapse key.
+  bool IsValid() const;
+
+  std::string token() const { return token_; }
+  std::string app_id() const { return app_id_; }
+  int64 device_user_id() const { return device_user_id_; }
+
+ private:
+  const std::string token_;
+  const std::string app_id_;
+  const int64 device_user_id_;
+};
+
+CollapseKey::CollapseKey(const mcs_proto::DataMessageStanza& message)
+    : token_(message.token()),
+      app_id_(message.category()),
+      device_user_id_(message.device_user_id()) {}
+
+CollapseKey::~CollapseKey() {}
+
+bool CollapseKey::IsValid() const {
+  // Device user id is optional, but the application id and token are not.
+  return !token_.empty() && !app_id_.empty();
+}
+
+bool CollapseKey::operator<(const CollapseKey& right) const {
+  if (device_user_id_ != right.device_user_id())
+    return device_user_id_ < right.device_user_id();
+  if (app_id_ != right.app_id())
+    return app_id_ < right.app_id();
+  return token_ < right.token();
+}
+
 struct ReliablePacketInfo {
   ReliablePacketInfo();
   ~ReliablePacketInfo();
 
   // The stream id with which the message was sent.
   uint32 stream_id;
 
   // If reliable delivery was requested, the persistent id of the message.
   std::string persistent_id;
 
@@ skipping 106 matching lines @@
   // 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*>::iterator
            iter = ordered_messages.begin();
        iter != ordered_messages.end(); ++iter) {
     ReliablePacketInfo* packet_info = new ReliablePacketInfo();
     packet_info->protobuf.reset(iter->second);
     packet_info->tag = GetMCSProtoTag(*iter->second);
     packet_info->persistent_id = base::Uint64ToString(iter->first);
     to_send_.push_back(make_linked_ptr(packet_info));
+
+    if (packet_info->tag == kDataMessageStanzaTag) {
+      mcs_proto::DataMessageStanza* data_message =
+          reinterpret_cast<mcs_proto::DataMessageStanza*>(
+              packet_info->protobuf.get());
+      CollapseKey collapse_key(*data_message);
+      if (collapse_key.IsValid())
+        collapse_key_map_[collapse_key] = packet_info;
+    }
   }
 }
 
 void MCSClient::Login(uint64 android_id, uint64 security_token) {
   DCHECK_EQ(state_, LOADED);
   DCHECK(android_id_ == 0 || android_id_ == android_id);
   DCHECK(security_token_ == 0 || security_token_ == security_token);
 
   if (android_id != android_id_ && security_token != security_token_) {
     DCHECK(android_id);
@@ skipping 18 matching lines @@
   if (message.size() > kMaxMessageBytes) {
     NotifyMessageSendStatus(message.GetProtobuf(), MESSAGE_TOO_LARGE);
     return;
   }
 
   scoped_ptr<ReliablePacketInfo> packet_info(new ReliablePacketInfo());
   packet_info->tag = message.tag();
   packet_info->protobuf = message.CloneProtobuf();
 
   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());
-    if (!gcm_store_->AddOutgoingMessage(
-            persistent_id,
-            MCSMessage(message.tag(),
-                       *(packet_info->protobuf)),
-            base::Bind(&MCSClient::OnGCMUpdateFinished,
-                       weak_ptr_factory_.GetWeakPtr()))) {
-      NotifyMessageSendStatus(message.GetProtobuf(),
-                              APP_QUEUE_SIZE_LIMIT_REACHED);
+    DCHECK_EQ(message.tag(), kDataMessageStanzaTag);
+
+    // First check if this message should replace a pending message with the
+    // same collapse key.
+    mcs_proto::DataMessageStanza* data_message =
+        reinterpret_cast<mcs_proto::DataMessageStanza*>(
+            packet_info->protobuf.get());
+    CollapseKey collapse_key(*data_message);
+    if (collapse_key.IsValid() && collapse_key_map_.count(collapse_key) > 0) {
+      ReliablePacketInfo* original_packet = collapse_key_map_[collapse_key];
+      DVLOG(1) << "Found matching collapse key, Reusing persistent id of "
+               << original_packet->persistent_id;
+      original_packet->protobuf = packet_info->protobuf.Pass();
+      SetPersistentId(original_packet->persistent_id,
+                      original_packet->protobuf.get());
+      gcm_store_->OverwriteOutgoingMessage(
+          original_packet->persistent_id,
+          message,
+          base::Bind(&MCSClient::OnGCMUpdateFinished,
+                     weak_ptr_factory_.GetWeakPtr()));
+
+      // The message is already queued, return.
       return;
+    } else {
+      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());
+      if (!gcm_store_->AddOutgoingMessage(
+               persistent_id,
+               MCSMessage(message.tag(), *(packet_info->protobuf)),
+               base::Bind(&MCSClient::OnGCMUpdateFinished,
+                          weak_ptr_factory_.GetWeakPtr()))) {
+        NotifyMessageSendStatus(message.GetProtobuf(),
+                                APP_QUEUE_SIZE_LIMIT_REACHED);
+        return;
+      }
     }
+
+    if (collapse_key.IsValid())
+      collapse_key_map_[collapse_key] = packet_info.get();
   } else if (!connection_factory_->IsEndpointReachable()) {
     DVLOG(1) << "No active connection, dropping message.";
     NotifyMessageSendStatus(message.GetProtobuf(), NO_CONNECTION_ON_ZERO_TTL);
     return;
   }
 
   to_send_.push_back(make_linked_ptr(packet_info.release()));
 
   // Notify that the messages has been succsfully queued for sending.
   // TODO(jianli): We should report QUEUED after writing to GCM store succeeds.
@@ skipping 52 matching lines @@
   // 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();
+    MCSPacketInternal packet = PopMessageForSend();
     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);
       NotifyMessageSendStatus(*packet->protobuf, TTL_EXCEEDED);
     }
@@ skipping 31 matching lines @@
   if (to_send_.empty())
     return;
 
   // 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;
 
-  MCSPacketInternal packet = to_send_.front();
-  to_send_.pop_front();
+  MCSPacketInternal packet = PopMessageForSend();
   if (HasTTLExpired(*packet->protobuf, clock_)) {
     DCHECK(!packet->persistent_id.empty());
     DVLOG(1) << "Dropping expired message " << packet->persistent_id << ".";
     NotifyMessageSendStatus(*packet->protobuf, TTL_EXCEEDED);
     gcm_store_->RemoveOutgoingMessage(
         packet->persistent_id,
         base::Bind(&MCSClient::OnGCMUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
     base::MessageLoop::current()->PostTask(
             FROM_HERE,
@@ skipping 289 matching lines @@
     StreamId device_stream_id = outgoing_packet->stream_id;
     HandleServerConfirmedReceipt(device_stream_id);
 
     to_resend_.pop_front();
   }
 
   // If the acknowledged ids aren't all there, they might be in the to_send_
   // queue (typically when a StreamAck confirms messages as part of a login
   // response).
   for (; iter != id_list.end() && !to_send_.empty(); ++iter) {
-    const MCSPacketInternal& outgoing_packet = to_send_.front();
+    const MCSPacketInternal& outgoing_packet = PopMessageForSend();
     DCHECK_EQ(outgoing_packet->persistent_id, *iter);
     NotifyMessageSendStatus(*outgoing_packet->protobuf, SENT);
 
     // No need to re-acknowledge any server messages this message already
     // acknowledged.
     StreamId device_stream_id = outgoing_packet->stream_id;
     HandleServerConfirmedReceipt(device_stream_id);
-
-    to_send_.pop_front();
   }
 
   DCHECK(iter == id_list.end());
 
   DVLOG(1) << "Server acked " << id_list.size()
            << " messages, " << to_resend_.size() << " remaining unacked.";
   gcm_store_->RemoveOutgoingMessages(
       id_list,
       base::Bind(&MCSClient::OnGCMUpdateFinished,
                  weak_ptr_factory_.GetWeakPtr()));
@@ skipping 47 matching lines @@
       data_message_stanza->device_user_id(),
       data_message_stanza->category(),
       data_message_stanza->id(),
       status);
 }
 
 void MCSClient::SetGCMStoreForTesting(GCMStore* gcm_store) {
   gcm_store_ = gcm_store;
 }
 
+MCSClient::MCSPacketInternal MCSClient::PopMessageForSend() {
+  MCSPacketInternal packet = to_send_.front();
+  to_send_.pop_front();
+
+  if (packet->tag == kDataMessageStanzaTag) {
+    mcs_proto::DataMessageStanza* data_message =
+        reinterpret_cast<mcs_proto::DataMessageStanza*>(packet->protobuf.get());
+    CollapseKey collapse_key(*data_message);
+    if (collapse_key.IsValid())
+      collapse_key_map_.erase(collapse_key);
+  }
+
+  return packet;
+}
+
 } // namespace gcm