Add a minimal AttachmentUploaderImpl.

This is a "dumbed down" implementation with several open TODOs.  Future
CLs will add auth support and wire it in to the rest of sync.

Wire protocol is in flux and will change once the server has been
implemented.


BUG=

Review URL: https://codereview.chromium.org/278263003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@271212 0039d316-1c4b-4281-b951-d872f2087c98

sync/internal_api/attachments/attachment_uploader_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_uploader_impl.h"
+
+#include "base/bind.h"
+#include "base/message_loop/message_loop.h"
+#include "base/threading/non_thread_safe.h"
+#include "net/base/load_flags.h"
+#include "net/http/http_status_code.h"
+#include "net/url_request/url_fetcher.h"
+#include "net/url_request/url_fetcher_delegate.h"
+#include "sync/api/attachments/attachment.h"
+#include "sync/protocol/sync.pb.h"
+#include "url/gurl.h"
+
+namespace {
+
+const char kContentType[] = "application/octet-stream";
+
+}  // namespace
+
+namespace syncer {
+
+// Encapsulates all the state associated with a single upload.
+class AttachmentUploaderImpl::UploadState : public net::URLFetcherDelegate,
+                                            public base::NonThreadSafe {
+ public:
+  // Construct an UploadState.
+  //
+  // |owner| is a pointer to the object that will own (and must outlive!) this
+  // |UploadState.
+  UploadState(const GURL& upload_url,
+              const scoped_refptr<net::URLRequestContextGetter>&
+                  url_request_context_getter,
+              const Attachment& attachment,
+              const UploadCallback& user_callback,
+              AttachmentUploaderImpl* owner);
+
+  virtual ~UploadState();
+
+  // Add |user_callback| to the list of callbacks to be invoked when this upload
+  // completed.
+  void AddUserCallback(const UploadCallback& user_callback);
+
+  // Return the Attachment this object is uploading.
+  const Attachment& GetAttachment();
+
+  // URLFetcher implementation.
+  virtual void OnURLFetchComplete(const net::URLFetcher* source) OVERRIDE;
+
+ private:
+  typedef std::vector<UploadCallback> UploadCallbackList;
+
+  GURL upload_url_;
+  const scoped_refptr<net::URLRequestContextGetter>&
+      url_request_context_getter_;
+  Attachment attachment_;
+  UploadCallbackList user_callbacks_;
+  scoped_ptr<net::URLFetcher> fetcher_;
+  // Pointer to the AttachmentUploaderImpl that owns this object.
+  AttachmentUploaderImpl* owner_;
+
+  DISALLOW_COPY_AND_ASSIGN(UploadState);
+};
+
+AttachmentUploaderImpl::UploadState::UploadState(
+    const GURL& upload_url,
+    const scoped_refptr<net::URLRequestContextGetter>&
+        url_request_context_getter,
+    const Attachment& attachment,
+    const UploadCallback& user_callback,
+    AttachmentUploaderImpl* owner)
+    : upload_url_(upload_url),
+      url_request_context_getter_(url_request_context_getter),
+      attachment_(attachment),
+      user_callbacks_(1, user_callback),
+      owner_(owner) {
+  DCHECK(url_request_context_getter_);
+  DCHECK(upload_url_.is_valid());
+  DCHECK(owner_);
+  fetcher_.reset(
+      net::URLFetcher::Create(upload_url_, net::URLFetcher::POST, this));
+  fetcher_->SetRequestContext(url_request_context_getter_.get());
+  // TODO(maniscalco): Is there a better way?  Copying the attachment data into
+  // a string feels wrong given how large attachments may be (several MBs).  If
+  // we may end up switching from URLFetcher to URLRequest, this copy won't be
+  // necessary.
+  scoped_refptr<base::RefCountedMemory> memory = attachment.GetData();
+  const std::string upload_content(memory->front_as<char>(), memory->size());
+  fetcher_->SetUploadData(kContentType, upload_content);
+  // TODO(maniscalco): Add authentication support (bug 371516).
+  fetcher_->SetLoadFlags(net::LOAD_DO_NOT_SAVE_COOKIES |
+                         net::LOAD_DO_NOT_SEND_COOKIES |
+                         net::LOAD_DISABLE_CACHE);
+  // TODO(maniscalco): Set an appropriate headers (User-Agent, Content-type, and
+  // Content-length) on the request and include the content's MD5,
+  // AttachmentId's unique_id and the "sync birthday" (bug 371521).
+  fetcher_->Start();
+}
+
+AttachmentUploaderImpl::UploadState::~UploadState() {
+}
+
+void AttachmentUploaderImpl::UploadState::AddUserCallback(
+    const UploadCallback& user_callback) {
+  DCHECK(CalledOnValidThread());
+  user_callbacks_.push_back(user_callback);
+}
+
+const Attachment& AttachmentUploaderImpl::UploadState::GetAttachment() {
+  DCHECK(CalledOnValidThread());
+  return attachment_;
+}
+
+void AttachmentUploaderImpl::UploadState::OnURLFetchComplete(
+    const net::URLFetcher* source) {
+  DCHECK(CalledOnValidThread());
+  // TODO(maniscalco): Once the protocol is better defined, deal with the
+  // various HTTP response code we may encounter.
+  UploadResult result = UPLOAD_UNSPECIFIED_ERROR;
+  if (source->GetResponseCode() == net::HTTP_OK) {
+    result = UPLOAD_SUCCESS;
+  }
+  // TODO(maniscalco): Update the attachment id with server address information
+  // before passing it to the callback (bug 371522).
+  AttachmentId updated_id = attachment_.GetId();
+  UploadCallbackList::const_iterator iter = user_callbacks_.begin();
+  UploadCallbackList::const_iterator end = user_callbacks_.end();
+  for (; iter != end; ++iter) {
+    base::MessageLoop::current()->PostTask(
+        FROM_HERE, base::Bind(*iter, result, updated_id));
+  }
+  // Destroy this object and return immediately.
+  owner_->DeleteUploadStateFor(attachment_.GetId().GetProto().unique_id());
+  return;
+}
+
+AttachmentUploaderImpl::AttachmentUploaderImpl(
+    const std::string& url_prefix,
+    const scoped_refptr<net::URLRequestContextGetter>&
+        url_request_context_getter)
+    : url_prefix_(url_prefix),
+      url_request_context_getter_(url_request_context_getter) {
+  DCHECK(CalledOnValidThread());
+}
+
+AttachmentUploaderImpl::~AttachmentUploaderImpl() {
+  DCHECK(CalledOnValidThread());
+}
+
+void AttachmentUploaderImpl::UploadAttachment(const Attachment& attachment,
+                                              const UploadCallback& callback) {
+  DCHECK(CalledOnValidThread());
+  const AttachmentId attachment_id = attachment.GetId();
+  const std::string unique_id = attachment_id.GetProto().unique_id();
+  DCHECK(!unique_id.empty());
+  StateMap::iterator iter = state_map_.find(unique_id);
+  if (iter == state_map_.end()) {
+    const GURL url = GetUploadURLForAttachmentId(attachment_id);
+    scoped_ptr<UploadState> upload_state(new UploadState(
+        url, url_request_context_getter_, attachment, callback, this));
+    state_map_.add(unique_id, upload_state.Pass());
+  } else {
+    DCHECK(
+        attachment.GetData()->Equals(iter->second->GetAttachment().GetData()));
+    // We already have an upload for this attachment.  "Join" it.
+    iter->second->AddUserCallback(callback);
+  }
+}
+
+GURL AttachmentUploaderImpl::GetUploadURLForAttachmentId(
+    const AttachmentId& attachment_id) const {
+  return GURL(url_prefix_ + attachment_id.GetProto().unique_id());
+}
+
+void AttachmentUploaderImpl::DeleteUploadStateFor(const UniqueId& unique_id) {
+  state_map_.erase(unique_id);
+}
+
+}  // namespace syncer

