Async: introduce sync executors

Sync executors don't pass Async::Future into the user-provided tasks, but
instead work with return values of the task methods, wrapping them into the
Async::Future internally. Sync tasks are of course possible since forever, but
not the API for those tasks is much cleaner, for users don't have to deal with
"future" in synchronous tasks, for instance when synchronously processing results
of an async task before passing the data to another async task.

2 files changed, 281 insertions, 27 deletions

diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
index 9240d28..4ebe65e 100644
--- a/async/autotests/asynctest.cpp
+++ b/async/autotests/asynctest.cpp
@@ -42,9 +42,14 @@ private Q_SLOTS:
     void testAsyncPromises();
     void testAsyncPromises2();
     void testNestedAsync();
+    void testAsyncThen();
+    void testSyncThen();
+    void testAsyncEach();
     void testSyncEach();
+    void testAsyncReduce();
     void testSyncReduce();
     void testErrorHandler();
+
 };
 
 void AsyncTest::testSyncPromises()
@@ -155,17 +160,92 @@ void AsyncTest::testNestedAsync()
     QTRY_VERIFY(done);
 }
 
-void AsyncTest::testSyncEach()
+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);
+        });
+
+    auto future = job.exec();
+    future.waitForFinished();
+
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 42);
+}
+
+
+void AsyncTest::testSyncThen()
+{
+    auto job = Async::start<int>(
+        []() -> int {
+            return 42;
+        }).then<int, int>(
+        [](int in) -> int {
+            return in * 2;
+        });
+
+    auto future = job.exec();
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 84);
+}
+
+void AsyncTest::testAsyncEach()
 {
     auto job = Async::start<QList<int>>(
         [](Async::Future<QList<int>> &future) {
-            future.setValue(QList<int>{ 1, 2, 3, 4 });
-            future.setFinished();
+            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);
         })
     .each<QList<int>, int>(
         [](const int &v, Async::Future<QList<int>> &future) {
-            future.setValue(QList<int>{ v + 1 });
-            future.setFinished();
+            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);
+        });
+
+    auto future = job.exec();
+    future.waitForFinished();
+
+    const QList<int> expected({ 2, 3, 4, 5 });
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), expected);
+}
+
+
+void AsyncTest::testSyncEach()
+{
+    auto job = Async::start<QList<int>>(
+        []() -> QList<int> {
+            return { 1, 2, 3, 4 };
+        })
+    .each<QList<int>, int>(
+        [](const int &v) -> QList<int> {
+            return { v + 1 };
         });
 
     Async::Future<QList<int>> future = job.exec();
@@ -175,19 +255,55 @@ void AsyncTest::testSyncEach()
     QCOMPARE(future.value(), expected);
 }
 
-void AsyncTest::testSyncReduce()
+void AsyncTest::testAsyncReduce()
 {
     auto job = Async::start<QList<int>>(
         [](Async::Future<QList<int>> &future) {
-            future.setValue(QList<int>{ 1, 2, 3, 4 });
-            future.setFinished();
+            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);
         })
     .reduce<int, QList<int>>(
         [](const QList<int> &list, Async::Future<int> &future) {
+            QTimer *timer = new QTimer();
+            QObject::connect(timer, &QTimer::timeout,
+                             [list, &future]() {
+                                 int sum = 0;
+                                 for (int i : list) sum += i;
+                                 future.setValue(sum);
+                                 future.setFinished();
+                             });
+            QObject::connect(timer, &QTimer::timeout,
+                             timer, &QObject::deleteLater);
+            timer->setSingleShot(true);
+            timer->start(0);
+        });
+
+    Async::Future<int> future = job.exec();
+    future.waitForFinished();
+
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 10);
+}
+
+void AsyncTest::testSyncReduce()
+{
+    auto job = Async::start<QList<int>>(
+        []() -> QList<int> {
+            return { 1, 2, 3, 4 };
+        })
+    .reduce<int, QList<int>>(
+        [](const QList<int> &list) -> int {
             int sum = 0;
             for (int i : list) sum += i;
-            future.setValue(sum);
-            future.setFinished();
+            return sum;
         });
 
     Async::Future<int> future = job.exec();
@@ -213,7 +329,7 @@ void AsyncTest::testErrorHandler()
     );
     auto future = job.exec();
     future.waitForFinished();
-    QVERIFY(error == 1);
+    QCOMPARE(error, 1);
     QVERIFY(future.isFinished());
 }
 
diff --git a/async/src/async.h b/async/src/async.h
index d15373b..336bae2 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -33,8 +33,6 @@
 
 
 /*
- * TODO: on .then and potentially others: support for ThenTask without future argument and return value which makes it implicitly a sync continuation.
- * Useful for typical value consumer continuations.
  * TODO: error continuation on .then and others.
  * TODO: instead of passing the future objects callbacks could be provided for result reporting (we can still use the future object internally
  */
@@ -47,13 +45,19 @@ class JobBase;
 
 template<typename Out, typename ... In>
 class Job;
