sync: Add NonBlockingSyncTypeProcessorCore (again)

This is a re-land of r258035, which was mistakenly reverted with
r258266.  Original CL description follows.

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.

TBR=zea@chromium.org
BUG=351005

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

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@258390 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',