[GCM] Add basic collapse key support for upstream

google_apis/gcm/engine/gcm_store_impl.cc

 // Copyright 2014 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/gcm_store_impl.h"
 
 #include "base/basictypes.h"
 #include "base/bind.h"
 #include "base/callback.h"
 #include "base/file_util.h"
@@ skipping 354 matching lines @@
     std::string outgoing_message;
     std::string key = MakeOutgoingKey(*iter);
     s = db_->Get(read_options,
                  MakeSlice(key),
                  &outgoing_message);
     if (!s.ok())
       break;
     mcs_proto::DataMessageStanza data_message;
     // Skip the initial tag byte and parse the rest to extract the message.
     if (data_message.ParseFromString(outgoing_message.substr(1))) {
-      DCHECK(!data_message.from().empty());
-      if (removed_message_counts.count(data_message.from()) != 0)
-        removed_message_counts[data_message.from()]++;
+      DCHECK(!data_message.category().empty());
+      if (removed_message_counts.count(data_message.category()) != 0)
+        removed_message_counts[data_message.category()]++;
       else
-        removed_message_counts[data_message.from()] = 1;
+        removed_message_counts[data_message.category()] = 1;
     }
     DVLOG(1) << "Removing outgoing message with id " << *iter;
     s = db_->Delete(write_options, MakeSlice(key));
     if (!s.ok())
       break;
   }
   if (s.ok()) {
     foreground_task_runner_->PostTask(FROM_HERE,
                                       base::Bind(callback,
                                                  true,
@@ skipping 291 matching lines @@
                  backend_,
                  persistent_ids,
                  callback));
 }
 
 bool GCMStoreImpl::AddOutgoingMessage(const std::string& persistent_id,
                                       const MCSMessage& message,
                                       const UpdateCallback& callback) {
   DCHECK_EQ(message.tag(), kDataMessageStanzaTag);
   std::string app_id = reinterpret_cast<const mcs_proto::DataMessageStanza*>(
-      &message.GetProtobuf())->from();
+                           &message.GetProtobuf())->category();
   DCHECK(!app_id.empty());
   if (app_message_counts_.count(app_id) == 0)
     app_message_counts_[app_id] = 0;
   if (app_message_counts_[app_id] < kMessagesPerAppLimit) {
     app_message_counts_[app_id]++;
 
     blocking_task_runner_->PostTask(
         FROM_HERE,
         base::Bind(&GCMStoreImpl::Backend::AddOutgoingMessage,
                    backend_,
                    persistent_id,
                    message,
                    base::Bind(&GCMStoreImpl::AddOutgoingMessageContinuation,
                               weak_ptr_factory_.GetWeakPtr(),
                               callback,
                               app_id)));
     return true;
   }
   return false;
 }
 
+void GCMStoreImpl::OverwriteOutgoingMessage(const std::string& persistent_id,
+                                            const MCSMessage& message,
+                                            const UpdateCallback& callback) {
+  DCHECK_EQ(message.tag(), kDataMessageStanzaTag);
+  std::string app_id = reinterpret_cast<const mcs_proto::DataMessageStanza*>(
+                           &message.GetProtobuf())->category();
+  DCHECK(!app_id.empty());
+  // There should already be pending messages for this app.
+  DCHECK(app_message_counts_.count(app_id));
+  // TODO(zea): consider verifying the specific message already exists.
+  blocking_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&GCMStoreImpl::Backend::AddOutgoingMessage,
+                 backend_,
+                 persistent_id,
+                 message,
+                 callback));
+}
+
 void GCMStoreImpl::RemoveOutgoingMessage(const std::string& persistent_id,
                                          const UpdateCallback& callback) {
   blocking_task_runner_->PostTask(
       FROM_HERE,
       base::Bind(&GCMStoreImpl::Backend::RemoveOutgoingMessages,
                  backend_,
                  PersistentIdList(1, persistent_id),
                  base::Bind(&GCMStoreImpl::RemoveOutgoingMessagesContinuation,
                             weak_ptr_factory_.GetWeakPtr(),
                             callback)));
@@ skipping 49 matching lines @@
   if (!result->success) {
     callback.Run(result.Pass());
     return;
   }
   int num_throttled_apps = 0;
   for (OutgoingMessageMap::const_iterator
            iter = result->outgoing_messages.begin();
        iter != result->outgoing_messages.end(); ++iter) {
     const mcs_proto::DataMessageStanza* data_message =
         reinterpret_cast<mcs_proto::DataMessageStanza*>(iter->second.get());
-    DCHECK(!data_message->from().empty());
-    if (app_message_counts_.count(data_message->from()) == 0)
-      app_message_counts_[data_message->from()] = 1;
+    DCHECK(!data_message->category().empty());
+    if (app_message_counts_.count(data_message->category()) == 0)
+      app_message_counts_[data_message->category()] = 1;
     else
-      app_message_counts_[data_message->from()]++;
-    if (app_message_counts_[data_message->from()] == kMessagesPerAppLimit)
+      app_message_counts_[data_message->category()]++;
+    if (app_message_counts_[data_message->category()] == kMessagesPerAppLimit)
       num_throttled_apps++;
   }
   UMA_HISTOGRAM_COUNTS("GCM.NumThrottledApps", num_throttled_apps);
   callback.Run(result.Pass());
 }
 
 void GCMStoreImpl::AddOutgoingMessageContinuation(
     const UpdateCallback& callback,
     const std::string& app_id,
     bool success) {
@@ skipping 16 matching lines @@
            removed_message_counts.begin();
        iter != removed_message_counts.end(); ++iter) {
     DCHECK_NE(app_message_counts_.count(iter->first), 0U);
     app_message_counts_[iter->first] -= iter->second;
     DCHECK_GE(app_message_counts_[iter->first], 0);
   }
   callback.Run(true);
 }
 
 }  // namespace gcm