-
 template<typename Out, typename ... In>
 using ThenTask = typename detail::identity<std::function<void(In ..., Async::Future<Out>&)>>::type;
+template<typename Out, typename ... In>
+using SyncThenTask = typename detail::identity<std::function<Out(In ...)>>::type;
 template<typename Out, typename In>
 using EachTask = typename detail::identity<std::function<void(In, Async::Future<Out>&)>>::type;
 template<typename Out, typename In>
+using SyncEachTask = typename detail::identity<std::function<Out(In)>>::type;
+template<typename Out, typename In>
 using ReduceTask = typename detail::identity<std::function<void(In, Async::Future<Out>&)>>::type;
+template<typename Out, typename In>
+using SyncReduceTask = typename detail::identity<std::function<Out(In)>>::type;
+
 using ErrorHandler = std::function<void(int, const QString &)>;
 
 namespace Private
@@ -134,6 +138,33 @@ public:
     ReduceExecutor(ReduceTask<Out, In> reduce, const ExecutorBasePtr &parent);
 };
 
+template<typename Out, typename ... In>
+class SyncThenExecutor : public Executor<typename PreviousOut<In ...>::type, Out, In ...>
+{
+public:
+    SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorHandler = ErrorHandler(), const ExecutorBasePtr &parent = ExecutorBasePtr());
+    void previousFutureReady();
+protected:
+    SyncThenTask<Out, In ...> mFunc;
+};
+
+template<typename Out, typename In>
+class SyncReduceExecutor : public SyncThenExecutor<Out, In>
+{
+public:
+    SyncReduceExecutor(SyncReduceTask<Out, In> reduce, const ExecutorBasePtr &parent = ExecutorBasePtr());
+};
+
+template<typename PrevOut, typename Out, typename In>
+class SyncEachExecutor : public Executor<PrevOut, Out, In>
+{
+public:
+    SyncEachExecutor(SyncEachTask<Out, In> each, const ExecutorBasePtr &parent = ExecutorBasePtr());
+    void previousFutureReady();
+protected:
+    SyncEachTask<Out, In> mFunc;
+};
+
 } // namespace Private
 
 /**
@@ -238,6 +269,9 @@ class Job : public JobBase
     template<typename OutOther>
     friend Job<OutOther> start(Async::ThenTask<OutOther> func);
 
+    template<typename OutOther>
+    friend Job<OutOther> start(Async::SyncThenTask<OutOther> func);
+
 public:
     template<typename OutOther, typename ... InOther>
     Job<OutOther, InOther ...> then(ThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc = ErrorHandler())
@@ -246,28 +280,45 @@ public:
             new Private::ThenExecutor<OutOther, InOther ...>(func, errorFunc, mExecutor)));
     }
 
+    template<typename OutOther, typename ... InOther>
+    Job<OutOther, InOther ...> then(SyncThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc = ErrorHandler())
+    {
+        return Job<OutOther, InOther ...>(Private::ExecutorBasePtr(
+            new Private::SyncThenExecutor<OutOther, InOther ...>(func, errorFunc, mExecutor)));
+    }
+
     template<typename OutOther, typename InOther>
     Job<OutOther, InOther> each(EachTask<OutOther, InOther> func)
     {
-        static_assert(detail::isIterable<Out>::value,
-                      "The 'Each' task can only be connected to a job that returns a list or an array.");
-        static_assert(detail::isIterable<OutOther>::value,
-                      "The result type of 'Each' task must be a list or an array.");
+        eachInvariants<OutOther>();
         return Job<OutOther, InOther>(Private::ExecutorBasePtr(
             new Private::EachExecutor<Out, OutOther, InOther>(func, mExecutor)));
     }
 
     template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> each(SyncEachTask<OutOther, InOther> func)
+    {
+        eachInvariants<OutOther>();
+        return Job<OutOther, InOther>(Private::ExecutorBasePtr(
+            new Private::SyncEachExecutor<Out, OutOther, InOther>(func, mExecutor)));
+    }
+
+    template<typename OutOther, typename InOther>
     Job<OutOther, InOther> reduce(ReduceTask<OutOther, InOther> func)
     {
-        static_assert(Async::detail::isIterable<Out>::value,
-                      "The 'Result' task can only be connected to a job that returns a list or an array");
-        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");
+        reduceInvariants<InOther>();
         return Job<OutOther, InOther>(Private::ExecutorBasePtr(
             new Private::ReduceExecutor<OutOther, InOther>(func, mExecutor)));
     }
 
+    template<typename OutOther, typename InOther>
+    Job<OutOther, InOther> reduce(SyncReduceTask<OutOther, InOther> func)
+    {
+        reduceInvariants<InOther>();
+        return Job<OutOther, InOther>(Private::ExecutorBasePtr(
+            new Private::SyncReduceExecutor<OutOther, InOther>(func, mExecutor)));
+    }
+
     Async::Future<Out> exec()
     {
         mExecutor->exec();
@@ -283,6 +334,24 @@ private:
     Job(Private::ExecutorBasePtr executor)
         : JobBase(executor)
     {}
+
+    template<typename OutOther>
+    void eachInvariants()
+    {
+        static_assert(detail::isIterable<Out>::value,
+                      "The 'Each' task can only be connected to a job that returns a list or an array.");
+        static_assert(detail::isIterable<OutOther>::value,
+                      "The result type of 'Each' task must be a list or an array.");
+    }
+
+    template<typename InOther>
+    void reduceInvariants()
+    {
+        static_assert(Async::detail::isIterable<Out>::value,
+                      "The 'Result' task can only be connected to a job that returns a list or an array");
+        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");
+    }
 };
 
 } // namespace Async
@@ -298,6 +367,12 @@ Job<Out> start(ThenTask<Out> func)
     return Job<Out>(Private::ExecutorBasePtr(new Private::ThenExecutor<Out>(func)));
 }
 
+template<typename Out>
+Job<Out> start(SyncThenTask<Out> func)
+{
+    return Job<Out>(Private::ExecutorBasePtr(new Private::SyncThenExecutor<Out>(func)));
+}
+
 namespace Private {
 
 template<typename PrevOut, typename Out, typename ... In>
@@ -378,22 +453,23 @@ void EachExecutor<PrevOut, Out, In>::previousFutureReady()
     }
 
     for (auto arg : this->mPrevFuture->value()) {
-        Async::Future<Out> future;
-        this->mFunc(arg, future);
+        auto future = new Async::Future<Out>;
+        this->mFunc(arg, *future);
         auto fw = new Async::FutureWatcher<Out>();
         mFutureWatchers.append(fw);
         QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
                          [out, future, fw, this]() {
-                             assert(future.isFinished());
+                             assert(future->isFinished());
                              const int index = mFutureWatchers.indexOf(fw);
                              assert(index > -1);
                              mFutureWatchers.removeAt(index);
-                             out->setValue(out->value() + future.value());
+                             out->setValue(out->value() + future->value());
+                             delete future;
                              if (mFutureWatchers.isEmpty()) {
                                  out->setFinished();
                              }
                          });
-        fw->setFuture(future);
+        fw->setFuture(*future);
     }
 }
 
@@ -403,6 +479,68 @@ ReduceExecutor<Out, In>::ReduceExecutor(ReduceTask<Out, In> reduce, const Execut
 {
 }
 
+template<typename Out, typename ... In>
+SyncThenExecutor<Out, In ...>::SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent)
+    : Executor<typename PreviousOut<In ...>::type, Out, In ...>(parent)
+{
+    this->mFunc = then;
+    this->mErrorFunc = errorHandler;
+}
+
+template<typename Out, typename ... In>
+void SyncThenExecutor<Out, In ...>::previousFutureReady()
+{
+    if (this->mPrevFuture) {
+        assert(this->mPrevFuture->isFinished());
+    }
+
+    if (this->mPrevFuture && this->mPrevFuture->errorCode()) {
+        if (this->mErrorFunc) {
+            this->mErrorFunc(this->mPrevFuture->errorCode(), this->mPrevFuture->errorMessage());
+            this->mResult->setFinished();
+        } else {
+            static_cast<Async::Future<Out>*>(this->mResult)->setError(this->mPrevFuture->errorCode(), this->mPrevFuture->errorMessage());
+            //propagate error if no error handler is available
+            Out result = this->mFunc(this->mPrevFuture ? this->mPrevFuture->value() : In() ...);
+            static_cast<Async::Future<Out>*>(this->mResult)->setValue(result);
+            this->mResult->setFinished();
+        }
+    } else {
+        Out result = this->mFunc(this->mPrevFuture ? this->mPrevFuture->value() : In() ...);
+        static_cast<Async::Future<Out>*>(this->mResult)->setValue(result);
+        this->mResult->setFinished();
+    }
+}
+
+template<typename PrevOut, typename Out, typename In>
+SyncEachExecutor<PrevOut, Out, In>::SyncEachExecutor(SyncEachTask<Out, In> each, const ExecutorBasePtr &parent)
+    : Executor<PrevOut, Out, In>(parent)
+{
+    this->mFunc = each;
+}
+
+template<typename PrevOut, typename Out, typename In>
+void SyncEachExecutor<PrevOut, Out, In>::previousFutureReady()
+{
+    assert(this->mPrevFuture->isFinished());
+    auto out = static_cast<Async::Future<Out>*>(this->mResult);
+    if (this->mPrevFuture->value().isEmpty()) {
+        out->setFinished();
+        return;
+    }
+
+    for (auto arg : this->mPrevFuture->value()) {
+        out->setValue(out->value() + this->mFunc(arg));
+    }
+    out->setFinished();
+}
+
+template<typename Out, typename In>
+SyncReduceExecutor<Out, In>::SyncReduceExecutor(SyncReduceTask<Out, In> reduce, const ExecutorBasePtr &parent)
+    : SyncThenExecutor<Out, In>(reduce, ErrorHandler(), parent)
+{
+}
+
 
 } // namespace Private