async.h

 
#pragma once
 
#include <functional>
#include <system/object_ext.h>
#include <system/action.h>
#include <system/threading/tasks/result_task.h>
#include <system/threading/tasks/task_status.h>
#include <system/details/dereference.h>
 
namespace System { namespace Details {
 
struct ASPOSECPP_SHARED_CLASS AsyncContextBase
{
    void Step(int32_t s) {stage = s;}
    ASPOSECPP_SHARED_API void Continue();
    template<typename T>
    bool Await(const T& awaitee, int32_t step)
    {
        return AwaitImpl(awaitee, step, [](const auto& awaiter) { awaiter.GetResult(); });
    }
    template <typename T, typename V>
    bool Await(const T& awaitee, V& value, int32_t step)
    {
        return AwaitImpl(awaitee, step, [&value](const auto& awaiter) { value = ExplicitCast<V>(awaiter.GetResult()); });
    }
 
    ASPOSECPP_SHARED_API void operator()();
 
    int32_t stage = 0;
 
protected:
    using Continuer = std::function<void()>;
 
    virtual Threading::Tasks::Task* get_Task() const = 0;
    virtual void Execute() = 0;
    template <typename AwaiteeT, typename ContinuerT>
    bool AwaitImpl(const AwaiteeT& awaitee, int32_t step, const ContinuerT& continuer)
    {
        stage = step;
 
        // Getting awaiter instance
        auto awaiter = Dereference(awaitee).GetAwaiter();
 
        // Trying to avoid sleeping if awaiter completed already
        if (awaiter.get_IsCompleted())
        {
            continuer(awaiter);
            return false;
        }
 
        // Going to suspend
        Suspend([awaiter, continuer]() { continuer(awaiter); });
 
        // Setting up resume context (we need to get it here because resuming should be executed in other context)
        auto continuationContext = GetResumeScheduler(awaiter.continueOnCapturedContext);
 
        // Setting up resume action
        awaiter.OnCompleted(Action<>([this, continuationContext] { Resume(continuationContext); }));
 
        return true;
    }
    ASPOSECPP_SHARED_API void Suspend(Continuer&& continuer);
    ASPOSECPP_SHARED_API Threading::Tasks::TaskScheduler* GetResumeScheduler(bool continueOnCapturedContext) const;
    ASPOSECPP_SHARED_API void Resume(Threading::Tasks::TaskScheduler* context);
 
    Continuer m_continuer;
    bool m_executing = false;
    Threading::Tasks::TaskScheduler* m_activator = nullptr;
};
 
using AsyncFunction = std::function<void(struct AsyncContext&)>;
 
struct AsyncContext : AsyncContextBase
{
    AsyncContext(Threading::Tasks::Task* task, const AsyncFunction& fnc) : m_task(task), m_function(fnc)
    {}
 
    ASPOSECPP_SHARED_API void Return();
 
protected:
    Threading::Tasks::Task* get_Task() const override {return m_task;}
    void Execute() override {m_function(*this);}
 
    Threading::Tasks::Task* m_task;
    AsyncFunction m_function;
};
 
template<typename T>
struct ResultAsyncContext;
 
template <typename T>
using ResultAsyncFunction = std::function<void(ResultAsyncContext<T>&)>;
 
template<typename T>
struct ResultAsyncContext : AsyncContextBase
{
    ResultAsyncContext(Threading::Tasks::ResultTask<T>* task, const ResultAsyncFunction<T>& fnc) : m_result_task(task), m_function(fnc)
    {}
 
    void Return(const T& result)
    {
        m_result_task->set_Result(result);
        m_result_task->set_Status(Threading::Tasks::TaskStatus::RanToCompletion);
    }
 
protected:
    Threading::Tasks::Task* get_Task() const override {return m_result_task;}
    void Execute() override {m_function(*this);}
 
    Threading::Tasks::ResultTask<T>* m_result_task;
    ResultAsyncFunction<T> m_function;
};
 
} // namespace Details
 
ASPOSECPP_SHARED_API TaskPtr MakeAsync(const Details::AsyncFunction& fnc);
 
template<typename T>
RTaskPtr<T> MakeAsync(const Details::ResultAsyncFunction<T>& fnc)
{
    auto result = System::MakeObject<Threading::Tasks::ResultTask<T>>();
    result->set_Function(Details::ResultAsyncContext<T>(result.GetPointer(), fnc));
    result->RunSynchronously();
    return result;
}
 
} // namespace System