sync: Implement per-type update processing

sync/engine/download.cc

Index: sync/engine/download.cc
diff --git a/sync/engine/download.cc b/sync/engine/download.cc
index acda8349edcf2981458b0b680a973da5c4a32b3e..d590a60a6d076249aba0fd8b8723cc1882d6a6d5 100644
--- a/sync/engine/download.cc
+++ b/sync/engine/download.cc
@@ -7,8 +7,8 @@
 #include <string>
 
 #include "base/command_line.h"
-#include "sync/engine/process_updates_command.h"
-#include "sync/engine/store_timestamps_command.h"
+#include "sync/engine/process_updates_util.h"
+#include "sync/engine/sync_directory_update_handler.h"
 #include "sync/engine/syncer.h"
 #include "sync/engine/syncer_proto_util.h"
 #include "sync/sessions/nudge_tracker.h"
@@ -27,6 +27,9 @@ using std::string;
 
 namespace {
 
+typedef std::map<ModelType, const sync_pb::DataTypeProgressMarker*>
+    TypeProgressMarkerMap;
+
 SyncerError HandleGetEncryptionKeyResponse(
     const sync_pb::ClientToServerResponse& update_response,
     syncable::Directory* dir) {
@@ -135,6 +138,76 @@ void InitDownloadUpdatesRequest(
   }
 }
 
+// Builds a map of ModelTypes to pointers to progress markers in the given
+// update response.  The map is returned in the |marker_map| parameter.
+void PartitionProgressMarkersByType(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet request_types,
+    TypeProgressMarkerMap* marker_map) {
+  for (int i = 0; i < gu_response.new_progress_marker_size(); ++i) {
+    int field_number = gu_response.new_progress_marker(i).data_type_id();
+    ModelType model_type = GetModelTypeFromSpecificsFieldNumber(field_number);
+    if (!IsRealDataType(model_type)) {
+      DLOG(WARNING) << "Unknown field number " << field_number;
+      continue;
+    }
+    if (!request_types.Has(model_type)) {
+      DLOG(WARNING)
+          << "Skipping unexpected progress marker for non-enabled type "
+          << ModelTypeToString(model_type);
+      continue;
+    }
+    marker_map->insert(
+        std::make_pair(model_type, &gu_response.new_progress_marker(i)));
+  }
+}
+
+// Examines the contents of the GetUpdates response message and forwards
+// relevant data to the UpdateHandlers for processing and persisting.
+bool ProcessUpdateResponseMessage(
+    const sync_pb::GetUpdatesResponse& gu_response,
+    ModelTypeSet proto_request_types,
+    UpdateHandlerMap* handler_map,
+    StatusController* status) {
+  TypeSyncEntityMap updates_by_type;
+  PartitionUpdatesByType(gu_response, proto_request_types, &updates_by_type);
+  DCHECK_EQ(proto_request_types.Size(), updates_by_type.size());
+
+  TypeProgressMarkerMap progress_by_type;
+  PartitionProgressMarkersByType(gu_response,
+                                 proto_request_types,
+                                 &progress_by_type);
+  if (proto_request_types.Size() != progress_by_type.size()) {
+    NOTREACHED() << "Missing progress markers in GetUpdates response.";
+    return false;
+  }
+
+  // Iterate over these maps in parallel, processing updates for each type.
+  TypeProgressMarkerMap::iterator pm_it = progress_by_type.begin();
+  TypeSyncEntityMap::iterator up_it = updates_by_type.begin();
+  for ( ; pm_it != progress_by_type.end() && up_it != updates_by_type.end();
+       ++pm_it, ++up_it) {
+    DCHECK_EQ(pm_it->first, up_it->first);
+    ModelType type = pm_it->first;
+
+    UpdateHandlerMap::iterator uh_it = handler_map->find(type);
+
+    if (uh_it != handler_map->end()) {
+      uh_it->second->ProcessGetUpdatesResponse(*pm_it->second,
+                                               up_it->second,
+                                               status);
+    } else {
+      DLOG(WARNING)
+          << "Ignoring received updates of a type we can't handle.  "
+          << "Type is: " << ModelTypeToString(type);
+      continue;
+    }
+  }
+  DCHECK(pm_it == progress_by_type.end() && up_it == updates_by_type.end());
+
+  return true;
+}
+
 }  // namespace
 
 void BuildNormalDownloadUpdates(
@@ -234,6 +307,7 @@ void BuildDownloadUpdatesForPoll(
 }
 
 SyncerError ExecuteDownloadUpdates(
+    ModelTypeSet request_types,
     SyncSession* session,
     sync_pb::ClientToServerMessage* msg) {
   sync_pb::ClientToServerResponse update_response;
@@ -251,30 +325,41 @@ SyncerError ExecuteDownloadUpdates(
   if (result != SYNCER_OK) {
     status->mutable_updates_response()->Clear();
     LOG(ERROR) << "PostClientToServerMessage() failed during GetUpdates";
-  } else {
-    status->mutable_updates_response()->CopyFrom(update_response);
-
-    DVLOG(1) << "GetUpdates "
-             << " returned " << update_response.get_updates().entries_size()
-             << " updates and indicated "
-             << update_response.get_updates().changes_remaining()
-             << " updates left on server.";
-
-    if (need_encryption_key ||
-        update_response.get_updates().encryption_keys_size() > 0) {
-      syncable::Directory* dir = session->context()->directory();
-      status->set_last_get_key_result(
-          HandleGetEncryptionKeyResponse(update_response, dir));
-    }
+    return result;
   }
 
-  ProcessUpdatesCommand process_updates;
-  process_updates.Execute(session);
+  status->mutable_updates_response()->CopyFrom(update_response);
 
-  StoreTimestampsCommand store_timestamps;
-  store_timestamps.Execute(session);
+  DVLOG(1) << "GetUpdates "
+           << " returned " << update_response.get_updates().entries_size()
+           << " updates and indicated "
+           << update_response.get_updates().changes_remaining()
+           << " updates left on server.";
 
-  return result;
+  if (need_encryption_key ||
+      update_response.get_updates().encryption_keys_size() > 0) {
+    syncable::Directory* dir = session->context()->directory();
+    status->set_last_get_key_result(
+        HandleGetEncryptionKeyResponse(update_response, dir));
+  }
+
+  const sync_pb::GetUpdatesResponse& gu_response =
+      update_response.get_updates();
+  status->increment_num_updates_downloaded_by(gu_response.entries_size());
+  DCHECK(gu_response.has_changes_remaining());
+  status->set_num_server_changes_remaining(gu_response.changes_remaining());
+
+  const ModelTypeSet proto_request_types =
+      Intersection(request_types, ProtocolTypes());
+
+  if (!ProcessUpdateResponseMessage(gu_response,
+                                    proto_request_types,
+                                    session->context()->update_handler_map(),
+                                    status)) {
+    return SERVER_RESPONSE_VALIDATION_FAILED;
+  } else {
+    return result;
+  }
 }
 
 }  // namespace syncer