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";