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

[Nicolas Zea, 2014/05/15 21:58:55]

Is this that much better than having NonBlockingTypeCore implement the interface
directly? I find this quite confusing, and it took me a while to figure out how
the core interface actually related to the core.

Why not use the normal Impl approach over this?

[rlarocque, 2014/05/15 22:28:54]

The most practical reason is that it allows me to hide both the object and the
thread that it lives on.  The NonBlockingTypeProcessor doesn't need to use
PostTask() when it sends messages to the NonBlockingTypeProcessorCore because
the wrapper hides that detail.

This makes unit tests a lot simpler.  It's possible to remove all task posting
from the interaction with the MockNonBlockingTypeProcessorCore.

[Nicolas Zea, 2014/05/19 23:54:15]

How about changing the naming then. It seems that the Interface is really a sync
engine construct. Will it eventually have more than just RequestCommits? If not,
naming it something based on the fact that it's a commit delegator of some sort
makes more sense.

Part of my confusion is that typically calling something XInterface means that
the X is actually implementing the interface, which isn't really true. This is
just a delegation class.

[rlarocque, 2014/05/20 01:39:47]

Right.  Let's discuss naming tomorrow.  There will be a few of these, so we
should figure out some standard way to name the pattern.

+    : 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.
+  (void)data_type_state;

[Nicolas Zea, 2014/05/19 23:54:15]

what is this doing?

[rlarocque, 2014/05/20 01:39:47]

Silences an unused variable warning.  I seem to remember that certain platforms
would choke on that warning in certain contexts.

This is my way of saying: "Yes, I know this variable isn't used anywhere, and
that's OK".  Hopefully it's informative to both readers and the compiler.

[Nicolas Zea, 2014/05/20 21:51:08]

I don't think the warning happens for parameters, just locally declared
variables. Can you double check that this is necessary?

[rlarocque, 2014/05/20 22:29:21]

Updated in the latest patch.  Let's see if the trybots accept it.

+
   // 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