File indexing completed on 2024-05-12 04:41:22

 /*
     SPDX-FileCopyrightText: 2014-2015 Daniel Vrátil <dvratil@redhat.com>
     SPDX-FileCopyrightText: 2015-2016 Daniel Vrátil <dvratil@kde.org>
     SPDX-FileCopyrightText: 2016 Christian Mollekopf <mollekopf@kolabsystems.com>
 
     SPDX-License-Identifier: LGPL-2.0-or-later
 */
 
 #ifndef KASYNC_JOB_IMPL_H
 #define KASYNC_JOB_IMPL_H
 
 #include "async.h"
 #include "traits_p.h"
 
 #include <QTimer>
 
 //@cond PRIVATE
 
 namespace KAsync
 {
 
 template<typename Out, typename ... In>
 template<typename ... InOther>
 Job<Out, In ...>::operator std::conditional_t<std::is_void<OutType>::value, IncompleteType, Job<void>> ()
 {
     return thenImpl<void, InOther ...>({JobContinuation<void, InOther ...>([](InOther ...){ return KAsync::null<void>(); })}, {});
 }
 
 template<typename Out, typename ... In>
 template<typename OutOther, typename ... InOther>
 Job<OutOther, In ...> Job<Out, In ...>::thenImpl(Private::ContinuationHolder<OutOther, InOther ...> workHelper,
                                                  Private::ExecutionFlag execFlag) const
 {
     thenInvariants<InOther ...>();
     return Job<OutOther, In ...>(QSharedPointer<Private::Executor<OutOther, InOther ...>>::create(
                 std::forward<Private::ContinuationHolder<OutOther, InOther ...>>(workHelper), mExecutor, execFlag));
 }
 
 template<typename Out, typename ... In>
 template<typename OutOther, typename ... InOther>
 Job<OutOther, In ...> Job<Out, In ...>::then(const Job<OutOther, InOther ...> &job) const
 {
     thenInvariants<InOther ...>();
     auto executor = job.mExecutor;
     executor->prepend(mExecutor);
     return Job<OutOther, In ...>(executor);
 }
 
 template<typename Out, typename ... In>
 Job<Out, In ...> Job<Out, In ...>::onError(SyncErrorContinuation<void> &&errorFunc) const
 {
     return Job<Out, In...>(QSharedPointer<Private::Executor<Out, Out>>::create(
                 // Extra indirection to allow propagating the result of a previous future when no
                 // error occurs
                 Private::ContinuationHolder<Out, Out>([errorFunc = std::move(errorFunc)](const Error &error, const Out &val) {
                     errorFunc(error);
                     return val;
                 }), mExecutor, Private::ExecutionFlag::ErrorCase));
 }
 
 template<> // Specialize for void jobs
 inline Job<void> Job<void>::onError(SyncErrorContinuation<void> &&errorFunc) const
 {
     return Job<void>(QSharedPointer<Private::Executor<void>>::create(
                 Private::ContinuationHolder<void>(std::forward<SyncErrorContinuation<void>>(errorFunc)),
                 mExecutor, Private::ExecutionFlag::ErrorCase));
 }
 
 template<typename Out, typename ... In>
 template<typename FirstIn>
 KAsync::Future<Out> Job<Out, In ...>::exec(FirstIn in)
 {
     // Inject a fake sync executor that will return the initial value
     Private::ExecutorBasePtr first = mExecutor;
     while (first->mPrev) {
         first = first->mPrev;
     }
 
     first->mPrev = QSharedPointer<Private::Executor<FirstIn>>::create(
             Private::ContinuationHolder<FirstIn>([val = std::move(in)](Future<FirstIn> &future) {
                  future.setResult(val);
             }));
 
     auto result = exec();
     // Remove the injected executor
     first->mPrev.reset();
     return result;
 }
 
 template<typename Out, typename ... In>
 KAsync::Future<Out> Job<Out, In ...>::exec()
 {
     Private::ExecutionPtr execution = mExecutor->exec(mExecutor, Private::ExecutionContext::Ptr::create());
     KAsync::Future<Out> result = *execution->result<Out>();
 
     return result;
 }
 
 template<typename Out, typename ... In>
 Job<Out, In ...>::Job(Private::ExecutorBasePtr executor)
     : JobBase(executor)
 {}
 
 template<typename Out, typename ... In>
 Job<Out, In ...>::Job(JobContinuation<Out, In ...> &&func)
     : JobBase(new Private::Executor<Out, In ...>(std::forward<JobContinuation<Out, In ...>>(func), {}))
 {
     qWarning() << "Creating job job";
     static_assert(sizeof...(In) <= 1, "Only one or zero input parameters are allowed.");
 }
 
 template<typename Out, typename ... In>
 template<typename OutOther>
 void Job<Out, In ...>::eachInvariants() const
 {
     static_assert(traits::isContainer<Out>::value,
                     "The 'Each' task can only be connected to a job that returns a list or an array.");
     static_assert(std::is_void<OutOther>::value || traits::isContainer<OutOther>::value,
                     "The result type of 'Each' task must be void, a list or an array.");
 }
 
 template<typename Out, typename ... In>
 template<typename InOtherFirst, typename ... InOtherTail>
 void Job<Out, In ...>::thenInvariants() const
 {
     static_assert(!std::is_void<Out>::value && (std::is_convertible<Out, InOtherFirst>::value || std::is_base_of<Out, InOtherFirst>::value),
                     "The return type of previous task must be compatible with input type of this task");
 }
 
 template<typename Out, typename ... In>
 template<typename ... InOther>
 auto Job<Out, In ...>::thenInvariants() const -> std::enable_if_t<(sizeof...(InOther) == 0)>
 {
 }
 
 template<template<typename> class Container>
 KAsync::Job<void> waitForCompletion(Container<KAsync::Future<void>> &futures)
 {
     struct Context {
         void removeWatcher(KAsync::FutureWatcher<void> *w)
         {
             pending.erase(std::remove_if(pending.begin(), pending.end(), [w](const auto &watcher) {
                 return w == watcher.get();
             }));
         }
 
         std::vector<std::unique_ptr<KAsync::FutureWatcher<void>>> pending;
     };
 
     return start<Context *>([]() {
             return new Context();
         })
         .template then<Context*, Context*>([futures](Context *context, KAsync::Future<Context *> &future) {
             for (KAsync::Future<void> subFuture : futures) {
                 if (subFuture.isFinished()) {
                     continue;
                 }
                 // FIXME bind lifetime all watcher to future (respectively the main job)
                 auto watcher = std::make_unique<KAsync::FutureWatcher<void>>();
                 QObject::connect(watcher.get(), &KAsync::FutureWatcher<void>::futureReady,
                                  [&future, watcher = watcher.get(), context]() {
                                     context->removeWatcher(watcher);
                                     if (context->pending.empty()) {
                                         future.setResult(context);
                                     }
                                 });
                 watcher->setFuture(subFuture);
                 context->pending.push_back(std::move(watcher));
             }
             if (context->pending.empty()) {
                 future.setResult(context);
             }
         })
         .template then<void, Context*>([](Context *context) {
             delete context;
         });
         // .finally<void>([context]() { delete context; });
 }
 
 template<typename List, typename ValueType>
 Job<void, List> forEach(KAsync::Job<void, ValueType> job)
 {
     auto cont = [job] (const List &values) mutable {
             auto error = QSharedPointer<KAsync::Error>::create();
             QVector<KAsync::Future<void>> list;
             for (const auto &v : values) {
                 auto future = job
                     .template then<void>([error] (const KAsync::Error &e) {
                         if (e && !*error) {
                             //TODO ideally we would aggregate the errors instead of just using the first one
                             *error = e;
                         }
                     })
                     .exec(v);
                 list.push_back(future);
             }
             return waitForCompletion(list)
                 .then<void>([error](KAsync::Future<void> &future) {
                     if (*error) {
                         future.setError(*error);
                     } else {
                         future.setFinished();
                     }
                 });
         };
     return Job<void, List>(QSharedPointer<Private::Executor<void, List>>::create(
                 Private::ContinuationHolder<void, List>(JobContinuation<void, List>(std::move(cont))), nullptr, Private::ExecutionFlag::GoodCase));
 }
 
 
 template<typename List, typename ValueType>
 Job<void, List> serialForEach(KAsync::Job<void, ValueType> job)
 {
     auto cont = [job] (const List &values) mutable {
             auto error = QSharedPointer<KAsync::Error>::create();
             auto serialJob = KAsync::null<void>();
             for (const auto &value : values) {
                 serialJob = serialJob.then<void>([value, job, error](KAsync::Future<void> &future) {
                     job.template then<void>([&future, error] (const KAsync::Error &e) {
                         if (e && !*error) {
                             //TODO ideally we would aggregate the errors instead of just using the first one
                             *error = e;
                         }
                         future.setFinished();
                     })
                     .exec(value);
                 });
             }
             return serialJob
                 .then<void>([error](KAsync::Future<void> &future) {
                     if (*error) {
                         future.setError(*error);
                     } else {
                         future.setFinished();
                     }
                 });
         };
     return Job<void, List>(QSharedPointer<Private::Executor<void, List>>::create(
             Private::ContinuationHolder<void, List>(JobContinuation<void, List>(std::move(cont))), nullptr, Private::ExecutionFlag::GoodCase));
 }
 
 template<typename List, typename ValueType>
 Job<void, List> forEach(JobContinuation<void, ValueType> &&func)
 {
     return forEach<List, ValueType>(KAsync::start<void, ValueType>(std::forward<JobContinuation<void, ValueType>>(func)));
 }
 
 template<typename List, typename ValueType>
 Job<void, List> serialForEach(JobContinuation<void, ValueType> &&func)
 {
     return serialForEach<List, ValueType>(KAsync::start<void, ValueType>(std::forward<JobContinuation<void, ValueType>>(func)));
 }
 
 template<typename Out>
 Job<Out> null()
 {
     return KAsync::start<Out>(
         [](KAsync::Future<Out> &future) {
             future.setFinished();
         });
 }
 
 template<typename Out>
 Job<Out> value(Out v)
 {
     return KAsync::start<Out>(
         [val = std::move(v)](KAsync::Future<Out> &future) {
             future.setResult(val);
         });
 }
 
 template<typename Out>
 Job<Out> error(int errorCode, const QString &errorMessage)
 {
     return error<Out>({errorCode, errorMessage});
 }
 
 template<typename Out>
 Job<Out> error(const char *message)
 {
     return error<Out>(Error(message));
 }
 
 template<typename Out>
 Job<Out> error(const Error &error)
 {
     return KAsync::start<Out>(
         [error](KAsync::Future<Out> &future) {
             future.setError(error);
         });
 }
 
 inline Job<void> doWhile(const Job<ControlFlowFlag> &body)
 {
     return KAsync::start<void>([body] (KAsync::Future<void> &future) {
         auto job = body.then<void, ControlFlowFlag>([&future, body](const KAsync::Error &error, ControlFlowFlag flag) {
             if (error) {
                 future.setError(error);
                 future.setFinished();
             } else if (flag == ControlFlowFlag::Continue) {
                 doWhile(body).then<void>([&future](const KAsync::Error &error) {
                     if (error) {
                         future.setError(error);
                     }
                     future.setFinished();
                 }).exec();
             } else {
                 future.setFinished();
             }
         }).exec();
     });
 }
 
 inline Job<void> doWhile(const JobContinuation<ControlFlowFlag> &body)
 {
     return doWhile(KAsync::start<ControlFlowFlag>([body] {
         return body();
     }));
 }
 
 inline Job<void> wait(int delay)
 {
     return KAsync::start<void>([delay](KAsync::Future<void> &future) {
         QTimer::singleShot(delay, [&future]() {
             future.setFinished();
         });
     });
 }
 } // namespace KAsync
 
 //@endcond
 
 #endif // KASYNC_JOB_IMPL_H