Renaming RMQStore to GCMStore and breaking out its interface

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/strings/string_number_conversions.h"
 #include "base/time/clock.h"
 #include "base/time/time.h"
 #include "google_apis/gcm/base/mcs_util.h"
 #include "google_apis/gcm/base/socket_stream.h"
 #include "google_apis/gcm/engine/connection_factory.h"
-#include "google_apis/gcm/engine/rmq_store.h"
 
 using namespace google::protobuf::io;
 
 namespace gcm {
 
 namespace {
 
 typedef scoped_ptr<google::protobuf::MessageLite> MCSProto;
 
 // The category of messages intended for the GCM client itself from MCS.
@@ skipping 54 matching lines @@
   MCSProto protobuf;
 };
 
 ReliablePacketInfo::ReliablePacketInfo()
   : stream_id(0), tag(0) {
 }
 ReliablePacketInfo::~ReliablePacketInfo() {}
 
 MCSClient::MCSClient(base::Clock* clock,
                      ConnectionFactory* connection_factory,
-                     RMQStore* rmq_store)
+                     GCMStore* gcm_store)
     : clock_(clock),
       state_(UNINITIALIZED),
       android_id_(0),
       security_token_(0),
       connection_factory_(connection_factory),
       connection_handler_(NULL),
       last_device_to_server_stream_id_received_(0),
       last_server_to_device_stream_id_received_(0),
       stream_id_out_(0),
       stream_id_in_(0),