sync/internal_api/attachments/attachment_uploader_impl_unittest.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_uploader_impl.h"
+
+#include "base/bind.h"
+#include "base/callback.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/ref_counted_memory.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/strings/stringprintf.h"
+#include "base/threading/non_thread_safe.h"
+#include "base/threading/thread.h"
+#include "net/test/embedded_test_server/embedded_test_server.h"
+#include "net/test/embedded_test_server/http_request.h"
+#include "net/test/embedded_test_server/http_response.h"
+#include "net/url_request/url_request_test_util.h"
+#include "sync/api/attachments/attachment.h"
+#include "sync/protocol/sync.pb.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace {
+
+const char kAttachmentData[] = "some data";
+
+}  // namespace
+
+namespace syncer {
+
+using net::test_server::BasicHttpResponse;
+using net::test_server::HttpRequest;
+using net::test_server::HttpResponse;
+
+class RequestHandler;
+
+// Text fixture for AttachmentUploaderImpl test.
+//
+// This fixture provides an embedded HTTP server for interacting with
+// AttachmentUploaderImpl.
+class AttachmentUploaderImplTest : public testing::Test,
+                                   public base::NonThreadSafe {
+ public:
+  void OnRequestReceived(const HttpRequest& request);
+
+ protected:
+  AttachmentUploaderImplTest();
+  virtual void SetUp();
+
+  // Run the message loop until UploadDone has been invoked |num_uploads| times.
+  void RunAndWaitFor(int num_uploads);
+
+  scoped_ptr<AttachmentUploader>& uploader();
+  const AttachmentUploader::UploadCallback& upload_callback() const;
+  std::vector<HttpRequest>& http_requests_received();
+  std::vector<AttachmentUploader::UploadResult>& upload_results();
+  std::vector<AttachmentId>& updated_attachment_ids();
+
+ private:
+  // An UploadCallback invoked by AttachmentUploaderImpl.
+  void UploadDone(const AttachmentUploader::UploadResult& result,
+                  const AttachmentId& updated_attachment_id);
+
+  base::MessageLoopForIO message_loop_;
+  scoped_refptr<net::URLRequestContextGetter> url_request_context_getter_;
+  scoped_ptr<RequestHandler> request_handler_;
+  scoped_ptr<AttachmentUploader> uploader_;
+  AttachmentUploader::UploadCallback upload_callback_;
+  net::test_server::EmbeddedTestServer server_;
+
+  // A closure that signals an upload has finished.
+  base::Closure signal_upload_done_;
+  std::vector<HttpRequest> http_requests_received_;
+  std::vector<AttachmentUploader::UploadResult> upload_results_;
+  std::vector<AttachmentId> updated_attachment_ids_;
+
+  // Must be last data member.
+  base::WeakPtrFactory<AttachmentUploaderImplTest> weak_ptr_factory_;
+};
+
+// Handles HTTP requests received by the EmbeddedTestServer.
+class RequestHandler : public base::NonThreadSafe {
+ public:
+  // Construct a RequestHandler that will PostTask to |test| using
+  // |test_task_runner|.
+  RequestHandler(
+      const scoped_refptr<base::SingleThreadTaskRunner>& test_task_runner,
+      const base::WeakPtr<AttachmentUploaderImplTest>& test);
+
+  ~RequestHandler();
+
+  scoped_ptr<HttpResponse> HandleRequest(const HttpRequest& request);
+
+ private:
+  scoped_refptr<base::SingleThreadTaskRunner> test_task_runner_;
+  base::WeakPtr<AttachmentUploaderImplTest> test_;
+};
+
+AttachmentUploaderImplTest::AttachmentUploaderImplTest()
+    : weak_ptr_factory_(this) {
+}
+
+void AttachmentUploaderImplTest::OnRequestReceived(const HttpRequest& request) {
+  DCHECK(CalledOnValidThread());
+  http_requests_received_.push_back(request);
+}
+
+void AttachmentUploaderImplTest::SetUp() {
+  DCHECK(CalledOnValidThread());
+  request_handler_.reset(new RequestHandler(message_loop_.message_loop_proxy(),
+                                            weak_ptr_factory_.GetWeakPtr()));
+  url_request_context_getter_ =
+      new net::TestURLRequestContextGetter(message_loop_.message_loop_proxy());
+
+  ASSERT_TRUE(server_.InitializeAndWaitUntilReady());
+  server_.RegisterRequestHandler(
+      base::Bind(&RequestHandler::HandleRequest,
+                 base::Unretained(request_handler_.get())));
+
+  std::string url_prefix(
+      base::StringPrintf("http://localhost:%d/uploads/", server_.port()));
+
+  uploader().reset(
+      new AttachmentUploaderImpl(url_prefix, url_request_context_getter_));
+  upload_callback_ = base::Bind(&AttachmentUploaderImplTest::UploadDone,
+                                base::Unretained(this));
+}
+
+void AttachmentUploaderImplTest::RunAndWaitFor(int num_uploads) {
+  for (int i = 0; i < num_uploads; ++i) {
+    // Run the loop until one upload completes.
+    base::RunLoop run_loop;
+    signal_upload_done_ = run_loop.QuitClosure();
+    run_loop.Run();
+  }
+}
+
+scoped_ptr<AttachmentUploader>& AttachmentUploaderImplTest::uploader() {
+  return uploader_;
+}
+
+const AttachmentUploader::UploadCallback&
+AttachmentUploaderImplTest::upload_callback() const {
+  return upload_callback_;
+}
+
+std::vector<HttpRequest>& AttachmentUploaderImplTest::http_requests_received() {
+  return http_requests_received_;
+}
+
+std::vector<AttachmentUploader::UploadResult>&
+AttachmentUploaderImplTest::upload_results() {
+  return upload_results_;
+}
+
+std::vector<AttachmentId>&
+AttachmentUploaderImplTest::updated_attachment_ids() {
+  return updated_attachment_ids_;
+}
+
+void AttachmentUploaderImplTest::UploadDone(
+    const AttachmentUploader::UploadResult& result,
+    const AttachmentId& updated_attachment_id) {
+  DCHECK(CalledOnValidThread());
+  upload_results_.push_back(result);
+  updated_attachment_ids_.push_back(updated_attachment_id);
+  signal_upload_done_.Run();
+}
+
+RequestHandler::RequestHandler(
+    const scoped_refptr<base::SingleThreadTaskRunner>& test_task_runner,
+    const base::WeakPtr<AttachmentUploaderImplTest>& test)
+    : test_task_runner_(test_task_runner), test_(test) {
+  DetachFromThread();
+}
+
+RequestHandler::~RequestHandler() {
+  DetachFromThread();
+}
+
+scoped_ptr<HttpResponse> RequestHandler::HandleRequest(
+    const HttpRequest& request) {
+  DCHECK(CalledOnValidThread());
+  test_task_runner_->PostTask(
+      FROM_HERE,
+      base::Bind(
+          &AttachmentUploaderImplTest::OnRequestReceived, test_, request));
+  scoped_ptr<BasicHttpResponse> http_response(new BasicHttpResponse);
+  http_response->set_code(net::HTTP_OK);
+  http_response->set_content("hello");
+  http_response->set_content_type("text/plain");
+  return http_response.PassAs<HttpResponse>();
+}
+
+// Verify the "happy case" of uploading an attachment.
+TEST_F(AttachmentUploaderImplTest, UploadAttachment_HappyCase) {
+  scoped_refptr<base::RefCountedString> some_data(new base::RefCountedString);
+  some_data->data() = kAttachmentData;
+  Attachment attachment = Attachment::Create(some_data);
+  uploader()->UploadAttachment(attachment, upload_callback());
+  RunAndWaitFor(1);
+
+  // See that the HTTP server received one request.
+  EXPECT_EQ(1U, http_requests_received().size());
+  const HttpRequest& http_request = http_requests_received().front();
+  EXPECT_EQ(net::test_server::METHOD_POST, http_request.method);
+  std::string expected_relative_url("/uploads/" +
+                                    attachment.GetId().GetProto().unique_id());
+  EXPECT_EQ(expected_relative_url, http_request.relative_url);
+  EXPECT_TRUE(http_request.has_content);
+  EXPECT_EQ(kAttachmentData, http_request.content);
+
+  // See that the UploadCallback received a result and updated AttachmentId.
+  EXPECT_EQ(1U, upload_results().size());
+  EXPECT_EQ(1U, updated_attachment_ids().size());
+  EXPECT_EQ(AttachmentUploader::UPLOAD_SUCCESS, upload_results().front());
+  EXPECT_EQ(attachment.GetId(), updated_attachment_ids().front());
+  // TODO(maniscalco): Once AttachmentUploaderImpl is capable of updating the
+  // AttachmentId with server address information about the attachment, add some
+  // checks here to verify it works properly (bug 371522).
+}
+
+// Verify two overlapping calls to upload the same attachment result in only one
+// HTTP request.
+TEST_F(AttachmentUploaderImplTest, UploadAttachment_Collapse) {
+  scoped_refptr<base::RefCountedString> some_data(new base::RefCountedString);
+  some_data->data() = kAttachmentData;
+  Attachment attachment1 = Attachment::Create(some_data);
+  Attachment attachment2 = attachment1;
+  uploader()->UploadAttachment(attachment1, upload_callback());
+  uploader()->UploadAttachment(attachment2, upload_callback());
+  // Wait for upload_callback() to be invoked twice.
+  RunAndWaitFor(2);
+  // See there was only one request.
+  EXPECT_EQ(1U, http_requests_received().size());
+}
+
+// Verify that the internal state associated with an upload is removed when the
+// uplaod finishes.  We do this by issuing two non-overlapping uploads for the
+// same attachment and see that it results in two HTTP requests.
+TEST_F(AttachmentUploaderImplTest, UploadAttachment_CleanUpAfterUpload) {
+  scoped_refptr<base::RefCountedString> some_data(new base::RefCountedString);
+  some_data->data() = kAttachmentData;
+  Attachment attachment1 = Attachment::Create(some_data);
+  Attachment attachment2 = attachment1;
+  uploader()->UploadAttachment(attachment1, upload_callback());
+  // Wait for upload_callback() to be invoked before starting the second upload.
+  RunAndWaitFor(1);
+  uploader()->UploadAttachment(attachment2, upload_callback());
+  // Wait for upload_callback() to be invoked a second time.
+  RunAndWaitFor(1);
+  // See there were two requests.
+  EXPECT_EQ(2U, http_requests_received().size());
+}
+
+}  // namespace syncer

