[GCM] Add basic collapse key support for upstream

google_apis/gcm/engine/mcs_client_unittest.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/files/scoped_temp_dir.h"
 #include "base/message_loop/message_loop.h"
 #include "base/run_loop.h"
 #include "base/strings/string_number_conversions.h"
@@ skipping 25 matching lines @@
 const int kTTLValue = 5 * 60;  // 5 minutes.
 
 // Helper for building arbitrary data messages.
 MCSMessage BuildDataMessage(const std::string& from,
                             const std::string& category,
                             const std::string& message_id,
                             int last_stream_id_received,
                             const std::string& persistent_id,
                             int ttl,
                             uint64 sent,
-                            int queued) {
+                            int queued,
+                            const std::string& token) {
   mcs_proto::DataMessageStanza data_message;
   data_message.set_id(message_id);
   data_message.set_from(from);
   data_message.set_category(category);
   data_message.set_last_stream_id_received(last_stream_id_received);
   if (!persistent_id.empty())
     data_message.set_persistent_id(persistent_id);
   data_message.set_ttl(ttl);
   data_message.set_sent(sent);
   data_message.set_queued(queued);
+  data_message.set_token(token);
   return MCSMessage(kDataMessageStanzaTag, data_message);
 }
 
 // MCSClient with overriden exposed persistent id logic.
 class TestMCSClient : public MCSClient {
  public:
   TestMCSClient(base::Clock* clock,
                 ConnectionFactory* connection_factory,
                 GCMStore* gcm_store)
     : MCSClient(clock, connection_factory, gcm_store),
@@ skipping 212 matching lines @@
   EXPECT_FALSE(connection_factory()->IsEndpointReachable());
   EXPECT_FALSE(init_success());
   EXPECT_FALSE(received_message());
 }
 
 // Send a message without RMQ support.
 TEST_F(MCSClientTest, SendMessageNoRMQ) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
-  MCSMessage message(BuildDataMessage("from", "category", "X", 1, "", 0, 1, 0));
+  MCSMessage message(
+      BuildDataMessage("from",
+                       "category",
+                       "X",
+                       1,
+                       "",
+                       0,
+                       1,
+                       0,
+                       ""));
   GetFakeHandler()->ExpectOutgoingMessage(message);
   mcs_client()->SendMessage(message);
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Send a message without RMQ support while disconnected. Message send should
 // fail immediately, invoking callback.
 TEST_F(MCSClientTest, SendMessageNoRMQWhileDisconnected) {
   BuildMCSClient();
   InitializeClient();
 
   EXPECT_TRUE(sent_message_id().empty());
-  MCSMessage message(BuildDataMessage("from", "category", "X", 1, "", 0, 1, 0));
+  MCSMessage message(
+      BuildDataMessage("from",
+                       "category",
+                       "X",
+                       1,
+                       "",
+                       0,
+                       1,
+                       0,
+                       ""));
   mcs_client()->SendMessage(message);
 
   // Message sent callback should be invoked, but no message should actually
   // be sent.
   EXPECT_EQ("X", sent_message_id());
   EXPECT_EQ(MCSClient::NO_CONNECTION_ON_ZERO_TTL, message_send_status());
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Send a message with RMQ support.
 TEST_F(MCSClientTest, SendMessageRMQ) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
-  MCSMessage message(
-      BuildDataMessage("from", "category", "X", 1, "1", kTTLValue, 1, 0));
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "X",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      ""));
   GetFakeHandler()->ExpectOutgoingMessage(message);
   mcs_client()->SendMessage(message);
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Send a message with RMQ support while disconnected. On reconnect, the message
 // should be resent.
 TEST_F(MCSClientTest, SendMessageRMQWhileDisconnected) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
   GetFakeHandler()->set_fail_send(true);
-  MCSMessage message(
-      BuildDataMessage("from", "category", "X", 1, "1", kTTLValue, 1, 0));
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "X",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      ""));
 
   // The initial (failed) send.
   GetFakeHandler()->ExpectOutgoingMessage(message);
   // The login request.
   GetFakeHandler()->ExpectOutgoingMessage(
       MCSMessage(kLoginRequestTag,
                  BuildLoginRequest(kAndroidId, kSecurityToken).
                      PassAs<const google::protobuf::MessageLite>()));
   // The second (re)send.
   MCSMessage message2(BuildDataMessage("from",
                                        "category",
                                        "X",
                                        1,
                                        "1",
                                        kTTLValue,
                                        1,
-                                       kTTLValue - 1));
+                                       kTTLValue - 1,
+                                       ""));
   GetFakeHandler()->ExpectOutgoingMessage(message2);
   mcs_client()->SendMessage(message);
   EXPECT_FALSE(GetFakeHandler()->AllOutgoingMessagesReceived());
   GetFakeHandler()->set_fail_send(false);
   clock()->Advance(base::TimeDelta::FromSeconds(kTTLValue - 1));
   connection_factory()->Connect();
   WaitForMCSEvent();  // Wait for the login to finish.
   PumpLoop();         // Wait for the send to happen.
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Send a message with RMQ support without receiving an acknowledgement. On
 // restart the message should be resent.
 TEST_F(MCSClientTest, SendMessageRMQOnRestart) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
   GetFakeHandler()->set_fail_send(true);
