sync: Add NonBlockingSyncTypeProcessorCore

Adds the class for handling sync updates and commits on the sync thread.
The current implementation is mostly a bunch of stubs.

In the future, this will be hooked up to a NonBlockingSyncTypeProcessor
that lives on the model type's thread.  These two classes will
collaborate to implement the non-blocking sync system.

This CL also adds some code to the ModelTypeRegistry and
SyncSessionContext to prepare for managing these
NonBlockingSyncTypeProcessor objects.  This new code is used only in
tests.

BUG=351005

Review URL: https://codereview.chromium.org/192953008

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@258035 0039d316-1c4b-4281-b951-d872f2087c98

12 files changed, 330 insertions, 36 deletions

diff --git a/sync/engine/commit_contributor.h b/sync/engine/commit_contributor.h
index 2d190af..5e9d156 100644
--- a/sync/engine/commit_contributor.h
+++ b/sync/engine/commit_contributor.h
@@ -8,6 +8,7 @@
 #include <cstddef>
 
 #include "base/memory/scoped_ptr.h"
+#include "sync/base/sync_export.h"
 
 namespace syncer {
 
@@ -21,7 +22,7 @@ class Directory;
 //
 // When asked, it can return CommitContribution objects that contain a set of
 // items to be committed from this source.
-class CommitContributor {
+class SYNC_EXPORT_PRIVATE CommitContributor {
  public:
   CommitContributor();
   virtual ~CommitContributor() = 0;
diff --git a/sync/engine/non_blocking_type_processor_core.cc b/sync/engine/non_blocking_type_processor_core.cc
new file mode 100644
index 0000000..6d6abc4
--- /dev/null
+++ b/sync/engine/non_blocking_type_processor_core.cc
@@ -0,0 +1,72 @@
+// 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 "sync/engine/non_blocking_type_processor_core.h"
+
+#include "base/logging.h"
+#include "sync/engine/commit_contribution.h"
+
+namespace syncer {
+
+NonBlockingTypeProcessorCore::NonBlockingTypeProcessorCore(
+    ModelType type) : type_(type), weak_ptr_factory_(this) {
+  progress_marker_.set_data_type_id(GetSpecificsFieldNumberFromModelType(type));
+}
+
+NonBlockingTypeProcessorCore::~NonBlockingTypeProcessorCore() {
+}
+
+ModelType NonBlockingTypeProcessorCore::GetModelType() const {
+  DCHECK(CalledOnValidThread());
+  return type_;
+}
+
+// UpdateHandler implementation.
+void NonBlockingTypeProcessorCore::GetDownloadProgress(
+    sync_pb::DataTypeProgressMarker* progress_marker) const {
+  DCHECK(CalledOnValidThread());
+  // TODO(rlarocque): Implement this properly.  crbug.com/351005.
+  VLOG(1) << "Getting progress for: " << ModelTypeToString(type_);
+  *progress_marker = progress_marker_;
+}
+
+void NonBlockingTypeProcessorCore::ProcessGetUpdatesResponse(
+    const sync_pb::DataTypeProgressMarker& progress_marker,
+    const SyncEntityList& applicable_updates,
+    sessions::StatusController* status) {
+  DCHECK(CalledOnValidThread());
+  // TODO(rlarocque): Implement this properly.  crbug.com/351005.
+  VLOG(1) << "Processing updates response for: " << ModelTypeToString(type_);
+  progress_marker_ = progress_marker;
+}
+
+void NonBlockingTypeProcessorCore::ApplyUpdates(
+    sessions::StatusController* status) {
+  DCHECK(CalledOnValidThread());
+  // TODO(rlarocque): Implement this properly.  crbug.com/351005.
+  VLOG(1) << "Applying updates for: " << ModelTypeToString(type_);
+}
+
+void NonBlockingTypeProcessorCore::PassiveApplyUpdates(
+    sessions::StatusController* status) {
+  NOTREACHED()
+      << "Non-blocking types should never apply updates on sync thread.  "
+      << "ModelType is: " << ModelTypeToString(type_);
+}
+
+// CommitContributor implementation.
+scoped_ptr<CommitContribution>
+NonBlockingTypeProcessorCore::GetContribution(size_t max_entries) {
+  DCHECK(CalledOnValidThread());
+  // TODO(rlarocque): Implement this properly.  crbug.com/351005.
+  VLOG(1) << "Getting commit contribution for: " << ModelTypeToString(type_);
+  return scoped_ptr<CommitContribution>();
+}
+
+base::WeakPtr<NonBlockingTypeProcessorCore>
+NonBlockingTypeProcessorCore::AsWeakPtr() {
+  return weak_ptr_factory_.GetWeakPtr();
+}
+
+}  // namespace syncer
diff --git a/sync/engine/non_blocking_type_processor_core.h b/sync/engine/non_blocking_type_processor_core.h
new file mode 100644
index 0000000..028bb3a
--- /dev/null
+++ b/sync/engine/non_blocking_type_processor_core.h
@@ -0,0 +1,73 @@
+// 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.
+
+#ifndef SYNC_ENGINE_NON_BLOCKING_TYPE_PROCESSOR_CORE_H_
+#define SYNC_ENGINE_NON_BLOCKING_TYPE_PROCESSOR_CORE_H_
+
+#include "base/memory/weak_ptr.h"
+#include "base/threading/non_thread_safe.h"
+#include "sync/base/sync_export.h"
+#include "sync/engine/commit_contributor.h"
+#include "sync/engine/update_handler.h"
+#include "sync/internal_api/public/base/model_type.h"
+#include "sync/protocol/sync.pb.h"
+
+namespace syncer {
+
+// A smart cache for sync types that use message passing (rather than
+// transactions and the syncable::Directory) to communicate with the sync
+// thread.
+//
+// When the non-blocking sync type wants to talk with the sync server, it will
+// send a message from its thread to this object on the sync thread.  This
+// object is responsible for helping to ensure the appropriate sync server
+// communication gets scheduled and executed.  The response, if any, will be
+// returned to the non-blocking sync type's thread eventually.
+//
+// This object also has a role to play in communications in the opposite
+// direction.  Sometimes the sync thread will receive changes from the sync
+// server and deliver them here.  This object will post this information back to
+// the appropriate component on the model type's thread.
+//
+// This object does more than just pass along messages.  It understands the sync
+// protocol, and it can make decisions when it sees conflicting messages.  For
+// example, if the sync server sends down an update for a sync entity that is
+// currently pending for commit, this object will detect this condition and
+// cancel the pending commit.
+class SYNC_EXPORT_PRIVATE NonBlockingTypeProcessorCore
+    : public UpdateHandler,
+      public CommitContributor,
+      public base::NonThreadSafe {
+ public:
+  explicit NonBlockingTypeProcessorCore(ModelType type);
+  virtual ~NonBlockingTypeProcessorCore();
+
+  ModelType GetModelType() const;
+
+  // UpdateHandler implementation.
+  virtual void GetDownloadProgress(
+      sync_pb::DataTypeProgressMarker* progress_marker) const OVERRIDE;
+  virtual void ProcessGetUpdatesResponse(
+      const sync_pb::DataTypeProgressMarker& progress_marker,
+      const SyncEntityList& applicable_updates,
+      sessions::StatusController* status) OVERRIDE;
+  virtual void ApplyUpdates(sessions::StatusController* status) OVERRIDE;
+  virtual void PassiveApplyUpdates(sessions::StatusController* status) OVERRIDE;
+
+  // CommitContributor implementation.
+  virtual scoped_ptr<CommitContribution> GetContribution(
+      size_t max_entries) OVERRIDE;
+
+  base::WeakPtr<NonBlockingTypeProcessorCore> AsWeakPtr();
+
+ private:
+  ModelType type_;
+  sync_pb::DataTypeProgressMarker progress_marker_;
+
+  base::WeakPtrFactory<NonBlockingTypeProcessorCore> weak_ptr_factory_;
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_ENGINE_NON_BLOCKING_TYPE_PROCESSOR_CORE_H_
diff --git a/sync/engine/sync_scheduler_impl.cc b/sync/engine/sync_scheduler_impl.cc
index 6ff250f..a9c6359 100644
--- a/sync/engine/sync_scheduler_impl.cc
+++ b/sync/engine/sync_scheduler_impl.cc
@@ -245,7 +245,7 @@ void SyncSchedulerImpl::Start(Mode mode) {
 }
 
 ModelTypeSet SyncSchedulerImpl::GetEnabledAndUnthrottledTypes() {
-  ModelTypeSet enabled_types = session_context_->enabled_types();
+  ModelTypeSet enabled_types = session_context_->GetEnabledTypes();
   ModelTypeSet enabled_protocol_types =
       Intersection(ProtocolTypes(), enabled_types);
   ModelTypeSet throttled_types = nudge_tracker_.GetThrottledTypes();
@@ -341,7 +341,7 @@ bool SyncSchedulerImpl::CanRunNudgeJobNow(JobPriority priority) {
     return false;
   }
 
-  const ModelTypeSet enabled_types = session_context_->enabled_types();
+  const ModelTypeSet enabled_types = session_context_->GetEnabledTypes();
   if (nudge_tracker_.GetThrottledTypes().HasAll(enabled_types)) {
     SDVLOG(1) << "Not running a nudge because we're fully type throttled.";
     return false;
@@ -458,7 +458,7 @@ void SyncSchedulerImpl::DoNudgeSyncSessionJob(JobPriority priority) {
   DCHECK(CanRunNudgeJobNow(priority));
 
   DVLOG(2) << "Will run normal mode sync cycle with types "
-           << ModelTypeSetToString(session_context_->enabled_types());
+           << ModelTypeSetToString(session_context_->GetEnabledTypes());
   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
   bool premature_exit = !syncer_->NormalSyncShare(
       GetEnabledAndUnthrottledTypes(),
@@ -501,10 +501,10 @@ void SyncSchedulerImpl::DoConfigurationSyncSessionJob(JobPriority priority) {
   }
 
   SDVLOG(2) << "Will run configure SyncShare with types "
-            << ModelTypeSetToString(session_context_->enabled_types());
+            << ModelTypeSetToString(session_context_->GetEnabledTypes());
   scoped_ptr<SyncSession> session(SyncSession::Build(session_context_, this));
   bool premature_exit = !syncer_->ConfigureSyncShare(
-      session_context_->enabled_types(),
+      session_context_->GetEnabledTypes(),
       pending_configure_params_->source,
       session.get());
   AdjustPolling(FORCE_RESET);
diff --git a/sync/engine/syncer.cc b/sync/engine/syncer.cc
index c752510..508a2d9 100644
--- a/sync/engine/syncer.cc
+++ b/sync/engine/syncer.cc
@@ -168,7 +168,7 @@ SyncerError Syncer::BuildAndPostCommits(ModelTypeSet requested_types,
     scoped_ptr<Commit> commit(
         Commit::Init(
             requested_types,
-            session->context()->enabled_types(),
+            session->context()->GetEnabledTypes(),
             session->context()->max_commit_batch_size(),
             session->context()->account_name(),
             session->context()->directory()->cache_guid(),
diff --git a/sync/engine/syncer_unittest.cc b/sync/engine/syncer_unittest.cc
index 2be9dab..d0a2ab4 100644
--- a/sync/engine/syncer_unittest.cc
+++ b/sync/engine/syncer_unittest.cc
@@ -194,7 +194,7 @@ class SyncerTest : public testing::Test,
 
     EXPECT_TRUE(
         syncer_->NormalSyncShare(
-            context_->enabled_types(),
+            context_->GetEnabledTypes(),
             nudge_tracker_,
             session_.get()));
   }
@@ -202,7 +202,7 @@ class SyncerTest : public testing::Test,
   void SyncShareConfigure() {
     ResetSession();
     EXPECT_TRUE(syncer_->ConfigureSyncShare(
-            context_->enabled_types(),
+            context_->GetEnabledTypes(),
             sync_pb::GetUpdatesCallerInfo::RECONFIGURATION,
             session_.get()));
   }
@@ -551,10 +551,10 @@ TEST_F(SyncerTest, GetCommitIdsFiltersThrottledEntries) {
 
   // Now sync without enabling bookmarks.
   mock_server_->ExpectGetUpdatesRequestTypes(
-      Difference(context_->enabled_types(), ModelTypeSet(BOOKMARKS)));
+      Difference(context_->GetEnabledTypes(), ModelTypeSet(BOOKMARKS)));
   ResetSession();
   syncer_->NormalSyncShare(
-      Difference(context_->enabled_types(), ModelTypeSet(BOOKMARKS)),
+      Difference(context_->GetEnabledTypes(), ModelTypeSet(BOOKMARKS)),
       nudge_tracker_,
       session_.get());
 
@@ -567,7 +567,7 @@ TEST_F(SyncerTest, GetCommitIdsFiltersThrottledEntries) {
   }
 
   // Sync again with bookmarks enabled.
-  mock_server_->ExpectGetUpdatesRequestTypes(context_->enabled_types());
+  mock_server_->ExpectGetUpdatesRequestTypes(context_->GetEnabledTypes());
   SyncShareNudge();
   {
     // It should have been committed.
diff --git a/sync/sessions/model_type_registry.cc b/sync/sessions/model_type_registry.cc
index 7035755..ab93c68 100644
--- a/sync/sessions/model_type_registry.cc
+++ b/sync/sessions/model_type_registry.cc
@@ -6,15 +6,14 @@
 
 #include "sync/engine/directory_commit_contributor.h"
 #include "sync/engine/directory_update_handler.h"
+#include "sync/engine/non_blocking_type_processor_core.h"
 
 namespace syncer {
 
 ModelTypeRegistry::ModelTypeRegistry(
     const std::vector<scoped_refptr<ModelSafeWorker> >& workers,
     syncable::Directory* directory)
-    : update_handler_deleter_(&update_handler_map_),
-      commit_contributor_deleter_(&commit_contributor_map_),
-      directory_(directory) {
+    : directory_(directory) {
   for (size_t i = 0u; i < workers.size(); ++i) {
     workers_map_.insert(
         std::make_pair(workers[i]->GetModelSafeGroup(), workers[i]));
@@ -25,11 +24,21 @@ ModelTypeRegistry::~ModelTypeRegistry() {}
 
 void ModelTypeRegistry::SetEnabledDirectoryTypes(
     const ModelSafeRoutingInfo& routing_info) {
-  STLDeleteValues(&update_handler_map_);
-  STLDeleteValues(&commit_contributor_map_);
-  update_handler_map_.clear();
-  commit_contributor_map_.clear();
+  // Remove all existing directory processors and delete them.
+  for (ModelTypeSet::Iterator it = enabled_directory_types_.First();
+       it.Good(); it.Inc()) {
+    size_t result1 = update_handler_map_.erase(it.Get());
+    size_t result2 = commit_contributor_map_.erase(it.Get());
+    DCHECK_EQ(1U, result1);
+    DCHECK_EQ(1U, result2);
+  }
+
+  // Clear the old instances of directory update handlers and commit
+  // contributors, deleting their contents in the processs.
+  directory_update_handlers_.clear();
+  directory_commit_contributors_.clear();
 
+  // Create new ones and add them to the appropriate containers.
   for (ModelSafeRoutingInfo::const_iterator routing_iter = routing_info.begin();
        routing_iter != routing_info.end(); ++routing_iter) {
     ModelType type = routing_iter->first;
@@ -44,6 +53,10 @@ void ModelTypeRegistry::SetEnabledDirectoryTypes(
     DirectoryUpdateHandler* updater =
         new DirectoryUpdateHandler(directory_, type, worker);
 
+    // These containers take ownership of their contents.
+    directory_commit_contributors_.push_back(committer);
+    directory_update_handlers_.push_back(updater);
+
     bool inserted1 =
         update_handler_map_.insert(std::make_pair(type, updater)).second;
     DCHECK(inserted1) << "Attempt to override existing type handler in map";
@@ -51,10 +64,53 @@ void ModelTypeRegistry::SetEnabledDirectoryTypes(
     bool inserted2 =
         commit_contributor_map_.insert(std::make_pair(type, committer)).second;
     DCHECK(inserted2) << "Attempt to override existing type handler in map";
+  }
+
+  enabled_directory_types_ = GetRoutingInfoTypes(routing_info);
+}
+
+void ModelTypeRegistry::InitializeNonBlockingType(syncer::ModelType type) {
+  DVLOG(1) << "Enabling an off-thread sync type: " << ModelTypeToString(type);
+
+  scoped_ptr<NonBlockingTypeProcessorCore> core(
+      new NonBlockingTypeProcessorCore(type));
+
+  DCHECK(update_handler_map_.find(type) == update_handler_map_.end());
+  DCHECK(commit_contributor_map_.find(type) == commit_contributor_map_.end());
+
+  update_handler_map_.insert(std::make_pair(type, core.get()));
+  commit_contributor_map_.insert(std::make_pair(type, core.get()));
+
+  // The container takes ownership.
+  non_blocking_type_processor_cores_.push_back(core.release());
+}
+
+void ModelTypeRegistry::RemoveNonBlockingType(ModelType type) {
+  DVLOG(1) << "Disabling an off-thread sync type: " << ModelTypeToString(type);
+  DCHECK(update_handler_map_.find(type) != update_handler_map_.end());
+  DCHECK(commit_contributor_map_.find(type) != commit_contributor_map_.end());
+
+  size_t updaters_erased = update_handler_map_.erase(type);
+  size_t committers_erased = commit_contributor_map_.erase(type);
 
+  DCHECK_EQ(1U, updaters_erased);
+  DCHECK_EQ(1U, committers_erased);
+
+  // Remove from the ScopedVector, deleting the core in the process.
+  for (ScopedVector<NonBlockingTypeProcessorCore>::iterator it =
+       non_blocking_type_processor_cores_.begin();
+       it != non_blocking_type_processor_cores_.end(); ++it) {
+    if ((*it)->GetModelType() == type) {
+      non_blocking_type_processor_cores_.erase(it);
+      break;
+    }
   }
 }
 
+ModelTypeSet ModelTypeRegistry::GetEnabledTypes() const {
+  return Union(GetEnabledDirectoryTypes(), GetEnabledNonBlockingTypes());
+}
+
 UpdateHandlerMap* ModelTypeRegistry::update_handler_map() {
   return &update_handler_map_;
 }
@@ -63,4 +119,18 @@ CommitContributorMap* ModelTypeRegistry::commit_contributor_map() {
   return &commit_contributor_map_;
 }
 
+ModelTypeSet ModelTypeRegistry::GetEnabledDirectoryTypes() const {
+  return enabled_directory_types_;
+}
+
+ModelTypeSet ModelTypeRegistry::GetEnabledNonBlockingTypes() const {
+  ModelTypeSet enabled_off_thread_types;
+  for (ScopedVector<NonBlockingTypeProcessorCore>::const_iterator it =
+           non_blocking_type_processor_cores_.begin();
+       it != non_blocking_type_processor_cores_.end(); ++it) {
+    enabled_off_thread_types.Put((*it)->GetModelType());
+  }
+  return enabled_off_thread_types;
+}
+
 }  // namespace syncer
diff --git a/sync/sessions/model_type_registry.h b/sync/sessions/model_type_registry.h
index ad34ada..65445d4 100644
--- a/sync/sessions/model_type_registry.h
+++ b/sync/sessions/model_type_registry.h
@@ -9,7 +9,7 @@
 
 #include "base/basictypes.h"
 #include "base/memory/scoped_ptr.h"
-#include "base/stl_util.h"
+#include "base/memory/scoped_vector.h"
 #include "sync/base/sync_export.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/internal_api/public/engine/model_safe_worker.h"
@@ -20,8 +20,11 @@ namespace syncable {
 class Directory;
 }  // namespace syncable
 
-class UpdateHandler;
 class CommitContributor;
+class DirectoryCommitContributor;
+class DirectoryUpdateHandler;
+class NonBlockingTypeProcessorCore;
+class UpdateHandler;
 
 typedef std::map<ModelType, UpdateHandler*> UpdateHandlerMap;
 typedef std::map<ModelType, CommitContributor*> CommitContributorMap;
@@ -37,20 +40,37 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry {
   // Sets the set of enabled types.
   void SetEnabledDirectoryTypes(const ModelSafeRoutingInfo& routing_info);
 
+  // Enables an off-thread type for syncing.
+  //
+  // Expects that the type is not currently enabled.
+  void InitializeNonBlockingType(syncer::ModelType type);
+
+  // Disables the syncing of an off-thread type.
+  //
+  // Expects that the type is currently enabled.
+  // Deletes the processor core associated with the type.
+  void RemoveNonBlockingType(syncer::ModelType type);
+
+  // Gets the set of enabled types.
+  ModelTypeSet GetEnabledTypes() const;
+
   // Simple getters.
   UpdateHandlerMap* update_handler_map();
   CommitContributorMap* commit_contributor_map();
 
  private:
-  // Classes to manage the types hooked up to receive and commit sync data.
-  UpdateHandlerMap update_handler_map_;
-  CommitContributorMap commit_contributor_map_;
+  ModelTypeSet GetEnabledNonBlockingTypes() const;
+  ModelTypeSet GetEnabledDirectoryTypes() const;
 
-  // Deleter for the |update_handler_map_|.
-  STLValueDeleter<UpdateHandlerMap> update_handler_deleter_;
+  // Sets of handlers and contributors.
+  ScopedVector<DirectoryCommitContributor> directory_commit_contributors_;
+  ScopedVector<DirectoryUpdateHandler> directory_update_handlers_;
+  ScopedVector<NonBlockingTypeProcessorCore> non_blocking_type_processor_cores_;
 
-  // Deleter for the |commit_contributor_map_|.
-  STLValueDeleter<CommitContributorMap> commit_contributor_deleter_;
+  // Maps of UpdateHandlers and CommitContributors.
+  // They do not own any of the objects they point to.
+  UpdateHandlerMap update_handler_map_;
+  CommitContributorMap commit_contributor_map_;
 
   // The known ModelSafeWorkers.
   std::map<ModelSafeGroup, scoped_refptr<ModelSafeWorker> > workers_map_;
@@ -58,6 +78,9 @@ class SYNC_EXPORT_PRIVATE ModelTypeRegistry {
   // The directory.  Not owned.
   syncable::Directory* directory_;
 
+  // The set of enabled directory types.
+  ModelTypeSet enabled_directory_types_;
+
   DISALLOW_COPY_AND_ASSIGN(ModelTypeRegistry);
 };
 
diff --git a/sync/sessions/model_type_registry_unittest.cc b/sync/sessions/model_type_registry_unittest.cc
index a063758..7f174e5 100644
--- a/sync/sessions/model_type_registry_unittest.cc
+++ b/sync/sessions/model_type_registry_unittest.cc
@@ -5,6 +5,7 @@
 #include <vector>
 
 #include "base/message_loop/message_loop.h"
+#include "sync/engine/non_blocking_type_processor_core.h"
 #include "sync/internal_api/public/base/model_type.h"
 #include "sync/sessions/model_type_registry.h"
 #include "sync/test/engine/fake_model_worker.h"
@@ -111,4 +112,59 @@ TEST_F(ModelTypeRegistryTest, SetEnabledDirectoryTypes_Clear) {
   registry()->SetEnabledDirectoryTypes(routing_info2);
 }
 
+TEST_F(ModelTypeRegistryTest, NonBlockingTypes) {
+  EXPECT_TRUE(registry()->GetEnabledTypes().Empty());
+
+  registry()->InitializeNonBlockingType(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::THEMES)));
+
+  registry()->InitializeNonBlockingType(syncer::SESSIONS);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::THEMES, syncer::SESSIONS)));
+
+  registry()->RemoveNonBlockingType(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(
+      ModelTypeSet(syncer::SESSIONS)));
+
+  // Allow ModelTypeRegistry destruction to delete the
+  // Sessions' NonBlockingTypeProcessorCore.
+}
+
+TEST_F(ModelTypeRegistryTest, NonBlockingTypesWithDirectoryTypes) {
+  ModelSafeRoutingInfo routing_info1;
+  routing_info1.insert(std::make_pair(NIGORI, GROUP_PASSIVE));
+  routing_info1.insert(std::make_pair(BOOKMARKS, GROUP_UI));
+  routing_info1.insert(std::make_pair(AUTOFILL, GROUP_DB));
+
+  ModelTypeSet current_types;
+  EXPECT_TRUE(registry()->GetEnabledTypes().Empty());
+
+  // Add the themes non-blocking type.
+  registry()->InitializeNonBlockingType(syncer::THEMES);
+  current_types.Put(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Add some directory types.
+  registry()->SetEnabledDirectoryTypes(routing_info1);
+  current_types.PutAll(GetRoutingInfoTypes(routing_info1));
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Add sessions non-blocking type.
+  registry()->InitializeNonBlockingType(syncer::SESSIONS);
+  current_types.Put(syncer::SESSIONS);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Remove themes non-blocking type.
+  registry()->RemoveNonBlockingType(syncer::THEMES);
+  current_types.Remove(syncer::THEMES);
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+
+  // Clear all directory types.
+  ModelSafeRoutingInfo routing_info2;
+  registry()->SetEnabledDirectoryTypes(routing_info2);
+  current_types.RemoveAll(GetRoutingInfoTypes(routing_info1));
+  EXPECT_TRUE(registry()->GetEnabledTypes().Equals(current_types));
+}
+
 }  // namespace syncer
diff --git a/sync/sessions/sync_session_context.cc b/sync/sessions/sync_session_context.cc
index 36e6343..54ae473 100644
--- a/sync/sessions/sync_session_context.cc
+++ b/sync/sessions/sync_session_context.cc
@@ -42,9 +42,12 @@ SyncSessionContext::SyncSessionContext(
 SyncSessionContext::~SyncSessionContext() {
 }
 
+ModelTypeSet SyncSessionContext::GetEnabledTypes() const {
+  return model_type_registry_->GetEnabledTypes();
+}
+
 void SyncSessionContext::SetRoutingInfo(
     const ModelSafeRoutingInfo& routing_info) {
-  enabled_types_ = GetRoutingInfoTypes(routing_info);
   model_type_registry_->SetEnabledDirectoryTypes(routing_info);
 }
 
diff --git a/sync/sessions/sync_session_context.h b/sync/sessions/sync_session_context.h
index f1afa7d..c7661ce 100644
--- a/sync/sessions/sync_session_context.h
+++ b/sync/sessions/sync_session_context.h
@@ -62,9 +62,7 @@ class SYNC_EXPORT_PRIVATE SyncSessionContext {
     return directory_;
   }
 
-  ModelTypeSet enabled_types() const {
-    return enabled_types_;
-  }
+  ModelTypeSet GetEnabledTypes() const;
 
   void SetRoutingInfo(const ModelSafeRoutingInfo& routing_info);
 
@@ -142,10 +140,6 @@ class SYNC_EXPORT_PRIVATE SyncSessionContext {
   ServerConnectionManager* const connection_manager_;
   syncable::Directory* const directory_;
 
-  // The set of enabled types.  Derrived from the routing info set with
-  // set_routing_info().
-  ModelTypeSet enabled_types_;
-
   // We use this to stuff extensions activity into CommitMessages so the server
   // can correlate commit traffic with extension-related bookmark mutations.
   scoped_refptr<ExtensionsActivity> extensions_activity_;
diff --git a/sync/sync_core.gypi b/sync/sync_core.gypi
index e973110..5ca59ee5 100644
--- a/sync/sync_core.gypi
+++ b/sync/sync_core.gypi
@@ -64,6 +64,8 @@
     'engine/net/server_connection_manager.h',
     'engine/net/url_translator.cc',
     'engine/net/url_translator.h',
+    'engine/non_blocking_type_processor_core.cc',
+    'engine/non_blocking_type_processor_core.h',
     'engine/nudge_source.cc',
     'engine/nudge_source.h',
     'engine/process_updates_util.cc',