Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(195)

Side by Side Diff: components/sync/core_impl/attachments/attachment_service_impl.cc

Issue 2399953002: [Sync] Move attachments code out of core/. (Closed)
Patch Set: Address comments. Created 4 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
OLDNEW
(Empty)
1 // Copyright 2014 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file.
4
5 #include "components/sync/core/attachments/attachment_service_impl.h"
6
7 #include <iterator>
8 #include <utility>
9
10 #include "base/bind.h"
11 #include "base/location.h"
12 #include "base/single_thread_task_runner.h"
13 #include "base/threading/thread_task_runner_handle.h"
14 #include "base/time/time.h"
15 #include "components/sync/api/attachments/attachment.h"
16 #include "components/sync/core/attachments/fake_attachment_downloader.h"
17 #include "components/sync/core/attachments/fake_attachment_uploader.h"
18
19 namespace syncer {
20
21 // GetOrDownloadAttachments starts multiple parallel DownloadAttachment calls.
22 // GetOrDownloadState tracks completion of these calls and posts callback for
23 // consumer once all attachments are either retrieved or reported unavailable.
24 class AttachmentServiceImpl::GetOrDownloadState
25 : public base::RefCounted<GetOrDownloadState>,
26 public base::NonThreadSafe {
27 public:
28 // GetOrDownloadState gets parameter from values passed to
29 // AttachmentService::GetOrDownloadAttachments.
30 // |attachment_ids| is a list of attachmens to retrieve.
31 // |callback| will be posted on current thread when all attachments retrieved
32 // or confirmed unavailable.
33 GetOrDownloadState(const AttachmentIdList& attachment_ids,
34 const GetOrDownloadCallback& callback);
35
36 // Attachment was just retrieved. Add it to retrieved attachments.
37 void AddAttachment(const Attachment& attachment);
38
39 // Both reading from local store and downloading attachment failed.
40 // Add it to unavailable set.
41 void AddUnavailableAttachmentId(const AttachmentId& attachment_id);
42
43 private:
44 friend class base::RefCounted<GetOrDownloadState>;
45 virtual ~GetOrDownloadState();
46
47 // If all attachment requests completed then post callback to consumer with
48 // results.
49 void PostResultIfAllRequestsCompleted();
50
51 GetOrDownloadCallback callback_;
52
53 // Requests for these attachments are still in progress.
54 AttachmentIdSet in_progress_attachments_;
55
56 AttachmentIdSet unavailable_attachments_;
57 std::unique_ptr<AttachmentMap> retrieved_attachments_;
58
59 DISALLOW_COPY_AND_ASSIGN(GetOrDownloadState);
60 };
61
62 AttachmentServiceImpl::GetOrDownloadState::GetOrDownloadState(
63 const AttachmentIdList& attachment_ids,
64 const GetOrDownloadCallback& callback)
65 : callback_(callback), retrieved_attachments_(new AttachmentMap()) {
66 std::copy(
67 attachment_ids.begin(), attachment_ids.end(),
68 std::inserter(in_progress_attachments_, in_progress_attachments_.end()));
69 PostResultIfAllRequestsCompleted();
70 }
71
72 AttachmentServiceImpl::GetOrDownloadState::~GetOrDownloadState() {
73 DCHECK(CalledOnValidThread());
74 }
75
76 void AttachmentServiceImpl::GetOrDownloadState::AddAttachment(
77 const Attachment& attachment) {
78 DCHECK(CalledOnValidThread());
79 DCHECK(retrieved_attachments_->find(attachment.GetId()) ==
80 retrieved_attachments_->end());
81 retrieved_attachments_->insert(
82 std::make_pair(attachment.GetId(), attachment));
83 DCHECK(in_progress_attachments_.find(attachment.GetId()) !=
84 in_progress_attachments_.end());
85 in_progress_attachments_.erase(attachment.GetId());
86 PostResultIfAllRequestsCompleted();
87 }
88
89 void AttachmentServiceImpl::GetOrDownloadState::AddUnavailableAttachmentId(
90 const AttachmentId& attachment_id) {
91 DCHECK(CalledOnValidThread());
92 DCHECK(unavailable_attachments_.find(attachment_id) ==
93 unavailable_attachments_.end());
94 unavailable_attachments_.insert(attachment_id);
95 DCHECK(in_progress_attachments_.find(attachment_id) !=
96 in_progress_attachments_.end());
97 in_progress_attachments_.erase(attachment_id);
98 PostResultIfAllRequestsCompleted();
99 }
100
101 void AttachmentServiceImpl::GetOrDownloadState::
102 PostResultIfAllRequestsCompleted() {
103 if (in_progress_attachments_.empty()) {
104 // All requests completed. Let's notify consumer.
105 GetOrDownloadResult result =
106 unavailable_attachments_.empty() ? GET_SUCCESS : GET_UNSPECIFIED_ERROR;
107 base::ThreadTaskRunnerHandle::Get()->PostTask(
108 FROM_HERE,
109 base::Bind(callback_, result, base::Passed(&retrieved_attachments_)));
110 }
111 }
112
113 AttachmentServiceImpl::AttachmentServiceImpl(
114 std::unique_ptr<AttachmentStoreForSync> attachment_store,
115 std::unique_ptr<AttachmentUploader> attachment_uploader,
116 std::unique_ptr<AttachmentDownloader> attachment_downloader,
117 Delegate* delegate,
118 const base::TimeDelta& initial_backoff_delay,
119 const base::TimeDelta& max_backoff_delay)
120 : attachment_store_(std::move(attachment_store)),
121 attachment_uploader_(std::move(attachment_uploader)),
122 attachment_downloader_(std::move(attachment_downloader)),
123 delegate_(delegate),
124 weak_ptr_factory_(this) {
125 DCHECK(CalledOnValidThread());
126 DCHECK(attachment_store_.get());
127
128 // TODO(maniscalco): Observe network connectivity change events. When the
129 // network becomes disconnected, consider suspending queue dispatch. When
130 // connectivity is restored, consider clearing any dispatch backoff (bug
131 // 411981).
132 upload_task_queue_.reset(new TaskQueue<AttachmentId>(
133 base::Bind(&AttachmentServiceImpl::BeginUpload,
134 weak_ptr_factory_.GetWeakPtr()),
135 initial_backoff_delay, max_backoff_delay));
136
137 net::NetworkChangeNotifier::AddNetworkChangeObserver(this);
138 }
139
140 AttachmentServiceImpl::~AttachmentServiceImpl() {
141 DCHECK(CalledOnValidThread());
142 net::NetworkChangeNotifier::RemoveNetworkChangeObserver(this);
143 }
144
145 // Static.
146 std::unique_ptr<AttachmentService> AttachmentServiceImpl::CreateForTest() {
147 std::unique_ptr<AttachmentStore> attachment_store =
148 AttachmentStore::CreateInMemoryStore();
149 std::unique_ptr<AttachmentUploader> attachment_uploader(
150 new FakeAttachmentUploader);
151 std::unique_ptr<AttachmentDownloader> attachment_downloader(
152 new FakeAttachmentDownloader());
153 std::unique_ptr<AttachmentService> attachment_service(
154 new AttachmentServiceImpl(
155 attachment_store->CreateAttachmentStoreForSync(),
156 std::move(attachment_uploader), std::move(attachment_downloader),
157 NULL, base::TimeDelta(), base::TimeDelta()));
158 return attachment_service;
159 }
160
161 void AttachmentServiceImpl::GetOrDownloadAttachments(
162 const AttachmentIdList& attachment_ids,
163 const GetOrDownloadCallback& callback) {
164 DCHECK(CalledOnValidThread());
165 scoped_refptr<GetOrDownloadState> state(
166 new GetOrDownloadState(attachment_ids, callback));
167 // SetModelTypeReference() makes attachments visible for model type.
168 // Needed when attachment doesn't have model type reference, but still
169 // available in local store.
170 attachment_store_->SetModelTypeReference(attachment_ids);
171 attachment_store_->Read(attachment_ids,
172 base::Bind(&AttachmentServiceImpl::ReadDone,
173 weak_ptr_factory_.GetWeakPtr(), state));
174 }
175
176 void AttachmentServiceImpl::ReadDone(
177 const scoped_refptr<GetOrDownloadState>& state,
178 const AttachmentStore::Result& result,
179 std::unique_ptr<AttachmentMap> attachments,
180 std::unique_ptr<AttachmentIdList> unavailable_attachment_ids) {
181 // Add read attachments to result.
182 for (AttachmentMap::const_iterator iter = attachments->begin();
183 iter != attachments->end(); ++iter) {
184 state->AddAttachment(iter->second);
185 }
186
187 AttachmentIdList::const_iterator iter = unavailable_attachment_ids->begin();
188 AttachmentIdList::const_iterator end = unavailable_attachment_ids->end();
189 if (result != AttachmentStore::STORE_INITIALIZATION_FAILED &&
190 attachment_downloader_.get()) {
191 // Try to download locally unavailable attachments.
192 for (; iter != end; ++iter) {
193 attachment_downloader_->DownloadAttachment(
194 *iter, base::Bind(&AttachmentServiceImpl::DownloadDone,
195 weak_ptr_factory_.GetWeakPtr(), state, *iter));
196 }
197 } else {
198 // No downloader so all locally unavailable attachments are unavailable.
199 for (; iter != end; ++iter) {
200 state->AddUnavailableAttachmentId(*iter);
201 }
202 }
203 }
204
205 void AttachmentServiceImpl::WriteDone(
206 const scoped_refptr<GetOrDownloadState>& state,
207 const Attachment& attachment,
208 const AttachmentStore::Result& result) {
209 switch (result) {
210 case AttachmentStore::SUCCESS:
211 state->AddAttachment(attachment);
212 break;
213 case AttachmentStore::UNSPECIFIED_ERROR:
214 case AttachmentStore::STORE_INITIALIZATION_FAILED:
215 state->AddUnavailableAttachmentId(attachment.GetId());
216 break;
217 }
218 }
219
220 void AttachmentServiceImpl::UploadDone(
221 const AttachmentUploader::UploadResult& result,
222 const AttachmentId& attachment_id) {
223 DCHECK(CalledOnValidThread());
224 AttachmentIdList ids;
225 ids.push_back(attachment_id);
226 switch (result) {
227 case AttachmentUploader::UPLOAD_SUCCESS:
228 attachment_store_->DropSyncReference(ids);
229 upload_task_queue_->MarkAsSucceeded(attachment_id);
230 if (delegate_) {
231 delegate_->OnAttachmentUploaded(attachment_id);
232 }
233 break;
234 case AttachmentUploader::UPLOAD_TRANSIENT_ERROR:
235 upload_task_queue_->MarkAsFailed(attachment_id);
236 upload_task_queue_->AddToQueue(attachment_id);
237 break;
238 case AttachmentUploader::UPLOAD_UNSPECIFIED_ERROR:
239 // TODO(pavely): crbug/372622: Deal with UploadAttachment failures.
240 attachment_store_->DropSyncReference(ids);
241 upload_task_queue_->MarkAsFailed(attachment_id);
242 break;
243 }
244 }
245
246 void AttachmentServiceImpl::DownloadDone(
247 const scoped_refptr<GetOrDownloadState>& state,
248 const AttachmentId& attachment_id,
249 const AttachmentDownloader::DownloadResult& result,
250 std::unique_ptr<Attachment> attachment) {
251 switch (result) {
252 case AttachmentDownloader::DOWNLOAD_SUCCESS: {
253 AttachmentList attachment_list;
254 attachment_list.push_back(*attachment.get());
255 attachment_store_->Write(
256 attachment_list,
257 base::Bind(&AttachmentServiceImpl::WriteDone,
258 weak_ptr_factory_.GetWeakPtr(), state, *attachment.get()));
259 break;
260 }
261 case AttachmentDownloader::DOWNLOAD_TRANSIENT_ERROR:
262 case AttachmentDownloader::DOWNLOAD_UNSPECIFIED_ERROR:
263 state->AddUnavailableAttachmentId(attachment_id);
264 break;
265 }
266 }
267
268 void AttachmentServiceImpl::BeginUpload(const AttachmentId& attachment_id) {
269 DCHECK(CalledOnValidThread());
270 AttachmentIdList attachment_ids;
271 attachment_ids.push_back(attachment_id);
272 attachment_store_->Read(attachment_ids,
273 base::Bind(&AttachmentServiceImpl::ReadDoneNowUpload,
274 weak_ptr_factory_.GetWeakPtr()));
275 }
276
277 void AttachmentServiceImpl::UploadAttachments(
278 const AttachmentIdList& attachment_ids) {
279 DCHECK(CalledOnValidThread());
280 if (!attachment_uploader_.get()) {
281 return;
282 }
283 attachment_store_->SetSyncReference(attachment_ids);
284
285 for (auto iter = attachment_ids.begin(); iter != attachment_ids.end();
286 ++iter) {
287 upload_task_queue_->AddToQueue(*iter);
288 }
289 }
290
291 void AttachmentServiceImpl::OnNetworkChanged(
292 net::NetworkChangeNotifier::ConnectionType type) {
293 if (type != net::NetworkChangeNotifier::CONNECTION_NONE) {
294 upload_task_queue_->ResetBackoff();
295 }
296 }
297
298 void AttachmentServiceImpl::ReadDoneNowUpload(
299 const AttachmentStore::Result& result,
300 std::unique_ptr<AttachmentMap> attachments,
301 std::unique_ptr<AttachmentIdList> unavailable_attachment_ids) {
302 DCHECK(CalledOnValidThread());
303 if (!unavailable_attachment_ids->empty()) {
304 // TODO(maniscalco): We failed to read some attachments. What should we do
305 // now?
306 AttachmentIdList::const_iterator iter = unavailable_attachment_ids->begin();
307 AttachmentIdList::const_iterator end = unavailable_attachment_ids->end();
308 for (; iter != end; ++iter) {
309 upload_task_queue_->Cancel(*iter);
310 }
311 attachment_store_->DropSyncReference(*unavailable_attachment_ids);
312 }
313
314 AttachmentMap::const_iterator iter = attachments->begin();
315 AttachmentMap::const_iterator end = attachments->end();
316 for (; iter != end; ++iter) {
317 attachment_uploader_->UploadAttachment(
318 iter->second, base::Bind(&AttachmentServiceImpl::UploadDone,
319 weak_ptr_factory_.GetWeakPtr()));
320 }
321 }
322
323 void AttachmentServiceImpl::SetTimerForTest(
324 std::unique_ptr<base::Timer> timer) {
325 upload_task_queue_->SetTimerForTest(std::move(timer));
326 }
327
328 } // namespace syncer
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698