[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 208 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.
+    mcs_proto::DataMessageStanza* data_message =
+        reinterpret_cast<mcs_proto::DataMessageStanza*>(
+            packet_info->protobuf.get());
+    std::string app_id = data_message->from();

[fgorski, 2014/01/27 17:39:38]

same here

[Nicolas Zea, 2014/01/31 11:58:43]

Done.

+    DCHECK(!app_id.empty());
+    std::string token = data_message->token();
+    if (!token.empty() &&
+        collapse_key_map_.count(app_id) != 0 &&
+        collapse_key_map_[app_id]->count(token) != 0) {
+      ReliablePacketInfo* original_packet =
+          (*collapse_key_map_[app_id])[data_message->token()];
+      DVLOG(1) << "Found matching collapse key, Reusing persistent id of "
+               << original_packet->persistent_id;

[fgorski, 2014/01/27 17:39:38]

Is it possible that you are assigning a persistent ID of a message that has
already been sent and is awaiting ack from the server? Because that would cause
the data carried in the message to be lost.

Did you think about that?

[Nicolas Zea, 2014/01/28 08:15:35]

SendMessage is only invoked for new messages, not for resending. As soon as a
message is sent over the wire (in MaybeSendMessage), we remove it's key from the
collapse key map, whether it gets sent sucessfully/acknowledged or not.

[fgorski, 2014/01/28 18:10:27]

Does that mean we can fill RMQ for the application with collapse key messages
using the same value of the key?

[Nicolas Zea, 2014/01/31 11:58:43]

Currently yes, if all the messages manage to send, but are never acknowledged by
the server. Otherwise we'll have an unbounded set of messages that are pending
acknowledgement.

+      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 (!token.empty()) {
+      if (!collapse_key_map_[app_id].get())
+        collapse_key_map_[app_id] = make_linked_ptr(new TokenMap());
+      (*collapse_key_map_[app_id])[token] = 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());
+    std::string app_id = data_message->from();

[fgorski, 2014/01/27 17:39:38]

ditto: category

[Nicolas Zea, 2014/01/31 11:58:43]

Done.

+    std::string token = data_message->token();
+    if (!token.empty())
+      collapse_key_map_[app_id]->erase(token);
+  }
   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 363 matching lines @@
   const mcs_proto::DataMessageStanza* data_message_stanza =
       reinterpret_cast<const mcs_proto::DataMessageStanza*>(&protobuf);
   message_sent_callback_.Run(
       data_message_stanza->device_user_id(),
       data_message_stanza->category(),
       data_message_stanza->id(),
       status);
 }
 
 } // namespace gcm