Async: allow appending existing Job objects to the Job chain

Now it's possible to do something like

Job<int, int> job = createSomeJob();
auto main = Async::start<int>(....).then(job);

Previously the 'job' would have to be wrapped in a ThenTask-like lambda (which
is what we still do internally), but with this new syntax it's possible to append
another job chain to existing chain easilly. This syntax is available for all
task types.

4 files changed, 181 insertions, 79 deletions

diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
index ed550ca..73026bb 100644
--- a/async/autotests/asynctest.cpp
+++ b/async/autotests/asynctest.cpp
@@ -47,12 +47,46 @@ private Q_SLOTS:
 
     void testAsyncThen();
     void testSyncThen();
+    void testJoinedThen();
+
     void testAsyncEach();
     void testSyncEach();
+    void testJoinedEach();
+
     void testAsyncReduce();
     void testSyncReduce();
+    void testJoinedReduce();
+
     void testErrorHandler();
 
+    void benchmarkSyncThenExecutor();
+
+private:
+    template<typename T>
+    class AsyncSimulator {
+    public:
+        AsyncSimulator(Async::Future<T> &future, const T &result)
+            : mFuture(future)
+            , mResult(result)
+        {
+            QObject::connect(&mTimer, &QTimer::timeout,
+                             [this]() {
+                                 mFuture.setValue(mResult);
+                                 mFuture.setFinished();
+                             });
+            QObject::connect(&mTimer, &QTimer::timeout,
+                             [this]() {
+                                 delete this;
+                             });
+            mTimer.setSingleShot(true);
+            mTimer.start(200);
+        }
+
+    private:
+        Async::Future<T> mFuture;
+        T mResult;
+        QTimer mTimer;
+    };
 };
 
 