sync/internal_api/attachments/fake_attachment_uploader.cc

@@ -7,6 +7,7 @@
 #include "base/bind.h"
 #include "base/message_loop/message_loop.h"
 #include "sync/api/attachments/attachment.h"
+#include "sync/protocol/sync.pb.h"
 
 namespace syncer {
 
@@ -21,6 +22,8 @@ FakeAttachmentUploader::~FakeAttachmentUploader() {
 void FakeAttachmentUploader::UploadAttachment(const Attachment& attachment,
                                               const UploadCallback& callback) {
   DCHECK(CalledOnValidThread());
+  DCHECK(!attachment.GetId().GetProto().unique_id().empty());
+
   UploadResult result = UPLOAD_SUCCESS;
   AttachmentId updated_id = attachment.GetId();
   // TODO(maniscalco): Update the attachment id with server address information

sync/internal_api/public/attachments/attachment_uploader_impl.h

+// 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.
+
+#ifndef SYNC_INTERNAL_API_PUBLIC_ATTACHMENTS_ATTACHMENT_UPLOADER_IMPL_H_
+#define SYNC_INTERNAL_API_PUBLIC_ATTACHMENTS_ATTACHMENT_UPLOADER_IMPL_H_
+
+#include "base/containers/scoped_ptr_hash_map.h"
+#include "base/threading/non_thread_safe.h"
+#include "net/url_request/url_request_context_getter.h"
+#include "sync/api/attachments/attachment_uploader.h"
+
+class GURL;
+
+namespace net {
+class URLRequestContextGetter;
+}  // namespace net
+
+namespace syncer {
+
+// An implementation of AttachmentUploader.
+class SYNC_EXPORT AttachmentUploaderImpl : public AttachmentUploader,
+                                           public base::NonThreadSafe {
+ public:
+  // |url_prefix| is the URL prefix (including trailing slash) to be used when
+  // |uploading attachments.
+  AttachmentUploaderImpl(const std::string& url_prefix,
+                         const scoped_refptr<net::URLRequestContextGetter>&
+                             url_request_context_getter);
+  virtual ~AttachmentUploaderImpl();
+
+  // AttachmentUploader implementation.
+  virtual void UploadAttachment(const Attachment& attachment,
+                                const UploadCallback& callback) OVERRIDE;
+
+ private:
+  class UploadState;
+  typedef std::string UniqueId;
+  typedef base::ScopedPtrHashMap<UniqueId, UploadState> StateMap;
+
+  GURL GetUploadURLForAttachmentId(const AttachmentId& attachment_id) const;
+  void DeleteUploadStateFor(const UniqueId& unique_id);
+
+  std::string url_prefix_;
+  scoped_refptr<net::URLRequestContextGetter> url_request_context_getter_;
+  StateMap state_map_;
+  DISALLOW_COPY_AND_ASSIGN(AttachmentUploaderImpl);
+};
+
+}  // namespace syncer
+
+#endif  // SYNC_INTERNAL_API_PUBLIC_ATTACHMENTS_ATTACHMENT_UPLOADER_IMPL_H_

sync/sync_internal_api.gypi

@@ -16,6 +16,7 @@
     '../url/url.gyp:url_lib',
   ],
   'sources': [
+    'internal_api/attachments/attachment_uploader_impl.cc',
     'internal_api/attachments/fake_attachment_store.cc',
     'internal_api/attachments/fake_attachment_uploader.cc',
     'internal_api/base_node.cc',
@@ -44,6 +45,7 @@
     'internal_api/js_sync_manager_observer.h',
     'internal_api/protocol_event_buffer.cc',
     'internal_api/protocol_event_buffer.h',
+    'internal_api/public/attachments/attachment_uploader_impl.h',
     'internal_api/public/attachments/fake_attachment_store.h',
     'internal_api/public/attachments/fake_attachment_uploader.h',
     'internal_api/public/base/ack_handle.cc',

sync/sync_tests.gypi

@@ -425,6 +425,7 @@
           '..',
         ],
         'sources': [
+          'internal_api/attachments/attachment_uploader_impl_unittest.cc',
           'internal_api/attachments/fake_attachment_store_unittest.cc',
           'internal_api/attachments/fake_attachment_uploader_unittest.cc',
           'internal_api/debug_info_event_listener_unittest.cc',