-      rmq_store_(rmq_store),
+      gcm_store_(gcm_store),
       weak_ptr_factory_(this) {
 }
 
 MCSClient::~MCSClient() {
 }
 
 void MCSClient::Initialize(
     const InitializationCompleteCallback& initialization_callback,
     const OnMessageReceivedCallback& message_received_callback,
     const OnMessageSentCallback& message_sent_callback,
-    const RMQStore::LoadResult& load_result) {
+    const GCMStore::LoadResult& load_result) {
   DCHECK_EQ(state_, UNINITIALIZED);
   initialization_callback_ = initialization_callback;
   message_received_callback_ = message_received_callback;
   message_sent_callback_ = message_sent_callback;
 
   connection_factory_->Initialize(
       base::Bind(&MCSClient::ResetStateAndBuildLoginRequest,
                  weak_ptr_factory_.GetWeakPtr()),
       base::Bind(&MCSClient::HandlePacketFromWire,
                  weak_ptr_factory_.GetWeakPtr()),
@@ skipping 46 matching lines @@
       expired_ttl_ids.push_back(iter->first);
       message_sent_callback_.Run("TTL expired for " + iter->first);
       delete iter->second;
       continue;
     }
 
     ordered_messages[timestamp] = iter->second;
   }
 
   if (!expired_ttl_ids.empty()) {
-    rmq_store_->RemoveOutgoingMessages(
+    gcm_store_->RemoveOutgoingMessages(
         expired_ttl_ids,
-        base::Bind(&MCSClient::OnRMQUpdateFinished,
+        base::Bind(&MCSClient::OnGCMUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
   }
 
   // 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*>::const_iterator
            iter = ordered_messages.begin();
        iter != ordered_messages.end(); ++iter) {
     ReliablePacketInfo* packet_info = new ReliablePacketInfo();
     packet_info->protobuf.reset(iter->second);
     packet_info->persistent_id = base::Uint64ToString(iter->first);
     to_send_.push_back(make_linked_ptr(packet_info));
   }
 
   // TODO(fgorski): that is likely the only place where the initialization
   // callback could be used.
   initialization_callback_.Run(true, android_id_, security_token_);
 }
 
 void MCSClient::Login(uint64 android_id, uint64 security_token) {
   if (android_id != android_id_ && security_token != security_token_) {
     DCHECK(android_id);
     DCHECK(security_token);
     DCHECK(restored_unackeds_server_ids_.empty());
     android_id_ = android_id;
     security_token_ = security_token;
-    rmq_store_->SetDeviceCredentials(
+    gcm_store_->SetDeviceCredentials(
         android_id_,
         security_token_,
-        base::Bind(&MCSClient::OnRMQUpdateFinished,
+        base::Bind(&MCSClient::OnGCMUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
   }
 
   state_ = CONNECTING;
   connection_factory_->Connect();
 }
 
 void MCSClient::SendMessage(const MCSMessage& message) {
   int ttl = GetTTL(message.GetProtobuf());
   DCHECK_GE(ttl, 0);
   if (to_send_.size() > kMaxSendQueueSize) {
     message_sent_callback_.Run("Message queue full.");
     return;
   }
   if (message.size() > kMaxMessageBytes) {
     message_sent_callback_.Run("Message too large.");
     return;
   }
 
   ReliablePacketInfo* packet_info = new ReliablePacketInfo();
   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());
-    rmq_store_->AddOutgoingMessage(persistent_id,
+    gcm_store_->AddOutgoingMessage(persistent_id,
                                    MCSMessage(message.tag(),
                                               *(packet_info->protobuf)),
-                                   base::Bind(&MCSClient::OnRMQUpdateFinished,
+                                   base::Bind(&MCSClient::OnGCMUpdateFinished,
                                               weak_ptr_factory_.GetWeakPtr()));
   } else if (!connection_factory_->IsEndpointReachable()) {
     DVLOG(1) << "No active connection, dropping message.";
     message_sent_callback_.Run("TTL expired");
     return;
   }
   to_send_.push_back(make_linked_ptr(packet_info));
   MaybeSendMessage();
 }
 
 void MCSClient::Destroy() {
-  rmq_store_->Destroy(base::Bind(&MCSClient::OnRMQUpdateFinished,
+  gcm_store_->Destroy(base::Bind(&MCSClient::OnGCMUpdateFinished,
                                  weak_ptr_factory_.GetWeakPtr()));
 }
 
 void MCSClient::ResetStateAndBuildLoginRequest(
     mcs_proto::LoginRequest* request) {
   DCHECK(android_id_);
   DCHECK(security_token_);
   stream_id_in_ = 0;
   stream_id_out_ = 1;
   last_device_to_server_stream_id_received_ = 0;
@@ skipping 52 matching lines @@
       // message from the persistent store.
       if (!packet->persistent_id.empty())
         expired_ttl_ids.push_back(packet->persistent_id);
       message_sent_callback_.Run("TTL expired");
     }
   }
 
   if (!expired_ttl_ids.empty()) {
     DVLOG(1) << "Connection reset, " << expired_ttl_ids.size()
              << " messages expired.";
-    rmq_store_->RemoveOutgoingMessages(
+    gcm_store_->RemoveOutgoingMessages(
         expired_ttl_ids,
-        base::Bind(&MCSClient::OnRMQUpdateFinished,
+        base::Bind(&MCSClient::OnGCMUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
   }
 
   to_send_.swap(new_to_send);
 
   DVLOG(1) << "Resetting state, with " << request->received_persistent_id_size()
            << " incoming acks pending, and " << to_send_.size()
            << " pending outgoing messages.";
 
   state_ = CONNECTING;
 }
 
 void MCSClient::SendHeartbeat() {
   SendMessage(MCSMessage(kHeartbeatPingTag, mcs_proto::HeartbeatPing()));
 }
 
-void MCSClient::OnRMQUpdateFinished(bool success) {
-  LOG_IF(ERROR, !success) << "RMQ Update failed!";
+void MCSClient::OnGCMUpdateFinished(bool success) {
+  LOG_IF(ERROR, !success) << "GCM Update failed!";
   // TODO(zea): Rebuild the store from scratch in case of persistence failure?
 }
 
 void MCSClient::MaybeSendMessage() {
   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();
   if (HasTTLExpired(*packet->protobuf, clock_)) {
     DCHECK(!packet->persistent_id.empty());
     DVLOG(1) << "Dropping expired message " << packet->persistent_id << ".";
     message_sent_callback_.Run("TTL expired for " + packet->persistent_id);
-    rmq_store_->RemoveOutgoingMessage(
+    gcm_store_->RemoveOutgoingMessage(
         packet->persistent_id,
-        base::Bind(&MCSClient::OnRMQUpdateFinished,
+        base::Bind(&MCSClient::OnGCMUpdateFinished,
                    weak_ptr_factory_.GetWeakPtr()));
     base::MessageLoop::current()->PostTask(
             FROM_HERE,
             base::Bind(&MCSClient::MaybeSendMessage,
                        weak_ptr_factory_.GetWeakPtr()));
     return;
   }
   DVLOG(1) << "Pending output message found, sending.";
   if (!packet->persistent_id.empty())
     to_resend_.push_back(packet);
@@ skipping 90 matching lines @@
     }
     for (StreamIdList::iterator iter = acked_stream_ids_to_remove.begin();
          iter != acked_stream_ids_to_remove.end(); ++iter) {
       acked_server_ids_.erase(*iter);
     }
   }
 
   ++stream_id_in_;
   if (!persistent_id.empty()) {
     unacked_server_ids_[stream_id_in_] = persistent_id;
-    rmq_store_->AddIncomingMessage(persistent_id,
-                                   base::Bind(&MCSClient::OnRMQUpdateFinished,
+    gcm_store_->AddIncomingMessage(persistent_id,
+                                   base::Bind(&MCSClient::OnGCMUpdateFinished,
                                               weak_ptr_factory_.GetWeakPtr()));
   }
 
   DVLOG(1) << "Received message of type " << protobuf->GetTypeName()
            << " with persistent id "
            << (persistent_id.empty() ? "NULL" : persistent_id)
            << ", stream id " << stream_id_in_ << " and last stream id received "
            << last_stream_id_received;
 
   if (unacked_server_ids_.size() > 0 &&
@@ skipping 126 matching lines @@
          to_resend_.front()->stream_id <= last_stream_id_received) {
     const MCSPacketInternal& outgoing_packet = to_resend_.front();
     acked_outgoing_persistent_ids.push_back(outgoing_packet->persistent_id);
     acked_outgoing_stream_ids.push_back(outgoing_packet->stream_id);
     to_resend_.pop_front();
   }
 
   DVLOG(1) << "Server acked " << acked_outgoing_persistent_ids.size()
            << " outgoing messages, " << to_resend_.size()
            << " remaining unacked";
-  rmq_store_->RemoveOutgoingMessages(
+  gcm_store_->RemoveOutgoingMessages(
       acked_outgoing_persistent_ids,
-      base::Bind(&MCSClient::OnRMQUpdateFinished,
+      base::Bind(&MCSClient::OnGCMUpdateFinished,
                  weak_ptr_factory_.GetWeakPtr()));
 
   HandleServerConfirmedReceipt(last_stream_id_received);
 }
 
 void MCSClient::HandleSelectiveAck(const PersistentIdList& id_list) {
   // First check the to_resend_ queue. Acknowledgments should always happen
   // in the order they were sent, so if messages are present they should match
   // the acknowledge list.
   PersistentIdList::const_iterator iter = id_list.begin();
@@ skipping 21 matching lines @@
     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.";
-  rmq_store_->RemoveOutgoingMessages(
+  gcm_store_->RemoveOutgoingMessages(
       id_list,
-      base::Bind(&MCSClient::OnRMQUpdateFinished,
+      base::Bind(&MCSClient::OnGCMUpdateFinished,
                  weak_ptr_factory_.GetWeakPtr()));
 
   // Resend any remaining outgoing messages, as they were not received by the
   // server.
   DVLOG(1) << "Resending " << to_resend_.size() << " messages.";
   while (!to_resend_.empty()) {
     to_send_.push_front(to_resend_.back());
     to_resend_.pop_back();
   }
 }
@@ skipping 11 matching lines @@
        iter != acked_server_ids_.end() &&
            iter->first <= device_stream_id;) {
     acked_incoming_ids.insert(acked_incoming_ids.end(),
                               iter->second.begin(),
                               iter->second.end());
     acked_server_ids_.erase(iter++);
   }
 
   DVLOG(1) << "Server confirmed receipt of " << acked_incoming_ids.size()
            << " acknowledged server messages.";
-  rmq_store_->RemoveIncomingMessages(
+  gcm_store_->RemoveIncomingMessages(
       acked_incoming_ids,
-      base::Bind(&MCSClient::OnRMQUpdateFinished,
+      base::Bind(&MCSClient::OnGCMUpdateFinished,
                  weak_ptr_factory_.GetWeakPtr()));
 }
 
 MCSClient::PersistentId MCSClient::GetNextPersistentId() {
   return base::Uint64ToString(base::TimeTicks::Now().ToInternalValue());
 }
 
 void MCSClient::OnConnectionResetByHeartbeat() {
   connection_factory_->SignalConnectionReset();
 }
 
 } // namespace gcm