sync: Implement NonBlockingTypeProcessorCore

sync/engine/non_blocking_type_processor.cc

Index: sync/engine/non_blocking_type_processor.cc
diff --git a/sync/engine/non_blocking_type_processor.cc b/sync/engine/non_blocking_type_processor.cc
index 0aaead452cf5f34febe2c9b2b1a536fa8e6b3f0e..c1982b3ec9f002cd2050cab14237d4e92aab39f3 100644
--- a/sync/engine/non_blocking_type_processor.cc
+++ b/sync/engine/non_blocking_type_processor.cc
@@ -51,7 +51,6 @@ void NonBlockingTypeProcessor::Enable(
   // TODO(rlarocque): At some point, this should be loaded from storage.
   data_type_state_.progress_marker.set_data_type_id(
       GetSpecificsFieldNumberFromModelType(type_));
-  data_type_state_.next_client_id = 0;
 
   sync_core_proxy_ = sync_core_proxy.Pass();
   sync_core_proxy_->ConnectTypeToCore(GetModelType(),
@@ -142,6 +141,10 @@ void NonBlockingTypeProcessor::FlushPendingCommitRequests() {
   if (!IsConnected())
     return;
 
+  // Don't send anything if the type is not ready to handle commits.
+  if (!data_type_state_.initial_sync_done)
+    return;
+
   // TODO(rlarocque): Do something smarter than iterate here.
   for (EntityMap::iterator it = entities_.begin(); it != entities_.end();
        ++it) {
@@ -184,6 +187,9 @@ void NonBlockingTypeProcessor::OnCommitCompletion(
 void NonBlockingTypeProcessor::OnUpdateReceived(
     const DataTypeState& data_type_state,
     const UpdateResponseDataList& response_list) {
+  bool initial_sync_just_finished =
+      !data_type_state_.initial_sync_done && data_type_state.initial_sync_done;
+
   data_type_state_ = data_type_state;
 
   for (UpdateResponseDataList::const_iterator list_it = response_list.begin();
@@ -215,6 +221,9 @@ void NonBlockingTypeProcessor::OnUpdateReceived(
     }
   }
 
+  if (initial_sync_just_finished)
+    FlushPendingCommitRequests();
+
   // TODO: Inform the model of the new or updated data.
 }