corounpacker.hpp

Go to the documentation of this file.

/*------------------------------------------------------------------------------
    Copyright Butterfly Energy Systems 2014-2015, 2022.
    Distributed under the Boost Software License, Version 1.0.
    http://www.boost.org/LICENSE_1_0.txt
------------------------------------------------------------------------------*/
 
#ifndef CPPWAMP_COROUNPACKER_HPP
#define CPPWAMP_COROUNPACKER_HPP
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
 
#include <boost/asio/spawn.hpp>
#include "config.hpp"
#include "unpacker.hpp"
#include "internal/callee.hpp"
 
namespace wamp
{
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename... TArgs>
class CoroEventUnpacker
{
public:
    using Slot = TSlot;
 
    explicit CoroEventUnpacker(Slot slot);
 
    void operator()(Event event) const;
 
private:
    using Yield = boost::asio::yield_context;
 
    template <int ...S>
    void invoke(Event&& event, internal::IntegerSequence<S...>) const;
 
    Slot slot_;
};
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
CoroEventUnpacker<DecayedSlot<TSlot>, TArgs...> unpackedCoroEvent(TSlot&& slot);
 
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename... TArgs>
class SimpleCoroEventUnpacker
{
public:
    using Slot = TSlot;
 
    explicit SimpleCoroEventUnpacker(Slot slot);
 
    void operator()(Event event) const;
 
private:
    using Yield = boost::asio::yield_context;
 
    template <int ...S>
    void invoke(Event&& event, internal::IntegerSequence<S...>) const;
 
    Slot slot_;
};
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename... TArgs>
using BasicCoroEventUnpacker CPPWAMP_DEPRECATED
    = SimpleCoroEventUnpacker<TSlot, TArgs...>;
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>
simpleCoroEvent(TSlot&& slot);
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>
basicCoroEvent(TSlot&& slot);
 
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename... TArgs>
class CoroInvocationUnpacker
{
public:
    using Slot = TSlot;
 
    explicit CoroInvocationUnpacker(Slot slot);
 
    Outcome operator()(Invocation inv) const;
 
private:
    using Yield = boost::asio::yield_context;
 
    template <int ...S>
    void invoke(Invocation&& inv, internal::IntegerSequence<S...>) const;
 
    Slot slot_;
};
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
CoroInvocationUnpacker<DecayedSlot<TSlot>, TArgs...>
unpackedCoroRpc(TSlot&& slot);
 
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename TResult, typename... TArgs>
class SimpleCoroInvocationUnpacker
{
public:
    using Slot = TSlot;
 
    using ResultType = TResult;
 
    explicit SimpleCoroInvocationUnpacker(Slot slot);
 
    Outcome operator()(Invocation inv) const;
 
private:
    using Yield = boost::asio::yield_context;
 
    template <int ...S>
    void invoke(TrueType, Invocation&& inv,
                internal::IntegerSequence<S...>) const;
 
    template <int ...S>
    void invoke(FalseType, Invocation&& inv,
                internal::IntegerSequence<S...>) const;
 
