[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 {

[jianli, 2014/02/04 22:55:14]

nit: empty line

[Nicolas Zea, 2014/02/12 23:35:46]

Done.

+  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 144 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 collapsed key.

[jianli, 2014/02/04 22:55:14]

nit: collapse

[Nicolas Zea, 2014/02/12 23:35:46]

Done.

+    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()));
   MaybeSendMessage();
 }
 
 void MCSClient::Destroy() {
@@ skipping 101 matching lines @@
 
   // 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();
+  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);
+  }
   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 367 matching lines @@
       data_message_stanza->category(),
       data_message_stanza->id(),
       status);
 }
 
 void MCSClient::SetGCMStoreForTesting(GCMStore* gcm_store) {
   gcm_store_ = gcm_store;
 }
 
 } // namespace gcm