-  MCSMessage message(
-      BuildDataMessage("from", "category", "X", 1, "1", kTTLValue, 1, 0));
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "X",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      ""));
 
   // The initial (failed) send.
   GetFakeHandler()->ExpectOutgoingMessage(message);
   GetFakeHandler()->set_fail_send(false);
   mcs_client()->SendMessage(message);
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Rebuild the client, which should resend the old message.
   StoreCredentials();
   BuildMCSClient();
   InitializeClient();
 
   clock()->Advance(base::TimeDelta::FromSeconds(kTTLValue - 1));
   MCSMessage message2(BuildDataMessage("from",
                                        "category",
                                        "X",
                                        1,
                                        "1",
                                        kTTLValue,
                                        1,
-                                       kTTLValue - 1));
+                                       kTTLValue - 1,
+                                       ""));
   LoginClient(std::vector<std::string>());
   GetFakeHandler()->ExpectOutgoingMessage(message2);
   PumpLoop();
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Send messages with RMQ support, followed by receiving a stream ack. On
 // restart nothing should be recent.
 TEST_F(MCSClientTest, SendMessageRMQWithStreamAck) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // Send some messages.
   for (int i = 1; i <= kMessageBatchSize; ++i) {
-    MCSMessage message(
-        BuildDataMessage("from",
-                         "category",
-                         "X",
-                         1,
-                         base::IntToString(i),
-                         kTTLValue,
-                         1,
-                         0));
+    MCSMessage message(BuildDataMessage("from",
+                                        "category",
+                                        "X",
+                                        1,
+                                        base::IntToString(i),
+                                        kTTLValue,
+                                        1,
+                                        0,
+                                        ""));
     GetFakeHandler()->ExpectOutgoingMessage(message);
     mcs_client()->SendMessage(message);
   }
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Receive the ack.
   scoped_ptr<mcs_proto::IqStanza> ack = BuildStreamAck();
   ack->set_last_stream_id_received(kMessageBatchSize + 1);
   GetFakeHandler()->ReceiveMessage(
       MCSMessage(kIqStanzaTag,
@@ skipping 12 matching lines @@
 // the login response. No messages should be resent.
 TEST_F(MCSClientTest, SendMessageRMQAckOnReconnect) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // Send some messages.
   std::vector<std::string> id_list;
   for (int i = 1; i <= kMessageBatchSize; ++i) {
     id_list.push_back(base::IntToString(i));
-        MCSMessage message(
-            BuildDataMessage("from",
-                             "category",
-                             id_list.back(),
-                             1,
-                             id_list.back(),
-                             kTTLValue,
-                             1,
-                             0));
+        MCSMessage message(BuildDataMessage("from",
+                                            "category",
+                                            id_list.back(),
+                                            1,
+                                            id_list.back(),
+                                            kTTLValue,
+                                            1,
+                                            0,
+                                            ""));
     GetFakeHandler()->ExpectOutgoingMessage(message);
     mcs_client()->SendMessage(message);
   }
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Rebuild the client, and receive an acknowledgment for the messages as
   // part of the login response.
   StoreCredentials();
   BuildMCSClient();
   InitializeClient();
@@ skipping 10 matching lines @@
 // be resent.
 TEST_F(MCSClientTest, SendMessageRMQPartialAckOnReconnect) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // Send some messages.
   std::vector<std::string> id_list;
   for (int i = 1; i <= kMessageBatchSize; ++i) {
     id_list.push_back(base::IntToString(i));
-    MCSMessage message(
-        BuildDataMessage("from",
-                         "category",
-                         id_list.back(),
-                         1,
-                         id_list.back(),
-                         kTTLValue,
-                         1,
-                         0));
+    MCSMessage message(BuildDataMessage("from",
+                                        "category",
+                                        id_list.back(),
+                                        1,
+                                        id_list.back(),
+                                        kTTLValue,
+                                        1,
+                                        0,
+                                        ""));
     GetFakeHandler()->ExpectOutgoingMessage(message);
     mcs_client()->SendMessage(message);
   }
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Rebuild the client, and receive an acknowledgment for the messages as
   // part of the login response.
   StoreCredentials();
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   std::vector<std::string> acked_ids, remaining_ids;
   acked_ids.insert(acked_ids.end(),
                    id_list.begin(),
                    id_list.begin() + kMessageBatchSize / 2);
   remaining_ids.insert(remaining_ids.end(),
                        id_list.begin() + kMessageBatchSize / 2,
                        id_list.end());
   for (int i = 1; i <= kMessageBatchSize / 2; ++i) {
     MCSMessage message(
         BuildDataMessage("from",
                          "category",
                          remaining_ids[i - 1],
                          2,
                          remaining_ids[i - 1],
                          kTTLValue,
-                         1, 0));
+                         1, 0,
+                         ""));
     GetFakeHandler()->ExpectOutgoingMessage(message);
   }
   scoped_ptr<mcs_proto::IqStanza> ack(BuildSelectiveAck(acked_ids));
   GetFakeHandler()->ReceiveMessage(
       MCSMessage(kIqStanzaTag,
                  ack.PassAs<const google::protobuf::MessageLite>()));
   WaitForMCSEvent();
   PumpLoop();
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Receive some messages. On restart, the login request should contain the
 // appropriate acknowledged ids.
 TEST_F(MCSClientTest, AckOnLogin) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // Receive some messages.
   std::vector<std::string> id_list;
   for (int i = 1; i <= kMessageBatchSize; ++i) {
     id_list.push_back(base::IntToString(i));
-    MCSMessage message(
-        BuildDataMessage("from",
-                         "category",
-                         "X",
-                         1,
-                         id_list.back(),
-                         kTTLValue,
-                         1,
-                         0));
+    MCSMessage message(BuildDataMessage("from",
+                                        "category",
+                                        "X",
+                                        1,
+                                        id_list.back(),
+                                        kTTLValue,
+                                        1,
+                                        0,
+                                        ""));
     GetFakeHandler()->ReceiveMessage(message);
     WaitForMCSEvent();
     PumpLoop();
   }
 
   // Restart the client.
   StoreCredentials();
   BuildMCSClient();
   InitializeClient();
   LoginClient(id_list);
 }
 
 // Receive some messages. On the next send, the outgoing message should contain
 // the appropriate last stream id received field to ack the received messages.
 TEST_F(MCSClientTest, AckOnSend) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // Receive some messages.
   std::vector<std::string> id_list;
   for (int i = 1; i <= kMessageBatchSize; ++i) {
     id_list.push_back(base::IntToString(i));
-    MCSMessage message(
-        BuildDataMessage("from",
-                         "category",
-                         id_list.back(),
-                         1,
-                         id_list.back(),
-                         kTTLValue,
-                         1,
-                         0));
+    MCSMessage message(BuildDataMessage("from",
+                                        "category",
+                                        id_list.back(),
+                                        1,
+                                        id_list.back(),
+                                        kTTLValue,
+                                        1,
+                                        0,
+                                        ""));
     GetFakeHandler()->ReceiveMessage(message);
     PumpLoop();
   }
 
   // Trigger a message send, which should acknowledge via stream ack.
   MCSMessage message(
       BuildDataMessage("from",
                        "category",
                        "X",
                        kMessageBatchSize + 1,
                        "1",
                        kTTLValue,
-                       1, 0));
+                       1, 0,
+                       ""));
   GetFakeHandler()->ExpectOutgoingMessage(message);
   mcs_client()->SendMessage(message);
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // Receive the ack limit in messages, which should trigger an automatic
 // stream ack. Receive a heartbeat to confirm the ack.
 TEST_F(MCSClientTest, AckWhenLimitReachedWithHeartbeat) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
 
   // The stream ack.
   scoped_ptr<mcs_proto::IqStanza> ack = BuildStreamAck();
   ack->set_last_stream_id_received(kAckLimitSize + 1);
   GetFakeHandler()->ExpectOutgoingMessage(
       MCSMessage(kIqStanzaTag,
                  ack.PassAs<const google::protobuf::MessageLite>()));
 
   // Receive some messages.
   std::vector<std::string> id_list;
   for (int i = 1; i <= kAckLimitSize; ++i) {
     id_list.push_back(base::IntToString(i));
-    MCSMessage message(
-        BuildDataMessage("from",
-                         "category",
-                         id_list.back(),
-                         1,
-                         id_list.back(),
-                         kTTLValue,
-                         1,
-                         0));
+    MCSMessage message(BuildDataMessage("from",
+                                        "category",
+                                        id_list.back(),
+                                        1,
+                                        id_list.back(),
+                                        kTTLValue,
+                                        1,
+                                        0,
+                                        ""));
     GetFakeHandler()->ReceiveMessage(message);
     WaitForMCSEvent();
     PumpLoop();
   }
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Receive a heartbeat confirming the ack (and receive the heartbeat ack).
   scoped_ptr<mcs_proto::HeartbeatPing> heartbeat(
       new mcs_proto::HeartbeatPing());
   heartbeat->set_last_stream_id_received(2);
