3 files changed, 280 insertions, 89 deletions

diff --git a/async/autotests/asynctest.cpp b/async/autotests/asynctest.cpp
index 65e604f..d709567 100644
--- a/async/autotests/asynctest.cpp
+++ b/async/autotests/asynctest.cpp
@@ -27,6 +27,8 @@
 #include <QtTest/QTest>
 #include <QDebug>
 
+#include <functional>
+
 class AsyncTest : public QObject
 {
     Q_OBJECT
@@ -62,6 +64,9 @@ private Q_SLOTS:
 
     void testProgressReporting();
     void testErrorHandler();
+    void testErrorPropagation();
+    void testErrorHandlerAsync();
+    void testErrorPropagationAsync();
 
     void testChainingRunningJob();
     void testChainingFinishedJob();
@@ -92,8 +97,25 @@ private:
             mTimer.start(200);
         }
 
+        AsyncSimulator(Async::Future<T> &future, std::function<void(Async::Future<T>&)> callback)
+            : mFuture(future)
+            , mCallback(callback)
+        {
+            QObject::connect(&mTimer, &QTimer::timeout,
+                             [this]() {
+                                 mCallback(mFuture);
+                             });
+            QObject::connect(&mTimer, &QTimer::timeout,
+                             [this]() {
+                                 delete this;
+                             });
+            mTimer.setSingleShot(true);
+            mTimer.start(200);
+        }
+
     private:
         Async::Future<T> mFuture;
+        std::function<void(Async::Future<T>&)> mCallback;
         T mResult;
         QTimer mTimer;
     };
@@ -390,18 +412,14 @@ void AsyncTest::testAsyncReduce()
         })
     .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);
+            new AsyncSimulator<int>(future,
+                [list](Async::Future<int> &future) {
+                    int sum = 0;
+                    for (int i : list) sum += i;
+                    future.setValue(sum);
+                    future.setFinished();
+                }
+            );
         });
 
     Async::Future<int> future = job.exec();
