Replace AttachmentStore's StoreAttachments with UploadAttachments.

sync/internal_api/attachments/attachment_service_impl.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/internal_api/public/attachments/attachment_service_impl.h"
 
+#include <iterator>
+
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
 #include "base/thread_task_runner_handle.h"
 #include "sync/api/attachments/attachment.h"
 #include "sync/api/attachments/fake_attachment_store.h"
 #include "sync/internal_api/public/attachments/fake_attachment_downloader.h"
 #include "sync/internal_api/public/attachments/fake_attachment_uploader.h"
 
 namespace syncer {
 
@@ skipping 117 matching lines @@
   scoped_ptr<AttachmentDownloader> attachment_downloader(
       new FakeAttachmentDownloader());
   scoped_ptr<syncer::AttachmentService> attachment_service(
       new syncer::AttachmentServiceImpl(attachment_store.Pass(),
                                         attachment_uploader.Pass(),
                                         attachment_downloader.Pass(),
                                         NULL));
   return attachment_service.Pass();
 }
 
+AttachmentStore* AttachmentServiceImpl::GetStore() {
+  return attachment_store_.get();
+}
+
 void AttachmentServiceImpl::GetOrDownloadAttachments(
     const AttachmentIdList& attachment_ids,
     const GetOrDownloadCallback& callback) {
   DCHECK(CalledOnValidThread());
   scoped_refptr<GetOrDownloadState> state(
       new GetOrDownloadState(attachment_ids, callback));
   attachment_store_->Read(attachment_ids,
                           base::Bind(&AttachmentServiceImpl::ReadDone,
                                      weak_ptr_factory_.GetWeakPtr(),
                                      state));
 }
 
 void AttachmentServiceImpl::DropAttachments(
     const AttachmentIdList& attachment_ids,
     const DropCallback& callback) {
   DCHECK(CalledOnValidThread());
   attachment_store_->Drop(attachment_ids,
                           base::Bind(&AttachmentServiceImpl::DropDone,
                                      weak_ptr_factory_.GetWeakPtr(),
                                      callback));
 }
 
-void AttachmentServiceImpl::StoreAttachments(const AttachmentList& attachments,
-                                             const StoreCallback& callback) {
-  DCHECK(CalledOnValidThread());
-  attachment_store_->Write(attachments,
-                           base::Bind(&AttachmentServiceImpl::WriteDone,
-                                      weak_ptr_factory_.GetWeakPtr(),
-                                      callback));
-  if (attachment_uploader_.get()) {
-    for (AttachmentList::const_iterator iter = attachments.begin();
-         iter != attachments.end();
-         ++iter) {
-      attachment_uploader_->UploadAttachment(
-          *iter,
-          base::Bind(&AttachmentServiceImpl::UploadDone,
-                     weak_ptr_factory_.GetWeakPtr()));
-    }
-  }
-}
-
 void AttachmentServiceImpl::ReadDone(
     const scoped_refptr<GetOrDownloadState>& state,
     const AttachmentStore::Result& result,
     scoped_ptr<AttachmentMap> attachments,
     scoped_ptr<AttachmentIdList> unavailable_attachment_ids) {
   // Add read attachments to result.
   for (AttachmentMap::const_iterator iter = attachments->begin();
        iter != attachments->end();
        ++iter) {
     state->AddAttachment(iter->second);
@@ skipping 24 matching lines @@
   AttachmentService::DropResult drop_result =
       AttachmentService::DROP_UNSPECIFIED_ERROR;
   if (result == AttachmentStore::SUCCESS) {
     drop_result = AttachmentService::DROP_SUCCESS;
   }
   // TODO(maniscalco): Deal with case where an error occurred (bug 361251).
   base::MessageLoop::current()->PostTask(FROM_HERE,
                                          base::Bind(callback, drop_result));
 }
 
-void AttachmentServiceImpl::WriteDone(const StoreCallback& callback,
-                                      const AttachmentStore::Result& result) {
-  AttachmentService::StoreResult store_result =
-      AttachmentService::STORE_UNSPECIFIED_ERROR;
-  if (result == AttachmentStore::SUCCESS) {
-    store_result = AttachmentService::STORE_SUCCESS;
-  }
-  // TODO(maniscalco): Deal with case where an error occurred (bug 361251).
-  base::MessageLoop::current()->PostTask(FROM_HERE,
-                                         base::Bind(callback, store_result));
-}
-
 void AttachmentServiceImpl::UploadDone(
     const AttachmentUploader::UploadResult& result,
     const AttachmentId& attachment_id) {
+  ids_in_queue_.erase(attachment_id);
   // TODO(pavely): crbug/372622: Deal with UploadAttachment failures.
   if (result != AttachmentUploader::UPLOAD_SUCCESS)
     return;
   if (delegate_) {
     delegate_->OnAttachmentUploaded(attachment_id);
   }
 }
 
 void AttachmentServiceImpl::DownloadDone(
     const scoped_refptr<GetOrDownloadState>& state,
     const AttachmentId& attachment_id,
     const AttachmentDownloader::DownloadResult& result,
     scoped_ptr<Attachment> attachment) {
   if (result == AttachmentDownloader::DOWNLOAD_SUCCESS) {
     state->AddAttachment(*attachment.get());
   } else {
     state->AddUnavailableAttachmentId(attachment_id);
   }
 }
 
+void AttachmentServiceImpl::UploadAttachments(
+    const AttachmentIdSet& attachment_ids) {
+  DCHECK(CalledOnValidThread());
+  if (!attachment_uploader_.get()) {
+    return;
+  }
+
+  // Enqueue the attachment ids that aren't already in the queue.
+  AttachmentIdSet::const_iterator iter = attachment_ids.begin();
+  AttachmentIdSet::const_iterator end = attachment_ids.end();
+  for (; iter != end; ++iter) {
+    if (ids_in_queue_.find(*iter) == ids_in_queue_.end()) {
+      queue_.push_back(*iter);
+      ids_in_queue_.insert(*iter);
+    }
+  }
+
+  ProcessQueuedUploads();
+}
+
+void AttachmentServiceImpl::ProcessQueuedUploads() {
+  DCHECK(CalledOnValidThread());
+  // TODO(maniscalco): Don't dequeue them all.  Instead, limit the number of
+  // concurrent uploads and apply backoff on failure.
+  while (!queue_.empty()) {
+    const AttachmentId id = queue_.front();
+    queue_.pop_front();
+    AttachmentIdList attachment_ids;
+    attachment_ids.push_back(id);
+    attachment_store_->Read(
+        attachment_ids,
+        base::Bind(&AttachmentServiceImpl::ReadDoneNowUpload,
+                   weak_ptr_factory_.GetWeakPtr()));
+  }
+}
+
+void AttachmentServiceImpl::ReadDoneNowUpload(
+    const AttachmentStore::Result& result,
+    scoped_ptr<AttachmentMap> attachments,
+    scoped_ptr<AttachmentIdList> unavailable_attachment_ids) {
+  DCHECK(CalledOnValidThread());
+  if (!unavailable_attachment_ids->empty()) {
+    // TODO(maniscalco): We failed to read some attachments. What should we do
+    // now?
+  }
+
+  AttachmentMap::const_iterator iter = attachments->begin();
+  AttachmentMap::const_iterator end = attachments->end();
+  for (; iter != end; ++iter) {
+    attachment_uploader_->UploadAttachment(
+        iter->second,
+        base::Bind(&AttachmentServiceImpl::UploadDone,
+                   weak_ptr_factory_.GetWeakPtr()));
+  }
+}
+
 }  // namespace syncer