FutureCoroutine.hpp

/*
** Copyright (C) 2022 CERN
**
** This software is provided 'as-is', without any express or implied
** warranty.  In no event will the authors be held liable for any damages
** arising from the use of this software.
**
** Permission is granted to anyone to use this software for any purpose,
** including commercial applications, and to alter it and redistribute it
** freely, subject to the following restrictions:
**
** 1. The origin of this software must not be misrepresented; you must not
**    claim that you wrote the original software. If you use this software
**    in a product, an acknowledgment in the product documentation would be
**    appreciated but is not required.
** 2. Altered source versions must be plainly marked as such, and must not be
**    misrepresented as being the original software.
** 3. This notice may not be removed or altered from any source distribution.
**
** Created on: 2022-07-26T16:50:02
** Author: Sylvain Fargier <sylvain.fargier@cern.ch>
*/
 
#ifndef FUTURECOROUTINE_HPP__
#define FUTURECOROUTINE_HPP__
 
#include <chrono>
#include <future>
 
#include <logger/Logger.hpp>
#include <oatpp/core/async/Coroutine.hpp>
 
#include "Units.hpp"
 
namespace smc {
namespace debug {
 
extern const std::chrono::milliseconds FutureCoroutineMaxInterval;
 
template<typename T>
class FutureCoroutine :
    public oatpp::async::CoroutineWithResult<FutureCoroutine<T>, const T &>
{
public:
    typedef oatpp::async::CoroutineWithResult<FutureCoroutine<T>, const T &>
        CoroutineWithResult;
 
    explicit FutureCoroutine(const std::shared_future<T> &future) :
        future{future}
    {}
 
    oatpp::async::Action act() override
    {
        if (!future.valid())
            return oatpp::async::AbstractCoroutine::error<oatpp::async::Error>(
                "failed");
        else if (future.wait_for(std::chrono::seconds(0)) ==
                 std::future_status::ready)
            return CoroutineWithResult::_return(future.get());
        else
        {
            if (delay < FutureCoroutineMaxInterval)
                delay = std::chrono::milliseconds(delay.count() << 1);
            return CoroutineWithResult::waitRepeat(delay);
        }
    }
 
    std::chrono::milliseconds delay{1};
    std::shared_future<T> future;
};
 
template<>
class FutureCoroutine<void> :
    public oatpp::async::Coroutine<FutureCoroutine<void>>
{
public:
    explicit FutureCoroutine(const std::shared_future<void> &future) :
        future{future}
    {}
 
    Action act() override
    {
        if (!future.valid())
            return error<Error>("failed");
        else if (future.wait_for(std::chrono::seconds(0)) ==
                 std::future_status::ready)
        {
            future.get(); // throw if needed
            return finish();
        }
        else
        {
            if (delay < FutureCoroutineMaxInterval)
                delay = std::chrono::milliseconds(delay.count() << 1);
            return waitRepeat(delay);
        }
    }
 
    std::chrono::milliseconds delay{1};
    std::shared_future<void> future;
};
 
template<typename T>
oatpp::async::CoroutineStarterForResult<const T &> wait(
    const std::shared_future<T> &future)
{
    return new FutureCoroutine<T>(future);
}
 
inline oatpp::async::CoroutineStarter wait(const std::shared_future<void> &future)
{
    return new FutureCoroutine<void>(future);
}
 
} // namespace debug
} // namespace smc
 
#endif