@@ -509,27 +527,145 @@ void AsyncTest::testProgressReporting()
 
 void AsyncTest::testErrorHandler()
 {
+
+    {
+        auto job = Async::start<int>(
+            [](Async::Future<int> &f) {
+                f.setError(1, "error");
+            });
+
+        auto future = job.exec();
+        QVERIFY(future.isFinished());
+        QCOMPARE(future.errorCode(), 1);
+        QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    }
+
+    {
+        int error = 0;
+        auto job = Async::start<int>(
+            [](Async::Future<int> &f) {
+                f.setError(1, "error");
+            },
+            [&error](int errorCode, const QString &errorMessage) {
+                error += errorCode;
+            }
+        );
+
+        auto future = job.exec();
+        QVERIFY(future.isFinished());
+        QCOMPARE(error, 1);
+        QCOMPARE(future.errorCode(), 1);
+        QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    }
+}
+
+void AsyncTest::testErrorPropagation()
+{
     int error = 0;
+    bool called = false;
     auto job = Async::start<int>(
         [](Async::Future<int> &f) {
             f.setError(1, "error");
         })
     .then<int, int>(
-        [](int v, Async::Future<int> &f) {
+        [&called](int v, Async::Future<int> &f) {
+            called = true;
             f.setFinished();
         },
         [&error](int errorCode, const QString &errorMessage) {
-            error = errorCode; 
+            error += errorCode;
         }
     );
     auto future = job.exec();
-    future.waitForFinished();
+    QVERIFY(future.isFinished());
+    QCOMPARE(future.errorCode(), 1);
+    QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    QCOMPARE(called, false);
     QCOMPARE(error, 1);
+}
+
+void AsyncTest::testErrorHandlerAsync()
+{
+    {
+        auto job = Async::start<int>(
+            [](Async::Future<int> &f) {
+                new AsyncSimulator<int>(f,
+                    [](Async::Future<int> &f) {
+                        f.setError(1, "error");
+                    }
+                );
+            }
+        );
+
+        auto future = job.exec();
+        future.waitForFinished();
+
+        QVERIFY(future.isFinished());
+        QCOMPARE(future.errorCode(), 1);
+        QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    }
+
+    {
+        int error = 0;
+        auto job = Async::start<int>(
+            [](Async::Future<int> &f) {
+                new AsyncSimulator<int>(f,
+                    [](Async::Future<int> &f) {
+                        f.setError(1, "error");
+                    }
+                );
+            },
+            [&error](int errorCode, const QString &errorMessage) {
+                error += errorCode;
+            }
+        );
+
+        auto future = job.exec();
+        future.waitForFinished();
+
+        QVERIFY(future.isFinished());
+        QCOMPARE(error, 1);
+        QCOMPARE(future.errorCode(), 1);
+        QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    }
+}
+
+void AsyncTest::testErrorPropagationAsync()
+{
+    int error = 0;
+    bool called = false;
+    auto job = Async::start<int>(
+        [](Async::Future<int> &f) {
+            new AsyncSimulator<int>(f,
+                [](Async::Future<int> &f) {
+                    f.setError(1, "error");
+                }
+            );
+        })
+    .then<int, int>(
+        [&called](int v, Async::Future<int> &f) {
+            called = true;
+            f.setFinished();
+        },
+        [&error](int errorCode, const QString &errorMessage) {
+            error += errorCode;
+        }
+    );
+
+    auto future = job.exec();
+    future.waitForFinished();
+
     QVERIFY(future.isFinished());
+    QCOMPARE(future.errorCode(), 1);
+    QCOMPARE(future.errorMessage(), QString::fromLatin1("error"));
+    QCOMPARE(called, false);
+    QCOMPARE(error, 1);
 }
 
 
 
+
+
 void AsyncTest::testChainingRunningJob()
 {
     int check = 0;
diff --git a/async/src/async.cpp b/async/src/async.cpp
index 5e26bd8..c780878 100644
--- a/async/src/async.cpp
+++ b/async/src/async.cpp
@@ -24,6 +24,50 @@
 
 using namespace Async;
 
+Private::Execution::Execution(const Private::ExecutorBasePtr &executor)
+    : executor(executor)
+    , resultBase(nullptr)
+    , isRunning(false)
+    , isFinished(false)
+{
+}
+
+Private::Execution::~Execution()
+{
+    if (resultBase) {
+        resultBase->releaseExecution();
+        delete resultBase;
+    }
+    prevExecution.reset();
+}
+
+void Private::Execution::setFinished()
+{
+    isFinished = true;
+    //executor.clear();
+}
+
+void Private::Execution::releaseFuture()
+{
+    resultBase = 0;
+}
+
+bool Private::Execution::errorWasHandled() const
+{
+    Execution * const exec = this;
+    while (exec) {
+        if (exec->executor->hasErrorFunc()) {
+            return true;
+        }
+        exec = exec->prevExecution.data();
+    }
+    return false;
+}
+
+
+
+
+
 Private::ExecutorBase::ExecutorBase(const ExecutorBasePtr &parent)
     : mPrev(parent)
 {
@@ -34,6 +78,8 @@ Private::ExecutorBase::~ExecutorBase()
 }
 
 
+
+
 JobBase::JobBase(const Private::ExecutorBasePtr &executor)
     : mExecutor(executor)
 {
diff --git a/async/src/async.h b/async/src/async.h
index 89ca0d0..adc0b69 100644
--- a/async/src/async.h
+++ b/async/src/async.h
@@ -96,27 +96,9 @@ class ExecutorBase;
 typedef QSharedPointer<ExecutorBase> ExecutorBasePtr;
 
 struct Execution {
-    Execution(const ExecutorBasePtr &executor)
-        : executor(executor)
-        , resultBase(nullptr)
-        , isRunning(false)
-        , isFinished(false)
-    {}
-
-    ~Execution()
-    {
-        if (resultBase) {
-            resultBase->releaseExecution();
-            delete resultBase;
-        }
-        prevExecution.reset();
-    }
-
-    void setFinished()
-    {
-        isFinished = true;
-        executor.clear();
-    }
+    Execution(const ExecutorBasePtr &executor);
+    ~Execution();
+    void setFinished();
 
     template<typename T>
     Async::Future<T>* result()
@@ -124,10 +106,8 @@ struct Execution {
         return static_cast<Async::Future<T>*>(resultBase);
     }
 
-    void releaseFuture()
-    {
-        resultBase = 0;
-    }
+    void releaseFuture();
+    bool errorWasHandled() const;
 
     ExecutorBasePtr executor;
     FutureBase *resultBase;
@@ -147,6 +127,8 @@ class ExecutorBase
     template<typename Out, typename ... In>
     friend class Async::Job;
 
+    friend class Execution;
+
 public:
     virtual ~ExecutorBase();
     virtual ExecutionPtr exec(const ExecutorBasePtr &self) = 0;
@@ -157,6 +139,9 @@ protected:
     template<typename T>
     Async::Future<T>* createFuture(const ExecutionPtr &execution) const;
 
+    virtual bool hasErrorFunc() const = 0;
+    virtual bool handleError(const ExecutionPtr &execution) = 0;
+
     ExecutorBasePtr mPrev;
 };
 
@@ -164,14 +149,17 @@ template<typename PrevOut, typename Out, typename ... In>
 class Executor : public ExecutorBase
 {
 protected:
-    Executor(ErrorHandler errorHandler, const Private::ExecutorBasePtr &parent)
+    Executor(ErrorHandler errorFunc, const Private::ExecutorBasePtr &parent)
         : ExecutorBase(parent)
-        , mErrorFunc(errorHandler)
+        , mErrorFunc(errorFunc)
     {}
+
     virtual ~Executor() {}
     virtual void run(const ExecutionPtr &execution) = 0;
 
     ExecutionPtr exec(const ExecutorBasePtr &self);
+    bool hasErrorFunc() const { return (bool) mErrorFunc; }
+    bool handleError(const ExecutionPtr &execution);
 
     std::function<void(int, const QString &)> mErrorFunc;
 };
@@ -180,7 +168,7 @@ template<typename Out, typename ... In>
 class ThenExecutor: public Executor<typename detail::prevOut<In ...>::type, Out, In ...>
 {
 public:
-    ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
     void run(const ExecutionPtr &execution);
 private:
     ThenTask<Out, In ...> mFunc;
@@ -190,7 +178,7 @@ template<typename PrevOut, typename Out, typename In>
 class EachExecutor : public Executor<PrevOut, Out, In>
 {
 public:
-    EachExecutor(EachTask<Out, In> each, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    EachExecutor(EachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
     void run(const ExecutionPtr &execution);
 private:
     EachTask<Out, In> mFunc;
@@ -201,7 +189,7 @@ template<typename Out, typename In>
 class ReduceExecutor : public ThenExecutor<Out, In>
 {
 public:
-    ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
 private:
     ReduceTask<Out, In> mFunc;
 };
@@ -210,7 +198,7 @@ template<typename Out, typename ... In>
 class SyncThenExecutor : public Executor<typename detail::prevOut<In ...>::type, Out, In ...>
 {
 public:
-    SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
     void run(const ExecutionPtr &execution);
 
 private:
@@ -223,7 +211,7 @@ template<typename Out, typename In>
 class SyncReduceExecutor : public SyncThenExecutor<Out, In>
 {
 public:
-    SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
 private:
     SyncReduceTask<Out, In> mFunc;
 };
@@ -232,7 +220,7 @@ template<typename PrevOut, typename Out, typename In>
 class SyncEachExecutor : public Executor<PrevOut, Out, In>
 {
 public:
-    SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorHandler, const ExecutorBasePtr &parent);
+    SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent);
     void run(const ExecutionPtr &execution);
 private:
     void run(Async::Future<Out> *future, const typename PrevOut::value_type &arg, std::false_type); // !std::is_void<Out>
@@ -253,10 +241,10 @@ private:
  *             where @p In is type of the result.
  */
 template<typename Out, typename ... In>
-Job<Out, In ...> start(ThenTask<Out, In ...> func);
+Job<Out, In ...> start(ThenTask<Out, In ...> func, ErrorHandler errorFunc = ErrorHandler());
 
 template<typename Out, typename ... In>
-Job<Out, In ...> start(SyncThenTask<Out, In ...> func);
+Job<Out, In ...> start(SyncThenTask<Out, In ...> func, ErrorHandler errorFunc = ErrorHandler());
 
 #ifdef WITH_KJOB
 template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
@@ -358,10 +346,10 @@ class Job : public JobBase
     friend class Job;
 
     template<typename OutOther, typename ... InOther>
-    friend Job<OutOther, InOther ...> start(Async::ThenTask<OutOther, InOther ...> func);
+    friend Job<OutOther, InOther ...> start(Async::ThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc);
 
     template<typename OutOther, typename ... InOther>
-    friend Job<OutOther, InOther ...> start(Async::SyncThenTask<OutOther, InOther ...> func);
+    friend Job<OutOther, InOther ...> start(Async::SyncThenTask<OutOther, InOther ...> func, ErrorHandler errorFunc);
 
 #ifdef WITH_KJOB
     template<typename ReturnType, typename KJobType, ReturnType (KJobType::*KJobResultMethod)(), typename ... Args>
@@ -524,17 +512,17 @@ private:
 namespace Async {
 
 template<typename Out, typename ... In>
-Job<Out, In ...> start(ThenTask<Out, In ...> func)
+Job<Out, In ...> start(ThenTask<Out, In ...> func, ErrorHandler error)
 {
     return Job<Out, In...>(Private::ExecutorBasePtr(
-        new Private::ThenExecutor<Out, In ...>(func, ErrorHandler(), Private::ExecutorBasePtr())));
+        new Private::ThenExecutor<Out, In ...>(func, error, Private::ExecutorBasePtr())));
 }
 
 template<typename Out, typename ... In>
-Job<Out, In ...> start(SyncThenTask<Out, In ...> func)
+Job<Out, In ...> start(SyncThenTask<Out, In ...> func, ErrorHandler error)
 {
     return Job<Out, In...>(Private::ExecutorBasePtr(
-        new Private::SyncThenExecutor<Out, In ...>(func, ErrorHandler(), Private::ExecutorBasePtr())));
+        new Private::SyncThenExecutor<Out, In ...>(func, error, Private::ExecutorBasePtr())));
 }
 
 #ifdef WITH_KJOB
@@ -596,17 +584,12 @@ ExecutionPtr Executor<PrevOut, Out, In ...>::exec(const ExecutorBasePtr &self)
 
     // chainup
     execution->prevExecution = mPrev ? mPrev->exec(mPrev) : ExecutionPtr();
-    /*
-    } else if (mPrev && !mPrevFuture) {
-        // If previous job is running or finished, just get it's future
-        mPrevFuture = static_cast<Async::Future<PrevOut>*>(mPrev->result());
-    }
-    */
 
     execution->resultBase = this->createFuture<Out>(execution);
     auto fw = new Async::FutureWatcher<Out>();
     QObject::connect(fw, &Async::FutureWatcher<Out>::futureReady,
                      [fw, execution, this]() {
+                         handleError(execution);
                          execution->setFinished();
                          delete fw;
                      });
@@ -614,44 +597,70 @@ ExecutionPtr Executor<PrevOut, Out, In ...>::exec(const ExecutorBasePtr &self)
 
     Async::Future<PrevOut> *prevFuture = execution->prevExecution ? execution->prevExecution->result<PrevOut>() : nullptr;
     if (!prevFuture || prevFuture->isFinished()) {
-        if (prevFuture && prevFuture->errorCode() != 0) {
-            if (mErrorFunc) {
-                mErrorFunc(prevFuture->errorCode(), prevFuture->errorMessage());
-                execution->resultBase->setFinished();
-                execution->setFinished();
-                return execution;
+        if (prevFuture) { // prevFuture implies execution->prevExecution
+            if (prevFuture->errorCode()) {
+                // Propagate the errorCode and message to the outer Future
+                execution->resultBase->setError(prevFuture->errorCode(), prevFuture->errorMessage());
+                if (!execution->errorWasHandled()) {
+                    if (handleError(execution)) {
+                        return execution;
+                    }
+                } else {
+                    return execution;
+                }
             } else {
-                // Propagate the error to next caller
+                // Propagate error (if any)
             }
         }
+
         execution->isRunning = true;
         run(execution);
     } else {
-        auto futureWatcher = new Async::FutureWatcher<PrevOut>();
-        QObject::connect(futureWatcher, &Async::FutureWatcher<PrevOut>::futureReady,
-                         [futureWatcher, execution, this]() {
-                             auto prevFuture = futureWatcher->future();
+        auto prevFutureWatcher = new Async::FutureWatcher<PrevOut>();
+        QObject::connect(prevFutureWatcher, &Async::FutureWatcher<PrevOut>::futureReady,
+                         [prevFutureWatcher, execution, this]() {
+                             auto prevFuture = prevFutureWatcher->future();
                              assert(prevFuture.isFinished());
-                             delete futureWatcher;
-                             if (prevFuture.errorCode() != 0) {
-                                 if (mErrorFunc) {
-                                     mErrorFunc(prevFuture.errorCode(), prevFuture.errorMessage());
-                                     execution->resultBase->setFinished();
-                                     return;
+                             delete prevFutureWatcher;
+                             auto prevExecutor = execution->executor->mPrev;
+                             if (prevFuture.errorCode()) {
+                                 execution->resultBase->setError(prevFuture.errorCode(), prevFuture.errorMessage());
+                                 if (!execution->errorWasHandled()) {
+                                    if (handleError(execution)) {
+                                        return;
+                                    }
                                  } else {
-                                     // Propagate the error to next caller
+                                     return;
                                  }
                              }
+
+
+                             // propagate error (if any)
                              execution->isRunning = true;
                              run(execution);
                          });
 
-        futureWatcher->setFuture(*static_cast<Async::Future<PrevOut>*>(prevFuture));
+        prevFutureWatcher->setFuture(*static_cast<Async::Future<PrevOut>*>(prevFuture));
     }
 
     return execution;
 }
 
+template<typename PrevOut, typename Out, typename ... In>
+bool Executor<PrevOut, Out, In ...>::handleError(const ExecutionPtr &execution)
+{
+    assert(execution->resultBase->isFinished());
+    if (execution->resultBase->errorCode()) {
+        if (mErrorFunc) {
+            mErrorFunc(execution->resultBase->errorCode(),
+                       execution->resultBase->errorMessage());
+            return true;
+        }
+    }
+
+    return false;
+}
+
 
 template<typename Out, typename ... In>
 ThenExecutor<Out, In ...>::ThenExecutor(ThenTask<Out, In ...> then, ErrorHandler error, const ExecutorBasePtr &parent)
@@ -715,14 +724,14 @@ void EachExecutor<PrevOut, Out, In>::run(const ExecutionPtr &execution)
 }
 
 template<typename Out, typename In>
-ReduceExecutor<Out, In>::ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler error, const ExecutorBasePtr &parent)
-    : ThenExecutor<Out, In>(reduce, error, parent)
+ReduceExecutor<Out, In>::ReduceExecutor(ReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
+    : ThenExecutor<Out, In>(reduce, errorFunc, parent)
 {
 }
 
 template<typename Out, typename ... In>
-SyncThenExecutor<Out, In ...>::SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorHandler, const ExecutorBasePtr &parent)
-    : Executor<typename detail::prevOut<In ...>::type, Out, In ...>(errorHandler, parent)
+SyncThenExecutor<Out, In ...>::SyncThenExecutor(SyncThenTask<Out, In ...> then, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
+    : Executor<typename detail::prevOut<In ...>::type, Out, In ...>(errorFunc, parent)
     , mFunc(then)
 {
 }
@@ -762,8 +771,8 @@ void SyncThenExecutor<Out, In ...>::run(const ExecutionPtr &execution, std::true
 }
 
 template<typename PrevOut, typename Out, typename In>
-SyncEachExecutor<PrevOut, Out, In>::SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorHandler, const ExecutorBasePtr &parent)
-    : Executor<PrevOut, Out, In>(errorHandler, parent)
+SyncEachExecutor<PrevOut, Out, In>::SyncEachExecutor(SyncEachTask<Out, In> each, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
+    : Executor<PrevOut, Out, In>(errorFunc, parent)
     , mFunc(each)
 {
 }
@@ -800,8 +809,8 @@ void SyncEachExecutor<PrevOut, Out, In>::run(Async::Future<Out> * /* unused */,
 }
 
 template<typename Out, typename In>
-SyncReduceExecutor<Out, In>::SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorHandler, const ExecutorBasePtr &parent)
-    : SyncThenExecutor<Out, In>(reduce, errorHandler, parent)
+SyncReduceExecutor<Out, In>::SyncReduceExecutor(SyncReduceTask<Out, In> reduce, ErrorHandler errorFunc, const ExecutorBasePtr &parent)
+    : SyncThenExecutor<Out, In>(reduce, errorFunc, parent)
 {
 }