@@ -86,16 +120,7 @@ void AsyncTest::testAsyncPromises()
 {
     auto job = Async::start<int>(
         [](Async::Future<int> &future) {
-            QTimer *timer = new QTimer();
-            QObject::connect(timer, &QTimer::timeout,
-                             [&]() {
-                                future.setValue(42);
-                                future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(200);
+            new AsyncSimulator<int>(future, 42);
         });
 
     Async::Future<int> future = job.exec();
@@ -110,16 +135,7 @@ void AsyncTest::testAsyncPromises2()
 
     auto job = Async::start<int>(
         [](Async::Future<int> &future) {
-            QTimer *timer = new QTimer();
-            QObject::connect(timer, &QTimer::timeout,
-                             [&]() {
-                                future.setValue(42);
-                                future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(200);
+            new AsyncSimulator<int>(future, 42);
         }
     ).then<int, int>([&done](int result, Async::Future<int> &future) {
         done = true;
@@ -139,16 +155,7 @@ void AsyncTest::testNestedAsync()
     auto job = Async::start<int>(
         [](Async::Future<int> &future) {
             auto innerJob = Async::start<int>([](Async::Future<int> &innerFuture) {
-                QTimer *timer = new QTimer();
-                QObject::connect(timer, &QTimer::timeout,
-                                [&]() {
-                                    innerFuture.setValue(42);
-                                    innerFuture.setFinished();
-                                });
-                QObject::connect(timer, &QTimer::timeout,
-                                timer, &QObject::deleteLater);
-                timer->setSingleShot(true);
-                timer->start(0);
+                new AsyncSimulator<int>(innerFuture, 42);
             }).then<void>([&future](Async::Future<void> &innerThenFuture) {
                 future.setFinished();
                 innerThenFuture.setFinished();
@@ -186,16 +193,7 @@ void AsyncTest::testAsyncThen()
 {
     auto job = Async::start<int>(
         [](Async::Future<int> &future) {
-            QTimer *timer = new QTimer;
-            QObject::connect(timer, &QTimer::timeout,
-                             [&]() {
-                                 future.setValue(42);
-                                 future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(0);
+            new AsyncSimulator<int>(future, 42);
         });
 
     auto future = job.exec();
@@ -221,33 +219,37 @@ void AsyncTest::testSyncThen()
     QCOMPARE(future.value(), 84);
 }
 
+void AsyncTest::testJoinedThen()
+{
+    auto job1 = Async::start<int, int>(
+        [](int in, Async::Future<int> &future) {
+            new AsyncSimulator<int>(future, in * 2);
+        });
+
+    auto job2 = Async::start<int>(
+        [](Async::Future<int> &future) {
+            new AsyncSimulator<int>(future, 42);
+        })
+    .then<int>(job1);
+
+    auto future = job2.exec();
+    future.waitForFinished();
+
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 84);
+}
+
+
+
 void AsyncTest::testAsyncEach()
 {
     auto job = Async::start<QList<int>>(
         [](Async::Future<QList<int>> &future) {
-            QTimer *timer = new QTimer;
-            QObject::connect(timer, &QTimer::timeout,
-                             [&future]() {
-                                 future.setValue({ 1, 2, 3, 4 });
-                                 future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(0);
+            new AsyncSimulator<QList<int>>(future, { 1, 2, 3, 4 });
         })
     .each<QList<int>, int>(
         [](const int &v, Async::Future<QList<int>> &future) {
-            QTimer *timer = new QTimer;
-            QObject::connect(timer, &QTimer::timeout,
-                             [v, &future]() {
-                                 future.setValue({ v + 1 });
-                                 future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(0);
+            new AsyncSimulator<QList<int>>(future, { v + 1 });
         });
 
     auto future = job.exec();
@@ -258,7 +260,6 @@ void AsyncTest::testAsyncEach()
     QCOMPARE(future.value(), expected);
 }
 
-
 void AsyncTest::testSyncEach()
 {
     auto job = Async::start<QList<int>>(
@@ -277,20 +278,35 @@ void AsyncTest::testSyncEach()
     QCOMPARE(future.value(), expected);
 }
 
+void AsyncTest::testJoinedEach()
+{
+    auto job1 = Async::start<QList<int>, int>(
+        [](int v, Async::Future<QList<int>> &future) {
+            new AsyncSimulator<QList<int>>(future, { v * 2 });
+        });
+
+    auto job = Async::start<QList<int>>(
+        []() -> QList<int> {
+            return { 1, 2, 3, 4 };
+        })
+    .each(job1);
+
+    auto future = job.exec();
+    future.waitForFinished();
+
+    const QList<int> expected({ 2, 4, 6, 8 });
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), expected);
+}
+
+
+
+
 void AsyncTest::testAsyncReduce()
 {
     auto job = Async::start<QList<int>>(
         [](Async::Future<QList<int>> &future) {
-            QTimer *timer = new QTimer();
-            QObject::connect(timer, &QTimer::timeout,
-                             [&future]() {
-                                 future.setValue({ 1, 2, 3, 4 });
-                                 future.setFinished();
-                             });
-            QObject::connect(timer, &QTimer::timeout,
-                             timer, &QObject::deleteLater);
-            timer->setSingleShot(true);
-            timer->start(0);
+            new AsyncSimulator<QList<int>>(future, { 1, 2, 3, 4 });
         })
     .reduce<int, QList<int>>(
         [](const QList<int> &list, Async::Future<int> &future) {
@@ -334,6 +350,32 @@ void AsyncTest::testSyncReduce()
     QCOMPARE(future.value(), 10);
 }
 
+
+void AsyncTest::testJoinedReduce()
+{
+    auto job1 = Async::start<int, QList<int>>(
+        [](const QList<int> &list, Async::Future<int> &future) {
+            int sum = 0;
+            for (int i : list) sum += i;
+            new AsyncSimulator<int>(future, sum);
+        });
+
+    auto job = Async::start<QList<int>>(
+        []() -> QList<int> {
+            return { 1, 2, 3, 4 };
+        })
+    .reduce(job1);
+
+    auto future = job.exec();
+    future.waitForFinished();
+
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 10);
+}
+
+
+
+
 void AsyncTest::testErrorHandler()
 {
     int error = 0;
@@ -356,6 +398,23 @@ void AsyncTest::testErrorHandler()
 }
 
 
+
+
+
+void AsyncTest::benchmarkSyncThenExecutor()
+{
+    auto job = Async::start<int>(
+        []() -> int {
+            return 0;
+        });
+
+    QBENCHMARK {
+       job.exec();
+    }
+}
+
+
+
 QTEST_MAIN(AsyncTest);
 
 #include "asynctest.moc"
diff --git a/async/src/async.h b/async/src/async.h
index 386722a..d21caf8 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -291,6 +291,12 @@ public:
             new Private::SyncThenExecutor<OutOther, InOther ...>(func, errorFunc, mExecutor)));
     }
 
+    template<typename OutOther, typename ... InOther>
+    Job<OutOther, InOther ...> then(Job<OutOther, InOther ...> otherJob, ErrorHandler errorFunc = ErrorHandler())
+    {
+        return then<OutOther, InOther ...>(nestedJobWrapper<OutOther, InOther ...>(otherJob), errorFunc);
+    }
+
     template<typename OutOther, typename InOther>
     Job<OutOther, InOther> each(EachTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
     {
@@ -308,6 +314,13 @@ public:
     }
 
     template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> each(Job<OutOther, InOther> otherJob, ErrorHandler errorFunc = ErrorHandler())
+    {
+        eachInvariants<OutOther>();
+        return each<OutOther, InOther>(nestedJobWrapper<OutOther, InOther>(otherJob), errorFunc);
+    }
+
+    template<typename OutOther, typename InOther>
     Job<OutOther, InOther> reduce(ReduceTask<OutOther, InOther> func, ErrorHandler errorFunc = ErrorHandler())
     {
         reduceInvariants<InOther>();
@@ -323,6 +336,12 @@ public:
             new Private::SyncReduceExecutor<OutOther, InOther>(func, errorFunc, mExecutor)));
     }
 
+    template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> reduce(Job<OutOther, InOther> otherJob, ErrorHandler errorFunc = ErrorHandler())
+    {
+        return reduce<OutOther, InOther>(nestedJobWrapper<OutOther, InOther>(otherJob), errorFunc);
+    }
+
     template<typename FirstIn>
     Async::Future<Out> exec(FirstIn in)
     {
@@ -377,6 +396,22 @@ private:
         static_assert(std::is_same<typename Out::value_type, typename InOther::value_type>::value,
                       "The return type of previous task must be compatible with input type of this task");
     }
+
+    template<typename OutOther, typename ... InOther>
+    inline std::function<void(InOther ..., Async::Future<OutOther>&)> nestedJobWrapper(Job<OutOther, InOther ...> otherJob) {
+        return [otherJob](InOther ... in, Async::Future<OutOther> &future) {
+            // copy by value is const
+            auto job = otherJob;
+            FutureWatcher<OutOther> *watcher = new FutureWatcher<OutOther>();
+            QObject::connect(watcher, &FutureWatcherBase::futureReady,
+                             [watcher, &future]() {
+                                 Async::detail::copyFutureValue(watcher->future(), future);
+                                 future.setFinished();
+                                 watcher->deleteLater();
+                             });
+            watcher->setFuture(job.exec(in ...));
+        };
+    }
 };
 
 } // namespace Async
diff --git a/async/src/async_impl.h b/async/src/async_impl.h
index 58f6ced..eccbc9b 100644
--- a/async/src/async_impl.h
+++ b/async/src/async_impl.h
@@ -19,6 +19,7 @@
 #define ASYNC_IMPL_H
 
 #include "async.h"
+#include <type_traits>
 
 namespace Async {
 
@@ -45,6 +46,20 @@ struct prevOut {
     using type = typename std::tuple_element<0, std::tuple<T ..., void>>::type;
 };
 
+template<typename T>
+inline typename std::enable_if<!std::is_void<T>::value, void>::type
+copyFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
+{
+    out.setValue(in.value());
+}
+
+template<typename T>
+inline typename std::enable_if<std::is_void<T>::value, void>::type
+copyFutureValue(const Async::Future<T> &in, Async::Future<T> &out)
+{
+    // noop
+}
+
 } // namespace Detail
 
 } // namespace Async
diff --git a/common/clientapi.h b/common/clientapi.h
index c1404da..aa47802 100644
--- a/common/clientapi.h
+++ b/common/clientapi.h
@@ -421,14 +421,7 @@ public:
                 //We have to bind an instance to the function callback. Since we use a shared pointer this keeps the result provider instance (and thus also the emitter) alive.
                 std::function<void(const typename DomainType::Ptr &)> addCallback = std::bind(&ResultProvider<typename DomainType::Ptr>::add, resultSet, std::placeholders::_1);
                 //We copy the facade pointer to keep it alive
-                //TODO JOBAPI: we should be able to just do, job = job.then(facade->load(..))
-                job = job.then<void>([facade, query, addCallback](Async::Future<void> &future) {
-                    Async::Job<void> j = facade->load(query, addCallback);
-                    j.then<void>([&future, facade](Async::Future<void> &f) {
-                        future.setFinished();
-                        f.setFinished();
-                    }).exec();
-                });
+                job = job.then(facade->load(query, addCallback));
             }
             job.then<void>([/* eventloop,  */resultSet](Async::Future<void> &future) {
                 qDebug() << "Query complete";