@@ skipping 18 matching lines @@
   LoginClient(std::vector<std::string>());
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 // If a message's TTL has expired by the time it reaches the front of the send
 // queue, it should be dropped.
 TEST_F(MCSClientTest, ExpiredTTLOnSend) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
-  MCSMessage message(
-      BuildDataMessage("from", "category", "X", 1, "1", kTTLValue, 1, 0));
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "X",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      ""));
 
   // Advance time to after the TTL.
   clock()->Advance(base::TimeDelta::FromSeconds(kTTLValue + 2));
   EXPECT_TRUE(sent_message_id().empty());
   mcs_client()->SendMessage(message);
 
   // No messages should be sent, but the callback should still be invoked.
   EXPECT_EQ("X", sent_message_id());
   EXPECT_EQ(MCSClient::TTL_EXCEEDED, message_send_status());
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
 TEST_F(MCSClientTest, ExpiredTTLOnRestart) {
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
   GetFakeHandler()->set_fail_send(true);
-  MCSMessage message(
-      BuildDataMessage("from", "category", "X", 1, "1", kTTLValue, 1, 0));
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "X",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      ""));
 
   // The initial (failed) send.
   GetFakeHandler()->ExpectOutgoingMessage(message);
   GetFakeHandler()->set_fail_send(false);
   mcs_client()->SendMessage(message);
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 
   // Move the clock forward and rebuild the client, which should fail the
   // message send on restart.
   clock()->Advance(base::TimeDelta::FromSeconds(kTTLValue + 2));
   StoreCredentials();
   BuildMCSClient();
   InitializeClient();
   LoginClient(std::vector<std::string>());
   PumpLoop();
   EXPECT_EQ("X", sent_message_id());
   EXPECT_EQ(MCSClient::TTL_EXCEEDED, message_send_status());
   EXPECT_TRUE(GetFakeHandler()->AllOutgoingMessagesReceived());
 }
 
