TaskGenerator.h

Go to the documentation of this file.

 
#include <RobotRaconteur/Generator.h>
 
#include "com__robotraconteur__action.h"
 
#pragma once
 
namespace RobotRaconteur
{
namespace Companion
{
namespace Util
{
template <typename StatusType>
class AsyncTaskGenerator : public virtual RobotRaconteur::Generator<RR_INTRUSIVE_PTR<StatusType>, void>,
                           public RR_ENABLE_SHARED_FROM_THIS<AsyncTaskGenerator<StatusType> >
{
  public:
    AsyncTaskGenerator(const RR_SHARED_PTR<RobotRaconteur::RobotRaconteurNode>& node, int32_t next_timeout,
                       int32_t watchdog_timeout)
        : node(node), next_timeout(next_timeout), watchdog_timeout(watchdog_timeout)
    {
        started = false;
        closed = false;
        aborted = false;
        completed = false;
        task_completed = false;
        send_update = false;
        if (next_timeout < 0)
        {
            throw RobotRaconteur::InvalidArgumentException("next_timeout must be >= 0");
        }
 
        if (watchdog_timeout > 0 && watchdog_timeout < next_timeout)
        {
            throw RobotRaconteur::InvalidArgumentException("watchdog_timeout must be >= next_timeout");
        }
    }
 
    RR_OVIRTUAL void AsyncNext(
        boost::function<void(const RR_INTRUSIVE_PTR<StatusType>&, const RobotRaconteur::RobotRaconteurExceptionPtr&)>
            handler,
        int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(timeout);
        boost::mutex::scoped_lock lock(this_lock);
        if (watchdog_timer)
        {
            try
            {
                watchdog_timer->Stop();
            }
            catch (std::exception&)
            {}
            watchdog_timer.reset();
        }
        if (aborted)
        {
            throw RobotRaconteur::OperationAbortedException("Operation was aborted");
        }
        if ((closed && !started) || completed)
        {
            throw RobotRaconteur::StopIterationException("");
        }
 
        if (!started)
        {
            StartTask();
            started = true;
            RR_INTRUSIVE_PTR<StatusType> ret(new StatusType());
            ret->action_status = com::robotraconteur::action::ActionStatusCode::running;
            FillStatus(ret);
            lock.unlock();
            handler(ret, nullptr);
            return;
        }
 
        if (next_handler)
        {
            throw RobotRaconteur::InvalidOperationException("Next call already in progress");
        }
 
        if (task_completed)
        {
            complete_gen(handler);
            return;
        }
 
        if (send_update)
        {
            send_update = false;
            RR_INTRUSIVE_PTR<StatusType> ret(new StatusType());
            ret->action_status = com::robotraconteur::action::ActionStatusCode::running;
            FillStatus(ret);
            lock.unlock();
            handler(ret, nullptr);
            return;
        }
 
        next_handler = handler;
 
        RR_WEAK_PTR<AsyncTaskGenerator<StatusType> > weak_this = this->shared_from_this();
        next_timer = RobotRaconteur::RobotRaconteurNode::s()->CreateTimer(
            boost::posix_time::milliseconds(next_timeout),
            boost::bind(&AsyncTaskGenerator::next_timer_handler, weak_this, RR_BOOST_PLACEHOLDERS(_1)), true);
        next_timer->Start();
        if (watchdog_timeout > 0)
        {
            watchdog_timer = RobotRaconteur::RobotRaconteurNode::s()->CreateTimer(
                boost::posix_time::milliseconds(watchdog_timeout),
                boost::bind(&AsyncTaskGenerator::watchdog_timer_handler, weak_this, RR_BOOST_PLACEHOLDERS(_1)), true);
            watchdog_timer->Start();
        }
    }
 
    RR_OVIRTUAL void AsyncClose(boost::function<void(const RobotRaconteur::RobotRaconteurExceptionPtr& err)> handler,
                                int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(timeout);
        boost::mutex::scoped_lock lock(this_lock);
        if (closed || aborted)
        {
            lock.unlock();
            handler(nullptr);
            return;
        }
        closed = true;
        if (started)
        {
            CloseRequested();
        }
        lock.unlock();
        handler(nullptr);
    }
 
    RR_OVIRTUAL void AsyncAbort(boost::function<void(const RobotRaconteur::RobotRaconteurExceptionPtr& err)> handler,
                                int32_t timeout = RR_TIMEOUT_INFINITE) RR_OVERRIDE
    {
        RR_UNUSED(timeout);
        boost::mutex::scoped_lock lock(this_lock);
        if (closed || aborted)
        {
            lock.unlock();
            handler(nullptr);
            return;
        }
        aborted = true;
        if (started)
        {
            AbortRequested();
        }
        lock.unlock();
        handler(nullptr);
    }
 
    RR_OVIRTUAL RR_INTRUSIVE_PTR<StatusType> Next() RR_OVERRIDE
    {
        // This is never called
        throw RobotRaconteur::NotImplementedException("Not implemented");
    }
    RR_OVIRTUAL void Close() RR_OVERRIDE
    {
        // This is never called
        throw RobotRaconteur::NotImplementedException("Not implemented");
    }
    RR_OVIRTUAL void Abort() RR_OVERRIDE
    {
        // This is never called
        throw RobotRaconteur::NotImplementedException("Not implemented");
    }
 
  protected:
    virtual void SetResult(RR_INTRUSIVE_PTR<StatusType>& result)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (task_completed)
            return;
        this->result = result;
        do_result();
    }

