2 files changed, 72 insertions, 22 deletions

diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
index 403fb83..7aedfc4 100644
--- a/async/autotests/asynctest.cpp
+++ b/async/autotests/asynctest.cpp
@@ -41,6 +41,7 @@ private Q_SLOTS:
     void testSyncPromises();
     void testAsyncPromises();
     void testSyncEach();
+    void testSyncReduce();
 };
 
 void AsyncTest::testSyncPromises()
@@ -65,27 +66,29 @@ void AsyncTest::testSyncPromises()
     job.exec();
     Async::Future<QString> future = job.result();
 
+    QVERIFY(future.isFinished());
     QCOMPARE(future.value(), QString::fromLatin1("RESULT IS 42"));
 }
 
 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);
-      });
+        [](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);
+        });
 
     job.exec();
     Async::Future<int> future = job.result();
+    QVERIFY(future.isFinished());
     QCOMPARE(future.value(), 42);
 }
 
@@ -105,9 +108,32 @@ void AsyncTest::testSyncEach()
     job.exec();
     Async::Future<QList<int>> future = job.result();
     const QList<int> expected({ 2, 3, 4, 5 });
+    QVERIFY(future.isFinished());
     QCOMPARE(future.value(), expected);
 }
 
+void AsyncTest::testSyncReduce()
+{
+    auto job = Async::start<QList<int>>(
+        [](Async::Future<QList<int>> &future) {
+            future.setValue(QList<int>{ 1, 2, 3, 4 });
+            future.setFinished();
+        })
+    .reduce<int, QList<int>>(
+        [](const QList<int> &list, Async::Future<int> &future) {
+            int sum = 0;
+            for (int i : list) sum += i;
+            future.setValue(sum);
+            future.setFinished();
+        });
+
+    job.exec();
+    Async::Future<int> future = job.result();
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.value(), 10);
+}
+
+
 
 QTEST_MAIN(AsyncTest);
 
diff --git a/async/src/async.h b/async/src/async.h
index 0f027f5..a976fa2 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -80,10 +80,6 @@ template<typename Out, typename ... In>
 class ThenExecutor:  public Executor
 {
 
-    typedef Out OutType;
-    typedef typename std::tuple_element<0, std::tuple<In ..., void>>::type InType;
-
-
 public:
     ThenExecutor(ThenTask<Out, In ...> then, Executor *parent = nullptr)
         : Executor(parent)
@@ -93,10 +89,12 @@ public:
 
     void exec()
     {
-        Async::Future<InType> *in = 0;
+        typedef typename std::tuple_element<0, std::tuple<In ..., void>>::type PrevOut;
+
+        Async::Future<PrevOut> *in = 0;
         if (mPrev) {
             mPrev->exec();
-            in = static_cast<Async::Future<InType>*>(mPrev->result());
+            in = static_cast<Async::Future<PrevOut>*>(mPrev->result());
             assert(in->isFinished());
         }
 
@@ -133,6 +131,7 @@ public:
             future.waitForFinished();
             out->setValue(out->value() + future.value());
         }
+        out->setFinished();
 
         mResult = out;
     }
@@ -141,6 +140,32 @@ private:
     std::function<void(const In&, Async::Future<Out>&)> mFunc;
 };
 
+template<typename Out, typename In>
+class ReduceExecutor : public Executor
+{
+public:
+    ReduceExecutor(ReduceTask<Out, In> reduce, Executor *parent = nullptr)
+        : Executor(parent)
+        , mFunc(reduce)
+    {
+    }
+
+    void exec()
+    {
+        assert(mPrev);
+        mPrev->exec();
+        Async::Future<In> *in = static_cast<Async::Future<In>*>(mPrev->result());
+
+        auto out = new Async::Future<Out>();
+        mFunc(in->value(), *out);
+        out->waitForFinished();
+        mResult = out;
+    }
+
+private:
+    std::function<void(const In &, Async::Future<Out> &)> mFunc;
+};
+
 class JobBase
 {
     template<typename Out, typename ... In>
@@ -187,8 +212,8 @@ public:
     Job<OutOther, InOther> reduce(ReduceTask<OutOther, InOther> func)
     {
         static_assert(Async::detail::isIterable<Out>::value,
-                      "The result type of 'Reduce' task must be a list or an array.");
-        //return Job<Out_, In_>::create(func, new ReduceEx, this);
+                      "The 'Result' task can only be connected to a job that returns a list or array");
+        return Job<OutOther, InOther>(new ReduceExecutor<OutOther, InOther>(func, mExecutor));
     }
 
     Async::Future<Out> result() const
@@ -212,8 +237,7 @@ private:
 template<typename Out, typename ... In>
 Async::Job<Out, In ...> Async::start(ThenTask<Out, In ...> func)
 {
-    Executor *exec = new ThenExecutor<Out, In ...>(func);
-    return Job<Out, In ...>(exec);
+    return Job<Out, In ...>(new ThenExecutor<Out, In ...>(func));
 }
 
 #endif // ASYNC_H