FCMSyncInvalidationListener is implementation for the SyncInvalidationLIstener.

FCMSyncInvalidationListener aims to substitude SyncInvalidationListener for
the Tango->FCM migration.

Bug: 801985
Change-Id: Ib2907b8d5431e4730e42947744c84036050cd253
Reviewed-on: https://chromium-review.googlesource.com/1111714
Commit-Queue: Tatiana Gornak <melandory@chromium.org>
Reviewed-by: Pavel Yatsuk <pavely@chromium.org>
Reviewed-by: Jan Krcal <jkrcal@chromium.org>
Cr-Commit-Position: refs/heads/master@{#571448}

components/invalidation/impl/BUILD.gn

@@ -65,6 +65,8 @@ static_library("impl") {
 
   if (!is_android) {
     sources += [
+      "fcm_sync_invalidation_listener.cc",
+      "fcm_sync_invalidation_listener.h",
       "gcm_invalidation_bridge.cc",
       "gcm_invalidation_bridge.h",
       "gcm_network_channel.cc",
@@ -147,6 +149,7 @@ source_set("unit_tests") {
     # Non-Android tests.
     sources += [
       "fake_invalidator_unittest.cc",
+      "fcm_sync_invalidation_listener_unittest.cc",
       "gcm_invalidation_bridge_unittest.cc",
       "gcm_network_channel_unittest.cc",
       "invalidation_notifier_unittest.cc",

components/invalidation/impl/fcm_sync_invalidation_listener.cc

+// Copyright 2018 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 "components/invalidation/impl/fcm_sync_invalidation_listener.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/logging.h"
+#include "components/invalidation/public/invalidation_util.h"
+#include "components/invalidation/public/object_id_invalidation_map.h"
+#include "components/prefs/pref_service.h"
+#include "google/cacheinvalidation/include/invalidation-client.h"
+#include "google/cacheinvalidation/include/types.h"
+
+namespace syncer {
+
+namespace {
+
+invalidation::ObjectId ConvertToObjectId(
+    const invalidation::InvalidationObjectId& invalidation_object_id) {
+  return invalidation::ObjectId(invalidation_object_id.source(),
+                                invalidation_object_id.name());
+}
+
+invalidation::InvalidationObjectId ConvertToInvalidationObjectId(
+    const invalidation::ObjectId& object_id) {
+  return invalidation::InvalidationObjectId(object_id.source(),
+                                            object_id.name());
+}
+
+ObjectIdSet ConvertToObjectIdSet(const InvalidationObjectIdSet& ids) {
+  ObjectIdSet object_ids;
+  for (const auto& id : ids)
+    object_ids.insert(ConvertToObjectId(id));
+  return object_ids;
+}
+
+InvalidationObjectIdSet ConvertToInvalidationObjectIdSet(
+    const ObjectIdSet& ids) {
+  InvalidationObjectIdSet invalidation_object_ids;
+  for (const auto& id : ids)
+    invalidation_object_ids.insert(ConvertToInvalidationObjectId(id));
+  return invalidation_object_ids;
+}
+
+}  // namespace
+
+FCMSyncInvalidationListener::Delegate::~Delegate() {}
+
+FCMSyncInvalidationListener::FCMSyncInvalidationListener(
+    std::unique_ptr<SyncNetworkChannel> network_channel)
+    : sync_network_channel_(std::move(network_channel)),
+      sync_system_resources_(sync_network_channel_.get(), nullptr),
+      delegate_(nullptr),
+      ticl_state_(DEFAULT_INVALIDATION_ERROR),
+      fcm_network_state_(DEFAULT_INVALIDATION_ERROR),
+      weak_factory_(this) {
+  sync_network_channel_->AddObserver(this);
+}
+
+FCMSyncInvalidationListener::~FCMSyncInvalidationListener() {
+  sync_network_channel_->RemoveObserver(this);
+  Stop();
+  DCHECK(!delegate_);
+}
+
+void FCMSyncInvalidationListener::Start(
+    CreateInvalidationClientCallback create_invalidation_client_callback,
+    Delegate* delegate,
+    std::unique_ptr<PerUserTopicRegistrationManager>
+        per_user_topic_registration_manager) {
+  DCHECK(delegate);
+  Stop();
+  sync_system_resources_.Start();
+  delegate_ = delegate;
+  per_user_topic_registration_manager_ =
+      std::move(per_user_topic_registration_manager);
+
+  invalidation_client_ = std::move(create_invalidation_client_callback)
+                             .Run(&sync_system_resources_, this);
+  invalidation_client_->Start();
+}
+
+void FCMSyncInvalidationListener::UpdateRegisteredIds(const ObjectIdSet& ids) {
+  registered_ids_ = ConvertToInvalidationObjectIdSet(ids);
+  if (ticl_state_ == INVALIDATIONS_ENABLED &&
+      per_user_topic_registration_manager_)
+    DoRegistrationUpdate();
+}
+
+void FCMSyncInvalidationListener::Ready(
+    invalidation::InvalidationClient* client) {
+  DCHECK_EQ(client, invalidation_client_.get());
+  ticl_state_ = INVALIDATIONS_ENABLED;
+  EmitStateChange();
+  DoRegistrationUpdate();
+}
+
+void FCMSyncInvalidationListener::Invalidate(
+    invalidation::InvalidationClient* client,
+    const invalidation::Invalidation& invalidation,
+    const invalidation::AckHandle& ack_handle) {
+  DCHECK_EQ(client, invalidation_client_.get());
+  client->Acknowledge(ack_handle);
+
+  const invalidation::ObjectId& id = invalidation.object_id();
+
+  std::string payload;
+  // payload() CHECK()'s has_payload(), so we must check it ourselves first.
+  if (invalidation.has_payload())
+    payload = invalidation.payload();
+
+  DVLOG(2) << "Received invalidation with version " << invalidation.version()
+           << " for " << ObjectIdToString(id);
+
+  ObjectIdInvalidationMap invalidations;
+  Invalidation inv = Invalidation::Init(id, invalidation.version(), payload);
+  inv.SetAckHandler(AsWeakPtr(), base::ThreadTaskRunnerHandle::Get());
+  invalidations.Insert(inv);
+
+  DispatchInvalidations(invalidations);
+}
+
+void FCMSyncInvalidationListener::InvalidateUnknownVersion(
+    invalidation::InvalidationClient* client,
+    const invalidation::ObjectId& object_id,
+    const invalidation::AckHandle& ack_handle) {
+  DCHECK_EQ(client, invalidation_client_.get());
+  DVLOG(1) << "InvalidateUnknownVersion";
+  client->Acknowledge(ack_handle);
+
+  ObjectIdInvalidationMap invalidations;
+  Invalidation unknown_version = Invalidation::InitUnknownVersion(object_id);
+  unknown_version.SetAckHandler(AsWeakPtr(),
+                                base::ThreadTaskRunnerHandle::Get());
+  invalidations.Insert(unknown_version);
+
+  DispatchInvalidations(invalidations);
+}
+
+// This should behave as if we got an invalidation with version
+// UNKNOWN_OBJECT_VERSION for all known data types.
+void FCMSyncInvalidationListener::InvalidateAll(
+    invalidation::InvalidationClient* client,
+    const invalidation::AckHandle& ack_handle) {
+  DCHECK_EQ(client, invalidation_client_.get());
+  DVLOG(1) << "InvalidateAll";
+  client->Acknowledge(ack_handle);
+
+  ObjectIdInvalidationMap invalidations;
+  for (const auto& registered_id : registered_ids_) {
+    Invalidation unknown_version =
+        Invalidation::InitUnknownVersion(ConvertToObjectId(registered_id));
+    unknown_version.SetAckHandler(AsWeakPtr(),
+                                  base::ThreadTaskRunnerHandle::Get());
+    invalidations.Insert(unknown_version);
+  }
+  DispatchInvalidations(invalidations);
+}
+
+void FCMSyncInvalidationListener::DispatchInvalidations(
+    const ObjectIdInvalidationMap& invalidations) {
+  ObjectIdInvalidationMap to_save = invalidations;
+  ObjectIdInvalidationMap to_emit = invalidations.GetSubsetWithObjectIds(
+      ConvertToObjectIdSet(registered_ids_));
+
+  SaveInvalidations(to_save);
+  EmitSavedInvalidations(to_emit);
+}
+
+void FCMSyncInvalidationListener::SaveInvalidations(
+    const ObjectIdInvalidationMap& to_save) {
+  ObjectIdSet objects_to_save = to_save.GetObjectIds();
+  for (ObjectIdSet::const_iterator it = objects_to_save.begin();
+       it != objects_to_save.end(); ++it) {
+    UnackedInvalidationsMap::iterator lookup =
+        unacked_invalidations_map_.find(*it);
+    if (lookup == unacked_invalidations_map_.end()) {
+      lookup = unacked_invalidations_map_
+                   .insert(std::make_pair(*it, UnackedInvalidationSet(*it)))
+                   .first;
+    }
+    lookup->second.AddSet(to_save.ForObject(*it));
+  }
+}
+
+void FCMSyncInvalidationListener::EmitSavedInvalidations(
+    const ObjectIdInvalidationMap& to_emit) {
+  DVLOG(2) << "Emitting invalidations: " << to_emit.ToString();
+  delegate_->OnInvalidate(to_emit);
+}
+
+void FCMSyncInvalidationListener::InformRegistrationStatus(
+    invalidation::InvalidationClient* client,
+    const invalidation::ObjectId& object_id,
+    InvalidationListener::RegistrationState new_state) {
+  // TODO(melandory): this method is irrelevant in new architecture and
+  // should be removed along with cacheinvalidation library.
+}
+
+void FCMSyncInvalidationListener::InformRegistrationFailure(
+    invalidation::InvalidationClient* client,
+    const invalidation::ObjectId& object_id,
+    bool is_transient,
+    const std::string& error_message) {
+  // TODO(melandory): this method is irrelevant in new architecture and
+  // should be removed along with cacheinvalidation library.
+}
+
+void FCMSyncInvalidationListener::ReissueRegistrations(
+    invalidation::InvalidationClient* client,
+    const std::string& prefix,
+    int prefix_length) {
+  // TODO(melandory): this method is irrelevant in new architecture and
+  // should be removed along with cacheinvalidation library.
+}
+
+void FCMSyncInvalidationListener::InformError(
+    invalidation::InvalidationClient* client,
+    const invalidation::ErrorInfo& error_info) {}
+
+void FCMSyncInvalidationListener::Acknowledge(const invalidation::ObjectId& id,
+                                              const syncer::AckHandle& handle) {
+  UnackedInvalidationsMap::iterator lookup =
+      unacked_invalidations_map_.find(id);
+  if (lookup == unacked_invalidations_map_.end()) {
+    DLOG(WARNING) << "Received acknowledgement for untracked object ID";
+    return;
+  }
+  lookup->second.Acknowledge(handle);
+}
+
+void FCMSyncInvalidationListener::Drop(const invalidation::ObjectId& id,
+                                       const syncer::AckHandle& handle) {
+  UnackedInvalidationsMap::iterator lookup =
+      unacked_invalidations_map_.find(id);
+  if (lookup == unacked_invalidations_map_.end()) {
+    DLOG(WARNING) << "Received drop for untracked object ID";
+    return;
+  }
+  lookup->second.Drop(handle);
+}
+
+void FCMSyncInvalidationListener::DoRegistrationUpdate() {
+  per_user_topic_registration_manager_->UpdateRegisteredIds(registered_ids_);
+
+  // TODO(melandory): remove unacked invalidations for unregistered objects.
+  ObjectIdInvalidationMap object_id_invalidation_map;
+  for (auto& unacked : unacked_invalidations_map_) {
+    if (registered_ids_.find(ConvertToInvalidationObjectId(unacked.first)) ==
+        registered_ids_.end()) {
+      continue;
+    }
+    unacked.second.ExportInvalidations(AsWeakPtr(),
+                                       base::ThreadTaskRunnerHandle::Get(),
+                                       &object_id_invalidation_map);
+  }
+
+  // There's no need to run these through DispatchInvalidations(); they've
+  // already been saved to storage (that's where we found them) so all we need
+  // to do now is emit them.
+  EmitSavedInvalidations(object_id_invalidation_map);
+}
+
+void FCMSyncInvalidationListener::StopForTest() {
+  Stop();
+}
+
+ObjectIdSet FCMSyncInvalidationListener::GetRegisteredIdsForTest() const {
+  return ConvertToObjectIdSet(registered_ids_);
+}
+
+base::WeakPtr<FCMSyncInvalidationListener>
+FCMSyncInvalidationListener::AsWeakPtr() {
+  return weak_factory_.GetWeakPtr();
+}
+
+void FCMSyncInvalidationListener::Stop() {
+  if (!invalidation_client_) {
+    return;
+  }
+
+  sync_system_resources_.Stop();
+  invalidation_client_->Stop();
+
+  invalidation_client_.reset();
+  delegate_ = nullptr;
+  per_user_topic_registration_manager_.reset();
+
+  ticl_state_ = DEFAULT_INVALIDATION_ERROR;
+  fcm_network_state_ = DEFAULT_INVALIDATION_ERROR;
+}
+
+InvalidatorState FCMSyncInvalidationListener::GetState() const {
+  if (ticl_state_ == INVALIDATION_CREDENTIALS_REJECTED ||
+      fcm_network_state_ == INVALIDATION_CREDENTIALS_REJECTED) {
+    // If either the ticl or the push client rejected our credentials,
+    // return INVALIDATION_CREDENTIALS_REJECTED.
+    return INVALIDATION_CREDENTIALS_REJECTED;
+  }
+  if (ticl_state_ == INVALIDATIONS_ENABLED &&
+      fcm_network_state_ == INVALIDATIONS_ENABLED) {
+    // If the ticl is ready and the push client notifications are
+    // enabled, return INVALIDATIONS_ENABLED.
+    return INVALIDATIONS_ENABLED;
+  }
+  // Otherwise, we have a transient error.
+  return TRANSIENT_INVALIDATION_ERROR;
+}
+
+void FCMSyncInvalidationListener::EmitStateChange() {
+  delegate_->OnInvalidatorStateChange(GetState());
+}
+
+void FCMSyncInvalidationListener::OnNetworkChannelStateChanged(
+    InvalidatorState invalidator_state) {
+  fcm_network_state_ = invalidator_state;
+  EmitStateChange();
+}
+
+}  // namespace syncer

components/invalidation/impl/fcm_sync_invalidation_listener.h

+// Copyright 2018 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.
+//
+// A simple wrapper around invalidation::InvalidationClient that
+// handles all the startup/shutdown details and hookups.
+
+#ifndef COMPONENTS_INVALIDATION_IMPL_FCM_SYNC_INVALIDATION_LISTENER_H_
+#define COMPONENTS_INVALIDATION_IMPL_FCM_SYNC_INVALIDATION_LISTENER_H_
+
+#include <memory>
+
+#include "base/callback_forward.h"
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "components/invalidation/impl/per_user_topic_registration_manager.h"
+#include "components/invalidation/impl/sync_system_resources.h"
+#include "components/invalidation/impl/unacked_invalidation_set.h"
+#include "components/invalidation/public/ack_handler.h"
+#include "components/invalidation/public/invalidation_object_id.h"
+#include "components/invalidation/public/invalidation_util.h"
+#include "components/invalidation/public/invalidator_state.h"
+#include "google/cacheinvalidation/include/invalidation-listener.h"
+#include "services/network/public/mojom/url_loader_factory.mojom.h"
+
+namespace syncer {
+
+class ObjectIdInvalidationMap;
+
+class INVALIDATION_EXPORT FCMSyncInvalidationListener
+    : public invalidation::InvalidationListener,
+      public SyncNetworkChannel::Observer,
+      public AckHandler {
+ public:
+  typedef base::OnceCallback<std::unique_ptr<invalidation::InvalidationClient>(
+      invalidation::SystemResources*,
+      invalidation::InvalidationListener*)>
+      CreateInvalidationClientCallback;
+
+  class INVALIDATION_EXPORT Delegate {
+   public:
+    virtual ~Delegate();
+
+    virtual void OnInvalidate(const ObjectIdInvalidationMap& invalidations) = 0;
+
+    virtual void OnInvalidatorStateChange(InvalidatorState state) = 0;
+  };
+
+  explicit FCMSyncInvalidationListener(
+      std::unique_ptr<SyncNetworkChannel> network_channel);
+
+  ~FCMSyncInvalidationListener() override;
+
+  void Start(
+      CreateInvalidationClientCallback create_invalidation_client_callback,
+      Delegate* delegate,
+      std::unique_ptr<PerUserTopicRegistrationManager>
+          per_user_topic_registration_manager);
+
+  // Update the set of object IDs that we're interested in getting
+  // notifications for. May be called at any time.
+  void UpdateRegisteredIds(const ObjectIdSet& ids);
+
+  // invalidation::InvalidationListener implementation.
+  void Ready(invalidation::InvalidationClient* client) override;
+  void Invalidate(invalidation::InvalidationClient* client,
+                  const invalidation::Invalidation& invalidation,
+                  const invalidation::AckHandle& ack_handle) override;
+  void InvalidateUnknownVersion(
+      invalidation::InvalidationClient* client,
+      const invalidation::ObjectId& object_id,
+      const invalidation::AckHandle& ack_handle) override;
+  void InvalidateAll(invalidation::InvalidationClient* client,
+                     const invalidation::AckHandle& ack_handle) override;
+  void InformRegistrationStatus(
+      invalidation::InvalidationClient* client,
+      const invalidation::ObjectId& object_id,
+      invalidation::InvalidationListener::RegistrationState reg_state) override;
+  void InformRegistrationFailure(invalidation::InvalidationClient* client,
+                                 const invalidation::ObjectId& object_id,
+                                 bool is_transient,
+                                 const std::string& error_message) override;
+  void ReissueRegistrations(invalidation::InvalidationClient* client,
+                            const std::string& prefix,
+                            int prefix_length) override;
+  void InformError(invalidation::InvalidationClient* client,
+                   const invalidation::ErrorInfo& error_info) override;
+
+  // AckHandler implementation.
+  void Acknowledge(const invalidation::ObjectId& id,
+                   const syncer::AckHandle& handle) override;
+  void Drop(const invalidation::ObjectId& id,
+            const syncer::AckHandle& handle) override;
+
+  // SyncNetworkChannel::Observer implementation.
+  void OnNetworkChannelStateChanged(
+      InvalidatorState invalidator_state) override;
+
+  void DoRegistrationUpdate();
+
+  void StopForTest();
+
+  ObjectIdSet GetRegisteredIdsForTest() const;
+
+  base::WeakPtr<FCMSyncInvalidationListener> AsWeakPtr();
+
+ private:
+  void Stop();
+
+  InvalidatorState GetState() const;
+
+  void EmitStateChange();
+
+  // Sends invalidations to their appropriate destination.
+  //
+  // If there are no observers registered for them, they will be saved for
+  // later.
+  //
+  // If there are observers registered, they will be saved (to make sure we
+  // don't drop them until they've been acted on) and emitted to the observers.
+  void DispatchInvalidations(const ObjectIdInvalidationMap& invalidations);
+
+  // Saves invalidations.
+  //
+  // This call isn't synchronous so we can't guarantee these invalidations will
+  // be safely on disk by the end of the call, but it should ensure that the
+  // data makes it to disk eventually.
+  void SaveInvalidations(const ObjectIdInvalidationMap& to_save);
+  // Emits previously saved invalidations to their registered observers.
+  void EmitSavedInvalidations(const ObjectIdInvalidationMap& to_emit);
+
+  // Generate a Dictionary with all the debugging information.
+  std::unique_ptr<base::DictionaryValue> CollectDebugData() const;
+
+  std::unique_ptr<SyncNetworkChannel> sync_network_channel_;
+  SyncSystemResources sync_system_resources_;
+  UnackedInvalidationsMap unacked_invalidations_map_;
+  Delegate* delegate_;
+  std::unique_ptr<invalidation::InvalidationClient> invalidation_client_;
+
+  // Stored to pass to |per_user_topic_registration_manager_| on start.
+  InvalidationObjectIdSet registered_ids_;
+
+  // The states of the ticl and FCN channel.
+  InvalidatorState ticl_state_;
+  InvalidatorState fcm_network_state_;
+
+  std::unique_ptr<PerUserTopicRegistrationManager>
+      per_user_topic_registration_manager_;
+
+  base::WeakPtrFactory<FCMSyncInvalidationListener> weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(FCMSyncInvalidationListener);
+};
+
+}  // namespace syncer
+
+#endif  // COMPONENTS_INVALIDATION_IMPL_FCM_SYNC_INVALIDATION_LISTENER_H_

components/invalidation/impl/fcm_sync_invalidation_listener_unittest.cc

+// Copyright 2018 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 <string>
+#include <vector>
+
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/stl_util.h"
+#include "base/threading/thread_task_runner_handle.h"
+#include "components/invalidation/impl/fake_invalidation_state_tracker.h"
+#include "components/invalidation/impl/fcm_sync_invalidation_listener.h"
+#include "components/invalidation/impl/json_unsafe_parser.h"
+#include "components/invalidation/impl/per_user_topic_registration_manager.h"
+#include "components/invalidation/impl/push_client_channel.h"
+#include "components/invalidation/impl/unacked_invalidation_set_test_util.h"
+#include "components/invalidation/public/invalidation_util.h"
+#include "components/invalidation/public/object_id_invalidation_map.h"
+#include "google/cacheinvalidation/include/invalidation-client.h"
+#include "google/cacheinvalidation/include/types.h"
+#include "jingle/notifier/listener/fake_push_client.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace syncer {
+
+namespace {
+
+using invalidation::AckHandle;
+using invalidation::ObjectId;
+
+const char kPayload1[] = "payload1";
+const char kPayload2[] = "payload2";
+
+const int64_t kVersion1 = 1LL;
+const int64_t kVersion2 = 2LL;
+
+const int kChromeSyncSourceId = 1004;
+
+struct AckHandleLessThan {
+  bool operator()(const AckHandle& lhs, const AckHandle& rhs) const {
+    return lhs.handle_data() < rhs.handle_data();
+  }
+};
+
+typedef std::set<AckHandle, AckHandleLessThan> AckHandleSet;
+
+// Fake invalidation::InvalidationClient implementation that keeps
+// track of registered IDs and acked handles.
+class FakeInvalidationClient : public invalidation::InvalidationClient {
+ public:
+  FakeInvalidationClient() : started_(false) {}
+  ~FakeInvalidationClient() override {}
+
+  void ClearAckedHandles() { acked_handles_.clear(); }
+
+  bool IsAckedHandle(const AckHandle& ack_handle) const {
+    return (acked_handles_.find(ack_handle) != acked_handles_.end());
+  }
+
+  // invalidation::InvalidationClient implementation.
+
+  void Start() override { started_ = true; }
+
+  void Stop() override { started_ = false; }
+
+  void Register(const ObjectId& object_id) override {
+    if (!started_) {
+      ADD_FAILURE();
+      return;
+    }
+    registered_ids_.insert(object_id);
+  }
+
+  void Register(const invalidation::vector<ObjectId>& object_ids) override {
+    if (!started_) {
+      ADD_FAILURE();
+      return;
+    }
+    registered_ids_.insert(object_ids.begin(), object_ids.end());
+  }
+
+  void Unregister(const ObjectId& object_id) override {
+    if (!started_) {
+      ADD_FAILURE();
+      return;
+    }
+    registered_ids_.erase(object_id);
+  }
+
+  void Unregister(const invalidation::vector<ObjectId>& object_ids) override {
+    if (!started_) {
+      ADD_FAILURE();
+      return;
+    }
+    for (invalidation::vector<ObjectId>::const_iterator it = object_ids.begin();
+         it != object_ids.end(); ++it) {
+      registered_ids_.erase(*it);
+    }
+  }
+
+  void Acknowledge(const AckHandle& ack_handle) override {
+    if (!started_) {
+      ADD_FAILURE();
+      return;
+    }
+    acked_handles_.insert(ack_handle);
+  }
+
+ private:
+  bool started_;
+  ObjectIdSet registered_ids_;
+  AckHandleSet acked_handles_;
+};
+
+// Fake delegate that keeps track of invalidation counts, payloads,
+// and state.
+class FakeDelegate : public FCMSyncInvalidationListener::Delegate {
+ public:
+  explicit FakeDelegate(FCMSyncInvalidationListener* listener)
+      : state_(TRANSIENT_INVALIDATION_ERROR) {}
+  ~FakeDelegate() override {}
+
+  size_t GetInvalidationCount(const ObjectId& id) const {
+    Map::const_iterator it = invalidations_.find(id);
+    if (it == invalidations_.end()) {
+      return 0;
+    } else {
+      return it->second.size();
+    }
+  }
+
+  int64_t GetVersion(const ObjectId& id) const {
+    Map::const_iterator it = invalidations_.find(id);
+    if (it == invalidations_.end()) {
+      ADD_FAILURE() << "No invalidations for ID " << ObjectIdToString(id);
+      return 0;
+    } else {
+      return it->second.back().version();
+    }
+  }
+
+  std::string GetPayload(const ObjectId& id) const {
+    Map::const_iterator it = invalidations_.find(id);
+    if (it == invalidations_.end()) {
+      ADD_FAILURE() << "No invalidations for ID " << ObjectIdToString(id);
+      return nullptr;
+    } else {
+      return it->second.back().payload();
+    }
+  }
+
+  bool IsUnknownVersion(const ObjectId& id) const {
+    Map::const_iterator it = invalidations_.find(id);
+    if (it == invalidations_.end()) {
+      ADD_FAILURE() << "No invalidations for ID " << ObjectIdToString(id);
+      return false;
+    } else {
+      return it->second.back().is_unknown_version();
+    }
+  }
+
+  bool StartsWithUnknownVersion(const ObjectId& id) const {
+    Map::const_iterator it = invalidations_.find(id);
+    if (it == invalidations_.end()) {
+      ADD_FAILURE() << "No invalidations for ID " << ObjectIdToString(id);
+      return false;
+    } else {
+      return it->second.front().is_unknown_version();
+    }
+  }
+
+  InvalidatorState GetInvalidatorState() const { return state_; }
+
+  void AcknowledgeNthInvalidation(const ObjectId& id, size_t n) {
+    List& list = invalidations_[id];
+    List::iterator it = list.begin() + n;
+    it->Acknowledge();
+  }
+
+  void AcknowledgeAll(const ObjectId& id) {
+    List& list = invalidations_[id];
+    for (List::iterator it = list.begin(); it != list.end(); ++it) {
+      it->Acknowledge();
+    }
+  }
+
+  void DropNthInvalidation(const ObjectId& id, size_t n) {
+    List& list = invalidations_[id];
+    List::iterator it = list.begin() + n;
+    it->Drop();
+    dropped_invalidations_map_.erase(id);
+    dropped_invalidations_map_.insert(std::make_pair(id, *it));
+  }
+
+  void RecoverFromDropEvent(const ObjectId& id) {
+    DropMap::iterator it = dropped_invalidations_map_.find(id);
+    if (it != dropped_invalidations_map_.end()) {
+      it->second.Acknowledge();
+      dropped_invalidations_map_.erase(it);
+    }
+  }
+
+  // FCMSyncInvalidationListener::Delegate implementation.
+  void OnInvalidate(const ObjectIdInvalidationMap& invalidation_map) override {
+    ObjectIdSet ids = invalidation_map.GetObjectIds();
+    for (ObjectIdSet::iterator it = ids.begin(); it != ids.end(); ++it) {
+      const SingleObjectInvalidationSet& incoming =
+          invalidation_map.ForObject(*it);
+      List& list = invalidations_[*it];
+      list.insert(list.end(), incoming.begin(), incoming.end());
+    }
+  }
+
+  void OnInvalidatorStateChange(InvalidatorState state) override {
+    state_ = state;
+  }
+
+ private:
+  typedef std::vector<Invalidation> List;
+  typedef std::map<ObjectId, List, ObjectIdLessThan> Map;
+  typedef std::map<ObjectId, Invalidation, ObjectIdLessThan> DropMap;
+
+  Map invalidations_;
+  InvalidatorState state_;
+  DropMap dropped_invalidations_map_;
+};
+
+std::unique_ptr<invalidation::InvalidationClient> CreateFakeInvalidationClient(
+    FakeInvalidationClient** fake_invalidation_client,
+    invalidation::SystemResources* resources,
+    invalidation::InvalidationListener* listener) {
+  std::unique_ptr<FakeInvalidationClient> fake_client =
+      std::make_unique<FakeInvalidationClient>();
+  *fake_invalidation_client = fake_client.get();
+  return fake_client;
+}
+
+class MockRegistrationManager : public PerUserTopicRegistrationManager {
+ public:
+  MockRegistrationManager()
+      : PerUserTopicRegistrationManager(
+            "fake_instance_id_token",
+            "fake_access_token",
+            nullptr /* pref_service */,
+            nullptr /* loader_factory */,
+            base::BindRepeating(&syncer::JsonUnsafeParser::Parse)) {}
+  ~MockRegistrationManager() override {}
+  MOCK_METHOD1(UpdateRegisteredIds, void(const InvalidationObjectIdSet& ids));
+};
+
+class FCMSyncInvalidationListenerTest : public testing::Test {
+ protected:
+  FCMSyncInvalidationListenerTest()
+      : kBookmarksId_(kChromeSyncSourceId, "BOOKMARK"),
+        kPreferencesId_(kChromeSyncSourceId, "PREFERENCE"),
+        kExtensionsId_(kChromeSyncSourceId, "EXTENSION"),
+        kAppsId_(kChromeSyncSourceId, "APP"),
+        fake_network_status_(new notifier::FakePushClient()),
+        fake_invalidation_client_(nullptr),
+        listener_(base::WrapUnique(
+            new PushClientChannel(base::WrapUnique(fake_network_status_)))),
+        fake_delegate_(&listener_) {}
+
+  void SetUp() override {
+    StartClient();
+
+    registered_ids_.insert(kBookmarksId_);
+    registered_ids_.insert(kPreferencesId_);
+    listener_.UpdateRegisteredIds(registered_ids_);
+  }
+
+  void TearDown() override { StopClient(); }
+
+  // Restart client without re-registering IDs.
+  void RestartClient() {
+    StopClient();
+    StartClient();
+  }
+
+  void StartClient() {
+    fake_invalidation_client_ = nullptr;
+    std::unique_ptr<MockRegistrationManager> mock_registration_manager =
+        std::make_unique<MockRegistrationManager>();
+    listener_.Start(base::BindOnce(&CreateFakeInvalidationClient,
+                                   &fake_invalidation_client_),
+                    &fake_delegate_, std::move(mock_registration_manager));
+    DCHECK(fake_invalidation_client_);
+  }
+
+  void StopClient() {
+    // listener_.StopForTest() stops the invalidation scheduler, which
+    // deletes any pending tasks without running them.  Some tasks
+    // "run and delete" another task, so they must be run in order to
+    // avoid leaking the inner task.  listener_.StopForTest() does not
+    // schedule any tasks, so it's both necessary and sufficient to
+    // drain the task queue before calling it.
+    fake_invalidation_client_ = nullptr;
+    listener_.StopForTest();
+  }
+
+  size_t GetInvalidationCount(const ObjectId& id) const {
+    return fake_delegate_.GetInvalidationCount(id);
+  }
+
+  int64_t GetVersion(const ObjectId& id) const {
+    return fake_delegate_.GetVersion(id);
+  }
+
+  std::string GetPayload(const ObjectId& id) const {
+    return fake_delegate_.GetPayload(id);
+  }
+
+  bool IsUnknownVersion(const ObjectId& id) const {
+    return fake_delegate_.IsUnknownVersion(id);
+  }
+
+  bool StartsWithUnknownVersion(const ObjectId& id) const {
+    return fake_delegate_.StartsWithUnknownVersion(id);
+  }
+
+  void AcknowledgeNthInvalidation(const ObjectId& id, size_t n) {
+    fake_delegate_.AcknowledgeNthInvalidation(id, n);
+  }
+
+  void DropNthInvalidation(const ObjectId& id, size_t n) {
+    return fake_delegate_.DropNthInvalidation(id, n);
+  }
+
+  void RecoverFromDropEvent(const ObjectId& id) {
+    return fake_delegate_.RecoverFromDropEvent(id);
+  }
+
+  InvalidatorState GetInvalidatorState() {
+    return fake_delegate_.GetInvalidatorState();
+  }
+
+  void AcknowledgeAll(const ObjectId& id) { fake_delegate_.AcknowledgeAll(id); }
+
+  ObjectIdSet GetRegisteredIds() const {
+    return listener_.GetRegisteredIdsForTest();
+  }
+
+  // |payload| can be NULL.
+  void FireInvalidate(const ObjectId& object_id,
+                      int64_t version,
+                      const char* payload) {
+    invalidation::Invalidation inv;
+    if (payload) {
+      inv = invalidation::Invalidation(object_id, version, payload);
+    } else {
+      inv = invalidation::Invalidation(object_id, version);
+    }
+    const AckHandle ack_handle("fakedata");
+    fake_invalidation_client_->ClearAckedHandles();
+    listener_.Invalidate(fake_invalidation_client_, inv, ack_handle);
+    EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
+  }
+
+  // |payload| can be NULL, but not |type_name|.
+  void FireInvalidateUnknownVersion(const ObjectId& object_id) {
+    const AckHandle ack_handle("fakedata_unknown");
+    fake_invalidation_client_->ClearAckedHandles();
+    listener_.InvalidateUnknownVersion(fake_invalidation_client_, object_id,
+                                       ack_handle);
+    EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
+  }
+
+  void FireInvalidateAll() {
+    const AckHandle ack_handle("fakedata_all");
+    fake_invalidation_client_->ClearAckedHandles();
+    listener_.InvalidateAll(fake_invalidation_client_, ack_handle);
+    EXPECT_TRUE(fake_invalidation_client_->IsAckedHandle(ack_handle));
+  }
+
+  void EnableNotifications() { fake_network_status_->EnableNotifications(); }
+
+  void DisableNotifications(notifier::NotificationsDisabledReason reason) {
+    fake_network_status_->DisableNotifications(reason);
+  }
+
+  const ObjectId kBookmarksId_;
+  const ObjectId kPreferencesId_;
+  const ObjectId kExtensionsId_;
+  const ObjectId kAppsId_;
+
+  ObjectIdSet registered_ids_;
+
+ private:
+  base::MessageLoop message_loop_;
+  notifier::FakePushClient* const fake_network_status_;
+
+ protected:
+  // A derrived test needs direct access to this.
+  FakeInvalidationStateTracker fake_tracker_;
+
+  // Tests need to access these directly.
+  FakeInvalidationClient* fake_invalidation_client_;
+  FCMSyncInvalidationListener listener_;
+
+ private:
+  FakeDelegate fake_delegate_;
+};
+
+// Invalidation tests.
+
+// Fire an invalidation without a payload.  It should be processed,
+// the payload should remain empty, and the version should be updated.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateNoPayload) {
+  const ObjectId& id = kBookmarksId_;
+
+  FireInvalidate(id, kVersion1, nullptr);
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kVersion1, GetVersion(id));
+  EXPECT_EQ("", GetPayload(id));
+}
+
+// Fire an invalidation with an empty payload.  It should be
+// processed, the payload should remain empty, and the version should
+// be updated.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateEmptyPayload) {
+  const ObjectId& id = kBookmarksId_;
+
+  FireInvalidate(id, kVersion1, "");
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kVersion1, GetVersion(id));
+  EXPECT_EQ("", GetPayload(id));
+}
+
+// Fire an invalidation with a payload.  It should be processed, and
+// both the payload and the version should be updated.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateWithPayload) {
+  const ObjectId& id = kPreferencesId_;
+
+  FireInvalidate(id, kVersion1, kPayload1);
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kVersion1, GetVersion(id));
+  EXPECT_EQ(kPayload1, GetPayload(id));
+}
+
+// Fire ten invalidations in a row.  All should be received.
+TEST_F(FCMSyncInvalidationListenerTest, ManyInvalidations_NoDrop) {
+  const int kRepeatCount = 10;
+  const ObjectId& id = kPreferencesId_;
+  int64_t initial_version = kVersion1;
+  for (int64_t i = initial_version; i < initial_version + kRepeatCount; ++i) {
+    FireInvalidate(id, i, kPayload1);
+  }
+  ASSERT_EQ(static_cast<size_t>(kRepeatCount), GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kPayload1, GetPayload(id));
+  EXPECT_EQ(initial_version + kRepeatCount - 1, GetVersion(id));
+}
+
+// Fire an invalidation for an unregistered object ID with a payload.  It should
+// still be processed, and both the payload and the version should be updated.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateBeforeRegistration_Simple) {
+  const ObjectId kUnregisteredId(kChromeSyncSourceId, "unregistered");
+  const ObjectId& id = kUnregisteredId;
+  ObjectIdSet ids;
+  ids.insert(id);
+
+  EXPECT_EQ(0U, GetInvalidationCount(id));
+
+  FireInvalidate(id, kVersion1, kPayload1);
+
+  ASSERT_EQ(0U, GetInvalidationCount(id));
+
+  EnableNotifications();
+  listener_.Ready(fake_invalidation_client_);
+  listener_.UpdateRegisteredIds(ids);
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kVersion1, GetVersion(id));
+  EXPECT_EQ(kPayload1, GetPayload(id));
+}
+
+// Fire ten invalidations before an object registers.  Some invalidations will
+// be dropped an replaced with an unknown version invalidation.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateBeforeRegistration_Drop) {
+  const int kRepeatCount =
+      UnackedInvalidationSet::kMaxBufferedInvalidations + 1;
+  const ObjectId kUnregisteredId(kChromeSyncSourceId, "unregistered");
+  const ObjectId& id = kUnregisteredId;
+  ObjectIdSet ids;
+  ids.insert(id);
+
+  EXPECT_EQ(0U, GetInvalidationCount(id));
+
+  int64_t initial_version = kVersion1;
+  for (int64_t i = initial_version; i < initial_version + kRepeatCount; ++i) {
+    FireInvalidate(id, i, kPayload1);
+  }
+
+  EnableNotifications();
+  listener_.Ready(fake_invalidation_client_);
+  listener_.UpdateRegisteredIds(ids);
+
+  ASSERT_EQ(UnackedInvalidationSet::kMaxBufferedInvalidations,
+            GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(initial_version + kRepeatCount - 1, GetVersion(id));
+  EXPECT_EQ(kPayload1, GetPayload(id));
+  EXPECT_TRUE(StartsWithUnknownVersion(id));
+}
+
+// Fire an invalidation, then fire another one with a lower version.  Both
+// should be received.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateVersion) {
+  const ObjectId& id = kPreferencesId_;
+
+  FireInvalidate(id, kVersion2, kPayload2);
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+  EXPECT_EQ(kVersion2, GetVersion(id));
+  EXPECT_EQ(kPayload2, GetPayload(id));
+
+  FireInvalidate(id, kVersion1, kPayload1);
+
+  ASSERT_EQ(2U, GetInvalidationCount(id));
+  ASSERT_FALSE(IsUnknownVersion(id));
+
+  EXPECT_EQ(kVersion1, GetVersion(id));
+  EXPECT_EQ(kPayload1, GetPayload(id));
+}
+
+// Fire an invalidation with an unknown version.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateUnknownVersion) {
+  const ObjectId& id = kBookmarksId_;
+
+  FireInvalidateUnknownVersion(id);
+
+  ASSERT_EQ(1U, GetInvalidationCount(id));
+  EXPECT_TRUE(IsUnknownVersion(id));
+}
+
+// Fire an invalidation for all enabled IDs.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateAll) {
+  FireInvalidateAll();
+
+  for (ObjectIdSet::const_iterator it = registered_ids_.begin();
+       it != registered_ids_.end(); ++it) {
+    ASSERT_EQ(1U, GetInvalidationCount(*it));
+    EXPECT_TRUE(IsUnknownVersion(*it));
+  }
+}
+
+// Test a simple scenario for multiple IDs.
+TEST_F(FCMSyncInvalidationListenerTest, InvalidateMultipleIds) {
+  FireInvalidate(kBookmarksId_, 3, nullptr);
+
+  ASSERT_EQ(1U, GetInvalidationCount(kBookmarksId_));
+  ASSERT_FALSE(IsUnknownVersion(kBookmarksId_));
+  EXPECT_EQ(3, GetVersion(kBookmarksId_));
+  EXPECT_EQ("", GetPayload(kBookmarksId_));
+
+  // kExtensionId is not registered, so the invalidation should not get through.
+  FireInvalidate(kExtensionsId_, 2, nullptr);
+  ASSERT_EQ(0U, GetInvalidationCount(kExtensionsId_));
+}
+
+// Without readying the client, disable notifications, then enable
+// them.  The listener should still think notifications are disabled.
+TEST_F(FCMSyncInvalidationListenerTest, EnableNotificationsNotReady) {
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  DisableNotifications(notifier::TRANSIENT_NOTIFICATION_ERROR);
+
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  EnableNotifications();
+
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+}
+
+// Enable notifications then Ready the invalidation client.  The
+// delegate should then be ready.
+TEST_F(FCMSyncInvalidationListenerTest, EnableNotificationsThenReady) {
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  EnableNotifications();
+
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  listener_.Ready(fake_invalidation_client_);
+
+  EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
+}
+
+// Ready the invalidation client then enable notifications.  The
+// delegate should then be ready.
+TEST_F(FCMSyncInvalidationListenerTest, ReadyThenEnableNotifications) {
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  listener_.Ready(fake_invalidation_client_);
+
+  EXPECT_EQ(TRANSIENT_INVALIDATION_ERROR, GetInvalidatorState());
+
+  EnableNotifications();
+
+  EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
+}
+
+// Enable notifications and ready the client.  Then disable
+// notifications with an auth error and re-enable notifications.  The
+// delegate should go into an auth error mode and then back out.
+TEST_F(FCMSyncInvalidationListenerTest, PushClientAuthError) {
+  EnableNotifications();
+  listener_.Ready(fake_invalidation_client_);
+
+  EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
+
+  DisableNotifications(notifier::NOTIFICATION_CREDENTIALS_REJECTED);
+
+  EXPECT_EQ(INVALIDATION_CREDENTIALS_REJECTED, GetInvalidatorState());
+
+  EnableNotifications();
+
+  EXPECT_EQ(INVALIDATIONS_ENABLED, GetInvalidatorState());
+}
+
+// A variant of FCMSyncInvalidationListenerTest that starts with some initial
+// state.  We make not attempt to abstract away the contents of this state.  The
+// tests that make use of this harness depend on its implementation details.
+class FCMSyncInvalidationListenerTest_WithInitialState
+    : public FCMSyncInvalidationListenerTest {
+ public:
+  void SetUp() override {
+    UnackedInvalidationSet bm_state(kBookmarksId_);
+    UnackedInvalidationSet ext_state(kExtensionsId_);
+
+    Invalidation bm_unknown = Invalidation::InitUnknownVersion(kBookmarksId_);
+    Invalidation bm_v100 = Invalidation::Init(kBookmarksId_, 100, "hundred");
+    bm_state.Add(bm_unknown);
+    bm_state.Add(bm_v100);
+
+    Invalidation ext_v10 = Invalidation::Init(kExtensionsId_, 10, "ten");
+    Invalidation ext_v20 = Invalidation::Init(kExtensionsId_, 20, "twenty");
+    ext_state.Add(ext_v10);
+    ext_state.Add(ext_v20);
+
+    initial_state.insert(std::make_pair(kBookmarksId_, bm_state));
+    initial_state.insert(std::make_pair(kExtensionsId_, ext_state));
+
+    fake_tracker_.SetSavedInvalidations(initial_state);
+
+    FCMSyncInvalidationListenerTest::SetUp();
+  }
+
+  UnackedInvalidationsMap initial_state;
+};
+
+}  // namespace
+
+}  // namespace syncer

