diff options
author | Christian Mollekopf <chrigi_1@fastmail.fm> | 2017-01-10 11:28:53 +0100 |
---|---|---|
committer | Christian Mollekopf <chrigi_1@fastmail.fm> | 2017-01-10 11:28:53 +0100 |
commit | 8e84c8a78b7e308cc2b09241af649851036d11de (patch) | |
tree | e4b8b964aa7400e9f1d88b04cadcabb423e039dd /examples | |
parent | 5064e5c9a705365524321e01686e73ac1bdf28a0 (diff) | |
download | sink-8e84c8a78b7e308cc2b09241af649851036d11de.tar.gz sink-8e84c8a78b7e308cc2b09241af649851036d11de.zip |
Improved imap mail sync algorithm.
* when requesting individual mails we sync the full content
* when requesting individual folders we get 2 weeks of full content +
headers for everything else.
* when requesting a sync for all folders we only get 2 weeks of full
content.
Getting the headers for 50k messages takes about 180s on my system with
kolabnow (network being the bottleneck), so that's managable. Getting
the full content would take in the range of hours.
This way we have something to show, and a way to request more data,
without making the system overly complex yet.
Certainly not the final solution, but a good start.
Diffstat (limited to 'examples')
-rw-r--r-- | examples/imapresource/imapresource.cpp | 293 | ||||
-rw-r--r-- | examples/imapresource/imapserverproxy.cpp | 12 | ||||
-rw-r--r-- | examples/imapresource/imapserverproxy.h | 2 |
3 files changed, 202 insertions, 105 deletions
diff --git a/examples/imapresource/imapresource.cpp b/examples/imapresource/imapresource.cpp index 4482a54..f07a62e 100644 --- a/examples/imapresource/imapresource.cpp +++ b/examples/imapresource/imapresource.cpp | |||
@@ -140,6 +140,24 @@ public: | |||
140 | } | 140 | } |
141 | } | 141 | } |
142 | 142 | ||
143 | static void setFlags(Sink::ApplicationDomain::Mail &mail, const KIMAP2::MessageFlags &flags) | ||
144 | { | ||
145 | mail.setUnread(!flags.contains(Imap::Flags::Seen)); | ||
146 | mail.setImportant(flags.contains(Imap::Flags::Flagged)); | ||
147 | } | ||
148 | |||
149 | KIMAP2::MessageFlags getFlags(const Sink::ApplicationDomain::Mail &mail) | ||
150 | { | ||
151 | KIMAP2::MessageFlags flags; | ||
152 | if (!mail.getUnread()) { | ||
153 | flags << Imap::Flags::Seen; | ||
154 | } | ||
155 | if (mail.getImportant()) { | ||
156 | flags << Imap::Flags::Flagged; | ||
157 | } | ||
158 | return flags; | ||
159 | } | ||
160 | |||
143 | void synchronizeMails(const QByteArray &folderRid, const Message &message) | 161 | void synchronizeMails(const QByteArray &folderRid, const Message &message) |
144 | { | 162 | { |
145 | auto time = QSharedPointer<QTime>::create(); | 163 | auto time = QSharedPointer<QTime>::create(); |
@@ -158,8 +176,7 @@ public: | |||
158 | auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier); | 176 | auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier); |
159 | mail.setFolder(folderLocalId); | 177 | mail.setFolder(folderLocalId); |
160 | mail.setMimeMessage(message.msg->encodedContent()); | 178 | mail.setMimeMessage(message.msg->encodedContent()); |
161 | mail.setUnread(!message.flags.contains(Imap::Flags::Seen)); | 179 | setFlags(mail, message.flags); |
162 | mail.setImportant(message.flags.contains(Imap::Flags::Flagged)); | ||
163 | 180 | ||
164 | createOrModify(bufferType, remoteId, mail); | 181 | createOrModify(bufferType, remoteId, mail); |
165 | // const auto elapsed = time->elapsed(); | 182 | // const auto elapsed = time->elapsed(); |
@@ -198,76 +215,87 @@ public: | |||
198 | SinkLog() << "Removed " << count << " mails in " << folderRid << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]"; | 215 | SinkLog() << "Removed " << count << " mails in " << folderRid << Sink::Log::TraceTime(elapsed) << " " << elapsed/qMax(count, 1) << " [ms/mail]"; |
199 | } | 216 | } |
200 | 217 | ||
201 | KAsync::Job<void> synchronizeFolder(QSharedPointer<ImapServerProxy> imap, const Imap::Folder &folder, const QDate &dateFilter) | 218 | KAsync::Job<void> synchronizeFolder(QSharedPointer<ImapServerProxy> imap, const Imap::Folder &folder, const QDate &dateFilter, bool fetchHeaderAlso = false) |
202 | { | 219 | { |
203 | QSet<qint64> uids; | ||
204 | SinkLogCtx(mLogCtx) << "Synchronizing mails: " << folderRid(folder); | 220 | SinkLogCtx(mLogCtx) << "Synchronizing mails: " << folderRid(folder); |
205 | if (folder.path().isEmpty()) { | 221 | SinkLogCtx(mLogCtx) << " fetching headers also: " << fetchHeaderAlso; |
222 | const auto folderRemoteId = folderRid(folder); | ||
223 | if (folder.path().isEmpty() || folderRemoteId.isEmpty()) { | ||
224 | SinkWarningCtx(mLogCtx) << "Invalid folder " << folderRemoteId << folder.path(); | ||
206 | return KAsync::error<void>("Invalid folder"); | 225 | return KAsync::error<void>("Invalid folder"); |
207 | } | 226 | } |
208 | auto capabilities = imap->getCapabilities(); | 227 | // auto capabilities = imap->getCapabilities(); |
209 | bool canDoIncrementalRemovals = false; | 228 | |
229 | //First we fetch flag changes for all messages. Since we don't know which messages are locally available we just get everything and only apply to what we have. | ||
210 | return KAsync::start<void>([=]() { | 230 | return KAsync::start<void>([=]() { |
211 | //First we fetch flag changes for all messages. Since we don't know which messages are locally available we just get everything and only apply to what we have. | 231 | auto uidNext = syncStore().readValue(folderRemoteId, "uidnext").toLongLong(); |
212 | auto uidNext = syncStore().readValue(folderRid(folder), "uidnext").toLongLong(); | ||
213 | bool ok = false; | 232 | bool ok = false; |
214 | const auto changedsince = syncStore().readValue(folderRid(folder), "changedsince").toLongLong(&ok); | 233 | const auto changedsince = syncStore().readValue(folderRemoteId, "changedsince").toLongLong(&ok); |
215 | SinkLog() << "About to update flags" << folder.path() << "changedsince: " << changedsince; | 234 | SinkLogCtx(mLogCtx) << "About to update flags" << folder.path() << "changedsince: " << changedsince; |
235 | //If we have any mails so far we start off by updating any changed flags using changedsince | ||
216 | if (ok) { | 236 | if (ok) { |
217 | return imap->fetchFlags(folder, KIMAP2::ImapSet(1, qMax(uidNext, qint64(1))), changedsince, [this, folder](const Message &message) { | 237 | return imap->fetchFlags(folder, KIMAP2::ImapSet(1, qMax(uidNext, qint64(1))), changedsince, [=](const Message &message) { |
218 | const auto folderLocalId = syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, folderRid(folder)); | 238 | const auto folderLocalId = syncStore().resolveRemoteId(ENTITY_TYPE_FOLDER, folderRemoteId); |
219 | const auto remoteId = assembleMailRid(folderLocalId, message.uid); | 239 | const auto remoteId = assembleMailRid(folderLocalId, message.uid); |
220 | 240 | ||
221 | SinkLog() << "Updating mail flags " << remoteId << message.flags; | 241 | SinkLogCtx(mLogCtx) << "Updating mail flags " << remoteId << message.flags; |
222 | 242 | ||
223 | auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier); | 243 | auto mail = Sink::ApplicationDomain::Mail::create(mResourceInstanceIdentifier); |
224 | mail.setUnread(!message.flags.contains(Imap::Flags::Seen)); | 244 | setFlags(mail, message.flags); |
225 | mail.setImportant(message.flags.contains(Imap::Flags::Flagged)); | ||
226 | 245 | ||
227 | modify(ENTITY_TYPE_MAIL, remoteId, mail); | 246 | modify(ENTITY_TYPE_MAIL, remoteId, mail); |
228 | }) | 247 | }) |
229 | .syncThen<void, SelectResult>([this, folder](const SelectResult &selectResult) { | 248 | .syncThen<void, SelectResult>([=](const SelectResult &selectResult) { |
230 | SinkLog() << "Flags updated. New changedsince value: " << selectResult.highestModSequence; | 249 | SinkLogCtx(mLogCtx) << "Flags updated. New changedsince value: " << selectResult.highestModSequence; |
231 | syncStore().writeValue(folderRid(folder), "changedsince", QByteArray::number(selectResult.highestModSequence)); | 250 | syncStore().writeValue(folderRemoteId, "changedsince", QByteArray::number(selectResult.highestModSequence)); |
232 | }); | 251 | }); |
233 | } else { | 252 | } else { |
234 | //We hit this path on initial sync | 253 | //We hit this path on initial sync and simply record the current changedsince value |
235 | return imap->select(imap->mailboxFromFolder(folder)) | 254 | return imap->select(imap->mailboxFromFolder(folder)) |
236 | .syncThen<void, SelectResult>([this, folder](const SelectResult &selectResult) { | 255 | .syncThen<void, SelectResult>([=](const SelectResult &selectResult) { |
237 | SinkLog() << "No flags to update. New changedsince value: " << selectResult.highestModSequence; | 256 | SinkLogCtx(mLogCtx) << "No flags to update. New changedsince value: " << selectResult.highestModSequence; |
238 | syncStore().writeValue(folderRid(folder), "changedsince", QByteArray::number(selectResult.highestModSequence)); | 257 | syncStore().writeValue(folderRemoteId, "changedsince", QByteArray::number(selectResult.highestModSequence)); |
239 | }); | 258 | }); |
240 | } | 259 | } |
241 | }) | 260 | }) |
261 | //Next we synchronize the full set that is given by the date limit. | ||
262 | //We fetch all data for this set. | ||
263 | //This will also pull in any new messages in subsequent runs. | ||
242 | .then<void>([=]() { | 264 | .then<void>([=]() { |
243 | auto job = [&] { | 265 | auto job = [&] { |
266 | SinkLogCtx(mLogCtx) << "Fetching messages since: " << dateFilter; | ||
244 | if (dateFilter.isValid()) { | 267 | if (dateFilter.isValid()) { |
245 | return imap->fetchUidsSince(imap->mailboxFromFolder(folder), dateFilter); | 268 | return imap->fetchUidsSince(imap->mailboxFromFolder(folder), dateFilter); |
246 | } else { | 269 | } else { |
247 | return imap->fetchUids(imap->mailboxFromFolder(folder)); | 270 | return imap->fetchUids(imap->mailboxFromFolder(folder)); |
248 | } | 271 | } |
249 | }(); | 272 | }(); |
250 | return job.then<void, QVector<qint64>>([this, folder, imap](const QVector<qint64> &uidsToFetch) { | 273 | return job.then<void, QVector<qint64>>([=](const QVector<qint64> &uidsToFetch) { |
251 | SinkTrace() << "Received result set " << uidsToFetch; | 274 | SinkTraceCtx(mLogCtx) << "Received result set " << uidsToFetch; |
252 | SinkTrace() << "About to fetch mail" << folder.path(); | 275 | SinkTraceCtx(mLogCtx) << "About to fetch mail" << folder.path(); |
253 | const auto uidNext = syncStore().readValue(folderRid(folder), "uidnext").toLongLong(); | 276 | const auto uidNext = syncStore().readValue(folderRemoteId, "uidnext").toLongLong(); |
277 | |||
278 | //Make sure the uids are sorted in reverse order and drop everything below uidNext (so we don't refetch what we already have | ||
254 | QVector<qint64> filteredAndSorted = uidsToFetch; | 279 | QVector<qint64> filteredAndSorted = uidsToFetch; |
255 | qSort(filteredAndSorted.begin(), filteredAndSorted.end(), qGreater<qint64>()); | 280 | qSort(filteredAndSorted.begin(), filteredAndSorted.end(), qGreater<qint64>()); |
256 | auto lowerBound = qLowerBound(filteredAndSorted.begin(), filteredAndSorted.end(), uidNext, qGreater<qint64>()); | 281 | auto lowerBound = qLowerBound(filteredAndSorted.begin(), filteredAndSorted.end(), uidNext, qGreater<qint64>()); |
257 | if (lowerBound != filteredAndSorted.end()) { | 282 | if (lowerBound != filteredAndSorted.end()) { |
258 | filteredAndSorted.erase(lowerBound, filteredAndSorted.end()); | 283 | filteredAndSorted.erase(lowerBound, filteredAndSorted.end()); |
259 | } | 284 | } |
285 | const qint64 lowerBoundUid = filteredAndSorted.isEmpty() ? 0 : filteredAndSorted.last(); | ||
260 | 286 | ||
261 | auto maxUid = QSharedPointer<qint64>::create(0); | 287 | auto maxUid = QSharedPointer<qint64>::create(0); |
262 | if (!filteredAndSorted.isEmpty()) { | 288 | if (!filteredAndSorted.isEmpty()) { |
263 | *maxUid = filteredAndSorted.first(); | 289 | *maxUid = filteredAndSorted.first(); |
264 | } | 290 | } |
265 | SinkTrace() << "Uids to fetch: " << filteredAndSorted; | 291 | SinkTraceCtx(mLogCtx) << "Uids to fetch: " << filteredAndSorted; |
266 | return imap->fetchMessages(folder, filteredAndSorted, [this, folder, maxUid](const Message &m) { | 292 | |
293 | bool headersOnly = false; | ||
294 | return imap->fetchMessages(folder, filteredAndSorted, headersOnly, [=](const Message &m) { | ||
267 | if (*maxUid < m.uid) { | 295 | if (*maxUid < m.uid) { |
268 | *maxUid = m.uid; | 296 | *maxUid = m.uid; |
269 | } | 297 | } |
270 | synchronizeMails(folderRid(folder), m); | 298 | synchronizeMails(folderRemoteId, m); |
271 | }, | 299 | }, |
272 | [this, maxUid, folder](int progress, int total) { | 300 | [this, maxUid, folder](int progress, int total) { |
273 | SinkLog() << "Progress: " << progress << " out of " << total; | 301 | SinkLog() << "Progress: " << progress << " out of " << total; |
@@ -276,30 +304,68 @@ public: | |||
276 | commit(); | 304 | commit(); |
277 | } | 305 | } |
278 | }) | 306 | }) |
279 | .syncThen<void>([this, maxUid, folder]() { | 307 | .syncThen<void>([=]() { |
280 | SinkLog() << "UIDMAX: " << *maxUid << folder.path(); | 308 | SinkLogCtx(mLogCtx) << "UIDMAX: " << *maxUid << folder.path(); |
281 | if (*maxUid > 0) { | 309 | if (*maxUid > 0) { |
282 | syncStore().writeValue(folderRid(folder) + "uidnext", QByteArray::number(*maxUid)); | 310 | syncStore().writeValue(folderRemoteId, "uidnext", QByteArray::number(*maxUid)); |
283 | } | 311 | } |
312 | syncStore().writeValue(folderRemoteId, "fullsetLowerbound", QByteArray::number(lowerBoundUid)); | ||
284 | commit(); | 313 | commit(); |
285 | }); | 314 | }); |
286 | }); | 315 | }); |
287 | }) | 316 | }) |
288 | .then<void>([=]() { | 317 | .then<void>([=] { |
289 | //TODO Remove what's no longer existing | 318 | bool ok = false; |
290 | if (canDoIncrementalRemovals) { | 319 | const auto headersFetched = !syncStore().readValue(folderRemoteId, "headersFetched").isEmpty(); |
291 | //TODO do an examine with QRESYNC and remove VANISHED messages | 320 | const auto fullsetLowerbound = syncStore().readValue(folderRemoteId, "fullsetLowerbound").toLongLong(&ok); |
292 | } else { | 321 | if (ok && !headersFetched) { |
293 | return imap->fetchUids(folder).syncThen<void, QVector<qint64>>([this, folder](const QVector<qint64> &uids) { | 322 | SinkLogCtx(mLogCtx) << "Fetching headers until: " << fullsetLowerbound; |
294 | SinkTrace() << "Syncing removals: " << folder.path(); | 323 | |
295 | synchronizeRemovals(folderRid(folder), uids.toList().toSet()); | 324 | return imap->fetchUids(imap->mailboxFromFolder(folder)) |
325 | .then<void, QVector<qint64>>([=] (const QVector<qint64> &uids) { | ||
326 | //sort in reverse order and remove everything greater than fullsetLowerbound | ||
327 | QVector<qint64> toFetch = uids; | ||
328 | qSort(toFetch.begin(), toFetch.end(), qGreater<qint64>()); | ||
329 | if (fullsetLowerbound) { | ||
330 | auto upperBound = qUpperBound(toFetch.begin(), toFetch.end(), fullsetLowerbound, qGreater<qint64>()); | ||
331 | if (upperBound != toFetch.begin()) { | ||
332 | toFetch.erase(toFetch.begin(), upperBound); | ||
333 | } | ||
334 | } | ||
335 | SinkLogCtx(mLogCtx) << "Fetching headers for: " << toFetch; | ||
336 | |||
337 | bool headersOnly = true; | ||
338 | return imap->fetchMessages(folder, toFetch, headersOnly, [=](const Message &m) { | ||
339 | synchronizeMails(folderRemoteId, m); | ||
340 | }, | ||
341 | [=](int progress, int total) { | ||
342 | SinkLogCtx(mLogCtx) << "Progress: " << progress << " out of " << total; | ||
343 | //commit every 100 messages | ||
344 | if ((progress % 100) == 0) { | ||
345 | commit(); | ||
346 | } | ||
347 | }); | ||
348 | }) | ||
349 | .syncThen<void>([=]() { | ||
350 | SinkLogCtx(mLogCtx) << "Headers fetched: " << folder.path(); | ||
351 | syncStore().writeValue(folderRemoteId, "headersFetched", "true"); | ||
296 | commit(); | 352 | commit(); |
297 | }); | 353 | }); |
354 | |||
355 | } else { | ||
356 | SinkLogCtx(mLogCtx) << "No additional headers to fetch."; | ||
298 | } | 357 | } |
299 | return KAsync::null<void>(); | 358 | return KAsync::null(); |
359 | }) | ||
360 | //Finally remove messages that are no longer existing on the server. | ||
361 | .then<void>([=]() { | ||
362 | //TODO do an examine with QRESYNC and remove VANISHED messages if supported instead | ||
363 | return imap->fetchUids(folder).syncThen<void, QVector<qint64>>([=](const QVector<qint64> &uids) { | ||
364 | SinkTraceCtx(mLogCtx) << "Syncing removals: " << folder.path(); | ||
365 | synchronizeRemovals(folderRemoteId, uids.toList().toSet()); | ||
366 | commit(); | ||
367 | }); | ||
300 | }); | 368 | }); |
301 | |||
302 | |||
303 | } | 369 | } |
304 | 370 | ||
305 | Sink::QueryBase applyMailDefaults(const Sink::QueryBase &query) | 371 | Sink::QueryBase applyMailDefaults(const Sink::QueryBase &query) |
@@ -337,6 +403,33 @@ public: | |||
337 | }); | 403 | }); |
338 | } | 404 | } |
339 | 405 | ||
406 | KAsync::Job<QVector<Folder>> getFolderList(QSharedPointer<ImapServerProxy> imap, const Sink::QueryBase &query) | ||
407 | { | ||
408 | if (query.hasFilter<ApplicationDomain::Mail::Folder>()) { | ||
409 | //If we have a folder filter fetch full payload of date-range & all headers | ||
410 | QVector<Folder> folders; | ||
411 | auto folderFilter = query.getFilter<ApplicationDomain::Mail::Folder>(); | ||
412 | auto localIds = resolveFilter(folderFilter); | ||
413 | auto folderRemoteIds = syncStore().resolveLocalIds(ApplicationDomain::getTypeName<ApplicationDomain::Folder>(), localIds); | ||
414 | for (const auto &r : folderRemoteIds) { | ||
415 | folders << Folder{r}; | ||
416 | } | ||
417 | return KAsync::value(folders); | ||
418 | } else { | ||
419 | //Otherwise fetch full payload for daterange | ||
420 | auto folderList = QSharedPointer<QVector<Folder>>::create(); | ||
421 | return imap->fetchFolders([folderList](const Folder &folder) { | ||
422 | if (!folder.noselect) { | ||
423 | *folderList << folder; | ||
424 | } | ||
425 | }) | ||
426 | .onError([](const KAsync::Error &error) { | ||
427 | SinkWarning() << "Folder list sync failed."; | ||
428 | }) | ||
429 | .syncThen<QVector<Folder>>([folderList]() { return *folderList; } ); | ||
430 | } | ||
431 | } | ||
432 | |||
340 | KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE | 433 | KAsync::Job<void> synchronizeWithSource(const Sink::QueryBase &query) Q_DECL_OVERRIDE |
341 | { | 434 | { |
342 | if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) { | 435 | if (query.type() == ApplicationDomain::getTypeName<ApplicationDomain::Folder>()) { |
@@ -368,51 +461,63 @@ public: | |||
368 | //* apply the date filter to the fetch | 461 | //* apply the date filter to the fetch |
369 | //if we have no folder filter: | 462 | //if we have no folder filter: |
370 | //* fetch list of folders from server directly and sync (because we have no guarantee that the folder sync was already processed by the pipeline). | 463 | //* fetch list of folders from server directly and sync (because we have no guarantee that the folder sync was already processed by the pipeline). |
371 | return KAsync::start<void>([this, query]() { | 464 | auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort); |
372 | auto imap = QSharedPointer<ImapServerProxy>::create(mServer, mPort); | 465 | return login(imap) |
373 | auto job = login(imap); | 466 | .then<void>([=] () -> KAsync::Job<void> { |
374 | job = job.then<QVector<Folder>>([this, imap, query]() { | 467 | if (!query.ids().isEmpty()) { |
375 | SinkLog() << "Login was successful"; | 468 | //If we have mail id's simply fetch the full payload of those mails |
376 | //FIXME If we were able to to flush in between we could just query the local store for the folder list. | 469 | QVector<qint64> toFetch; |
377 | // | 470 | auto mailRemoteIds = syncStore().resolveLocalIds(ApplicationDomain::getTypeName<ApplicationDomain::Mail>(), query.ids()); |
378 | if (query.hasFilter<ApplicationDomain::Mail::Folder>()) { | 471 | QByteArray folderRemoteId; |
379 | QVector<Folder> folders; | 472 | for (const auto &r : mailRemoteIds) { |
380 | auto folderFilter = query.getFilter<ApplicationDomain::Mail::Folder>(); | 473 | const auto f = folderIdFromMailRid(r); |
381 | auto localIds = resolveFilter(folderFilter); | 474 | if (folderRemoteId.isEmpty()) { |
382 | auto folderRemoteIds = syncStore().resolveLocalIds(ApplicationDomain::getTypeName<ApplicationDomain::Folder>(), localIds); | 475 | folderRemoteId = f; |
383 | for (const auto &r : folderRemoteIds) { | 476 | } else { |
384 | folders << Folder{r}; | 477 | if (folderRemoteId != f) { |
478 | SinkWarningCtx(mLogCtx) << "Not all messages come from the same folder " << r << folderRemoteId; | ||
479 | } | ||
385 | } | 480 | } |
386 | return KAsync::value(folders); | 481 | toFetch << uidFromMailRid(r); |
387 | } else { | ||
388 | auto folderList = QSharedPointer<QVector<Folder>>::create(); | ||
389 | return imap->fetchFolders([folderList](const Folder &folder) { | ||
390 | *folderList << folder; | ||
391 | }) | ||
392 | .onError([](const KAsync::Error &error) { | ||
393 | SinkWarning() << "Folder list sync failed."; | ||
394 | }) | ||
395 | .syncThen<QVector<Folder>>([this, folderList]() { | ||
396 | return *folderList; | ||
397 | }); | ||
398 | } | ||
399 | }) | ||
400 | .serialEach<void>([this, imap, query](const Folder &folder) { | ||
401 | if (folder.noselect) { | ||
402 | return KAsync::null<void>(); | ||
403 | } | 482 | } |
404 | QDate dateFilter; | 483 | SinkLog() << "Fetching messages: " << toFetch; |
405 | auto filter = query.getFilter<ApplicationDomain::Mail::Date>(); | 484 | bool headersOnly = false; |
406 | if (filter.value.canConvert<QDate>()) { | 485 | return imap->fetchMessages(Folder{folderRemoteId}, toFetch, headersOnly, [=](const Message &m) { |
407 | dateFilter = filter.value.value<QDate>(); | 486 | synchronizeMails(folderRemoteId, m); |
408 | } | 487 | }, |
409 | return synchronizeFolder(imap, folder, dateFilter) | 488 | [=](int progress, int total) { |
410 | .onError([folder](const KAsync::Error &error) { | 489 | SinkLogCtx(mLogCtx) << "Progress: " << progress << " out of " << total; |
411 | SinkWarning() << "Failed to sync folder: " << folder.path() << "Error: " << error.errorMessage; | 490 | //commit every 100 messages |
491 | if ((progress % 100) == 0) { | ||
492 | commit(); | ||
493 | } | ||
494 | }) | ||
495 | .syncThen<void>([=]() { | ||
496 | commit(); | ||
497 | }); | ||
498 | } else { | ||
499 | //Otherwise we sync the folder(s) | ||
500 | bool syncHeaders = query.hasFilter<ApplicationDomain::Mail::Folder>(); | ||
501 | //FIXME If we were able to to flush in between we could just query the local store for the folder list. | ||
502 | return getFolderList(imap, query) | ||
503 | .then<void, QVector<Folder>>([=] (const QVector<Folder> &folders) { | ||
504 | //Synchronize folders | ||
505 | return KAsync::value(folders) | ||
506 | .serialEach<void>([=](const Folder &folder) { | ||
507 | SinkLog() << "Syncing folder " << folder.path(); | ||
508 | QDate dateFilter; | ||
509 | auto filter = query.getFilter<ApplicationDomain::Mail::Date>(); | ||
510 | if (filter.value.canConvert<QDate>()) { | ||
511 | dateFilter = filter.value.value<QDate>(); | ||
512 | SinkLog() << " with date-range " << dateFilter; | ||
513 | } | ||
514 | return synchronizeFolder(imap, folder, dateFilter, syncHeaders) | ||
515 | .onError([folder](const KAsync::Error &error) { | ||
516 | SinkWarning() << "Failed to sync folder: " << folder.path() << "Error: " << error.errorMessage; | ||
517 | }); | ||
412 | }); | 518 | }); |
413 | }); | 519 | }); |
414 | 520 | } | |
415 | return job; | ||
416 | }); | 521 | }); |
417 | } | 522 | } |
418 | return KAsync::error<void>("Nothing to do"); | 523 | return KAsync::error<void>("Nothing to do"); |
@@ -425,13 +530,7 @@ public: | |||
425 | if (operation == Sink::Operation_Creation) { | 530 | if (operation == Sink::Operation_Creation) { |
426 | QString mailbox = syncStore().resolveLocalId(ENTITY_TYPE_FOLDER, mail.getFolder()); | 531 | QString mailbox = syncStore().resolveLocalId(ENTITY_TYPE_FOLDER, mail.getFolder()); |
427 | QByteArray content = KMime::LFtoCRLF(mail.getMimeMessage()); | 532 | QByteArray content = KMime::LFtoCRLF(mail.getMimeMessage()); |
428 | QByteArrayList flags; | 533 | auto flags = getFlags(mail); |
429 | if (!mail.getUnread()) { | ||
430 | flags << Imap::Flags::Seen; | ||
431 | } | ||
432 | if (mail.getImportant()) { | ||
433 | flags << Imap::Flags::Flagged; | ||
434 | } | ||
435 | QDateTime internalDate = mail.getDate(); | 534 | QDateTime internalDate = mail.getDate(); |
436 | return login.then(imap->append(mailbox, content, flags, internalDate)) | 535 | return login.then(imap->append(mailbox, content, flags, internalDate)) |
437 | .addToContext(imap) | 536 | .addToContext(imap) |
@@ -458,13 +557,7 @@ public: | |||
458 | 557 | ||
459 | SinkTrace() << "Modifying a mail: " << oldRemoteId << " in the mailbox: " << mailbox << changedProperties; | 558 | SinkTrace() << "Modifying a mail: " << oldRemoteId << " in the mailbox: " << mailbox << changedProperties; |
460 | 559 | ||
461 | QByteArrayList flags; | 560 | auto flags = getFlags(mail); |
462 | if (!mail.getUnread()) { | ||
463 | flags << Imap::Flags::Seen; | ||
464 | } | ||
465 | if (mail.getImportant()) { | ||
466 | flags << Imap::Flags::Flagged; | ||
467 | } | ||
468 | 561 | ||
469 | const bool messageMoved = changedProperties.contains(ApplicationDomain::Mail::Folder::name); | 562 | const bool messageMoved = changedProperties.contains(ApplicationDomain::Mail::Folder::name); |
470 | const bool messageChanged = changedProperties.contains(ApplicationDomain::Mail::MimeMessage::name); | 563 | const bool messageChanged = changedProperties.contains(ApplicationDomain::Mail::MimeMessage::name); |
diff --git a/examples/imapresource/imapserverproxy.cpp b/examples/imapresource/imapserverproxy.cpp index af1f4d1..a172c93 100644 --- a/examples/imapresource/imapserverproxy.cpp +++ b/examples/imapresource/imapserverproxy.cpp | |||
@@ -441,18 +441,18 @@ KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, qint64 ui | |||
441 | SinkTrace() << " Total: " << uidsToFetch.size(); | 441 | SinkTrace() << " Total: " << uidsToFetch.size(); |
442 | SinkTrace() << " Uids to fetch: " << uidsToFetch; | 442 | SinkTrace() << " Uids to fetch: " << uidsToFetch; |
443 | SinkTrace() << " Took: " << Sink::Log::TraceTime(time->elapsed()); | 443 | SinkTrace() << " Took: " << Sink::Log::TraceTime(time->elapsed()); |
444 | return fetchMessages(folder, uidsToFetch, callback, progress); | 444 | return fetchMessages(folder, uidsToFetch, false, callback, progress); |
445 | }); | 445 | }); |
446 | 446 | ||
447 | }); | 447 | }); |
448 | } | 448 | } |
449 | 449 | ||
450 | KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, std::function<void(const Message &)> callback, std::function<void(int, int)> progress) | 450 | KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, bool headersOnly, std::function<void(const Message &)> callback, std::function<void(int, int)> progress) |
451 | { | 451 | { |
452 | auto time = QSharedPointer<QTime>::create(); | 452 | auto time = QSharedPointer<QTime>::create(); |
453 | time->start(); | 453 | time->start(); |
454 | Q_ASSERT(!mPersonalNamespaceSeparator.isNull()); | 454 | Q_ASSERT(!mPersonalNamespaceSeparator.isNull()); |
455 | return select(mailboxFromFolder(folder)).then<void, SelectResult>([this, callback, folder, time, progress, uidsToFetch](const SelectResult &selectResult) -> KAsync::Job<void> { | 455 | return select(mailboxFromFolder(folder)).then<void, SelectResult>([this, callback, folder, time, progress, uidsToFetch, headersOnly](const SelectResult &selectResult) -> KAsync::Job<void> { |
456 | 456 | ||
457 | SinkTrace() << "Fetching messages" << folder.path(); | 457 | SinkTrace() << "Fetching messages" << folder.path(); |
458 | SinkTrace() << " Total: " << uidsToFetch.size(); | 458 | SinkTrace() << " Total: " << uidsToFetch.size(); |
@@ -467,7 +467,11 @@ KAsync::Job<void> ImapServerProxy::fetchMessages(const Folder &folder, const QVe | |||
467 | } | 467 | } |
468 | KIMAP2::FetchJob::FetchScope scope; | 468 | KIMAP2::FetchJob::FetchScope scope; |
469 | scope.parts.clear(); | 469 | scope.parts.clear(); |
470 | scope.mode = KIMAP2::FetchJob::FetchScope::Full; | 470 | if (headersOnly) { |
471 | scope.mode = KIMAP2::FetchJob::FetchScope::Headers; | ||
472 | } else { | ||
473 | scope.mode = KIMAP2::FetchJob::FetchScope::Full; | ||
474 | } | ||
471 | 475 | ||
472 | KIMAP2::ImapSet set; | 476 | KIMAP2::ImapSet set; |
473 | set.add(uidsToFetch); | 477 | set.add(uidsToFetch); |
diff --git a/examples/imapresource/imapserverproxy.h b/examples/imapresource/imapserverproxy.h index fa2d022..140c5b2 100644 --- a/examples/imapresource/imapserverproxy.h +++ b/examples/imapresource/imapserverproxy.h | |||
@@ -163,7 +163,7 @@ public: | |||
163 | KAsync::Job<void> fetchFolders(std::function<void(const Folder &)> callback); | 163 | KAsync::Job<void> fetchFolders(std::function<void(const Folder &)> callback); |
164 | KAsync::Job<void> fetchMessages(const Folder &folder, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>()); | 164 | KAsync::Job<void> fetchMessages(const Folder &folder, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>()); |
165 | KAsync::Job<void> fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>()); | 165 | KAsync::Job<void> fetchMessages(const Folder &folder, qint64 uidNext, std::function<void(const Message &)> callback, std::function<void(int, int)> progress = std::function<void(int, int)>()); |
166 | KAsync::Job<void> fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, std::function<void(const Message &)> callback, std::function<void(int, int)> progress); | 166 | KAsync::Job<void> fetchMessages(const Folder &folder, const QVector<qint64> &uidsToFetch, bool headersOnly, std::function<void(const Message &)> callback, std::function<void(int, int)> progress); |
167 | KAsync::Job<SelectResult> fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const Message &)> callback); | 167 | KAsync::Job<SelectResult> fetchFlags(const Folder &folder, const KIMAP2::ImapSet &set, qint64 changedsince, std::function<void(const Message &)> callback); |
168 | KAsync::Job<QVector<qint64>> fetchUids(const Folder &folder); | 168 | KAsync::Job<QVector<qint64>> fetchUids(const Folder &folder); |
169 | 169 | ||