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

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.