components/invalidation/impl/per_user_topic_registration_manager.h

@@ -45,7 +45,7 @@ class INVALIDATION_EXPORT PerUserTopicRegistrationManager {
 
   static void RegisterProfilePrefs(PrefRegistrySimple* registry);
 
-  void UpdateRegisteredIds(const InvalidationObjectIdSet& ids);
+  virtual void UpdateRegisteredIds(const InvalidationObjectIdSet& ids);
 
   InvalidationObjectIdSet GetRegisteredIds() const;
 

components/invalidation/impl/sync_system_resources.cc

@@ -287,9 +287,10 @@ SyncSystemResources::SyncSystemResources(
       logger_(new SyncLogger()),
       internal_scheduler_(new SyncInvalidationScheduler()),
       listener_scheduler_(new SyncInvalidationScheduler()),
-      storage_(new SyncStorage(state_writer, internal_scheduler_.get())),
-      sync_network_channel_(sync_network_channel) {
-}
+      storage_(state_writer
+                   ? new SyncStorage(state_writer, internal_scheduler_.get())
+                   : nullptr),
+      sync_network_channel_(sync_network_channel) {}
 
 SyncSystemResources::~SyncSystemResources() {
   Stop();
@@ -323,7 +324,7 @@ SyncLogger* SyncSystemResources::logger() {
 }
 
 SyncStorage* SyncSystemResources::storage() {
-  return storage_.get();
+  return storage_ ? storage_.get() : nullptr;
 }
 
 SyncNetworkChannel* SyncSystemResources::network() {