+// Sending two messages with the same collapse key and same app id while
+// disconnected should only send the latter of the two on reconnection.
+TEST_F(MCSClientTest, CollapseKeysSameApp) {
+  BuildMCSClient();
+  InitializeClient();
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "message id 1",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      "token"));
+  mcs_client()->SendMessage(message);

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

Speaking to the earlier comment. If connection breaks (assuming client was
logged in, and MaybeSendMessage was fired) after sending this, the message is
still in storage, with a collapse_key = "token", but we cannot overwrite it with
the next one, as it is no longer in the map. Is that correct?

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

Correct, they're treated as completely independent messages. As far as I
understand they need to be acknowledged separately (they have different
persistent id and stream id after all).

+
+  MCSMessage message2(BuildDataMessage("from",
+                                       "category",
+                                       "message id 2",
+                                       1,
+                                       "1",
+                                       kTTLValue,
+                                       1,
+                                       0,
+                                       "token"));
+  mcs_client()->SendMessage(message2);
+
+  LoginClient(std::vector<std::string>());
+  GetFakeHandler()->ExpectOutgoingMessage(message2);
+  PumpLoop();
+}
+
+// Sending two messages with the same collapse key and different app id while
+// disconnected should not perform any collapsing.
+TEST_F(MCSClientTest, CollapseKeysDifferentApp) {
+  BuildMCSClient();
+  InitializeClient();
+  MCSMessage message(BuildDataMessage("from",
+                                      "category",
+                                      "message id 1",
+                                      1,
+                                      "1",
+                                      kTTLValue,
+                                      1,
+                                      0,
+                                      "token"));
+  mcs_client()->SendMessage(message);
+
+  MCSMessage message2(BuildDataMessage("from a differnt app",
+                                       "category",
+                                       "message id 2",
+                                       1,
+                                       "2",
+                                       kTTLValue,
+                                       1,
+                                       0,
+                                       "token"));
+  mcs_client()->SendMessage(message2);
+
+  LoginClient(std::vector<std::string>());
+  GetFakeHandler()->ExpectOutgoingMessage(message);
+  GetFakeHandler()->ExpectOutgoingMessage(message2);
+  PumpLoop();
+}
+
 } // namespace
 
 }  // namespace gcm