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

6 files changed, 158 insertions, 12 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.