sync: Implement NonBlockingTypeProcessorCore

sync/sessions/model_type_registry.cc

 // 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/sessions/model_type_registry.h"
 
 #include "base/bind.h"
 #include "base/message_loop/message_loop_proxy.h"
 #include "base/observer_list.h"
 #include "sync/engine/directory_commit_contributor.h"
 #include "sync/engine/directory_update_handler.h"
 #include "sync/engine/non_blocking_sync_common.h"
 #include "sync/engine/non_blocking_type_processor.h"
 #include "sync/engine/non_blocking_type_processor_core.h"
 #include "sync/engine/non_blocking_type_processor_core_interface.h"
 #include "sync/sessions/directory_type_debug_info_emitter.h"
 
 namespace syncer {
 
 namespace {
 
+class NonBlockingTypeProcessorWrapper
+    : public NonBlockingTypeProcessorInterface {
+ public:
+  NonBlockingTypeProcessorWrapper(
+      base::WeakPtr<NonBlockingTypeProcessor> processor,
+      scoped_refptr<base::SequencedTaskRunner> processor_task_runner);
+  virtual ~NonBlockingTypeProcessorWrapper();
+
+  virtual void ReceiveCommitResponse(
+      const DataTypeState& type_state,
+      const CommitResponseDataList& response_list) OVERRIDE;
+  virtual void ReceiveUpdateResponse(
+      const DataTypeState& type_state,
+      const UpdateResponseDataList& response_list) OVERRIDE;
+
+ private:
+  base::WeakPtr<NonBlockingTypeProcessor> processor_;
+  scoped_refptr<base::SequencedTaskRunner> processor_task_runner_;
+};
+
+NonBlockingTypeProcessorWrapper::NonBlockingTypeProcessorWrapper(
+    base::WeakPtr<NonBlockingTypeProcessor> processor,
+    scoped_refptr<base::SequencedTaskRunner> processor_task_runner)
+    : processor_(processor), processor_task_runner_(processor_task_runner) {
+}
+
+NonBlockingTypeProcessorWrapper::~NonBlockingTypeProcessorWrapper() {
+}
+
+void NonBlockingTypeProcessorWrapper::ReceiveCommitResponse(
+    const DataTypeState& type_state,
+    const CommitResponseDataList& response_list) {
+  processor_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&NonBlockingTypeProcessor::OnCommitCompletion,
+                 processor_,
+                 type_state,
+                 response_list));
+}
+
+void NonBlockingTypeProcessorWrapper::ReceiveUpdateResponse(
+    const DataTypeState& type_state,
+    const UpdateResponseDataList& response_list) {
+  processor_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(&NonBlockingTypeProcessor::OnUpdateReceived,
+                 processor_,
+                 type_state,
+                 response_list));
+}
+
 class NonBlockingTypeProcessorCoreWrapper
     : public NonBlockingTypeProcessorCoreInterface {
  public:
   NonBlockingTypeProcessorCoreWrapper(
       base::WeakPtr<NonBlockingTypeProcessorCore> core,
       scoped_refptr<base::SequencedTaskRunner> sync_thread);
   virtual ~NonBlockingTypeProcessorCoreWrapper();
 
   virtual void RequestCommits(const CommitRequestDataList& list) OVERRIDE;
 
  private:
   base::WeakPtr<NonBlockingTypeProcessorCore> core_;
   scoped_refptr<base::SequencedTaskRunner> sync_thread_;
 };
 
 NonBlockingTypeProcessorCoreWrapper::NonBlockingTypeProcessorCoreWrapper(
     base::WeakPtr<NonBlockingTypeProcessorCore> core,
     scoped_refptr<base::SequencedTaskRunner> sync_thread)
     : core_(core), sync_thread_(sync_thread) {
 }
 
 NonBlockingTypeProcessorCoreWrapper::~NonBlockingTypeProcessorCoreWrapper() {
 }
 
 void NonBlockingTypeProcessorCoreWrapper::RequestCommits(
     const CommitRequestDataList& list) {
   sync_thread_->PostTask(
       FROM_HERE,
-      base::Bind(&NonBlockingTypeProcessorCore::RequestCommits, core_, list));
+      base::Bind(&NonBlockingTypeProcessorCore::EnqueueForCommit, core_, list));
 }
 
 }  // namespace
 
 ModelTypeRegistry::ModelTypeRegistry() : directory_(NULL) {}
 
 ModelTypeRegistry::ModelTypeRegistry(
     const std::vector<scoped_refptr<ModelSafeWorker> >& workers,
     syncable::Directory* directory)
     : directory_(directory) {
@@ skipping 71 matching lines @@
 }
 
 void ModelTypeRegistry::InitializeNonBlockingType(
     ModelType type,
     const DataTypeState& data_type_state,
     scoped_refptr<base::SequencedTaskRunner> type_task_runner,
     base::WeakPtr<NonBlockingTypeProcessor> processor) {
   DVLOG(1) << "Enabling an off-thread sync type: " << ModelTypeToString(type);
 
   // Initialize CoreProcessor -> Processor communication channel.
+  scoped_ptr<NonBlockingTypeProcessorInterface> processor_interface(
+      new NonBlockingTypeProcessorWrapper(processor, type_task_runner));
   scoped_ptr<NonBlockingTypeProcessorCore> core(
-      new NonBlockingTypeProcessorCore(type, type_task_runner, processor));
-
-  // TODO(rlarocque): DataTypeState should be forwarded to core here.
+      new NonBlockingTypeProcessorCore(
+          type, data_type_state, processor_interface.Pass()));
 
   // Initialize Processor -> CoreProcessor communication channel.
   scoped_ptr<NonBlockingTypeProcessorCoreInterface> core_interface(
       new NonBlockingTypeProcessorCoreWrapper(
           core->AsWeakPtr(),
           scoped_refptr<base::SequencedTaskRunner>(
               base::MessageLoopProxy::current())));
   type_task_runner->PostTask(FROM_HERE,
                              base::Bind(&NonBlockingTypeProcessor::OnConnect,
                                         processor,
@@ skipping 85 matching lines @@
   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