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 "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;
 
 // TODO(zea): get these values from MCS settings.
@@ skipping 55 matching lines @@
 
   // The protobuf of the message itself.
   MCSProto protobuf;
 };
 
 ReliablePacketInfo::ReliablePacketInfo()
   : stream_id(0), tag(0) {
 }
 ReliablePacketInfo::~ReliablePacketInfo() {}
 
-MCSClient::MCSClient(ConnectionFactory* connection_factory, RMQStore* rmq_store)
+MCSClient::MCSClient(ConnectionFactory* connection_factory, GCMStore* gcm_store)
     : 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),
       heartbeat_interval_(
           base::TimeDelta::FromSeconds(kHeartbeatDefaultSeconds)),
       heartbeat_timer_(true, true),
-      weak_ptr_factory_(this) {
-}
+      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 59 matching lines @@
 }
 
 void MCSClient::Login(uint64 android_id, uint64 security_token) {
   DCHECK_EQ(state_, LOADED);
   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, bool use_rmq) {
   DCHECK_EQ(state_, CONNECTED);
   if (to_send_.size() > kMaxSendQueueSize) {
@@ skipping 11 matching lines @@
 
   ReliablePacketInfo* packet_info = new ReliablePacketInfo();
   packet_info->protobuf = message.CloneProtobuf();
 
   if (use_rmq) {
     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,
-                                   MCSMessage(message.tag(),
-                                              *(packet_info->protobuf)),
-                                   base::Bind(&MCSClient::OnRMQUpdateFinished,
-                                              weak_ptr_factory_.GetWeakPtr()));
+    gcm_store_->AddOutgoingMessage(
+        persistent_id,
+        MCSMessage(message.tag(), *(packet_info->protobuf)),
+        base::Bind(&MCSClient::OnGCMUpdateFinished,
+                   weak_ptr_factory_.GetWeakPtr()));
   } else {
     // Check that there is an active connection to the endpoint.
     if (!connection_handler_->CanSendMessage()) {
       base::MessageLoop::current()->PostTask(
           FROM_HERE,
           base::Bind(message_sent_callback_, "Unable to reach endpoint"));
       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 44 matching lines @@
   heartbeat_timer_.Stop();
 
   state_ = CONNECTING;
 }
 
 void MCSClient::SendHeartbeat() {
   SendMessage(MCSMessage(kHeartbeatPingTag, mcs_proto::HeartbeatPing()),
               false);
 }
 
-void MCSClient::OnRMQUpdateFinished(bool success) {
+void MCSClient::OnGCMUpdateFinished(bool success) {
   LOG_IF(ERROR, !success) << "RMQ Update failed!";
   // TODO(zea): Rebuild the store from scratch in case of persistence failure?
 }
 
 void MCSClient::MaybeSendMessage() {
   if (to_send_.empty())
     return;
 
   if (!connection_handler_->CanSendMessage())
     return;
@@ skipping 98 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 119 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());
 }
 
 } // namespace gcm