Implement sync in the NonBlockingTypeProcessor

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 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));
+}
+
+}  // namespace
+
 ModelTypeRegistry::ModelTypeRegistry() : directory_(NULL) {}
 
 ModelTypeRegistry::ModelTypeRegistry(
     const std::vector<scoped_refptr<ModelSafeWorker> >& workers,
     syncable::Directory* directory)
     : directory_(directory) {
   for (size_t i = 0u; i < workers.size(); ++i) {
     workers_map_.insert(
         std::make_pair(workers[i]->GetModelSafeGroup(), workers[i]));
   }
@@ skipping 61 matching lines @@
     DCHECK(inserted2) << "Attempt to override existing type handler in map";
   }
 
   enabled_directory_types_ = GetRoutingInfoTypes(routing_info);
   DCHECK(Intersection(GetEnabledDirectoryTypes(),
                       GetEnabledNonBlockingTypes()).Empty());
 }
 
 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<NonBlockingTypeProcessorCore> core(
       new NonBlockingTypeProcessorCore(type, type_task_runner, processor));
 
+  // TODO(rlarocque): DataTypeState should be forwarded to core here.
+
   // Initialize Processor -> CoreProcessor communication channel.
-  type_task_runner->PostTask(
-      FROM_HERE,
-      base::Bind(&NonBlockingTypeProcessor::OnConnect,
-                 processor,
-                 core->AsWeakPtr(),
-                 scoped_refptr<base::SequencedTaskRunner>(
-                     base::MessageLoopProxy::current())));
+  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,
+                                        base::Passed(&core_interface)));
 
   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());
 
@@ skipping 74 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