    Slot slot_;
};
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TSlot, typename TResult, typename... TArgs>
using BasicCoroInvocationUnpacker =
    SimpleCoroInvocationUnpacker<TSlot, TResult, TArgs...>;
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TResult, typename... TArgs, typename TSlot>
SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>
simpleCoroRpc(TSlot&& slot);
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TResult, typename... TArgs, typename TSlot>
SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>
basicCoroRpc(TSlot&& slot);
 
//******************************************************************************
// Internal helper types
//******************************************************************************
 
namespace internal
{
 
//------------------------------------------------------------------------------
// This is caught internally by Client while dispatching RPCs and is never
// propagated through the API.
//------------------------------------------------------------------------------
struct UnpackCoroError : public Error
{
    UnpackCoroError() : Error("wamp.error.invalid_argument") {}
};
 
} // namespace internal
 
 
//******************************************************************************
// CoroEventUnpacker implementation
//******************************************************************************
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
CoroEventUnpacker<S,A...>::CoroEventUnpacker(Slot slot)
    : slot_(std::move(slot))
{}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
void CoroEventUnpacker<S,A...>::operator()(Event event) const
{
    if (event.args().size() < sizeof...(A))
    {
        std::ostringstream oss;
        oss << "Expected " << sizeof...(A)
            << " args, but only got " << event.args().size();
        throw internal::UnpackCoroError().withArgs(oss.str());
    }
 
    // Use the integer parameter pack technique shown in
    // http://stackoverflow.com/a/7858971/245265
    using Seq = typename internal::GenIntegerSequence<sizeof...(A)>::type;
    invoke(std::move(event), Seq());
}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
template <int ...Seq>
void CoroEventUnpacker<S,A...>::invoke(Event&& event,
                                       internal::IntegerSequence<Seq...>) const
{
    struct Spawned
    {
        Slot slot;
        Event event;
 
        void operator()(Yield yield)
        {
            std::tuple<ValueTypeOf<A>...> args;
            try
            {
                event.convertToTuple(args);
                slot(std::move(event), std::get<Seq>(std::move(args))...,
                     yield);
            }
            catch (const error::BadType&)
            {
                /*  Do nothing. This is to prevent the publisher crashing
                subscribers when it passes Variant objects having
                incorrect schema. */
            }
        }
    };
 
    auto executor =
        boost::asio::get_associated_executor(slot_, event.executor());
    boost::asio::spawn(executor, Spawned{slot_, std::move(event)});
}
 
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
CoroEventUnpacker<DecayedSlot<TSlot>, TArgs...> unpackedCoroEvent(TSlot&& slot)
{
    return CoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>(
        std::forward<TSlot>(slot));
}
 
 
//******************************************************************************
// SimpleCoroEventUnpacker implementation
//******************************************************************************
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
SimpleCoroEventUnpacker<S,A...>::SimpleCoroEventUnpacker(Slot slot)
    : slot_(std::move(slot))
{}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
void SimpleCoroEventUnpacker<S,A...>::operator()(Event event) const
{
    if (event.args().size() < sizeof...(A))
    {
        std::ostringstream oss;
        oss << "Expected " << sizeof...(A)
            << " args, but only got " << event.args().size();
        throw internal::UnpackCoroError().withArgs(oss.str());
    }
 
    // Use the integer parameter pack technique shown in
    // http://stackoverflow.com/a/7858971/245265
    using Seq = typename internal::GenIntegerSequence<sizeof...(A)>::type;
    invoke(std::move(event), Seq());
}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
template <int ...Seq>
void
SimpleCoroEventUnpacker<S,A...>::invoke(Event&& event,
                                       internal::IntegerSequence<Seq...>) const
{
    struct Spawned
    {
        Slot slot;
        Event event;
 
        void operator()(Yield yield)
        {
            std::tuple<ValueTypeOf<A>...> args;
            try
            {
                event.convertToTuple(args);
                slot(std::get<Seq>(std::move(args))..., yield);
            }
            catch (const error::BadType&)
            {
                /* Do nothing. This is to prevent the publisher crashing
                   subscribers when it passes Variant objects having
                   incorrect schema. */
            }
        }
    };
 
    auto executor =
        boost::asio::get_associated_executor(slot_, event.executor());
    boost::asio::spawn(executor, Spawned{slot_, std::move(event)});
}
 
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>
simpleCoroEvent(TSlot&& slot)
{
    return SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>(
        std::forward<TSlot>(slot));
}
 
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>
basicCoroEvent(TSlot&& slot)
{
    return SimpleCoroEventUnpacker<DecayedSlot<TSlot>, TArgs...>(
        std::forward<TSlot>(slot));
}
 
 
//******************************************************************************
// CoroInvocationUnpacker implementation
//******************************************************************************
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
CoroInvocationUnpacker<S,A...>::CoroInvocationUnpacker(Slot slot)
    : slot_(std::move(slot))
{}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
Outcome CoroInvocationUnpacker<S,A...>::operator()(Invocation inv) const
{
    if (inv.args().size() < sizeof...(A))
    {
        std::ostringstream oss;
        oss << "Expected " << sizeof...(A)
            << " args, but only got " << inv.args().size();
        throw internal::UnpackCoroError().withArgs(oss.str());
    }
 
    // Use the integer parameter pack technique shown in
    // http://stackoverflow.com/a/7858971/245265
    using Seq = typename internal::GenIntegerSequence<sizeof...(A)>::type;
    invoke(std::move(inv), Seq());
 
    return deferment;
}
 
//------------------------------------------------------------------------------
template <typename S, typename... A>
template <int ...Seq>
void
CoroInvocationUnpacker<S,A...>::invoke(Invocation&& inv,
                                       internal::IntegerSequence<Seq...>) const
{
    class Spawned
    {
    public:
        Spawned(const Slot& slot, Invocation&& inv)
            : slot_(slot),
              calleePtr_(inv.callee_),
              reqId_(inv.requestId()),
              inv_(std::move(inv))
        {}
 
        void operator()(Yield yield)
        {
            try
            {
                std::tuple<ValueTypeOf<A>...> args;
                inv_.convertToTuple(args);
                Outcome outcome = slot_(std::move(inv_),
                                        std::get<Seq>(std::move(args))...,
                                        yield);
 
                switch (outcome.type())
                {
                case Outcome::Type::deferred:
                    // Do nothing
                    break;
 
                case Outcome::Type::result:
                    safeYield(std::move(outcome).asResult());
                    break;
 
                case Outcome::Type::error:
                    safeYield(std::move(outcome).asError());
                    break;
 
                default:
                    assert(false && "Unexpected wamp::Outcome::Type enumerator");
                }
 
            }
            catch (Error e)
            {
                safeYield(std::move(e));
            }
            catch (const error::BadType& e)
            {
                // Forward Variant conversion exceptions as ERROR messages.
                safeYield(Error(e));
            }
        }
 
    private:
        void safeYield(Result&& result)
        {
            auto callee = calleePtr_.lock();
            if (callee)
                callee->safeYield(reqId_, std::move(result));
        }
 
        void safeYield(Error&& error)
        {
            auto callee = calleePtr_.lock();
            if (callee)
                callee->safeYield(reqId_, std::move(error));
        }
 
        Slot slot_;
        std::weak_ptr<internal::Callee> calleePtr_;
        RequestId reqId_;
        Invocation inv_;
    };
 
    auto executor =
        boost::asio::get_associated_executor(slot_, inv.executor());
    boost::asio::spawn(executor, Spawned{slot_, std::move(inv)});
}
 
//------------------------------------------------------------------------------
template <typename... TArgs, typename TSlot>
CoroInvocationUnpacker<DecayedSlot<TSlot>, TArgs...>
unpackedCoroRpc(TSlot&& slot)
{
    return CoroInvocationUnpacker<DecayedSlot<TSlot>, TArgs...>(
        std::forward<TSlot>(slot) );
}
 
//******************************************************************************
// SimpleCoroInvocationUnpacker implementation
//******************************************************************************
 
//------------------------------------------------------------------------------
template <typename S, typename R, typename... A>
SimpleCoroInvocationUnpacker<S,R,A...>::SimpleCoroInvocationUnpacker(Slot slot)
    : slot_(std::move(slot))
{}
 
//------------------------------------------------------------------------------
template <typename S, typename R, typename... A>
Outcome
SimpleCoroInvocationUnpacker<S,R,A...>::operator()(Invocation inv) const
{
    if (inv.args().size() < sizeof...(A))
    {
        std::ostringstream oss;
        oss << "Expected " << sizeof...(A)
            << " args, but only got " << inv.args().size();
        throw internal::UnpackCoroError().withArgs(oss.str());
    }
 
    // Use the integer parameter pack technique shown in
    // http://stackoverflow.com/a/7858971/245265
    using Seq = typename internal::GenIntegerSequence<sizeof...(A)>::type;
    using IsVoidResult = BoolConstant<std::is_same<ResultType, void>::value>;
    invoke(IsVoidResult{}, std::move(inv), Seq());
 
    return deferment;
}
 
//------------------------------------------------------------------------------
template <typename S, typename R, typename... A>
template <int ...Seq>
void SimpleCoroInvocationUnpacker<S,R,A...>::invoke(
    TrueType, Invocation&& inv, internal::IntegerSequence<Seq...>) const
{
    struct Spawned
    {
        Slot slot;
        Invocation inv;
 
        void operator()(Yield yield)
        {
            try
            {
                std::tuple<ValueTypeOf<A>...> args;
                inv.convertToTuple(args);
                slot(std::get<Seq>(std::move(args))..., yield);
                inv.yield();
            }
            catch (const Error& e)
            {
                inv.yield(e);
            }
            catch (const error::BadType& e)
            {
                // Forward Variant conversion exceptions as ERROR messages.
                inv.yield(Error(e));
            }
        }
    };
 
    auto executor =
        boost::asio::get_associated_executor(slot_, inv.executor());
    boost::asio::spawn(executor, Spawned{slot_, std::move(inv)});
}
 
//------------------------------------------------------------------------------
template <typename S, typename R, typename... A>
template <int ...Seq>
void SimpleCoroInvocationUnpacker<S,R,A...>::invoke(
    FalseType, Invocation&& inv, internal::IntegerSequence<Seq...>) const
{
    struct Spawned
    {
        Slot slot;
        Invocation inv;
 
        void operator()(Yield yield)
        {
            try
            {
                std::tuple<ValueTypeOf<A>...> args;
                inv.convertToTuple(args);
                ResultType result = slot(std::get<Seq>(std::move(args))...,
                                         yield);
                inv.yield(Result().withArgs(std::move(result)));
            }
            catch (const Error& e)
            {
                inv.yield(e);
            }
            catch (const error::BadType& e)
            {
                // Forward Variant conversion exceptions as ERROR messages.
                inv.yield(Error(e));
            }
        }
    };
 
    auto executor = inv.executor();
    boost::asio::spawn(executor, Spawned{slot_, std::move(inv)});
}
 
//------------------------------------------------------------------------------
template <typename TResult, typename... TArgs, typename TSlot>
SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>
simpleCoroRpc(TSlot&& slot)
{
    return SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>(
        std::forward<TSlot>(slot) );
}
 
//------------------------------------------------------------------------------
template <typename TResult, typename... TArgs, typename TSlot>
SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>
basicCoroRpc(TSlot&& slot)
{
    return SimpleCoroInvocationUnpacker<DecayedSlot<TSlot>, TResult, TArgs...>(
        std::forward<TSlot>(slot) );
}
 
} // namespace wamp
 
#endif // CPPWAMP_COROUNPACKER_HPP