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.

2 files changed, 50 insertions, 0 deletions

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