summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--CMakeLists.txt12
-rw-r--r--async/CMakeLists.txt12
-rw-r--r--async/autotests/CMakeLists.txt11
-rw-r--r--async/autotests/asynctest.cpp859
-rw-r--r--async/autotests/kjobtest.cpp69
-rw-r--r--async/autotests/testkjob.cpp28
-rw-r--r--async/autotests/testkjob.h48
-rw-r--r--async/src/CMakeLists.txt17
-rw-r--r--async/src/async.cpp148
-rw-r--r--async/src/async.h872
-rw-r--r--async/src/async_impl.h81
-rw-r--r--async/src/debug.cpp75
-rw-r--r--async/src/debug.h80
-rw-r--r--async/src/future.cpp158
-rw-r--r--async/src/future.h255
-rw-r--r--common/CMakeLists.txt38
-rw-r--r--common/clientapi.cpp2
-rw-r--r--common/clientapi.h16
-rw-r--r--common/facade.h35
-rw-r--r--common/genericresource.cpp32
-rw-r--r--common/genericresource.h4
-rw-r--r--common/pipeline.cpp9
-rw-r--r--common/pipeline.h5
-rw-r--r--common/resource.cpp8
-rw-r--r--common/resource.h7
-rw-r--r--common/resourceaccess.cpp30
-rw-r--r--common/resourceaccess.h11
-rw-r--r--examples/dummyresource/CMakeLists.txt2
-rw-r--r--examples/dummyresource/facade.cpp4
-rw-r--r--examples/dummyresource/facade.h2
-rw-r--r--examples/dummyresource/resourcefacade.cpp16
-rw-r--r--examples/dummyresource/resourcefacade.h11
-rw-r--r--examples/dummyresource/resourcefactory.cpp4
-rw-r--r--examples/dummyresource/resourcefactory.h5
-rw-r--r--synchronizer/CMakeLists.txt2
-rw-r--r--synchronizer/listener.cpp2
-rw-r--r--tests/CMakeLists.txt4
-rw-r--r--tests/clientapitest.cpp18
-rw-r--r--tests/hawd/dataset.cpp1
39 files changed, 146 insertions, 2847 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt
index bf7f7ae..bd4a229 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -1,4 +1,4 @@
1cmake_minimum_required(VERSION 2.8) 1cmake_minimum_required(VERSION 2.8.12)
2 2
3 3
4# ECM setup 4# ECM setup
@@ -19,9 +19,9 @@ include(KDEInstallDirs)
19# include(KDECMakeSettings) 19# include(KDECMakeSettings)
20 20
21find_package(Qt5 COMPONENTS REQUIRED Core Widgets Network) 21find_package(Qt5 COMPONENTS REQUIRED Core Widgets Network)
22 22find_package(KF5 COMPONENTS REQUIRED Async)
23find_package(FlatBuffers REQUIRED) 23find_package(FlatBuffers REQUIRED)
24function(generate_flatbuffers) 24function(generate_flatbuffers _target)
25 foreach(fbs ${ARGN}) 25 foreach(fbs ${ARGN})
26 message("making ${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h from ${CMAKE_CURRENT_SOURCE_DIR}/${fbs}.fbs") 26 message("making ${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h from ${CMAKE_CURRENT_SOURCE_DIR}/${fbs}.fbs")
27 add_custom_command( 27 add_custom_command(
@@ -31,7 +31,8 @@ function(generate_flatbuffers)
31 ) 31 )
32 set_source_files_properties(${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h PROPERTIES GENERATED 1) 32 set_source_files_properties(${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h PROPERTIES GENERATED 1)
33 string(REGEX REPLACE "/" "_" target_name ${fbs}) 33 string(REGEX REPLACE "/" "_" target_name ${fbs})
34 add_custom_target(generate_bindings${target_name} ALL DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h) 34 add_custom_target(${_target}-generate_bindings${target_name} ALL DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/${fbs}_generated.h)
35 add_dependencies(${_target} ${_target}-generate_bindings${target_name})
35 endforeach(fbs) 36 endforeach(fbs)
36endfunction(generate_flatbuffers) 37endfunction(generate_flatbuffers)
37 38
@@ -58,7 +59,4 @@ add_subdirectory(examples)
58# some tests 59# some tests
59add_subdirectory(tests) 60add_subdirectory(tests)
60 61
61# async library prototype
62add_subdirectory(async)
63
64feature_summary(WHAT ALL FATAL_ON_MISSING_REQUIRED_PACKAGES) 62feature_summary(WHAT ALL FATAL_ON_MISSING_REQUIRED_PACKAGES)
diff --git a/async/CMakeLists.txt b/async/CMakeLists.txt
deleted file mode 100644
index a6b53f8..0000000
--- a/async/CMakeLists.txt
+++ /dev/null
@@ -1,12 +0,0 @@
1project(libasync)
2
3option(WITH_KJOB "Enable native support for KJob in libasync API (enabled by default)" ON)
4
5if (WITH_KJOB)
6 set(MINUMUM_FRAMEWORKS_VERSION "5.8.0")
7 find_package(KF5CoreAddons REQUIRED ${MINUMUM_FRAMEWORKS_VERSION})
8 add_definitions(-DWITH_KJOB)
9endif()
10
11add_subdirectory(src)
12add_subdirectory(autotests) \ No newline at end of file
diff --git a/async/autotests/CMakeLists.txt b/async/autotests/CMakeLists.txt
deleted file mode 100644
index 8116f13..0000000
--- a/async/autotests/CMakeLists.txt
+++ /dev/null
@@ -1,11 +0,0 @@
1include_directories(../src ${CMAKE_CURRENT_BINARY_DIR})
2
3add_executable(asynctest asynctest.cpp)
4qt5_use_modules(asynctest Test)
5target_link_libraries(asynctest akonadi2async Qt5::Core Qt5::Test)
6
7if (WITH_KJOB)
8 add_executable(kjobtest kjobtest.cpp testkjob.cpp)
9 qt5_use_modules(kjobtest Test)
10 target_link_libraries(kjobtest akonadi2async Qt5::Core Qt5::Test KF5::CoreAddons)
11endif () \ No newline at end of file
diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
deleted file mode 100644
index 9bc9f6b..0000000
--- a/async/autotests/asynctest.cpp
+++ /dev/null
@@ -1,859 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This program is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License or (at your option) version 3 or any later version
8 * accepted by the membership of KDE e.V. (or its successor approved
9 * by the membership of KDE e.V.), which shall act as a proxy
10 * defined in Section 14 of version 3 of the license.
11 *
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with this program. If not, see <http://www.gnu.org/licenses/>.
19 *
20 */
21
22#include "../src/async.h"
23
24#include <QObject>
25#include <QString>
26#include <QTimer>
27#include <QtTest/QTest>
28#include <QDebug>
29
30#include <functional>
31
32class AsyncTest : public QObject
33{
34 Q_OBJECT
35
36public:
37 AsyncTest()
38 {}
39
40 ~AsyncTest()
41 {}
42
43private Q_SLOTS:
44 void testSyncPromises();
45 void testAsyncPromises();
46 void testAsyncPromises2();
47 void testNestedAsync();
48 void testStartValue();
49
50 void testAsyncThen();
51 void testSyncThen();
52 void testJoinedThen();
53 void testVoidThen();
54
55 void testAsyncEach();
56 void testSyncEach();
57 void testJoinedEach();
58 void testVoidEachThen();
59 void testAsyncVoidEachThen();
60
61 void testAsyncReduce();
62 void testSyncReduce();
63 void testJoinedReduce();
64 void testVoidReduce();
65
66 void testProgressReporting();
67 void testErrorHandler();
68 void testErrorPropagation();
69 void testErrorHandlerAsync();
70 void testErrorPropagationAsync();
71 void testNestedErrorPropagation();
72
73 void testChainingRunningJob();
74 void testChainingFinishedJob();
75
76 void testLifetimeWithoutHandle();
77 void testLifetimeWithHandle();
78
79 void benchmarkSyncThenExecutor();
80
81private:
82 template<typename T>
83 class AsyncSimulator {
84 public:
85 AsyncSimulator(Async::Future<T> &future, const T &result)
86 : mFuture(future)
87 , mResult(result)
88 {
89 QObject::connect(&mTimer, &QTimer::timeout,
90 [this]() {
91 mFuture.setValue(mResult);
92 mFuture.setFinished();
93 });
94 QObject::connect(&mTimer, &QTimer::timeout,
95 [this]() {
96 delete this;
97 });
98 mTimer.setSingleShot(true);
99 mTimer.start(200);
100 }
101
102 AsyncSimulator(Async::Future<T> &future, std::function<void(Async::Future<T>&)> callback)
103 : mFuture(future)
104 , mCallback(callback)
105 {
106 QObject::connect(&mTimer, &QTimer::timeout,
107 [this]() {
108 mCallback(mFuture);
109 });
110 QObject::connect(&mTimer, &QTimer::timeout,
111 [this]() {
112 delete this;
113 });
114 mTimer.setSingleShot(true);
115 mTimer.start(200);
116 }
117
118 private:
119 Async::Future<T> mFuture;
120 std::function<void(Async::Future<T>&)> mCallback;
121 T mResult;
122 QTimer mTimer;
123 };
124};
125
126
127template<>
128class AsyncTest::AsyncSimulator<void> {
129public:
130 AsyncSimulator(Async::Future<void> &future)
131 : mFuture(future)
132 {
133 QObject::connect(&mTimer, &QTimer::timeout,
134 [this]() {
135 mFuture.setFinished();
136 });
137 QObject::connect(&mTimer, &QTimer::timeout,
138 [this]() {
139 delete this;
140 });
141 mTimer.setSingleShot(true);
142 mTimer.start(200);
143 }
144
145private:
146 Async::Future<void> mFuture;
147 QTimer mTimer;
148};
149
150
151
152void AsyncTest::testSyncPromises()
153{
154 auto baseJob = Async::start<int>(
155 [](Async::Future<int> &f) {
156 f.setValue(42);
157 f.setFinished();
158 })
159 .then<QString, int>(
160 [](int v, Async::Future<QString> &f) {
161 f.setValue("Result is " + QString::number(v));
162 f.setFinished();
163 });
164
165 auto job = baseJob.then<QString, QString>(
166 [](const QString &v, Async::Future<QString> &f) {
167 f.setValue(v.toUpper());
168 f.setFinished();
169 });
170
171 Async::Future<QString> future = job.exec();
172
173 QVERIFY(future.isFinished());
174 QCOMPARE(future.value(), QString::fromLatin1("RESULT IS 42"));
175}
176
177void AsyncTest::testAsyncPromises()
178{
179 auto job = Async::start<int>(
180 [](Async::Future<int> &future) {
181 new AsyncSimulator<int>(future, 42);
182 });
183
184 Async::Future<int> future = job.exec();
185
186 future.waitForFinished();
187 QCOMPARE(future.value(), 42);
188}
189
190void AsyncTest::testAsyncPromises2()
191{
192 bool done = false;
193
194 auto job = Async::start<int>(
195 [](Async::Future<int> &future) {
196 new AsyncSimulator<int>(future, 42);
197 }
198 ).then<int, int>([&done](int result, Async::Future<int> &future) {
199 done = true;
200 future.setValue(result);
201 future.setFinished();
202 });
203 auto future = job.exec();
204
205 QTRY_VERIFY(done);
206 QCOMPARE(future.value(), 42);
207}
208
209void AsyncTest::testNestedAsync()
210{
211 bool done = false;
212
213 auto job = Async::start<int>(
214 [](Async::Future<int> &future) {
215 auto innerJob = Async::start<int>([](Async::Future<int> &innerFuture) {
216 new AsyncSimulator<int>(innerFuture, 42);
217 }).then<void>([&future](Async::Future<void> &innerThenFuture) {
218 future.setFinished();
219 innerThenFuture.setFinished();
220 });
221 innerJob.exec().waitForFinished();
222 }
223 ).then<int, int>([&done](int result, Async::Future<int> &future) {
224 done = true;
225 future.setValue(result);
226 future.setFinished();
227 });
228 job.exec();
229
230 QTRY_VERIFY(done);
231}
232
233void AsyncTest::testStartValue()
234{
235 auto job = Async::start<int, int>(
236 [](int in, Async::Future<int> &future) {
237 future.setValue(in);
238 future.setFinished();
239 });
240
241 auto future = job.exec(42);
242 QVERIFY(future.isFinished());
243 QCOMPARE(future.value(), 42);
244}
245
246
247
248
249
250void AsyncTest::testAsyncThen()
251{
252 auto job = Async::start<int>(
253 [](Async::Future<int> &future) {
254 new AsyncSimulator<int>(future, 42);
255 });
256
257 auto future = job.exec();
258 future.waitForFinished();
259
260 QVERIFY(future.isFinished());
261 QCOMPARE(future.value(), 42);
262}
263
264
265void AsyncTest::testSyncThen()
266{
267 auto job = Async::start<int>(
268 []() -> int {
269 return 42;
270 })
271 .then<int, int>(
272 [](int in) -> int {
273 return in * 2;
274 });
275
276 auto future = job.exec();
277 QVERIFY(future.isFinished());
278 QCOMPARE(future.value(), 84);
279}
280
281void AsyncTest::testJoinedThen()
282{
283 auto job1 = Async::start<int, int>(
284 [](int in, Async::Future<int> &future) {
285 new AsyncSimulator<int>(future, in * 2);
286 });
287
288 auto job2 = Async::start<int>(
289 [](Async::Future<int> &future) {
290 new AsyncSimulator<int>(future, 42);
291 })
292 .then<int>(job1);
293
294 auto future = job2.exec();
295 future.waitForFinished();
296
297 QVERIFY(future.isFinished());
298 QCOMPARE(future.value(), 84);
299}
300
301void AsyncTest::testVoidThen()
302{
303 int check = 0;
304
305 auto job = Async::start<void>(
306 [&check](Async::Future<void> &future) {
307 new AsyncSimulator<void>(future);
308 ++check;
309 })
310 .then<void>(
311 [&check](Async::Future<void> &future) {
312 new AsyncSimulator<void>(future);
313 ++check;
314 })
315 .then<void>(
316 [&check]() {
317 ++check;
318 });
319
320 auto future = job.exec();
321 future.waitForFinished();
322
323 QVERIFY(future.isFinished());
324 QCOMPARE(check, 3);
325}
326
327
328
329void AsyncTest::testAsyncEach()
330{
331 auto job = Async::start<QList<int>>(
332 [](Async::Future<QList<int>> &future) {
333 new AsyncSimulator<QList<int>>(future, { 1, 2, 3, 4 });
334 })
335 .each<QList<int>, int>(
336 [](const int &v, Async::Future<QList<int>> &future) {
337 new AsyncSimulator<QList<int>>(future, { v + 1 });
338 });
339
340 auto future = job.exec();
341 future.waitForFinished();
342
343 const QList<int> expected({ 2, 3, 4, 5 });
344 QVERIFY(future.isFinished());
345 QCOMPARE(future.value(), expected);
346}
347
348void AsyncTest::testSyncEach()
349{
350 auto job = Async::start<QList<int>>(
351 []() -> QList<int> {
352 return { 1, 2, 3, 4 };
353 })
354 .each<QList<int>, int>(
355 [](const int &v) -> QList<int> {
356 return { v + 1 };
357 });
358
359 Async::Future<QList<int>> future = job.exec();
360
361 const QList<int> expected({ 2, 3, 4, 5 });
362 QVERIFY(future.isFinished());
363 QCOMPARE(future.value(), expected);
364}
365
366void AsyncTest::testJoinedEach()
367{
368 auto job1 = Async::start<QList<int>, int>(
369 [](int v, Async::Future<QList<int>> &future) {
370 new AsyncSimulator<QList<int>>(future, { v * 2 });
371 });
372
373 auto job = Async::start<QList<int>>(
374 []() -> QList<int> {
375 return { 1, 2, 3, 4 };
376 })
377 .each(job1);
378
379 auto future = job.exec();
380 future.waitForFinished();
381
382 const QList<int> expected({ 2, 4, 6, 8 });
383 QVERIFY(future.isFinished());
384 QCOMPARE(future.value(), expected);
385}
386
387void AsyncTest::testVoidEachThen()
388{
389 QList<int> check;
390 auto job = Async::start<QList<int>>(
391 []() -> QList<int> {
392 return { 1, 2, 3, 4 };
393 }).each<void, int>(
394 [&check](const int &v) {
395 check << v;
396 }).then<void>([](){});
397
398 auto future = job.exec();
399
400 const QList<int> expected({ 1, 2, 3, 4 });
401 QVERIFY(future.isFinished());
402 QCOMPARE(check, expected);
403}
404
405void AsyncTest::testAsyncVoidEachThen()
406{
407 bool completedJob = false;
408 QList<int> check;
409 auto job = Async::start<QList<int>>(
410 [](Async::Future<QList<int> > &future) {
411 new AsyncSimulator<QList<int>>(future, { 1, 2, 3, 4 });
412 }).each<void, int>(
413 [&check](const int &v, Async::Future<void> &future) {
414 check << v;
415 new AsyncSimulator<void>(future);
416 }).then<void>([&completedJob](Async::Future<void> &future) {
417 completedJob = true;
418 future.setFinished();
419 });
420
421 auto future = job.exec();
422 future.waitForFinished();
423
424 const QList<int> expected({ 1, 2, 3, 4 });
425 QVERIFY(future.isFinished());
426 QVERIFY(completedJob);
427 QCOMPARE(check, expected);
428}
429
430
431
432
433
434void AsyncTest::testAsyncReduce()
435{
436 auto job = Async::start<QList<int>>(
437 [](Async::Future<QList<int>> &future) {
438 new AsyncSimulator<QList<int>>(future, { 1, 2, 3, 4 });
439 })
440 .reduce<int, QList<int>>(
441 [](const QList<int> &list, Async::Future<int> &future) {
442 new AsyncSimulator<int>(future,
443 [list](Async::Future<int> &future) {
444 int sum = 0;
445 for (int i : list) sum += i;
446 future.setValue(sum);
447 future.setFinished();
448 }
449 );
450 });
451
452 Async::Future<int> future = job.exec();
453 future.waitForFinished();
454
455 QVERIFY(future.isFinished());
456 QCOMPARE(future.value(), 10);
457}
458
459void AsyncTest::testSyncReduce()
460{
461 auto job = Async::start<QList<int>>(
462 []() -> QList<int> {
463 return { 1, 2, 3, 4 };
464 })
465 .reduce<int, QList<int>>(
466 [](const QList<int> &list) -> int {
467 int sum = 0;
468 for (int i : list) sum += i;
469 return sum;
470 });
471
472 Async::Future<int> future = job.exec();
473
474 QVERIFY(future.isFinished());
475 QCOMPARE(future.value(), 10);
476}
477
478
479void AsyncTest::testJoinedReduce()
480{
481 auto job1 = Async::start<int, QList<int>>(
482 [](const QList<int> &list, Async::Future<int> &future) {
483 int sum = 0;
484 for (int i : list) sum += i;
485 new AsyncSimulator<int>(future, sum);
486 });
487
488 auto job = Async::start<QList<int>>(
489 []() -> QList<int> {
490 return { 1, 2, 3, 4 };
491 })
492 .reduce(job1);
493
494 auto future = job.exec();
495 future.waitForFinished();
496
497 QVERIFY(future.isFinished());
498 QCOMPARE(future.value(), 10);
499}
500
501void AsyncTest::testVoidReduce()
502{
503// This must not compile (reduce with void result makes no sense)
504#ifdef TEST_BUILD_FAIL
505 auto job = Async::start<QList<int>>(
506 []() -> QList<int> {
507 return { 1, 2, 3, 4 };
508 })
509 .reduce<void, QList<int>>(
510 [](const QList<int> &list) -> int {
511 return;
512 });
513
514 auto future = job.exec();
515 QVERIFY(future.isFinished());
516#endif
517}
518
519
520void AsyncTest::testProgressReporting()
521{
522 static int progress;
523 progress = 0;
524
525 auto job = Async::start<void>(
526 [](Async::Future<void> &f) {
527 QTimer *timer = new QTimer();
528 connect(timer, &QTimer::timeout,
529 [&f, timer]() {
530 f.setProgress(++progress);
531 if (progress == 100) {
532 timer->stop();
533 timer->deleteLater();
534 f.setFinished();
535 }
536 });
537 timer->start(1);
538 });
539
540 int progressCheck = 0;
541 Async::FutureWatcher<void> watcher;
542 connect(&watcher, &Async::FutureWatcher<void>::futureProgress,
543 [&progressCheck](qreal progress) {
544 progressCheck++;
545 // FIXME: Don't use Q_ASSERT in unit tests
546 Q_ASSERT((int) progress == progressCheck);
547 });
548 watcher.setFuture(job.exec());
549 watcher.future().waitForFinished();
550
551 QVERIFY(watcher.future().isFinished());
552 QCOMPARE(progressCheck, 100);
553}
554
555void AsyncTest::testErrorHandler()
556{
557
558 {
559 auto job = Async::start<int>(
560 [](Async::Future<int> &f) {
561 f.setError(1, "error");
562 });
563
564 auto future = job.exec();
565 QVERIFY(future.isFinished());
566 QCOMPARE(future.errorCode(), 1);
567 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
568 }
569
570 {
571 int error = 0;
572 auto job = Async::start<int>(
573 [](Async::Future<int> &f) {
574 f.setError(1, "error");
575 },
576 [&error](int errorCode, const QString &errorMessage) {
577 error += errorCode;
578 }
579 );
580
581 auto future = job.exec();
582 QVERIFY(future.isFinished());
583 QCOMPARE(error, 1);
584 QCOMPARE(future.errorCode(), 1);
585 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
586 }
587}
588
589void AsyncTest::testErrorPropagation()
590{
591 int error = 0;
592 bool called = false;
593 auto job = Async::start<int>(
594 [](Async::Future<int> &f) {
595 f.setError(1, "error");
596 })
597 .then<int, int>(
598 [&called](int v, Async::Future<int> &f) {
599 called = true;
600 f.setFinished();
601 },
602 [&error](int errorCode, const QString &errorMessage) {
603 error += errorCode;
604 }
605 );
606 auto future = job.exec();
607 QVERIFY(future.isFinished());
608 QCOMPARE(future.errorCode(), 1);
609 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
610 QCOMPARE(called, false);
611 QCOMPARE(error, 1);
612}
613
614void AsyncTest::testErrorHandlerAsync()
615{
616 {
617 auto job = Async::start<int>(
618 [](Async::Future<int> &f) {
619 new AsyncSimulator<int>(f,
620 [](Async::Future<int> &f) {
621 f.setError(1, "error");
622 }
623 );
624 }
625 );
626
627 auto future = job.exec();
628 future.waitForFinished();
629
630 QVERIFY(future.isFinished());
631 QCOMPARE(future.errorCode(), 1);
632 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
633 }
634
635 {
636 int error = 0;
637 auto job = Async::start<int>(
638 [](Async::Future<int> &f) {
639 new AsyncSimulator<int>(f,
640 [](Async::Future<int> &f) {
641 f.setError(1, "error");
642 }
643 );
644 },
645 [&error](int errorCode, const QString &errorMessage) {
646 error += errorCode;
647 }
648 );
649
650 auto future = job.exec();
651 future.waitForFinished();
652
653 QVERIFY(future.isFinished());
654 QCOMPARE(error, 1);
655 QCOMPARE(future.errorCode(), 1);
656 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
657 }
658}
659
660void AsyncTest::testErrorPropagationAsync()
661{
662 int error = 0;
663 bool called = false;
664 auto job = Async::start<int>(
665 [](Async::Future<int> &f) {
666 new AsyncSimulator<int>(f,
667 [](Async::Future<int> &f) {
668 f.setError(1, "error");
669 }
670 );
671 })
672 .then<int, int>(
673 [&called](int v, Async::Future<int> &f) {
674 called = true;
675 f.setFinished();
676 },
677 [&error](int errorCode, const QString &errorMessage) {
678 error += errorCode;
679 }
680 );
681
682 auto future = job.exec();
683 future.waitForFinished();
684
685 QVERIFY(future.isFinished());
686 QCOMPARE(future.errorCode(), 1);
687 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
688 QCOMPARE(called, false);
689 QCOMPARE(error, 1);
690}
691
692void AsyncTest::testNestedErrorPropagation()
693{
694 int error = 0;
695 auto job = Async::start<void>([](){})
696 .then<void>(Async::error<void>(1, "error")) //Nested job that throws error
697 .then<void>([](Async::Future<void> &future) {
698 //We should never get here
699 Q_ASSERT(false);
700 },
701 [&error](int errorCode, const QString &errorMessage) {
702 error += errorCode;
703 }
704 );
705 auto future = job.exec();
706
707 future.waitForFinished();
708
709 QVERIFY(future.isFinished());
710 QCOMPARE(future.errorCode(), 1);
711 QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
712 QCOMPARE(error, 1);
713}
714
715
716
717
718void AsyncTest::testChainingRunningJob()
719{
720 int check = 0;
721
722 auto job = Async::start<int>(
723 [&check](Async::Future<int> &future) {
724 QTimer *timer = new QTimer();
725 QObject::connect(timer, &QTimer::timeout,
726 [&future, &check]() {
727 ++check;
728 future.setValue(42);
729 future.setFinished();
730 });
731 QObject::connect(timer, &QTimer::timeout,
732 timer, &QObject::deleteLater);
733 timer->setSingleShot(true);
734 timer->start(500);
735 });
736
737 auto future1 = job.exec();
738 QTest::qWait(200);
739
740 auto job2 = job.then<int, int>(
741 [&check](int in) -> int {
742 ++check;
743 return in * 2;
744 });
745
746 auto future2 = job2.exec();
747 QVERIFY(!future1.isFinished());
748 future2.waitForFinished();
749
750 QEXPECT_FAIL("", "Chaining new job to a running job no longer executes the new job. "
751 "This is a trade-off for being able to re-execute single job multiple times.",
752 Abort);
753
754 QCOMPARE(check, 2);
755
756 QVERIFY(future1.isFinished());
757 QVERIFY(future2.isFinished());
758 QCOMPARE(future1.value(), 42);
759 QCOMPARE(future2.value(), 84);
760}
761
762void AsyncTest::testChainingFinishedJob()
763{
764 int check = 0;
765
766 auto job = Async::start<int>(
767 [&check]() -> int {
768 ++check;
769 return 42;
770 });
771
772 auto future1 = job.exec();
773 QVERIFY(future1.isFinished());
774
775 auto job2 = job.then<int, int>(
776 [&check](int in) -> int {
777 ++check;
778 return in * 2;
779 });
780
781 auto future2 = job2.exec();
782 QVERIFY(future2.isFinished());
783
784 QEXPECT_FAIL("", "Resuming finished job by chaining a new job and calling exec() is no longer suppported. "
785 "This is a trade-off for being able to re-execute single job multiple times.",
786 Abort);
787
788 QCOMPARE(check, 2);
789
790 QCOMPARE(future1.value(), 42);
791 QCOMPARE(future2.value(), 84);
792}
793
794/*
795 * We want to be able to execute jobs without keeping a handle explicitly alive.
796 * If the future handle inside the continuation would keep the executor alive, that would probably already work.
797 */
798void AsyncTest::testLifetimeWithoutHandle()
799{
800 bool done = false;
801 {
802 auto job = Async::start<void>([&done](Async::Future<void> &future) {
803 QTimer *timer = new QTimer();
804 QObject::connect(timer, &QTimer::timeout,
805 [&future, &done]() {
806 done = true;
807 future.setFinished();
808 });
809 QObject::connect(timer, &QTimer::timeout,
810 timer, &QObject::deleteLater);
811 timer->setSingleShot(true);
812 timer->start(500);
813 });
814 job.exec();
815 }
816
817 QTRY_VERIFY(done);
818}
819
820/*
821 * The future handle should keep the executor alive, and the future reference should probably not become invalid inside the continuation,
822 * until the job is done (alternatively a copy of the future inside the continuation should work as well).
823 */
824void AsyncTest::testLifetimeWithHandle()
825{
826 Async::Future<void> future;
827 {
828 auto job = Async::start<void>([](Async::Future<void> &future) {
829 QTimer *timer = new QTimer();
830 QObject::connect(timer, &QTimer::timeout,
831 [&future]() {
832 future.setFinished();
833 });
834 QObject::connect(timer, &QTimer::timeout,
835 timer, &QObject::deleteLater);
836 timer->setSingleShot(true);
837 timer->start(500);
838 });
839 future = job.exec();
840 }
841
842 QTRY_VERIFY(future.isFinished());
843}
844
845void AsyncTest::benchmarkSyncThenExecutor()
846{
847 auto job = Async::start<int>(
848 []() -> int {
849 return 0;
850 });
851
852 QBENCHMARK {
853 job.exec();
854 }
855}
856
857QTEST_MAIN(AsyncTest);
858
859#include "asynctest.moc"
diff --git a/async/autotests/kjobtest.cpp b/async/autotests/kjobtest.cpp
deleted file mode 100644
index be92d68..0000000
--- a/async/autotests/kjobtest.cpp
+++ /dev/null
@@ -1,69 +0,0 @@
1/*
2 * Copyright 2015 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This program is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License or (at your option) version 3 or any later version
8 * accepted by the membership of KDE e.V. (or its successor approved
9 * by the membership of KDE e.V.), which shall act as a proxy
10 * defined in Section 14 of version 3 of the license.
11 *
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with this program. If not, see <http://www.gnu.org/licenses/>.
19 *
20 */
21
22#include "../src/async.h"
23#include "testkjob.h"
24
25#include <QtTest/QTest>
26
27class KJobTest : public QObject
28{
29 Q_OBJECT
30
31public:
32 KJobTest()
33 {}
34
35 ~KJobTest()
36 {}
37
38private Q_SLOTS:
39 void testSingleKJob();
40 void testKJobChain();
41
42};
43
44void KJobTest::testSingleKJob()
45{
46 auto job = Async::start<int, TestKJob, &TestKJob::result, int>();
47
48 auto future = job.exec(42);
49 future.waitForFinished();
50
51 QVERIFY(future.isFinished());
52 QCOMPARE(future.value(), 42);
53}
54
55void KJobTest::testKJobChain()
56{
57 auto job = Async::start<int, TestKJob, &TestKJob::result, int>()
58 .then<int, TestKJob, &TestKJob::result, int>();
59
60 auto future = job.exec(42);
61 future.waitForFinished();
62
63 QVERIFY(future.isFinished());
64 QCOMPARE(future.value(), 42);
65}
66
67QTEST_MAIN(KJobTest)
68
69#include "kjobtest.moc" \ No newline at end of file
diff --git a/async/autotests/testkjob.cpp b/async/autotests/testkjob.cpp
deleted file mode 100644
index b86f913..0000000
--- a/async/autotests/testkjob.cpp
+++ /dev/null
@@ -1,28 +0,0 @@
1#include "testkjob.h"
2
3TestKJob::TestKJob(int result)
4 : mResult(result)
5{
6 connect(&mTimer, &QTimer::timeout,
7 this, &TestKJob::onTimeout);
8 mTimer.setSingleShot(true);
9 mTimer.setInterval(200);
10}
11
12TestKJob::~TestKJob()
13{}
14
15void TestKJob::start()
16{
17 mTimer.start();
18}
19
20int TestKJob::result()
21{
22 return mResult;
23}
24
25void TestKJob::onTimeout()
26{
27 emitResult();
28} \ No newline at end of file
diff --git a/async/autotests/testkjob.h b/async/autotests/testkjob.h
deleted file mode 100644
index eead98e..0000000
--- a/async/autotests/testkjob.h
+++ /dev/null
@@ -1,48 +0,0 @@
1/*
2 * Copyright 2015 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This program is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License or (at your option) version 3 or any later version
8 * accepted by the membership of KDE e.V. (or its successor approved
9 * by the membership of KDE e.V.), which shall act as a proxy
10 * defined in Section 14 of version 3 of the license.
11 *
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU General Public License for more details.
16 *
17 * You should have received a copy of the GNU General Public License
18 * along with this program. If not, see <http://www.gnu.org/licenses/>.
19 *
20 */
21
22#ifndef TESTKJOB_H
23#define TESTKJOB_H
24
25#include <KJob>
26#include <QTimer>
27
28class TestKJob : public KJob
29{
30 Q_OBJECT
31
32public:
33 TestKJob(int result);
34 ~TestKJob();
35
36 void start();
37
38 int result();
39
40private Q_SLOTS:
41 void onTimeout();
42
43private:
44 int mResult;
45 QTimer mTimer;
46};
47
48#endif // TESTKJOB_H \ No newline at end of file
diff --git a/async/src/CMakeLists.txt b/async/src/CMakeLists.txt
deleted file mode 100644
index becc8ee..0000000
--- a/async/src/CMakeLists.txt
+++ /dev/null
@@ -1,17 +0,0 @@
1project(akonadi2async)
2
3include_directories(${CMAKE_CURRENT_BINARY_DIR})
4
5set(async_SRCS
6 async.cpp
7 future.cpp
8 debug.cpp
9)
10
11add_library(${PROJECT_NAME} SHARED ${async_SRCS})
12target_link_libraries(${PROJECT_NAME} PUBLIC Qt5::Core)
13if (WITH_KJOB)
14 target_link_libraries(${PROJECT_NAME} PUBLIC KF5::CoreAddons)
15endif ()
16
17install(TARGETS ${PROJECT_NAME} ${KDE_INSTALL_TARGETS_DEFAULT_ARGS})
diff --git a/async/src/async.cpp b/async/src/async.cpp
deleted file mode 100644
index 0e86a84..0000000
--- a/async/src/async.cpp
+++ /dev/null
@@ -1,148 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#include "async.h"
19
20#include <QCoreApplication>
21#include <QDebug>
22#include <QEventLoop>
23#include <QTimer>
24
25using namespace Async;
26
27Private::Execution::Execution(const Private::ExecutorBasePtr &executor)
28 : executor(executor)
29 , resultBase(nullptr)
30 , isRunning(false)
31 , isFinished(false)
32{
33}
34
35Private::Execution::~Execution()
36{
37 if (resultBase) {
38 resultBase->releaseExecution();
39 delete resultBase;
40 }
41 prevExecution.reset();
42}
43
44void Private::Execution::setFinished()
45{
46 isFinished = true;
47 //executor.clear();
48#ifndef QT_NO_DEBUG
49 if (tracer) {
50 delete tracer;
51 }
52#endif
53}
54
55void Private::Execution::releaseFuture()
56{
57 resultBase = 0;
58}
59
60bool Private::Execution::errorWasHandled() const
61{
62 Execution *exec = const_cast<Execution*>(this);
63 while (exec) {
64 if (exec->executor->hasErrorFunc()) {
65 return true;
66 }
67 exec = exec->prevExecution.data();
68 }
69 return false;
70}
71
72
73
74
75
76Private::ExecutorBase::ExecutorBase(const ExecutorBasePtr &parent)
77 : mPrev(parent)
78{
79}
80
81Private::ExecutorBase::~ExecutorBase()
82{
83}
84
85
86
87
88JobBase::JobBase(const Private::ExecutorBasePtr &executor)
89 : mExecutor(executor)
90{
91}
92
93JobBase::~JobBase()
94{
95}
96
97static void asyncWhile(const std::function<void(std::function<void(bool)>)> &body, const std::function<void()> &completionHandler) {
98 body([body, completionHandler](bool complete) {
99 if (complete) {
100 completionHandler();
101 } else {
102 asyncWhile(body, completionHandler);
103 }
104 });
105}
106
107Job<void> Async::dowhile(Condition condition, ThenTask<void> body)
108{
109 return Async::start<void>([body, condition](Async::Future<void> &future) {
110 asyncWhile([condition, body](std::function<void(bool)> whileCallback) {
111 Async::start<void>(body).then<void>([whileCallback, condition]() {
112 whileCallback(!condition());
113 }).exec();
114 },
115 [&future]() { //while complete
116 future.setFinished();
117 });
118 });
119}
120
121Job<void> Async::dowhile(ThenTask<bool> body)
122{
123 return Async::start<void>([body](Async::Future<void> &future) {
124 asyncWhile([body](std::function<void(bool)> whileCallback) {
125 Async::start<bool>(body).then<bool, bool>([whileCallback](bool result) {
126 whileCallback(!result);
127 //FIXME this return value is only required because .then<bool, void> doesn't work
128 return true;
129 }).exec();
130 },
131 [&future]() { //while complete
132 future.setFinished();
133 });
134 });
135}
136
137Job<void> Async::wait(int delay)
138{
139 auto timer = QSharedPointer<QTimer>::create();
140 return Async::start<void>([timer, delay](Async::Future<void> &future) {
141 timer->setSingleShot(true);
142 QObject::connect(timer.data(), &QTimer::timeout, [&future]() {
143 future.setFinished();
144 });
145 timer->start(delay);
146 });
147}
148
diff --git a/async/src/async.h b/async/src/async.h
deleted file mode 100644
index b1f1121..0000000
--- a/async/src/async.h
+++ /dev/null
@@ -1,872 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#ifndef ASYNC_H
19#define ASYNC_H
20
21#include <functional>
22#include <list>
23#include <type_traits>
24#include <cassert>
25#include <iterator>
26
27#include "future.h"
28#include "debug.h"
29#include "async_impl.h"
30
31#include <QVector>
32#include <QObject>
33#include <QSharedPointer>
34
35#include <QDebug>
36
37#ifdef WITH_KJOB
38#include <KJob>
39#endif
40
41
42/*
43 * API to help write async code.
44 *
45 * This API is based around jobs that take lambdas to execute asynchronous tasks. Each async operation can take a continuation,
46 * that can then be used to execute further async operations. That way it is possible to build async chains of operations,
47 * that can be stored and executed later on. Jobs can be composed, similarly to functions.
48 *
49 * Relations between the components:
50 * * Job: API wrapper around Executors chain. Can be destroyed while still running,
51 * because the actual execution happens in the background
52 * * Executor: Describes task to execute. Executors form a linked list matching the
53 * order in which they will be executed. The Executor chain is destroyed when
54 * the parent Job is destroyed. However if the Job is still running it is
55 * guaranteed that the Executor chain will not be destroyed until the execution
56 * is finished.
57 * * Execution: The running execution of the task stored in Executor. Each call to Job::exec()
58 * instantiates new Execution chain, which makes it possible for the Job to be
59 * executed multiple times (even in parallel).
60 * * Future: Representation of the result that is being calculated
61 *
62 *
63 * TODO: Composed progress reporting
64 * TODO: Possibility to abort a job through future (perhaps optional?)
65 * TODO: Support for timeout, specified during exec call, after which the error handler gets called with a defined errorCode.
66 */
67namespace Async {
68
69template<typename PrevOut, typename Out, typename ... In>
70class Executor;
71
72class JobBase;
73
74template<typename Out, typename ... In>
75class Job;
76
77template<typename Out, typename ... In>
78using ThenTask = typename detail::identity<std::function<void(In ..., Async::Future<Out>&)>>::type;
79template<typename Out, typename ... In>
80using SyncThenTask = typename detail::identity<std::function<Out(In ...)>>::type;
81template<typename Out, typename In>
82using EachTask = typename detail::identity<std::function<void(In, Async::Future<Out>&)>>::type;
83template<typename Out, typename In>
84using SyncEachTask = typename detail::identity<std::function<Out(In)>>::type;
85template<typename Out, typename In>
86using ReduceTask = typename detail::identity<std::function<void(In, Async::Future<Out>&)>>::type;
87template<typename Out, typename In>
88using SyncReduceTask = typename detail::identity<std::function<Out(In)>>::type;
89
90using ErrorHandler = std::function<void(int, const QString &)>;
91using Condition = std::function<bool()>;
92
93namespace Private
94{
95
96class ExecutorBase;
97typedef QSharedPointer<ExecutorBase> ExecutorBasePtr;
98
99struct Execution {
100 Execution(const ExecutorBasePtr &executor);
101 ~Execution();
102 void setFinished();
103
104 template<typename T>
105 Async::Future<T>* result() const
106 {
107 return static_cast<Async::Future<T>*>(resultBase);
108 }
109
110 void releaseFuture();
111 bool errorWasHandled() const;
112
113 ExecutorBasePtr executor;
114 FutureBase *resultBase;
115 bool isRunning;
116 bool isFinished;
117
118 ExecutionPtr prevExecution;
119
120#ifndef QT_NO_DEBUG
121 Tracer *tracer;
122#endif
123};
124
125
126typedef QSharedPointer<Execution> ExecutionPtr;
127
128class ExecutorBase
129{
130 template<typename PrevOut, typename Out, typename ... In>
131 friend class Executor;
132
133 template<typename Out, typename ... In>
134 friend class Async::Job;
135
136 friend class Execution;
137 friend class Async::Tracer;
138
139public:
140 virtual ~ExecutorBase();
141 virtual ExecutionPtr exec(const ExecutorBasePtr &self) = 0;
142
143protected:
144 ExecutorBase(const ExecutorBasePtr &parent);
145
146 template<typename T>
147 Async::Future<T>* createFuture(const ExecutionPtr &execution) const;
148
149 virtual bool hasErrorFunc() const = 0;
150 virtual bool handleError(const ExecutionPtr &execution) = 0;
151
152 ExecutorBasePtr mPrev;
153
154#ifndef QT_NO_DEBUG
155 QString mExecutorName;
156#endif
157};
158
159template<typename PrevOut, typename Out, typename ... In>
160class Executor : public ExecutorBase
161{
162protected:
163 Executor(ErrorHandler errorFunc, const Private::ExecutorBasePtr &parent)
164 : ExecutorBase(parent)
165 , mErrorFunc(errorFunc)
166 {}
167
168 virtual ~Executor() {}
169 virtual void run(const ExecutionPtr &execution) = 0;
170
171 ExecutionPtr exec(const ExecutorBasePtr &self);
172 bool hasErrorFunc() const { return (bool) mErrorFunc; }
173 bool handleError(const ExecutionPtr &execution);
174
175 std::function<void(int, const QString &)> mErrorFunc;
176};
177
178template<typename Out, typename ... In>
179class ThenExecutor: public Executor<typename detail::prevOut<In ...>::type, Out, In ...>
180{
181public:
182 ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
183 void run(const ExecutionPtr &execution);
184private:
185 ThenTask<Out, In ...> mFunc;
186};
187
188template<typename PrevOut, typename Out, typename In>
189class EachExecutor : public Executor<PrevOut, Out, In>
190{
191public:
192 EachExecutor(EachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
193 void run(const ExecutionPtr &execution);
194private:
195 EachTask<Out, In> mFunc;
196 QVector<Async::FutureWatcher<Out>*> mFutureWatchers;
197};
198
199template<typename Out, typename In>
200class ReduceExecutor : public ThenExecutor<Out, In>
201{
202public:
203 ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
204private:
205 ReduceTask<Out, In> mFunc;
206};
207
208template<typename Out, typename ... In>
209class SyncThenExecutor : public Executor<typename detail::prevOut<In ...>::type, Out, In ...>
210{
211public:
212 SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
213 void run(const ExecutionPtr &execution);
214
215private:
216 void run(const ExecutionPtr &execution, std::false_type); // !std::is_void<Out>
217 void run(const ExecutionPtr &execution, std::true_type); // std::is_void<Out>
218 SyncThenTask<Out, In ...> mFunc;
219};
220
221template<typename Out, typename In>
222class SyncReduceExecutor : public SyncThenExecutor<Out, In>
223{
224public:
225 SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
226private:
227 SyncReduceTask<Out, In> mFunc;
228};
229
230template<typename PrevOut, typename Out, typename In>
231class SyncEachExecutor : public Executor<PrevOut, Out, In>
232{
233public:
234 SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
235 void run(const ExecutionPtr &execution);
236private:
237 void run(Async::Future<Out> *future, const typename PrevOut::value_type &arg, std::false_type); // !std::is_void<Out>
238 void run(Async::Future<Out> *future, const typename PrevOut::value_type &arg, std::true_type); // std::is_void<Out>
239 SyncEachTask<Out, In> mFunc;
240};
241
242} // namespace Private
243
244/**
245 * Start an asynchronous job sequence.
246 *
247 * Async::start() is your starting point to build a chain of jobs to be executed
248 * asynchronously.
249 *
250 * @param func An asynchronous function to be executed. The function must have
251 * void return type, and accept exactly one argument of type @p Async::Future<In>,
252 * where @p In is type of the result.
253 */
254template<typename Out, typename ... In>
255Job<Out, In ...> start(ThenTask<Out, In ...> func, ErrorHandler errorFunc = ErrorHandler());
256
257template<typename Out, typename ... In>
258Job<Out, In ...> start(SyncThenTask<Out, In ...> func, ErrorHandler errorFunc = ErrorHandler());
259
260#ifdef WITH_KJOB
261template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
262Job<ReturnType, Args ...> start();
263#endif
264
265/**
266 * Async while loop.
267 *
268 * The loop continues while @param condition returns true.
269 */
270Job<void> dowhile(Condition condition, ThenTask<void> func);
271
272/**
273 * Async while loop.
274 *
275 * Loop continues while body returns true.
276 */
277Job<void> dowhile(ThenTask<bool> body);
278
279/**
280 * Iterate over a container.
281 *
282 * Use in conjunction with .each
283 */
284template<typename Out>
285Job<Out> iterate(const Out &container);
286
287/**
288 * Async delay.
289 */
290Job<void> wait(int delay);
291
292/**
293 * A null job.
294 *
295 * An async noop.
296 *
297 */
298template<typename Out>
299Job<Out> null();
300
301/**
302 * An error job.
303 *
304 * An async error.
305 *
306 */
307template<typename Out>
308Job<Out> error(int errorCode = 1, const QString &errorMessage = QString());
309
310class JobBase
311{
312 template<typename Out, typename ... In>
313 friend class Job;
314
315public:
316 JobBase(const Private::ExecutorBasePtr &executor);
317 ~JobBase();
318
319protected:
320 Private::ExecutorBasePtr mExecutor;
321};
322
323/**
324 * An Asynchronous job
325 *
326 * A single instance of Job represents a single method that will be executed
327 * asynchrously. The Job is started by @p Job::exec(), which returns @p Async::Future
328 * immediatelly. The Future will be set to finished state once the asynchronous
329 * task has finished. You can use @p Async::Future::waitForFinished() to wait for
330 * for the Future in blocking manner.
331 *
332 * It is possible to chain multiple Jobs one after another in different fashion
333 * (sequential, parallel, etc.). Calling Job::exec() will then return a pending
334 * @p Async::Future, and will execute the entire chain of jobs.
335 *
336 * @code
337 * auto job = Job::start<QList<int>>(
338 * [](Async::Future<QList<int>> &future) {
339 * MyREST::PendingUsers *pu = MyREST::requestListOfUsers();
340 * QObject::connect(pu, &PendingOperation::finished,
341 * [&](PendingOperation *pu) {
342 * future->setValue(dynamic_cast<MyREST::PendingUsers*>(pu)->userIds());
343 * future->setFinished();
344 * });
345 * })
346 * .each<QList<MyREST::User>, int>(
347 * [](const int &userId, Async::Future<QList<MyREST::User>> &future) {
348 * MyREST::PendingUser *pu = MyREST::requestUserDetails(userId);
349 * QObject::connect(pu, &PendingOperation::finished,
350 * [&](PendingOperation *pu) {
351 * future->setValue(Qlist<MyREST::User>() << dynamic_cast<MyREST::PendingUser*>(pu)->user());
352 * future->setFinished();
353 * });
354 * });
355 *
356 * Async::Future<QList<MyREST::User>> usersFuture = job.exec();
357 * usersFuture.waitForFinished();
358 * QList<MyRest::User> users = usersFuture.value();
359 * @endcode
360 *
361 * In the example above, calling @p job.exec() will first invoke the first job,
362 * which will retrieve a list of IDs, and then will invoke the second function
363 * for each single entry in the list returned by the first function.
364 */
365template<typename Out, typename ... In>
366class Job : public JobBase
367{
368 template<typename OutOther, typename ... InOther>
369 friend class Job;
370
371 template<typename OutOther, typename ... InOther>
372 friend Job<OutOther, InOther ...> start(Async::ThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc);
373
374 template<typename OutOther, typename ... InOther>
375 friend Job<OutOther, InOther ...> start(Async::SyncThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc);
376
377#ifdef WITH_KJOB
378 template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
379 friend Job<ReturnType, Args ...> start();
380#endif
381
382public:
383 template<typename OutOther, typename ... InOther>
384 Job<OutOther, InOther ...> then(ThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc = ErrorHandler())
385 {
386 return Job<OutOther, InOther ...>(Private::ExecutorBasePtr(
387 new Private::ThenExecutor<OutOther, InOther ...>(func, errorFunc, mExecutor)));
388 }
389
390 template<typename OutOther, typename ... InOther>
391 Job<OutOther, InOther ...> then(SyncThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc = ErrorHandler())
392 {
393 return Job<OutOther, InOther ...>(Private::ExecutorBasePtr(
394 new Private::SyncThenExecutor<OutOther, InOther ...>(func, errorFunc, mExecutor)));
395 }
396
397 template<typename OutOther, typename ... InOther>
398 Job<OutOther, InOther ...> then(Job<OutOther, InOther ...> otherJob, ErrorHandler errorFunc = ErrorHandler())
399 {
400 return then<OutOther, InOther ...>(nestedJobWrapper<OutOther, InOther ...>(otherJob), errorFunc);
401 }
402
403#ifdef WITH_KJOB
404 template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
405 Job<ReturnType, Args ...> then()
406 {
407 return start<ReturnType, KJobType, KJobResultMethod, Args ...>();
408 }
409#endif
410
411 template<typename OutOther, typename InOther>
412 Job<OutOther, InOther> each(EachTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
413 {
414 eachInvariants<OutOther>();
415 return Job<OutOther, InOther>(Private::ExecutorBasePtr(
416 new Private::EachExecutor<Out, OutOther, InOther>(func, errorFunc, mExecutor)));
417 }
418
419 template<typename OutOther, typename InOther>
420 Job<OutOther, InOther> each(SyncEachTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
421 {
422 eachInvariants<OutOther>();
423 return Job<OutOther, InOther>(Private::ExecutorBasePtr(
424 new Private::SyncEachExecutor<Out, OutOther, InOther>(func, errorFunc, mExecutor)));
425 }
426
427 template<typename OutOther, typename InOther>
428 Job<OutOther, InOther> each(Job<OutOther, InOther> otherJob, ErrorHandler errorFunc = ErrorHandler())
429 {
430 eachInvariants<OutOther>();
431 return each<OutOther, InOther>(nestedJobWrapper<OutOther, InOther>(otherJob), errorFunc);
432 }
433
434 template<typename OutOther, typename InOther>
435 Job<OutOther, InOther> reduce(ReduceTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
436 {
437 reduceInvariants<InOther>();
438 return Job<OutOther, InOther>(Private::ExecutorBasePtr(
439 new Private::ReduceExecutor<OutOther, InOther>(func, errorFunc, mExecutor)));
440 }
441
442 template<typename OutOther, typename InOther>
443 Job<OutOther, InOther> reduce(SyncReduceTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
444 {
445 reduceInvariants<InOther>();
446 return Job<OutOther, InOther>(Private::ExecutorBasePtr(
447 new Private::SyncReduceExecutor<OutOther, InOther>(func, errorFunc, mExecutor)));
448 }
449
450 template<typename OutOther, typename InOther>
451 Job<OutOther, InOther> reduce(Job<OutOther, InOther> otherJob, ErrorHandler errorFunc = ErrorHandler())
452 {
453 return reduce<OutOther, InOther>(nestedJobWrapper<OutOther, InOther>(otherJob), errorFunc);
454 }
455
456 template<typename FirstIn>
457 Async::Future<Out> exec(FirstIn in)
458 {
459 // Inject a fake sync executor that will return the initial value
460 Private::ExecutorBasePtr first = mExecutor;
461 while (first->mPrev) {
462 first = first->mPrev;
463 }
464 auto init = new Private::SyncThenExecutor<FirstIn>(
465 [in]() -> FirstIn {
466 return in;
467 },
468 ErrorHandler(), Private::ExecutorBasePtr());
469 first->mPrev = Private::ExecutorBasePtr(init);
470
471 auto result = exec();
472 // Remove the injected executor
473 first->mPrev.reset();
474 return result;
475 }
476
477 Async::Future<Out> exec()
478 {
479 Private::ExecutionPtr execution = mExecutor->exec(mExecutor);
480 Async::Future<Out> result = *execution->result<Out>();
481
482 return result;
483 }
484
485private:
486 Job(Private::ExecutorBasePtr executor)
487 : JobBase(executor)
488 {}
489
490 template<typename OutOther>
491 void eachInvariants()
492 {
493 static_assert(detail::isIterable<Out>::value,
494 "The 'Each' task can only be connected to a job that returns a list or an array.");
495 static_assert(std::is_void<OutOther>::value || detail::isIterable<OutOther>::value,
496 "The result type of 'Each' task must be void, a list or an array.");
497 }
498
499 template<typename InOther>
500 void reduceInvariants()
501 {
502 static_assert(Async::detail::isIterable<Out>::value,
503 "The 'Result' task can only be connected to a job that returns a list or an array");
504 static_assert(std::is_same<typename Out::value_type, typename InOther::value_type>::value,
505 "The return type of previous task must be compatible with input type of this task");
506 }
507
508 template<typename OutOther, typename ... InOther>
509 inline std::function<void(InOther ..., Async::Future<OutOther>&)> nestedJobWrapper(Job<OutOther, InOther ...> otherJob) {
510 return [otherJob](InOther ... in, Async::Future<OutOther> &future) {
511 // copy by value is const
512 auto job = otherJob;
513 FutureWatcher<OutOther> *watcher = new FutureWatcher<OutOther>();
514 QObject::connect(watcher, &FutureWatcherBase::futureReady,
515 [watcher, future]() {
516 // FIXME: We pass future by value, because using reference causes the
517 // future to get deleted before this lambda is invoked, leading to crash
518 // in copyFutureValue()
519 // copy by value is const
520 auto outFuture = future;
521 Async::detail::copyFutureValue(watcher->future(), outFuture);
522 if (watcher->future().errorCode()) {
523 outFuture.setError(watcher->future().errorCode(), watcher->future().errorMessage());
524 } else {
525 outFuture.setFinished();
526 }
527 delete watcher;
528 });
529 watcher->setFuture(job.exec(in ...));
530 };
531 }
532};
533
534} // namespace Async
535
536
537// ********** Out of line definitions ****************
538
539namespace Async {
540
541template<typename Out, typename ... In>
542Job<Out, In ...> start(ThenTask<Out, In ...> func, ErrorHandler error)
543{
544 return Job<Out, In...>(Private::ExecutorBasePtr(
545 new Private::ThenExecutor<Out, In ...>(func, error, Private::ExecutorBasePtr())));
546}
547
548template<typename Out, typename ... In>
549Job<Out, In ...> start(SyncThenTask<Out, In ...> func, ErrorHandler error)
550{
551 return Job<Out, In...>(Private::ExecutorBasePtr(
552 new Private::SyncThenExecutor<Out, In ...>(func, error, Private::ExecutorBasePtr())));
553}
554
555#ifdef WITH_KJOB
556template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
557Job<ReturnType, Args ...> start()
558{
559 return Job<ReturnType, Args ...>(Private::ExecutorBasePtr(
560 new Private::ThenExecutor<ReturnType, Args ...>([](const Args & ... args, Async::Future<ReturnType> &future)
561 {
562 KJobType *job = new KJobType(args ...);
563 job->connect(job, &KJob::finished,
564 [&future](KJob *job) {
565 if (job->error()) {
566 future.setError(job->error(), job->errorString());
567 } else {
568 future.setValue((static_cast<KJobType*>(job)->*KJobResultMethod)());
569 future.setFinished();
570 }
571 });
572 job->start();
573 }, ErrorHandler(), Private::ExecutorBasePtr())));
574}
575#endif
576
577
578template<typename Out>
579Job<Out> null()
580{
581 return Async::start<Out>(
582 [](Async::Future<Out> &future) {
583 future.setFinished();
584 });
585}
586
587template<typename Out>
588Job<Out> error(int errorCode, const QString &errorMessage)
589{
590 return Async::start<Out>(
591 [errorCode, errorMessage](Async::Future<Out> &future) {
592 future.setError(errorCode, errorMessage);
593 });
594}
595
596template<typename Out>
597Job<Out> iterate(const Out &container)
598{
599 return Async::start<Out>(
600 [container]() {
601 return container;
602 });
603}
604
605
606namespace Private {
607
608template<typename T>
609Async::Future<T>* ExecutorBase::createFuture(const ExecutionPtr &execution) const
610{
611 return new Async::Future<T>(execution);
612}
613
614template<typename PrevOut, typename Out, typename ... In>
615ExecutionPtr Executor<PrevOut, Out, In ...>::exec(const ExecutorBasePtr &self)
616{
617 // Passing 'self' to execution ensures that the Executor chain remains
618 // valid until the entire execution is finished
619 ExecutionPtr execution = ExecutionPtr::create(self);
620#ifndef QT_NO_DEBUG
621 execution->tracer = new Tracer(execution.data()); // owned by execution
622#endif
623
624 // chainup
625 execution->prevExecution = mPrev ? mPrev->exec(mPrev) : ExecutionPtr();
626
627 execution->resultBase = ExecutorBase::createFuture<Out>(execution);
628 auto fw = new Async::FutureWatcher<Out>();
629 QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
630 [fw, execution, this]() {
631 handleError(execution);
632 execution->setFinished();
633 delete fw;
634 });
635 fw->setFuture(*execution->result<Out>());
636
637 Async::Future<PrevOut> *prevFuture = execution->prevExecution ? execution->prevExecution->result<PrevOut>() : nullptr;
638 if (!prevFuture || prevFuture->isFinished()) {
639 if (prevFuture) { // prevFuture implies execution->prevExecution
640 if (prevFuture->errorCode()) {
641 // Propagate the errorCode and message to the outer Future
642 execution->resultBase->setError(prevFuture->errorCode(), prevFuture->errorMessage());
643 if (!execution->errorWasHandled()) {
644 if (handleError(execution)) {
645 return execution;
646 }
647 } else {
648 return execution;
649 }
650 } else {
651 // Propagate error (if any)
652 }
653 }
654
655 execution->isRunning = true;
656 run(execution);
657 } else {
658 auto prevFutureWatcher = new Async::FutureWatcher<PrevOut>();
659 QObject::connect(prevFutureWatcher, &Async::FutureWatcher<PrevOut>::futureReady,
660 [prevFutureWatcher, execution, this]() {
661 auto prevFuture = prevFutureWatcher->future();
662 assert(prevFuture.isFinished());
663 delete prevFutureWatcher;
664 auto prevExecutor = execution->executor->mPrev;
665 if (prevFuture.errorCode()) {
666 execution->resultBase->setError(prevFuture.errorCode(), prevFuture.errorMessage());
667 if (!execution->errorWasHandled()) {
668 if (handleError(execution)) {
669 return;
670 }
671 } else {
672 return;
673 }
674 }
675
676
677 // propagate error (if any)
678 execution->isRunning = true;
679 run(execution);
680 });
681
682 prevFutureWatcher->setFuture(*static_cast<Async::Future<PrevOut>*>(prevFuture));
683 }
684
685 return execution;
686}
687
688template<typename PrevOut, typename Out, typename ... In>
689bool Executor<PrevOut, Out, In ...>::handleError(const ExecutionPtr &execution)
690{
691 assert(execution->resultBase->isFinished());
692 if (execution->resultBase->errorCode()) {
693 if (mErrorFunc) {
694 mErrorFunc(execution->resultBase->errorCode(),
695 execution->resultBase->errorMessage());
696 return true;
697 }
698 }
699
700 return false;
701}
702
703
704template<typename Out, typename ... In>
705ThenExecutor<Out, In ...>::ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler error, const ExecutorBasePtr &parent)
706 : Executor<typename detail::prevOut<In ...>::type, Out, In ...>(error, parent)
707 , mFunc(then)
708{
709 STORE_EXECUTOR_NAME("ThenExecutor", Out, In ...);
710}
711
712template<typename Out, typename ... In>
713void ThenExecutor<Out, In ...>::run(const ExecutionPtr &execution)
714{
715 Async::Future<typename detail::prevOut<In ...>::type> *prevFuture = nullptr;
716 if (execution->prevExecution) {
717 prevFuture = execution->prevExecution->result<typename detail::prevOut<In ...>::type>();
718 assert(prevFuture->isFinished());
719 }
720
721 ThenExecutor<Out, In ...>::mFunc(prevFuture ? prevFuture->value() : In() ..., *execution->result<Out>());
722}
723
724template<typename PrevOut, typename Out, typename In>
725EachExecutor<PrevOut, Out, In>::EachExecutor(EachTask<Out, In> each, ErrorHandler error, const ExecutorBasePtr &parent)
726 : Executor<PrevOut, Out, In>(error, parent)
727 , mFunc(each)
728{
729 STORE_EXECUTOR_NAME("EachExecutor", PrevOut, Out, In);
730}
731
732template<typename PrevOut, typename Out, typename In>
733void EachExecutor<PrevOut, Out, In>::run(const ExecutionPtr &execution)
734{
735 assert(execution->prevExecution);
736 auto prevFuture = execution->prevExecution->result<PrevOut>();
737 assert(prevFuture->isFinished());
738
739 auto out = execution->result<Out>();
740 if (prevFuture->value().isEmpty()) {
741 out->setFinished();
742 return;
743 }
744
745 for (auto arg : prevFuture->value()) {
746 //We have to manually manage the lifetime of these temporary futures
747 Async::Future<Out> *future = new Async::Future<Out>();
748 EachExecutor<PrevOut, Out, In>::mFunc(arg, *future);
749 auto fw = new Async::FutureWatcher<Out>();
750 mFutureWatchers.append(fw);
751 QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
752 [out, fw, this, future]() {
753 assert(fw->future().isFinished());
754 const int index = mFutureWatchers.indexOf(fw);
755 assert(index > -1);
756 mFutureWatchers.removeAt(index);
757 Async::detail::aggregateFutureValue<Out>(fw->future(), *out);
758 if (mFutureWatchers.isEmpty()) {
759 out->setFinished();
760 }
761 delete fw;
762 delete future;
763 });
764 fw->setFuture(*future);
765 }
766}
767
768template<typename Out, typename In>
769ReduceExecutor<Out, In>::ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
770 : ThenExecutor<Out, In>(reduce, errorFunc, parent)
771{
772 STORE_EXECUTOR_NAME("ReduceExecutor", Out, In);
773}
774
775template<typename Out, typename ... In>
776SyncThenExecutor<Out, In ...>::SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
777 : Executor<typename detail::prevOut<In ...>::type, Out, In ...>(errorFunc, parent)
778 , mFunc(then)
779{
780 STORE_EXECUTOR_NAME("SyncThenExecutor", Out, In ...);
781}
782
783template<typename Out, typename ... In>
784void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution)
785{
786 if (execution->prevExecution) {
787 assert(execution->prevExecution->resultBase->isFinished());
788 }
789
790 run(execution, std::is_void<Out>());
791 execution->resultBase->setFinished();
792}
793
794template<typename Out, typename ... In>
795void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution, std::false_type)
796{
797 Async::Future<typename detail::prevOut<In ...>::type> *prevFuture =
798 execution->prevExecution
799 ? execution->prevExecution->result<typename detail::prevOut<In ...>::type>()
800 : nullptr;
801 (void) prevFuture; // silence 'set but not used' warning
802 Async::Future<Out> *future = execution->result<Out>();
803 future->setValue(SyncThenExecutor<Out, In...>::mFunc(prevFuture ? prevFuture->value() : In() ...));
804}
805
806template<typename Out, typename ... In>
807void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution, std::true_type)
808{
809 Async::Future<typename detail::prevOut<In ...>::type> *prevFuture =
810 execution->prevExecution
811 ? execution->prevExecution->result<typename detail::prevOut<In ...>::type>()
812 : nullptr;
813 (void) prevFuture; // silence 'set but not used' warning
814 SyncThenExecutor<Out, In ...>::mFunc(prevFuture ? prevFuture->value() : In() ...);
815}
816
817template<typename PrevOut, typename Out, typename In>
818SyncEachExecutor<PrevOut, Out, In>::SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
819 : Executor<PrevOut, Out, In>(errorFunc, parent)
820 , mFunc(each)
821{
822 STORE_EXECUTOR_NAME("SyncEachExecutor", PrevOut, Out, In);
823}
824
825template<typename PrevOut, typename Out, typename In>
826void SyncEachExecutor<PrevOut, Out, In>::run(const ExecutionPtr &execution)
827{
828 assert(execution->prevExecution);
829 auto *prevFuture = execution->prevExecution->result<PrevOut>();
830 assert(prevFuture->isFinished());
831
832 auto out = execution->result<Out>();
833 if (prevFuture->value().isEmpty()) {
834 out->setFinished();
835 return;
836 }
837
838 for (auto arg : prevFuture->value()) {
839 run(out, arg, std::is_void<Out>());
840 }
841 out->setFinished();
842}
843
844template<typename PrevOut, typename Out, typename In>
845void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> *out, const typename PrevOut::value_type &arg, std::false_type)
846{
847 out->setValue(out->value() + SyncEachExecutor<PrevOut, Out, In>::mFunc(arg));
848}
849
850template<typename PrevOut, typename Out, typename In>
851void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> * /* unused */, const typename PrevOut::value_type &arg, std::true_type)
852{
853 SyncEachExecutor<PrevOut, Out, In>::mFunc(arg);
854}
855
856template<typename Out, typename In>
857SyncReduceExecutor<Out, In>::SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
858 : SyncThenExecutor<Out, In>(reduce, errorFunc, parent)
859{
860 STORE_EXECUTOR_NAME("SyncReduceExecutor", Out, In);
861}
862
863
864} // namespace Private
865
866} // namespace Async
867
868
869
870#endif // ASYNC_H
871
872
diff --git a/async/src/async_impl.h b/async/src/async_impl.h
deleted file mode 100644
index 8c74193..0000000
--- a/async/src/async_impl.h
+++ /dev/null
@@ -1,81 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#ifndef ASYNC_IMPL_H
19#define ASYNC_IMPL_H
20
21#include "async.h"
22#include <type_traits>
23
24namespace Async {
25
26namespace detail {
27
28template<typename T>
29struct identity
30{
31 typedef T type;
32};
33
34template<typename T, typename Enable = void>
35struct isIterable {
36 enum { value = 0 };
37};
38
39template<typename T>
40struct isIterable<T, typename std::conditional<false, typename T::iterator, void>::type> {
41 enum { value = 1 };
42};
43
44template<typename ... T>
45struct prevOut {
46 using type = typename std::tuple_element<0, std::tuple<T ..., void>>::type;
47};
48
49template<typename T>
50inline typename std::enable_if<!std::is_void<T>::value, void>::type
51copyFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
52{
53 out.setValue(in.value());
54}
55
56template<typename T>
57inline typename std::enable_if<std::is_void<T>::value, void>::type
58copyFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
59{
60 // noop
61}
62
63template<typename T>
64inline typename std::enable_if<!std::is_void<T>::value, void>::type
65aggregateFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
66{
67 out.setValue(out.value() + in.value());
68}
69
70template<typename T>
71inline typename std::enable_if<std::is_void<T>::value, void>::type
72aggregateFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
73{
74 // noop
75}
76
77} // namespace Detail
78
79} // namespace Async
80
81#endif // ASYNC_IMPL_H
diff --git a/async/src/debug.cpp b/async/src/debug.cpp
deleted file mode 100644
index fdf2fa1..0000000
--- a/async/src/debug.cpp
+++ /dev/null
@@ -1,75 +0,0 @@
1/*
2 * Copyright 2015 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#include "debug.h"
19#include "async.h"
20
21#include <QStringBuilder>
22
23#ifdef __GNUG__
24#include <cxxabi.h>
25#include <memory>
26#endif
27
28namespace Async
29{
30
31Q_LOGGING_CATEGORY(Debug, "org.kde.async", QtWarningMsg);
32Q_LOGGING_CATEGORY(Trace, "org.kde.async.trace", QtWarningMsg);
33
34QString demangleName(const char *name)
35{
36#ifdef __GNUG__
37 int status = 1; // uses -3 to 0 error codes
38 std::unique_ptr<char, void(*)(void*)> demangled(abi::__cxa_demangle(name, 0, 0, &status), std::free);
39 if (status == 0) {
40 return QString(demangled.get());
41 }
42#endif
43 return QString(name);
44}
45
46}
47
48using namespace Async;
49
50int Tracer::lastId = 0;
51
52Tracer::Tracer(Private::Execution *execution)
53 : mId(lastId++)
54 , mExecution(execution)
55{
56 msg(Async::Tracer::Start);
57}
58
59Tracer::~Tracer()
60{
61 msg(Async::Tracer::End);
62 // FIXME: Does this work on parallel executions?
63 --lastId;
64 --mId;
65}
66
67void Tracer::msg(Tracer::MsgType msgType)
68{
69#ifndef QT_NO_DEBUG
70 qCDebug(Trace).nospace() << (QString().fill(QLatin1Char(' '), mId * 2) %
71 (msgType == Async::Tracer::Start ? " START " : " END ") %
72 QString::number(mId) % " " %
73 mExecution->executor->mExecutorName);
74#endif
75}
diff --git a/async/src/debug.h b/async/src/debug.h
deleted file mode 100644
index c453eb3..0000000
--- a/async/src/debug.h
+++ /dev/null
@@ -1,80 +0,0 @@
1/*
2 * Copyright 2015 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#ifndef ASYNC_DEBUG_H
19#define ASYNC_DEBUG_H
20
21#include <QLoggingCategory>
22#include <QStringBuilder>
23
24#ifndef QT_NO_DEBUG
25#include <typeinfo>
26#endif
27
28namespace Async
29{
30
31Q_DECLARE_LOGGING_CATEGORY(Debug)
32Q_DECLARE_LOGGING_CATEGORY(Trace)
33
34QString demangleName(const char *name);
35
36namespace Private
37{
38class Execution;
39}
40
41class Tracer
42{
43public:
44 Tracer(Private::Execution *execution);
45 ~Tracer();
46
47private:
48 enum MsgType {
49 Start,
50 End
51 };
52 void msg(MsgType);
53
54 int mId;
55 Private::Execution *mExecution;
56
57 static int lastId;
58};
59
60}
61
62#ifndef QT_NO_DEBUG
63 template<typename T>
64 QString storeExecutorNameExpanded() {
65 return Async::demangleName(typeid(T).name());
66 }
67
68 template<typename T, typename ... Tail>
69 typename std::enable_if<sizeof ... (Tail) != 0, QString>::type
70 storeExecutorNameExpanded() {
71 return storeExecutorNameExpanded<T>() % QStringLiteral(", ") % storeExecutorNameExpanded<Tail ...>();
72 }
73
74 #define STORE_EXECUTOR_NAME(name, ...) \
75 ExecutorBase::mExecutorName = QStringLiteral(name) % QStringLiteral("<") % storeExecutorNameExpanded<__VA_ARGS__>() % QStringLiteral(">")
76#else
77 #define STORE_EXECUTOR_NAME(...)
78#endif
79
80#endif // ASYNC_DEBUG_H \ No newline at end of file
diff --git a/async/src/future.cpp b/async/src/future.cpp
deleted file mode 100644
index 4f3cd80..0000000
--- a/async/src/future.cpp
+++ /dev/null
@@ -1,158 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#include "future.h"
19#include "async.h"
20
21using namespace Async;
22
23FutureBase::PrivateBase::PrivateBase(const Private::ExecutionPtr &execution)
24 : finished(false)
25 , errorCode(0)
26 , mExecution(execution)
27{
28}
29
30FutureBase::PrivateBase::~PrivateBase()
31{
32 Private::ExecutionPtr executionPtr = mExecution.toStrongRef();
33 if (executionPtr) {
34 executionPtr->releaseFuture();
35 releaseExecution();
36 }
37}
38
39void FutureBase::PrivateBase::releaseExecution()
40{
41 mExecution.clear();
42}
43
44
45
46FutureBase::FutureBase()
47 : d(nullptr)
48{
49}
50
51FutureBase::FutureBase(FutureBase::PrivateBase *dd)
52 : d(dd)
53{
54}
55
56FutureBase::FutureBase(const Async::FutureBase &other)
57 : d(other.d)
58{
59}
60
61FutureBase::~FutureBase()
62{
63}
64
65void FutureBase::releaseExecution()
66{
67 d->releaseExecution();
68}
69
70void FutureBase::setFinished()
71{
72 if (isFinished()) {
73 return;
74 }
75 d->finished = true;
76 for (auto watcher : d->watchers) {
77 if (watcher) {
78 watcher->futureReadyCallback();
79 }
80 }
81}
82
83bool FutureBase::isFinished() const
84{
85 return d->finished;
86}
87
88void FutureBase::setError(int code, const QString &message)
89{
90 d->errorCode = code;
91 d->errorMessage = message;
92 setFinished();
93}
94
95int FutureBase::errorCode() const
96{
97 return d->errorCode;
98}
99
100QString FutureBase::errorMessage() const
101{
102 return d->errorMessage;
103}
104
105void FutureBase::setProgress(int processed, int total)
106{
107 setProgress((qreal) processed / total);
108}
109
110void FutureBase::setProgress(qreal progress)
111{
112 for (auto watcher : d->watchers) {
113 if (watcher) {
114 watcher->futureProgressCallback(progress);
115 }
116 }
117}
118
119
120
121void FutureBase::addWatcher(FutureWatcherBase* watcher)
122{
123 d->watchers.append(QPointer<FutureWatcherBase>(watcher));
124}
125
126
127
128
129
130FutureWatcherBase::FutureWatcherBase(QObject *parent)
131 : QObject(parent)
132 , d(new FutureWatcherBase::Private)
133{
134}
135
136FutureWatcherBase::~FutureWatcherBase()
137{
138 delete d;
139}
140
141void FutureWatcherBase::futureReadyCallback()
142{
143 Q_EMIT futureReady();
144}
145
146void FutureWatcherBase::futureProgressCallback(qreal progress)
147{
148 Q_EMIT futureProgress(progress);
149}
150
151void FutureWatcherBase::setFutureImpl(const FutureBase &future)
152{
153 d->future = future;
154 d->future.addWatcher(this);
155 if (future.isFinished()) {
156 futureReadyCallback();
157 }
158}
diff --git a/async/src/future.h b/async/src/future.h
deleted file mode 100644
index ff199ef..0000000
--- a/async/src/future.h
+++ /dev/null
@@ -1,255 +0,0 @@
1/*
2 * Copyright 2014 Daniel Vrátil <dvratil@redhat.com>
3 *
4 * This library is free software; you can redistribute it and/or
5 * modify it under the terms of the GNU Library General Public License as
6 * published by the Free Software Foundation; either version 2 of
7 * the License, or (at your option) any later version.
8 *
9 * This library is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU Library General Public License for more details.
13 *
14 * You should have received a copy of the GNU Library General Public License
15 * along with this library. If not, see <http://www.gnu.org/licenses/>.
16 */
17
18#ifndef FUTURE_H
19#define FUTURE_H
20
21class QEventLoop;
22
23#include <type_traits>
24
25#include <QSharedDataPointer>
26#include <QPointer>
27#include <QVector>
28#include <QEventLoop>
29
30namespace Async {
31
32class FutureWatcherBase;
33template<typename T>
34class FutureWatcher;
35
36namespace Private {
37class Execution;
38class ExecutorBase;
39
40typedef QSharedPointer<Execution> ExecutionPtr;
41} // namespace Private
42
43class FutureBase
44{
45 friend class Async::Private::Execution;
46 friend class FutureWatcherBase;
47
48public:
49 virtual ~FutureBase();
50
51 void setFinished();
52 bool isFinished() const;
53 void setError(int code = 1, const QString &message = QString());
54 int errorCode() const;
55 QString errorMessage() const;
56
57 void setProgress(qreal progress);
58 void setProgress(int processed, int total);
59
60protected:
61 class PrivateBase : public QSharedData
62 {
63 public:
64 PrivateBase(const Async::Private::ExecutionPtr &execution);
65 virtual ~PrivateBase();
66
67 void releaseExecution();
68
69 bool finished;
70 int errorCode;
71 QString errorMessage;
72
73 QVector<QPointer<FutureWatcherBase>> watchers;
74 private:
75 QWeakPointer<Async::Private::Execution> mExecution;
76 };
77
78 FutureBase();
79 FutureBase(FutureBase::PrivateBase *dd);
80 FutureBase(const FutureBase &other);
81
82 void addWatcher(Async::FutureWatcherBase *watcher);
83 void releaseExecution();
84
85protected:
86 QExplicitlySharedDataPointer<PrivateBase> d;
87};
88
89template<typename T>
90class FutureWatcher;
91
92template<typename T>
93class Future;
94
95template<typename T>
96class FutureGeneric : public FutureBase
97{
98 friend class FutureWatcher<T>;
99
100public:
101 void waitForFinished() const
102 {
103 if (isFinished()) {
104 return;
105 }
106 FutureWatcher<T> watcher;
107 QEventLoop eventLoop;
108 QObject::connect(&watcher, &Async::FutureWatcher<T>::futureReady,
109 &eventLoop, &QEventLoop::quit);
110 watcher.setFuture(*static_cast<const Async::Future<T>*>(this));
111 eventLoop.exec();
112 }
113
114protected:
115 FutureGeneric(const Async::Private::ExecutionPtr &execution)
116 : FutureBase(new Private(execution))
117 {}
118
119 FutureGeneric(const FutureGeneric<T> &other)
120 : FutureBase(other)
121 {}
122
123protected:
124 class Private : public FutureBase::PrivateBase
125 {
126 public:
127 Private(const Async::Private::ExecutionPtr &execution)
128 : FutureBase::PrivateBase(execution)
129 {}
130
131 typename std::conditional<std::is_void<T>::value, int /* dummy */, T>::type
132 value;
133 };
134};
135
136
137template<typename T>
138class Future : public FutureGeneric<T>
139{
140 friend class Async::Private::ExecutorBase;
141
142 template<typename T_>
143 friend class Async::FutureWatcher;
144
145public:
146 Future()
147 : FutureGeneric<T>(Async::Private::ExecutionPtr())
148 {}
149
150 Future(const Future<T> &other)
151 : FutureGeneric<T>(other)
152 {}
153
154 void setValue(const T &value)
155 {
156 static_cast<typename FutureGeneric<T>::Private*>(this->d.data())->value = value;
157 }
158
159 T value() const
160 {
161 return static_cast<typename FutureGeneric<T>::Private*>(this->d.data())->value;
162 }
163
164protected:
165 Future(const Async::Private::ExecutionPtr &execution)
166 : FutureGeneric<T>(execution)
167 {}
168
169};
170
171template<>
172class Future<void> : public FutureGeneric<void>
173{
174 friend class Async::Private::ExecutorBase;
175
176public:
177 Future()
178 : FutureGeneric<void>(Async::Private::ExecutionPtr())
179 {}
180
181 Future(const Future<void> &other)
182 : FutureGeneric<void>(other)
183 {}
184
185protected:
186 Future(const Async::Private::ExecutionPtr &execution)
187 : FutureGeneric<void>(execution)
188 {}
189};
190
191
192
193
194
195class FutureWatcherBase : public QObject
196{
197 Q_OBJECT
198
199 friend class FutureBase;
200
201Q_SIGNALS:
202 void futureReady();
203 void futureProgress(qreal progress);
204
205protected:
206 FutureWatcherBase(QObject *parent = nullptr);
207 virtual ~FutureWatcherBase();
208
209 void futureReadyCallback();
210 void futureProgressCallback(qreal progress);
211
212 void setFutureImpl(const Async::FutureBase &future);
213
214protected:
215 class Private {
216 public:
217 Async::FutureBase future;
218 };
219
220 Private * const d;
221
222private:
223 Q_DISABLE_COPY(FutureWatcherBase);
224};
225
226template<typename T>
227class FutureWatcher : public FutureWatcherBase
228{
229 friend class Async::FutureGeneric<T>;
230
231public:
232 FutureWatcher(QObject *parent = nullptr)
233 : FutureWatcherBase(parent)
234 {}
235
236 ~FutureWatcher()
237 {}
238
239 void setFuture(const Async::Future<T> &future)
240 {
241 setFutureImpl(*static_cast<const Async::FutureBase*>(&future));
242 }
243
244 Async::Future<T> future() const
245 {
246 return *static_cast<Async::Future<T>*>(&d->future);
247 }
248
249private:
250 Q_DISABLE_COPY(FutureWatcher<T>);
251};
252
253} // namespace Async
254
255#endif // FUTURE_H
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index f6847a7..2ece210 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -1,19 +1,6 @@
1include_directories(${CMAKE_CURRENT_BINARY_DIR})
2
1project(akonadi2common) 3project(akonadi2common)
2generate_flatbuffers(
3 commands/commandcompletion
4 commands/createentity
5 commands/deleteentity
6 commands/fetchentity
7 commands/handshake
8 commands/modifyentity
9 commands/revisionupdate
10 commands/synchronize
11 commands/notification
12 domain/event
13 entity
14 metadata
15 queuedcommand
16)
17 4
18if (STORAGE_unqlite) 5if (STORAGE_unqlite)
19 add_definitions(-DUNQLITE_ENABLE_THREADS -fpermissive) 6 add_definitions(-DUNQLITE_ENABLE_THREADS -fpermissive)
@@ -41,8 +28,27 @@ set(command_SRCS
41 ${storage_SRCS}) 28 ${storage_SRCS})
42 29
43add_library(${PROJECT_NAME} SHARED ${command_SRCS}) 30add_library(${PROJECT_NAME} SHARED ${command_SRCS})
31
32generate_flatbuffers(
33 ${PROJECT_NAME}
34
35 commands/commandcompletion
36 commands/createentity
37 commands/deleteentity
38 commands/fetchentity
39 commands/handshake
40 commands/modifyentity
41 commands/revisionupdate
42 commands/synchronize
43 commands/notification
44 domain/event
45 entity
46 metadata
47 queuedcommand
48)
49
44generate_export_header(${PROJECT_NAME} BASE_NAME Akonadi2Common EXPORT_FILE_NAME akonadi2common_export.h) 50generate_export_header(${PROJECT_NAME} BASE_NAME Akonadi2Common EXPORT_FILE_NAME akonadi2common_export.h)
45SET_TARGET_PROPERTIES(${PROJECT_NAME} PROPERTIES LINKER_LANGUAGE CXX) 51SET_TARGET_PROPERTIES(${PROJECT_NAME} PROPERTIES LINKER_LANGUAGE CXX)
46qt5_use_modules(${PROJECT_NAME} Network) 52qt5_use_modules(${PROJECT_NAME} Network)
47target_link_libraries(${PROJECT_NAME} ${storage_LIBS} akonadi2async) 53target_link_libraries(${PROJECT_NAME} ${storage_LIBS} KF5::Async)
48install(TARGETS ${PROJECT_NAME} ${KDE_INSTALL_TARGETS_DEFAULT_ARGS}) 54install(TARGETS ${PROJECT_NAME} ${KDE_INSTALL_TARGETS_DEFAULT_ARGS})
diff --git a/common/clientapi.cpp b/common/clientapi.cpp
index e4608c8..d287fcf 100644
--- a/common/clientapi.cpp
+++ b/common/clientapi.cpp
@@ -42,7 +42,7 @@ QByteArray getTypeName<AkonadiResource>()
42void Store::shutdown(const QByteArray &identifier) 42void Store::shutdown(const QByteArray &identifier)
43{ 43{
44 Trace() << "shutdown"; 44 Trace() << "shutdown";
45 ResourceAccess::connectToServer(identifier).then<void, QSharedPointer<QLocalSocket>>([identifier](const QSharedPointer<QLocalSocket> &socket, Async::Future<void> &future) { 45 ResourceAccess::connectToServer(identifier).then<void, QSharedPointer<QLocalSocket>>([identifier](const QSharedPointer<QLocalSocket> &socket, KAsync::Future<void> &future) {
46 //We can't currently reuse the socket 46 //We can't currently reuse the socket
47 socket->close(); 47 socket->close();
48 auto resourceAccess = QSharedPointer<Akonadi2::ResourceAccess>::create(identifier); 48 auto resourceAccess = QSharedPointer<Akonadi2::ResourceAccess>::create(identifier);
diff --git a/common/clientapi.h b/common/clientapi.h
index 1bd8bdc..c098bb5 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -28,8 +28,10 @@
28#include <QEventLoop> 28#include <QEventLoop>
29#include <functional> 29#include <functional>
30#include <memory> 30#include <memory>
31
32#include <Async/Async>
33
31#include "threadboundary.h" 34#include "threadboundary.h"
32#include "async/src/async.h"
33#include "resultprovider.h" 35#include "resultprovider.h"
34 36
35namespace async { 37namespace async {
@@ -228,10 +230,10 @@ class StoreFacade {
228public: 230public:
229 virtual ~StoreFacade(){}; 231 virtual ~StoreFacade(){};
230 QByteArray type() const { return ApplicationDomain::getTypeName<DomainType>(); } 232 QByteArray type() const { return ApplicationDomain::getTypeName<DomainType>(); }
231 virtual Async::Job<void> create(const DomainType &domainObject) = 0; 233 virtual KAsync::Job<void> create(const DomainType &domainObject) = 0;
232 virtual Async::Job<void> modify(const DomainType &domainObject) = 0; 234 virtual KAsync::Job<void> modify(const DomainType &domainObject) = 0;
233 virtual Async::Job<void> remove(const DomainType &domainObject) = 0; 235 virtual KAsync::Job<void> remove(const DomainType &domainObject) = 0;
234 virtual Async::Job<void> load(const Query &query, const QSharedPointer<ResultProvider<typename DomainType::Ptr> > &resultProvider) = 0; 236 virtual KAsync::Job<void> load(const Query &query, const QSharedPointer<ResultProvider<typename DomainType::Ptr> > &resultProvider) = 0;
235}; 237};
236 238
237 239
@@ -341,8 +343,8 @@ public:
341 //The result provider must be threadsafe. 343 //The result provider must be threadsafe.
342 async::run([query, resultSet](){ 344 async::run([query, resultSet](){
343 // Query all resources and aggregate results 345 // Query all resources and aggregate results
344 Async::iterate(query.resources) 346 KAsync::iterate(query.resources)
345 .template each<void, QByteArray>([query, resultSet](const QByteArray &resource, Async::Future<void> &future) { 347 .template each<void, QByteArray>([query, resultSet](const QByteArray &resource, KAsync::Future<void> &future) {
346 //TODO pass resource identifier to factory 348 //TODO pass resource identifier to factory
347 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource); 349 auto facade = FacadeFactory::instance().getFacade<DomainType>(resource);
348 if (facade) { 350 if (facade) {
diff --git a/common/facade.h b/common/facade.h
index 8c6578f..5743aa2 100644
--- a/common/facade.h
+++ b/common/facade.h
@@ -23,7 +23,8 @@
23 23
24#include <QByteArray> 24#include <QByteArray>
25 25
26#include "async/src/async.h" 26#include <Async/Async>
27
27#include "resourceaccess.h" 28#include "resourceaccess.h"
28#include "commands.h" 29#include "commands.h"
29#include "createentity_generated.h" 30#include "createentity_generated.h"
@@ -44,13 +45,13 @@ class QueryRunner : public QObject
44{ 45{
45 Q_OBJECT 46 Q_OBJECT
46public: 47public:
47 typedef std::function<Async::Job<qint64>(qint64 oldRevision, qint64 newRevision)> QueryFunction; 48 typedef std::function<KAsync::Job<qint64>(qint64 oldRevision, qint64 newRevision)> QueryFunction;
48 49
49 QueryRunner(const Akonadi2::Query &query) : mLatestRevision(0) {}; 50 QueryRunner(const Akonadi2::Query &query) : mLatestRevision(0) {};
50 /** 51 /**
51 * Starts query 52 * Starts query
52 */ 53 */
53 Async::Job<void> run(qint64 newRevision = 0) 54 KAsync::Job<void> run(qint64 newRevision = 0)
54 { 55 {
55 //TODO: JOBAPI: that last empty .then should not be necessary 56 //TODO: JOBAPI: that last empty .then should not be necessary
56 return queryFunction(mLatestRevision, newRevision).then<void, qint64>([this](qint64 revision) { 57 return queryFunction(mLatestRevision, newRevision).then<void, qint64>([this](qint64 revision) {
@@ -120,7 +121,7 @@ public:
120 return Akonadi2::ApplicationDomain::getTypeName<DomainType>(); 121 return Akonadi2::ApplicationDomain::getTypeName<DomainType>();
121 } 122 }
122 123
123 Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE 124 KAsync::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE
124 { 125 {
125 if (!mDomainTypeAdaptorFactory) { 126 if (!mDomainTypeAdaptorFactory) {
126 Warning() << "No domain type adaptor factory available"; 127 Warning() << "No domain type adaptor factory available";
@@ -130,25 +131,25 @@ public:
130 return sendCreateCommand(bufferTypeForDomainType(), QByteArray::fromRawData(reinterpret_cast<const char*>(entityFbb.GetBufferPointer()), entityFbb.GetSize())); 131 return sendCreateCommand(bufferTypeForDomainType(), QByteArray::fromRawData(reinterpret_cast<const char*>(entityFbb.GetBufferPointer()), entityFbb.GetSize()));
131 } 132 }
132 133
133 Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE 134 KAsync::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE
134 { 135 {
135 //TODO 136 //TODO
136 return Async::null<void>(); 137 return KAsync::null<void>();
137 } 138 }
138 139
139 Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE 140 KAsync::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE
140 { 141 {
141 //TODO 142 //TODO
142 return Async::null<void>(); 143 return KAsync::null<void>();
143 } 144 }
144 145
145 //TODO JOBAPI return job from sync continuation to execute it as subjob? 146 //TODO JOBAPI return job from sync continuation to execute it as subjob?
146 Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > &resultProvider) Q_DECL_OVERRIDE 147 KAsync::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > &resultProvider) Q_DECL_OVERRIDE
147 { 148 {
148 auto runner = QSharedPointer<QueryRunner>::create(query); 149 auto runner = QSharedPointer<QueryRunner>::create(query);
149 QWeakPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > weakResultProvider = resultProvider; 150 QWeakPointer<Akonadi2::ResultProvider<typename DomainType::Ptr> > weakResultProvider = resultProvider;
150 runner->setQuery([this, weakResultProvider, query] (qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> { 151 runner->setQuery([this, weakResultProvider, query] (qint64 oldRevision, qint64 newRevision) -> KAsync::Job<qint64> {
151 return Async::start<qint64>([this, weakResultProvider, query, oldRevision, newRevision](Async::Future<qint64> &future) { 152 return KAsync::start<qint64>([this, weakResultProvider, query, oldRevision, newRevision](KAsync::Future<qint64> &future) {
152 auto resultProvider = weakResultProvider.toStrongRef(); 153 auto resultProvider = weakResultProvider.toStrongRef();
153 if (!resultProvider) { 154 if (!resultProvider) {
154 Warning() << "Tried executing query after result provider is already gone"; 155 Warning() << "Tried executing query after result provider is already gone";
@@ -175,7 +176,7 @@ public:
175 } 176 }
176 177
177 //We have to capture the runner to keep it alive 178 //We have to capture the runner to keep it alive
178 return synchronizeResource(query.syncOnDemand, query.processAll).template then<void>([runner](Async::Future<void> &future) { 179 return synchronizeResource(query.syncOnDemand, query.processAll).template then<void>([runner](KAsync::Future<void> &future) {
179 runner->run().then<void>([&future]() { 180 runner->run().then<void>([&future]() {
180 future.setFinished(); 181 future.setFinished();
181 }).exec(); 182 }).exec();
@@ -183,7 +184,7 @@ public:
183 } 184 }
184 185
185protected: 186protected:
186 Async::Job<void> sendCreateCommand(const QByteArray &resourceBufferType, const QByteArray &buffer) 187 KAsync::Job<void> sendCreateCommand(const QByteArray &resourceBufferType, const QByteArray &buffer)
187 { 188 {
188 flatbuffers::FlatBufferBuilder fbb; 189 flatbuffers::FlatBufferBuilder fbb;
189 //This is the resource buffer type and not the domain type 190 //This is the resource buffer type and not the domain type
@@ -195,7 +196,7 @@ protected:
195 return mResourceAccess->sendCommand(Akonadi2::Commands::CreateEntityCommand, fbb); 196 return mResourceAccess->sendCommand(Akonadi2::Commands::CreateEntityCommand, fbb);
196 } 197 }
197 198
198 Async::Job<void> synchronizeResource(bool sync, bool processAll) 199 KAsync::Job<void> synchronizeResource(bool sync, bool processAll)
199 { 200 {
200 //TODO check if a sync is necessary 201 //TODO check if a sync is necessary
201 //TODO Only sync what was requested 202 //TODO Only sync what was requested
@@ -203,17 +204,17 @@ protected:
203 //TODO the synchronization should normally not be necessary: We just return what is already available. 204 //TODO the synchronization should normally not be necessary: We just return what is already available.
204 205
205 if (sync || processAll) { 206 if (sync || processAll) {
206 return Async::start<void>([=](Async::Future<void> &future) { 207 return KAsync::start<void>([=](KAsync::Future<void> &future) {
207 mResourceAccess->open(); 208 mResourceAccess->open();
208 mResourceAccess->synchronizeResource(sync, processAll).then<void>([&future]() { 209 mResourceAccess->synchronizeResource(sync, processAll).then<void>([&future]() {
209 future.setFinished(); 210 future.setFinished();
210 }).exec(); 211 }).exec();
211 }); 212 });
212 } 213 }
213 return Async::null<void>(); 214 return KAsync::null<void>();
214 } 215 }
215 216
216 virtual Async::Job<qint64> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision) { return Async::null<qint64>(); }; 217 virtual KAsync::Job<qint64> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision) { return KAsync::null<qint64>(); };
217 218
218protected: 219protected:
219 //TODO use one resource access instance per application => make static 220 //TODO use one resource access instance per application => make static
diff --git a/common/genericresource.cpp b/common/genericresource.cpp
index ea6413b..2394b80 100644
--- a/common/genericresource.cpp
+++ b/common/genericresource.cpp
@@ -61,33 +61,33 @@ private slots:
61 }).exec(); 61 }).exec();
62 } 62 }
63 63
64 Async::Job<void> processQueuedCommand(const Akonadi2::QueuedCommand *queuedCommand) 64 KAsync::Job<void> processQueuedCommand(const Akonadi2::QueuedCommand *queuedCommand)
65 { 65 {
66 Log() << "Processing command: " << Akonadi2::Commands::name(queuedCommand->commandId()); 66 Log() << "Processing command: " << Akonadi2::Commands::name(queuedCommand->commandId());
67 //Throw command into appropriate pipeline 67 //Throw command into appropriate pipeline
68 switch (queuedCommand->commandId()) { 68 switch (queuedCommand->commandId()) {
69 case Akonadi2::Commands::DeleteEntityCommand: 69 case Akonadi2::Commands::DeleteEntityCommand:
70 //mPipeline->removedEntity 70 //mPipeline->removedEntity
71 return Async::null<void>(); 71 return KAsync::null<void>();
72 case Akonadi2::Commands::ModifyEntityCommand: 72 case Akonadi2::Commands::ModifyEntityCommand:
73 //mPipeline->modifiedEntity 73 //mPipeline->modifiedEntity
74 return Async::null<void>(); 74 return KAsync::null<void>();
75 case Akonadi2::Commands::CreateEntityCommand: 75 case Akonadi2::Commands::CreateEntityCommand:
76 return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size()); 76 return mPipeline->newEntity(queuedCommand->command()->Data(), queuedCommand->command()->size());
77 default: 77 default:
78 return Async::error<void>(-1, "Unhandled command"); 78 return KAsync::error<void>(-1, "Unhandled command");
79 } 79 }
80 return Async::null<void>(); 80 return KAsync::null<void>();
81 } 81 }
82 82
83 //Process all messages of this queue 83 //Process all messages of this queue
84 Async::Job<void> processQueue(MessageQueue *queue) 84 KAsync::Job<void> processQueue(MessageQueue *queue)
85 { 85 {
86 //TODO use something like: 86 //TODO use something like:
87 //Async::foreach("pass iterator here").each("process value here").join(); 87 //KAsync::foreach("pass iterator here").each("process value here").join();
88 //Async::foreach("pass iterator here").parallel("process value here").join(); 88 //KAsync::foreach("pass iterator here").parallel("process value here").join();
89 return Async::dowhile( 89 return KAsync::dowhile(
90 [this, queue](Async::Future<bool> &future) { 90 [this, queue](KAsync::Future<bool> &future) {
91 if (queue->isEmpty()) { 91 if (queue->isEmpty()) {
92 future.setValue(false); 92 future.setValue(false);
93 future.setFinished(); 93 future.setFinished();
@@ -133,13 +133,13 @@ private slots:
133 ); 133 );
134 } 134 }
135 135
136 Async::Job<void> processPipeline() 136 KAsync::Job<void> processPipeline()
137 { 137 {
138 //Go through all message queues 138 //Go through all message queues
139 auto it = QSharedPointer<QListIterator<MessageQueue*> >::create(mCommandQueues); 139 auto it = QSharedPointer<QListIterator<MessageQueue*> >::create(mCommandQueues);
140 return Async::dowhile( 140 return KAsync::dowhile(
141 [it]() { return it->hasNext(); }, 141 [it]() { return it->hasNext(); },
142 [it, this](Async::Future<void> &future) { 142 [it, this](KAsync::Future<void> &future) {
143 auto queue = it->next(); 143 auto queue = it->next();
144 processQueue(queue).then<void>([&future]() { 144 processQueue(queue).then<void>([&future]() {
145 Trace() << "Queue processed"; 145 Trace() << "Queue processed";
@@ -206,12 +206,12 @@ void GenericResource::processCommand(int commandId, const QByteArray &data, uint
206 enqueueCommand(mUserQueue, commandId, data); 206 enqueueCommand(mUserQueue, commandId, data);
207} 207}
208 208
209Async::Job<void> GenericResource::processAllMessages() 209KAsync::Job<void> GenericResource::processAllMessages()
210{ 210{
211 //We have to wait for all items to be processed to ensure the synced items are available when a query gets executed. 211 //We have to wait for all items to be processed to ensure the synced items are available when a query gets executed.
212 //TODO: report errors while processing sync? 212 //TODO: report errors while processing sync?
213 //TODO JOBAPI: A helper that waits for n events and then continues? 213 //TODO JOBAPI: A helper that waits for n events and then continues?
214 return Async::start<void>([this](Async::Future<void> &f) { 214 return KAsync::start<void>([this](KAsync::Future<void> &f) {
215 if (mSynchronizerQueue.isEmpty()) { 215 if (mSynchronizerQueue.isEmpty()) {
216 f.setFinished(); 216 f.setFinished();
217 } else { 217 } else {
@@ -219,7 +219,7 @@ Async::Job<void> GenericResource::processAllMessages()
219 f.setFinished(); 219 f.setFinished();
220 }); 220 });
221 } 221 }
222 }).then<void>([this](Async::Future<void> &f) { 222 }).then<void>([this](KAsync::Future<void> &f) {
223 if (mUserQueue.isEmpty()) { 223 if (mUserQueue.isEmpty()) {
224 f.setFinished(); 224 f.setFinished();
225 } else { 225 } else {
diff --git a/common/genericresource.h b/common/genericresource.h
index 36fa567..ac28575 100644
--- a/common/genericresource.h
+++ b/common/genericresource.h
@@ -38,8 +38,8 @@ public:
38 virtual ~GenericResource(); 38 virtual ~GenericResource();
39 39
40 virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline) Q_DECL_OVERRIDE; 40 virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline) Q_DECL_OVERRIDE;
41 virtual Async::Job<void> synchronizeWithSource(Pipeline *pipeline) Q_DECL_OVERRIDE = 0; 41 virtual KAsync::Job<void> synchronizeWithSource(Pipeline *pipeline) Q_DECL_OVERRIDE = 0;
42 virtual Async::Job<void> processAllMessages() Q_DECL_OVERRIDE; 42 virtual KAsync::Job<void> processAllMessages() Q_DECL_OVERRIDE;
43 43
44 virtual void configurePipeline(Pipeline *pipeline) Q_DECL_OVERRIDE; 44 virtual void configurePipeline(Pipeline *pipeline) Q_DECL_OVERRIDE;
45 int error() const; 45 int error() const;
diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index e2f23ed..21cf1c5 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -29,7 +29,6 @@
29#include "metadata_generated.h" 29#include "metadata_generated.h"
30#include "createentity_generated.h" 30#include "createentity_generated.h"
31#include "entitybuffer.h" 31#include "entitybuffer.h"
32#include "async/src/async.h"
33#include "log.h" 32#include "log.h"
34 33
35namespace Akonadi2 34namespace Akonadi2
@@ -94,7 +93,7 @@ void Pipeline::null()
94 // state.step(); 93 // state.step();
95} 94}
96 95
97Async::Job<void> Pipeline::newEntity(void const *command, size_t size) 96KAsync::Job<void> Pipeline::newEntity(void const *command, size_t size)
98{ 97{
99 Log() << "Pipeline: New Entity"; 98 Log() << "Pipeline: New Entity";
100 99
@@ -107,7 +106,7 @@ Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
107 flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size); 106 flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(command), size);
108 if (!Akonadi2::Commands::VerifyCreateEntityBuffer(verifyer)) { 107 if (!Akonadi2::Commands::VerifyCreateEntityBuffer(verifyer)) {
109 qWarning() << "invalid buffer, not a create entity buffer"; 108 qWarning() << "invalid buffer, not a create entity buffer";
110 return Async::error<void>(); 109 return KAsync::error<void>();
111 } 110 }
112 } 111 }
113 auto createEntity = Akonadi2::Commands::GetCreateEntity(command); 112 auto createEntity = Akonadi2::Commands::GetCreateEntity(command);
@@ -118,7 +117,7 @@ Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
118 flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size()); 117 flatbuffers::Verifier verifyer(reinterpret_cast<const uint8_t *>(createEntity->delta()->Data()), createEntity->delta()->size());
119 if (!Akonadi2::VerifyEntityBuffer(verifyer)) { 118 if (!Akonadi2::VerifyEntityBuffer(verifyer)) {
120 qWarning() << "invalid buffer, not an entity buffer"; 119 qWarning() << "invalid buffer, not an entity buffer";
121 return Async::error<void>(); 120 return KAsync::error<void>();
122 } 121 }
123 } 122 }
124 auto entity = Akonadi2::GetEntity(createEntity->delta()->Data()); 123 auto entity = Akonadi2::GetEntity(createEntity->delta()->Data());
@@ -139,7 +138,7 @@ Async::Job<void> Pipeline::newEntity(void const *command, size_t size)
139 storage().setMaxRevision(newRevision); 138 storage().setMaxRevision(newRevision);
140 Log() << "Pipeline: wrote entity: "<< newRevision; 139 Log() << "Pipeline: wrote entity: "<< newRevision;
141 140
142 return Async::start<void>([this, key, entityType](Async::Future<void> &future) { 141 return KAsync::start<void>([this, key, entityType](KAsync::Future<void> &future) {
143 PipelineState state(this, NewPipeline, key, d->newPipeline[entityType], [&future]() { 142 PipelineState state(this, NewPipeline, key, d->newPipeline[entityType], [&future]() {
144 future.setFinished(); 143 future.setFinished();
145 }); 144 });
diff --git a/common/pipeline.h b/common/pipeline.h
index a574d27..b695bde 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -27,7 +27,8 @@
27 27
28#include <akonadi2common_export.h> 28#include <akonadi2common_export.h>
29#include <storage.h> 29#include <storage.h>
30#include "async/src/async.h" 30
31#include <Async/Async>
31 32
32#include "entity_generated.h" 33#include "entity_generated.h"
33 34
@@ -53,7 +54,7 @@ public:
53 54
54 void null(); 55 void null();
55 56
56 Async::Job<void> newEntity(void const *command, size_t size); 57 KAsync::Job<void> newEntity(void const *command, size_t size);
57 void modifiedEntity(const QString &entityType, const QByteArray &key, void *data, size_t size); 58 void modifiedEntity(const QString &entityType, const QByteArray &key, void *data, size_t size);
58 void deletedEntity(const QString &entityType, const QByteArray &key); 59 void deletedEntity(const QString &entityType, const QByteArray &key);
59 60
diff --git a/common/resource.cpp b/common/resource.cpp
index e158a40..bd69afd 100644
--- a/common/resource.cpp
+++ b/common/resource.cpp
@@ -53,16 +53,16 @@ void Resource::processCommand(int commandId, const QByteArray &data, uint size,
53 pipeline->null(); 53 pipeline->null();
54} 54}
55 55
56Async::Job<void> Resource::synchronizeWithSource(Pipeline *pipeline) 56KAsync::Job<void> Resource::synchronizeWithSource(Pipeline *pipeline)
57{ 57{
58 return Async::start<void>([pipeline](Async::Future<void> &f) { 58 return KAsync::start<void>([pipeline](KAsync::Future<void> &f) {
59 pipeline->null(); 59 pipeline->null();
60 }); 60 });
61} 61}
62 62
63Async::Job<void> Resource::processAllMessages() 63KAsync::Job<void> Resource::processAllMessages()
64{ 64{
65 return Async::null<void>(); 65 return KAsync::null<void>();
66} 66}
67 67
68class ResourceFactory::Private 68class ResourceFactory::Private
diff --git a/common/resource.h b/common/resource.h
index 18a6827..ea1e9d8 100644
--- a/common/resource.h
+++ b/common/resource.h
@@ -21,7 +21,8 @@
21#include <akonadi2common_export.h> 21#include <akonadi2common_export.h>
22#include <clientapi.h> 22#include <clientapi.h>
23#include <pipeline.h> 23#include <pipeline.h>
24#include <async/src/async.h> 24
25#include <Async/Async>
25 26
26namespace Akonadi2 27namespace Akonadi2
27{ 28{
@@ -36,8 +37,8 @@ public:
36 virtual ~Resource(); 37 virtual ~Resource();
37 38
38 virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline); 39 virtual void processCommand(int commandId, const QByteArray &data, uint size, Pipeline *pipeline);
39 virtual Async::Job<void> synchronizeWithSource(Pipeline *pipeline); 40 virtual KAsync::Job<void> synchronizeWithSource(Pipeline *pipeline);
40 virtual Async::Job<void> processAllMessages(); 41 virtual KAsync::Job<void> processAllMessages();
41 42
42 virtual void configurePipeline(Pipeline *pipeline); 43 virtual void configurePipeline(Pipeline *pipeline);
43 44
diff --git a/common/resourceaccess.cpp b/common/resourceaccess.cpp
index b7d569b..feffcf4 100644
--- a/common/resourceaccess.cpp
+++ b/common/resourceaccess.cpp
@@ -70,8 +70,8 @@ class ResourceAccess::Private
70{ 70{
71public: 71public:
72 Private(const QByteArray &name, ResourceAccess *ra); 72 Private(const QByteArray &name, ResourceAccess *ra);
73 Async::Job<void> tryToConnect(); 73 KAsync::Job<void> tryToConnect();
74 Async::Job<void> initializeSocket(); 74 KAsync::Job<void> initializeSocket();
75 QByteArray resourceName; 75 QByteArray resourceName;
76 QSharedPointer<QLocalSocket> socket; 76 QSharedPointer<QLocalSocket> socket;
77 QByteArray partialMessageBuffer; 77 QByteArray partialMessageBuffer;
@@ -89,10 +89,10 @@ ResourceAccess::Private::Private(const QByteArray &name, ResourceAccess *q)
89} 89}
90 90
91//Connects to server and returns connected socket on success 91//Connects to server and returns connected socket on success
92Async::Job<QSharedPointer<QLocalSocket> > ResourceAccess::connectToServer(const QByteArray &identifier) 92KAsync::Job<QSharedPointer<QLocalSocket> > ResourceAccess::connectToServer(const QByteArray &identifier)
93{ 93{
94 auto s = QSharedPointer<QLocalSocket>::create(); 94 auto s = QSharedPointer<QLocalSocket>::create();
95 return Async::start<QSharedPointer<QLocalSocket> >([identifier, s](Async::Future<QSharedPointer<QLocalSocket> > &future) { 95 return KAsync::start<QSharedPointer<QLocalSocket> >([identifier, s](KAsync::Future<QSharedPointer<QLocalSocket> > &future) {
96 s->setServerName(identifier); 96 s->setServerName(identifier);
97 auto context = new QObject; 97 auto context = new QObject;
98 QObject::connect(s.data(), &QLocalSocket::connected, context, [&future, &s, context]() { 98 QObject::connect(s.data(), &QLocalSocket::connected, context, [&future, &s, context]() {
@@ -109,15 +109,15 @@ Async::Job<QSharedPointer<QLocalSocket> > ResourceAccess::connectToServer(const
109 }); 109 });
110} 110}
111 111
112Async::Job<void> ResourceAccess::Private::tryToConnect() 112KAsync::Job<void> ResourceAccess::Private::tryToConnect()
113{ 113{
114 return Async::dowhile([this]() -> bool { 114 return KAsync::dowhile([this]() -> bool {
115 //TODO abort after N retries? 115 //TODO abort after N retries?
116 return !socket; 116 return !socket;
117 }, 117 },
118 [this](Async::Future<void> &future) { 118 [this](KAsync::Future<void> &future) {
119 Trace() << "Loop"; 119 Trace() << "Loop";
120 Async::wait(50) 120 KAsync::wait(50)
121 .then(connectToServer(resourceName)) 121 .then(connectToServer(resourceName))
122 .then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) { 122 .then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) {
123 Q_ASSERT(s); 123 Q_ASSERT(s);
@@ -130,9 +130,9 @@ Async::Job<void> ResourceAccess::Private::tryToConnect()
130 }); 130 });
131} 131}
132 132
133Async::Job<void> ResourceAccess::Private::initializeSocket() 133KAsync::Job<void> ResourceAccess::Private::initializeSocket()
134{ 134{
135 return Async::start<void>([this](Async::Future<void> &future) { 135 return KAsync::start<void>([this](KAsync::Future<void> &future) {
136 Trace() << "Trying to connect"; 136 Trace() << "Trying to connect";
137 connectToServer(resourceName).then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) { 137 connectToServer(resourceName).then<void, QSharedPointer<QLocalSocket> >([this, &future](const QSharedPointer<QLocalSocket> &s) {
138 Trace() << "Connected to resource, without having to start it."; 138 Trace() << "Connected to resource, without having to start it.";
@@ -189,9 +189,9 @@ void ResourceAccess::registerCallback(uint messageId, const std::function<void(i
189 d->resultHandler.insert(messageId, callback); 189 d->resultHandler.insert(messageId, callback);
190} 190}
191 191
192Async::Job<void> ResourceAccess::sendCommand(int commandId) 192KAsync::Job<void> ResourceAccess::sendCommand(int commandId)
193{ 193{
194 return Async::start<void>([this, commandId](Async::Future<void> &f) { 194 return KAsync::start<void>([this, commandId](KAsync::Future<void> &f) {
195 auto continuation = [&f](int error, const QString &errorMessage) { 195 auto continuation = [&f](int error, const QString &errorMessage) {
196 if (error) { 196 if (error) {
197 f.setError(error, errorMessage); 197 f.setError(error, errorMessage);
@@ -205,11 +205,11 @@ Async::Job<void> ResourceAccess::sendCommand(int commandId)
205 }); 205 });
206} 206}
207 207
208Async::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb) 208KAsync::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb)
209{ 209{
210 //The flatbuffer is transient, but we want to store it until the job is executed 210 //The flatbuffer is transient, but we want to store it until the job is executed
211 QByteArray buffer(reinterpret_cast<const char*>(fbb.GetBufferPointer()), fbb.GetSize()); 211 QByteArray buffer(reinterpret_cast<const char*>(fbb.GetBufferPointer()), fbb.GetSize());
212 return Async::start<void>([commandId, buffer, this](Async::Future<void> &f) { 212 return KAsync::start<void>([commandId, buffer, this](KAsync::Future<void> &f) {
213 auto callback = [&f](int error, const QString &errorMessage) { 213 auto callback = [&f](int error, const QString &errorMessage) {
214 if (error) { 214 if (error) {
215 f.setError(error, errorMessage); 215 f.setError(error, errorMessage);
@@ -225,7 +225,7 @@ Async::Job<void> ResourceAccess::sendCommand(int commandId, flatbuffers::FlatBu
225 }); 225 });
226} 226}
227 227
228Async::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool localSync) 228KAsync::Job<void> ResourceAccess::synchronizeResource(bool sourceSync, bool localSync)
229{ 229{
230 auto command = Akonadi2::CreateSynchronize(d->fbb, sourceSync, localSync); 230 auto command = Akonadi2::CreateSynchronize(d->fbb, sourceSync, localSync);
231 Akonadi2::FinishSynchronizeBuffer(d->fbb, command); 231 Akonadi2::FinishSynchronizeBuffer(d->fbb, command);
diff --git a/common/resourceaccess.h b/common/resourceaccess.h
index c16a9d2..b779db9 100644
--- a/common/resourceaccess.h
+++ b/common/resourceaccess.h
@@ -24,8 +24,9 @@
24#include <QObject> 24#include <QObject>
25#include <QTimer> 25#include <QTimer>
26 26
27#include <Async/Async>
28
27#include <flatbuffers/flatbuffers.h> 29#include <flatbuffers/flatbuffers.h>
28#include <async/src/async.h>
29 30
30namespace Akonadi2 31namespace Akonadi2
31{ 32{
@@ -43,13 +44,13 @@ public:
43 QByteArray resourceName() const; 44 QByteArray resourceName() const;
44 bool isReady() const; 45 bool isReady() const;
45 46
46 Async::Job<void> sendCommand(int commandId); 47 KAsync::Job<void> sendCommand(int commandId);
47 Async::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb); 48 KAsync::Job<void> sendCommand(int commandId, flatbuffers::FlatBufferBuilder &fbb);
48 Async::Job<void> synchronizeResource(bool remoteSync, bool localSync); 49 KAsync::Job<void> synchronizeResource(bool remoteSync, bool localSync);
49 /** 50 /**
50 * Tries to connect to server, and returns a connected socket on success. 51 * Tries to connect to server, and returns a connected socket on success.
51 */ 52 */
52 static Async::Job<QSharedPointer<QLocalSocket> > connectToServer(const QByteArray &identifier); 53 static KAsync::Job<QSharedPointer<QLocalSocket> > connectToServer(const QByteArray &identifier);
53 54
54public Q_SLOTS: 55public Q_SLOTS:
55 void open(); 56 void open();
diff --git a/examples/dummyresource/CMakeLists.txt b/examples/dummyresource/CMakeLists.txt
index 5e911d2..36b8e54 100644
--- a/examples/dummyresource/CMakeLists.txt
+++ b/examples/dummyresource/CMakeLists.txt
@@ -3,9 +3,9 @@ project(akonadi2_resource_dummy)
3add_definitions(-DQT_PLUGIN) 3add_definitions(-DQT_PLUGIN)
4include_directories(${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR}) 4include_directories(${CMAKE_CURRENT_SOURCE_DIR} ${CMAKE_CURRENT_BINARY_DIR})
5 5
6generate_flatbuffers(dummycalendar)
7 6
8add_library(${PROJECT_NAME} SHARED facade.cpp resourcefactory.cpp domainadaptor.cpp resourcefacade.cpp) 7add_library(${PROJECT_NAME} SHARED facade.cpp resourcefactory.cpp domainadaptor.cpp resourcefacade.cpp)
8generate_flatbuffers(${PROJECT_NAME} dummycalendar)
9qt5_use_modules(${PROJECT_NAME} Core Network) 9qt5_use_modules(${PROJECT_NAME} Core Network)
10target_link_libraries(${PROJECT_NAME} akonadi2common) 10target_link_libraries(${PROJECT_NAME} akonadi2common)
11 11
diff --git a/examples/dummyresource/facade.cpp b/examples/dummyresource/facade.cpp
index 176d489..c0143c0 100644
--- a/examples/dummyresource/facade.cpp
+++ b/examples/dummyresource/facade.cpp
@@ -127,9 +127,9 @@ void DummyResourceFacade::readValue(QSharedPointer<Akonadi2::Storage> storage, c
127 }); 127 });
128} 128}
129 129
130Async::Job<qint64> DummyResourceFacade::load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision) 130KAsync::Job<qint64> DummyResourceFacade::load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision)
131{ 131{
132 return Async::start<qint64>([=]() { 132 return KAsync::start<qint64>([=]() {
133 //Now that the sync is complete we can execute the query 133 //Now that the sync is complete we can execute the query
134 const auto preparedQuery = prepareQuery(query); 134 const auto preparedQuery = prepareQuery(query);
135 135
diff --git a/examples/dummyresource/facade.h b/examples/dummyresource/facade.h
index d0af487..41fb3f3 100644
--- a/examples/dummyresource/facade.h
+++ b/examples/dummyresource/facade.h
@@ -34,7 +34,7 @@ class DummyResourceFacade : public Akonadi2::GenericFacade<Akonadi2::Application
34public: 34public:
35 DummyResourceFacade(); 35 DummyResourceFacade();
36 virtual ~DummyResourceFacade(); 36 virtual ~DummyResourceFacade();
37 Async::Job<qint64> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision) Q_DECL_OVERRIDE; 37 KAsync::Job<qint64> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider, qint64 oldRevision, qint64 newRevision) Q_DECL_OVERRIDE;
38 38
39private: 39private:
40 void readValue(QSharedPointer<Akonadi2::Storage> storage, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback, std::function<bool(const std::string &key, DummyCalendar::DummyEvent const *buffer, Akonadi2::ApplicationDomain::Buffer::Event const *local)>); 40 void readValue(QSharedPointer<Akonadi2::Storage> storage, const QByteArray &key, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback, std::function<bool(const std::string &key, DummyCalendar::DummyEvent const *buffer, Akonadi2::ApplicationDomain::Buffer::Event const *local)>);
diff --git a/examples/dummyresource/resourcefacade.cpp b/examples/dummyresource/resourcefacade.cpp
index 870bea2..af293d4 100644
--- a/examples/dummyresource/resourcefacade.cpp
+++ b/examples/dummyresource/resourcefacade.cpp
@@ -40,38 +40,38 @@ QSharedPointer<QSettings> DummyResourceConfigFacade::getSettings()
40 return QSharedPointer<QSettings>::create(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/akonadi2/" + "org.kde." + instanceIdentifier + "/settings.ini", QSettings::IniFormat); 40 return QSharedPointer<QSettings>::create(QStandardPaths::writableLocation(QStandardPaths::GenericDataLocation) + "/akonadi2/" + "org.kde." + instanceIdentifier + "/settings.ini", QSettings::IniFormat);
41} 41}
42 42
43Async::Job<void> DummyResourceConfigFacade::create(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) 43KAsync::Job<void> DummyResourceConfigFacade::create(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject)
44{ 44{
45 //TODO create resource instance 45 //TODO create resource instance
46 //This can be generalized in a base implementation 46 //This can be generalized in a base implementation
47 return Async::null<void>(); 47 return KAsync::null<void>();
48} 48}
49 49
50Async::Job<void> DummyResourceConfigFacade::modify(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) 50KAsync::Job<void> DummyResourceConfigFacade::modify(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject)
51{ 51{
52 //modify configuration 52 //modify configuration
53 //This part is likely resource specific, but could be partially generalized 53 //This part is likely resource specific, but could be partially generalized
54 return Async::start<void>([domainObject, this]() { 54 return KAsync::start<void>([domainObject, this]() {
55 auto settings = getSettings(); 55 auto settings = getSettings();
56 //TODO Write properties to file 56 //TODO Write properties to file
57 }); 57 });
58} 58}
59 59
60Async::Job<void> DummyResourceConfigFacade::remove(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) 60KAsync::Job<void> DummyResourceConfigFacade::remove(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject)
61{ 61{
62 //TODO remove resource instance 62 //TODO remove resource instance
63 //This can be generalized in a base implementation 63 //This can be generalized in a base implementation
64 return Async::null<void>(); 64 return KAsync::null<void>();
65} 65}
66 66
67Async::Job<void> DummyResourceConfigFacade::load(const Akonadi2::Query &query, const QSharedPointer<async::ResultProvider<typename Akonadi2::ApplicationDomain::AkonadiResource::Ptr> > &resultProvider) 67KAsync::Job<void> DummyResourceConfigFacade::load(const Akonadi2::Query &query, const QSharedPointer<async::ResultProvider<typename Akonadi2::ApplicationDomain::AkonadiResource::Ptr> > &resultProvider)
68{ 68{
69 //Read configuration and list all available instances. 69 //Read configuration and list all available instances.
70 //This includes runtime information about runing instances etc. 70 //This includes runtime information about runing instances etc.
71 //Part of this is generic, and part is accessing the resource specific configuration. 71 //Part of this is generic, and part is accessing the resource specific configuration.
72 //FIXME this currently does not support live queries (because we're not inheriting from GenericFacade) 72 //FIXME this currently does not support live queries (because we're not inheriting from GenericFacade)
73 //FIXME only read what was requested in the query? 73 //FIXME only read what was requested in the query?
74 return Async::start<void>([resultProvider, this]() { 74 return KAsync::start<void>([resultProvider, this]() {
75 auto settings = getSettings(); 75 auto settings = getSettings();
76 auto memoryAdaptor = QSharedPointer<Akonadi2::ApplicationDomain::MemoryBufferAdaptor>::create(); 76 auto memoryAdaptor = QSharedPointer<Akonadi2::ApplicationDomain::MemoryBufferAdaptor>::create();
77 //TODO copy settings to adaptor 77 //TODO copy settings to adaptor
diff --git a/examples/dummyresource/resourcefacade.h b/examples/dummyresource/resourcefacade.h
index 0ba60ab..5d0291c 100644
--- a/examples/dummyresource/resourcefacade.h
+++ b/examples/dummyresource/resourcefacade.h
@@ -20,7 +20,8 @@
20#pragma once 20#pragma once
21 21
22#include "common/clientapi.h" 22#include "common/clientapi.h"
23#include "async/src/async.h" 23
24#include <Async/Async>
24 25
25class QSettings; 26class QSettings;
26 27
@@ -30,13 +31,13 @@ public:
30 DummyResourceConfigFacade(); 31 DummyResourceConfigFacade();
31 ~DummyResourceConfigFacade(); 32 ~DummyResourceConfigFacade();
32 //Create an instance 33 //Create an instance
33 Async::Job<void> create(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE; 34 KAsync::Job<void> create(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE;
34 //Modify configuration 35 //Modify configuration
35 Async::Job<void> modify(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE; 36 KAsync::Job<void> modify(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE;
36 //Remove instance 37 //Remove instance
37 Async::Job<void> remove(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE; 38 KAsync::Job<void> remove(const Akonadi2::ApplicationDomain::AkonadiResource &domainObject) Q_DECL_OVERRIDE;
38 //Read configuration and available instances 39 //Read configuration and available instances
39 Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<async::ResultProvider<typename Akonadi2::ApplicationDomain::AkonadiResource::Ptr> > &resultProvider) Q_DECL_OVERRIDE; 40 KAsync::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<async::ResultProvider<typename Akonadi2::ApplicationDomain::AkonadiResource::Ptr> > &resultProvider) Q_DECL_OVERRIDE;
40 41
41private: 42private:
42 QSharedPointer<QSettings> getSettings(); 43 QSharedPointer<QSettings> getSettings();
diff --git a/examples/dummyresource/resourcefactory.cpp b/examples/dummyresource/resourcefactory.cpp
index a4cd68d..e16a693 100644
--- a/examples/dummyresource/resourcefactory.cpp
+++ b/examples/dummyresource/resourcefactory.cpp
@@ -162,9 +162,9 @@ void findByRemoteId(QSharedPointer<Akonadi2::Storage> storage, const QString &ri
162 }); 162 });
163} 163}
164 164
165Async::Job<void> DummyResource::synchronizeWithSource(Akonadi2::Pipeline *pipeline) 165KAsync::Job<void> DummyResource::synchronizeWithSource(Akonadi2::Pipeline *pipeline)
166{ 166{
167 return Async::start<void>([this, pipeline](Async::Future<void> &f) { 167 return KAsync::start<void>([this, pipeline](KAsync::Future<void> &f) {
168 //TODO use a read-only transaction during the complete sync to sync against a defined revision 168 //TODO use a read-only transaction during the complete sync to sync against a defined revision
169 auto storage = QSharedPointer<Akonadi2::Storage>::create(Akonadi2::Store::storageLocation(), "org.kde.dummy"); 169 auto storage = QSharedPointer<Akonadi2::Storage>::create(Akonadi2::Store::storageLocation(), "org.kde.dummy");
170 for (auto it = s_dataSource.constBegin(); it != s_dataSource.constEnd(); it++) { 170 for (auto it = s_dataSource.constBegin(); it != s_dataSource.constEnd(); it++) {
diff --git a/examples/dummyresource/resourcefactory.h b/examples/dummyresource/resourcefactory.h
index 7b7783e..f5caf61 100644
--- a/examples/dummyresource/resourcefactory.h
+++ b/examples/dummyresource/resourcefactory.h
@@ -20,9 +20,10 @@
20#pragma once 20#pragma once
21 21
22#include "common/genericresource.h" 22#include "common/genericresource.h"
23#include "async/src/async.h"
24#include "common/messagequeue.h" 23#include "common/messagequeue.h"
25 24
25#include <Async/Async>
26
26#include <flatbuffers/flatbuffers.h> 27#include <flatbuffers/flatbuffers.h>
27 28
28//TODO: a little ugly to have this in two places, once here and once in Q_PLUGIN_METADATA 29//TODO: a little ugly to have this in two places, once here and once in Q_PLUGIN_METADATA
@@ -32,7 +33,7 @@ class DummyResource : public Akonadi2::GenericResource
32{ 33{
33public: 34public:
34 DummyResource(); 35 DummyResource();
35 Async::Job<void> synchronizeWithSource(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE; 36 KAsync::Job<void> synchronizeWithSource(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE;
36 void configurePipeline(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE; 37 void configurePipeline(Akonadi2::Pipeline *pipeline) Q_DECL_OVERRIDE;
37}; 38};
38 39
diff --git a/synchronizer/CMakeLists.txt b/synchronizer/CMakeLists.txt
index 1f95802..fe938e9 100644
--- a/synchronizer/CMakeLists.txt
+++ b/synchronizer/CMakeLists.txt
@@ -8,6 +8,6 @@ set(akonadi2synchronizer_SRCS
8) 8)
9 9
10add_executable(${PROJECT_NAME} ${akonadi2synchronizer_SRCS}) 10add_executable(${PROJECT_NAME} ${akonadi2synchronizer_SRCS})
11target_link_libraries(${PROJECT_NAME} akonadi2common) 11target_link_libraries(${PROJECT_NAME} akonadi2common KF5::Async)
12qt5_use_modules(${PROJECT_NAME} Widgets Network) 12qt5_use_modules(${PROJECT_NAME} Widgets Network)
13install(TARGETS ${PROJECT_NAME} ${KDE_INSTALL_TARGETS_DEFAULT_ARGS}) 13install(TARGETS ${PROJECT_NAME} ${KDE_INSTALL_TARGETS_DEFAULT_ARGS})
diff --git a/synchronizer/listener.cpp b/synchronizer/listener.cpp
index 8527ec2..71723c9 100644
--- a/synchronizer/listener.cpp
+++ b/synchronizer/listener.cpp
@@ -220,7 +220,7 @@ void Listener::processCommand(int commandId, uint messageId, Client &client, uin
220 Warning() << "No resource loaded"; 220 Warning() << "No resource loaded";
221 break; 221 break;
222 } 222 }
223 auto job = Async::null<void>(); 223 auto job = KAsync::null<void>();
224 if (buffer->sourceSync()) { 224 if (buffer->sourceSync()) {
225 job = m_resource->synchronizeWithSource(m_pipeline); 225 job = m_resource->synchronizeWithSource(m_pipeline);
226 } 226 }
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 7b97264..35b3996 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -8,11 +8,10 @@ include_directories(
8 ${CMAKE_CURRENT_SOURCE_DIR}/../examples/ 8 ${CMAKE_CURRENT_SOURCE_DIR}/../examples/
9 ) 9 )
10 10
11generate_flatbuffers(calendar)
12
13macro(manual_tests) 11macro(manual_tests)
14 foreach(_testname ${ARGN}) 12 foreach(_testname ${ARGN})
15 add_executable(${_testname} ${_testname}.cpp) 13 add_executable(${_testname} ${_testname}.cpp)
14 generate_flatbuffers(${_testname} calendar)
16 qt5_use_modules(${_testname} Core Test Concurrent) 15 qt5_use_modules(${_testname} Core Test Concurrent)
17 target_link_libraries(${_testname} akonadi2common libhawd) 16 target_link_libraries(${_testname} akonadi2common libhawd)
18 endforeach(_testname) 17 endforeach(_testname)
@@ -21,6 +20,7 @@ endmacro(manual_tests)
21macro(auto_tests) 20macro(auto_tests)
22 foreach(_testname ${ARGN}) 21 foreach(_testname ${ARGN})
23 add_executable(${_testname} ${_testname}.cpp) 22 add_executable(${_testname} ${_testname}.cpp)
23 generate_flatbuffers(${_testname} calendar)
24 add_test(${_testname} ${_testname}) 24 add_test(${_testname} ${_testname})
25 qt5_use_modules(${_testname} Core Test Concurrent) 25 qt5_use_modules(${_testname} Core Test Concurrent)
26 target_link_libraries(${_testname} akonadi2common libhawd) 26 target_link_libraries(${_testname} akonadi2common libhawd)
diff --git a/tests/clientapitest.cpp b/tests/clientapitest.cpp
index 1a5d873..1ff683f 100644
--- a/tests/clientapitest.cpp
+++ b/tests/clientapitest.cpp
@@ -24,13 +24,13 @@ class DummyResourceFacade : public Akonadi2::StoreFacade<Akonadi2::ApplicationDo
24{ 24{
25public: 25public:
26 ~DummyResourceFacade(){}; 26 ~DummyResourceFacade(){};
27 Async::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); }; 27 KAsync::Job<void> create(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
28 Async::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); }; 28 KAsync::Job<void> modify(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
29 Async::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return Async::null<void>(); }; 29 KAsync::Job<void> remove(const Akonadi2::ApplicationDomain::Event &domainObject) Q_DECL_OVERRIDE { return KAsync::null<void>(); };
30 30
31 Async::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback) 31 KAsync::Job<qint64> load(const Akonadi2::Query &query, const std::function<void(const Akonadi2::ApplicationDomain::Event::Ptr &)> &resultCallback)
32 { 32 {
33 return Async::start<qint64>([this, resultCallback](Async::Future<qint64> &future) { 33 return KAsync::start<qint64>([this, resultCallback](KAsync::Future<qint64> &future) {
34 qDebug() << "load called"; 34 qDebug() << "load called";
35 for(const auto &result : results) { 35 for(const auto &result : results) {
36 resultCallback(result); 36 resultCallback(result);
@@ -40,16 +40,16 @@ public:
40 }); 40 });
41 } 41 }
42 42
43 Async::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider) Q_DECL_OVERRIDE 43 KAsync::Job<void> load(const Akonadi2::Query &query, const QSharedPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > &resultProvider) Q_DECL_OVERRIDE
44 { 44 {
45 auto runner = QSharedPointer<QueryRunner>::create(query); 45 auto runner = QSharedPointer<QueryRunner>::create(query);
46 //The runner only lives as long as the resultProvider 46 //The runner only lives as long as the resultProvider
47 resultProvider->setQueryRunner(runner); 47 resultProvider->setQueryRunner(runner);
48 QWeakPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > weakResultProvider = resultProvider; 48 QWeakPointer<Akonadi2::ResultProvider<Akonadi2::ApplicationDomain::Event::Ptr> > weakResultProvider = resultProvider;
49 capturedResultProvider = resultProvider; 49 capturedResultProvider = resultProvider;
50 runner->setQuery([this, weakResultProvider, query](qint64 oldRevision, qint64 newRevision) -> Async::Job<qint64> { 50 runner->setQuery([this, weakResultProvider, query](qint64 oldRevision, qint64 newRevision) -> KAsync::Job<qint64> {
51 qDebug() << "Creating query for revisions: " << oldRevision << newRevision; 51 qDebug() << "Creating query for revisions: " << oldRevision << newRevision;
52 return Async::start<qint64>([this, weakResultProvider, query](Async::Future<qint64> &future) { 52 return KAsync::start<qint64>([this, weakResultProvider, query](KAsync::Future<qint64> &future) {
53 auto resultProvider = weakResultProvider.toStrongRef(); 53 auto resultProvider = weakResultProvider.toStrongRef();
54 if (!resultProvider) { 54 if (!resultProvider) {
55 Warning() << "Tried executing query after result provider is already gone"; 55 Warning() << "Tried executing query after result provider is already gone";
@@ -81,7 +81,7 @@ public:
81 }); 81 });
82 } 82 }
83 83
84 return Async::start<void>([runner](Async::Future<void> &future) { 84 return KAsync::start<void>([runner](KAsync::Future<void> &future) {
85 runner->run().then<void>([&future]() { 85 runner->run().then<void>([&future]() {
86 //TODO if not live query, destroy runner. 86 //TODO if not live query, destroy runner.
87 future.setFinished(); 87 future.setFinished();
diff --git a/tests/hawd/dataset.cpp b/tests/hawd/dataset.cpp
index 9f1d307..b068b66 100644
--- a/tests/hawd/dataset.cpp
+++ b/tests/hawd/dataset.cpp
@@ -22,6 +22,7 @@
22 22
23#include <QDateTime> 23#include <QDateTime>
24#include <QDebug> 24#include <QDebug>
25#include <QDataStream>
25 26
26#include <iostream> 27#include <iostream>
27 28
generated by cgit v1.2.3 (git 2.25.1) at 2025-04-30 11:18:23 +0000