sync: Implement per-type update processing

sync/engine/download.cc

 // Copyright 2013 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/engine/download.h"
 
 #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"
 #include "sync/syncable/directory.h"
 #include "sync/syncable/nigori_handler.h"
 #include "sync/syncable/syncable_read_transaction.h"
 
 using sync_pb::DebugInfo;
 
 namespace syncer {
 
 using sessions::StatusController;
 using sessions::SyncSession;
 using sessions::SyncSessionContext;
 using std::string;
 
 namespace {
 
+typedef std::map<ModelType, const sync_pb::DataTypeProgressMarker*>
+    TypeProgressMarkerMap;
+
 SyncerError HandleGetEncryptionKeyResponse(
     const sync_pb::ClientToServerResponse& update_response,
     syncable::Directory* dir) {
   bool success = false;
   if (update_response.get_updates().encryption_keys_size() == 0) {
     LOG(ERROR) << "Failed to receive encryption key from server.";
     return SERVER_RESPONSE_VALIDATION_FAILED;
   }
   syncable::ReadTransaction trans(FROM_HERE, dir);
   syncable::NigoriHandler* nigori_handler = dir->GetNigoriHandler();
@@ skipping 88 matching lines @@
   for (ModelTypeSet::Iterator it = request_types.First();
        it.Good(); it.Inc()) {
     if (ProxyTypes().Has(it.Get()))
       continue;
     sync_pb::DataTypeProgressMarker* progress_marker =
         get_updates->add_from_progress_marker();
     dir->GetDownloadProgress(it.Get(), progress_marker);
   }
 }
 
+// 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(
     SyncSession* session,
     bool create_mobile_bookmarks_folder,
     ModelTypeSet request_types,
     const sessions::NudgeTracker& nudge_tracker,
     sync_pb::ClientToServerMessage* client_to_server_message) {
   InitDownloadUpdatesRequest(
       session,
@@ skipping 79 matching lines @@
 
   // Set legacy GetUpdatesMessage.GetUpdatesCallerInfo information.
   get_updates->mutable_caller_info()->set_source(
       sync_pb::GetUpdatesCallerInfo::PERIODIC);
 
   // Set the new and improved version of source, too.
   get_updates->set_get_updates_origin(sync_pb::SyncEnums::PERIODIC);
 }
 
 SyncerError ExecuteDownloadUpdates(
+    ModelTypeSet request_types,
     SyncSession* session,
     sync_pb::ClientToServerMessage* msg) {
   sync_pb::ClientToServerResponse update_response;
   StatusController* status = session->mutable_status_controller();
   bool need_encryption_key = ShouldRequestEncryptionKey(session->context());
 
   SyncerError result = SyncerProtoUtil::PostClientToServerMessage(
       msg,
       &update_response,
       session);
 
   DVLOG(2) << SyncerProtoUtil::ClientToServerResponseDebugString(
       update_response);
 
   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