    virtual void SetResultException(RR_SHARED_PTR<RobotRaconteur::RobotRaconteurException>& exp)
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (task_completed)
            return;
        this->exception_result = exp;
        do_result();
    }
 
    void do_result()
    {
        task_completed = true;
        auto h = next_handler;
        next_handler.clear();
        if (h)
        {
            try
            {
                next_timer->Stop();
            }
            catch (std::exception&)
            {}
 
            complete_gen(h);
            return;
        }
    }
 
    void complete_gen(
        boost::function<void(const RR_INTRUSIVE_PTR<StatusType>&, const RobotRaconteur::RobotRaconteurExceptionPtr&)>
            handler)
    {
        completed = true;
 
        if (exception_result)
        {
            auto exp = exception_result;
            exception_result.reset();
            handler(nullptr, exp);
            return;
        }
        RR_INTRUSIVE_PTR<StatusType> ret = result;
        result.reset();
        handler(ret, nullptr);
    }

    void SendUpdate()
    {
        boost::mutex::scoped_lock lock(this_lock);
        if (task_completed || aborted || closed)
            return;
        if (!next_handler)
        {
            send_update = true;
            return;
        }
        try
        {
            next_timer->Stop();
        }
        catch (std::exception&)
        {}
        do_send_update();
    }

    virtual void StartTask() = 0;

    virtual void CloseRequested() {}

    virtual void AbortRequested() {}

    virtual void FillStatus(RR_INTRUSIVE_PTR<StatusType>& status) {}
 
    static void next_timer_handler(RR_WEAK_PTR<AsyncTaskGenerator> weak_this, const RobotRaconteur::TimerEvent& evt)
    {
        RR_UNUSED(evt);
        RR_SHARED_PTR<AsyncTaskGenerator> shared_this = weak_this.lock();
        if (!shared_this)
            return;
        boost::mutex::scoped_lock lock(shared_this->this_lock);
        shared_this->do_send_update();
    }
 
    void do_send_update()
    {
        auto h = next_handler;
        next_handler.clear();
        if (h)
        {
            RR_INTRUSIVE_PTR<StatusType> ret(new StatusType());
            ret->action_status = com::robotraconteur::action::ActionStatusCode::running;
            FillStatus(ret);
            h(ret, nullptr);
            return;
        }
    }
 
    static void watchdog_timer_handler(RR_WEAK_PTR<AsyncTaskGenerator> weak_this, const RobotRaconteur::TimerEvent& evt)
    {
        RR_SHARED_PTR<AsyncTaskGenerator> shared_this = weak_this.lock();
        if (!shared_this)
            return;
        if (evt.stopped)
            return;
        boost::mutex::scoped_lock lock(shared_this->this_lock);
        if (shared_this->task_completed)
            return;
        shared_this->aborted = true;
        shared_this->AbortRequested();
    }
 
    RR_WEAK_PTR<RobotRaconteur::RobotRaconteurNode> node;
    RR_SHARED_PTR<RobotRaconteur::Timer> next_timer;
    RR_SHARED_PTR<RobotRaconteur::Timer> watchdog_timer;
    bool started;
    bool closed;
    bool aborted;
    bool completed;
    bool task_completed;
    bool send_update;
    int32_t next_timeout;
    int32_t watchdog_timeout;
    boost::mutex this_lock;
    boost::function<void(const RR_INTRUSIVE_PTR<StatusType>&, const RobotRaconteur::RobotRaconteurExceptionPtr&)>
        next_handler;
    RR_INTRUSIVE_PTR<StatusType> result;
    RR_SHARED_PTR<RobotRaconteurException> exception_result;
};

template <typename StatusType>
class SyncTaskGenerator : public AsyncTaskGenerator<StatusType>
{
  public:
    SyncTaskGenerator(RR_SHARED_PTR<RobotRaconteur::RobotRaconteurNode> node, int32_t next_timeout,
                      int32_t watchdog_timeout)
        : AsyncTaskGenerator<StatusType>(node, next_timeout, watchdog_timeout)
    {}
 
  protected:
    virtual RR_INTRUSIVE_PTR<StatusType> RunTask() = 0;
 
    void run_task_thread()
    {
        try
        {
            RR_INTRUSIVE_PTR<StatusType> ret = RunTask();
            this->SetResult(ret);
        }
        catch (std::exception& exp)
        {
            RR_SHARED_PTR<RobotRaconteur::RobotRaconteurException> exp1 =
                RobotRaconteur::RobotRaconteurExceptionUtil::ExceptionToSharedPtr(exp);
            this->SetResultException(exp1);
        }
    }
 
    RR_OVIRTUAL void StartTask() RR_OVERRIDE
    {
        RR_SHARED_PTR<SyncTaskGenerator> this_ = RR_DYNAMIC_POINTER_CAST<SyncTaskGenerator>(this->shared_from_this());
        boost::thread(boost::bind(&SyncTaskGenerator::run_task_thread, this_));
    }
};
} // namespace Util
} // namespace Companion
} // namespace RobotRaconteur