corosession.hpp

Go to the documentation of this file.

/*------------------------------------------------------------------------------
    Copyright Butterfly Energy Systems 2014-2015, 2018, 2022.
    Distributed under the Boost Software License, Version 1.0.
    http://www.boost.org/LICENSE_1_0.txt
------------------------------------------------------------------------------*/
 
#ifndef CPPWAMP_CORO_COROSESSION_HPP
#define CPPWAMP_CORO_COROSESSION_HPP
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
 
#include <boost/asio/post.hpp>
#include <boost/asio/spawn.hpp>
#include "../asyncresult.hpp"
#include "../config.hpp"
#include "../session.hpp"
 
namespace wamp
{
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename TIgnoredBase = Session>
class CPPWAMP_DEPRECATED CoroSession: public Session
{
public:
    using Ptr = std::shared_ptr<CoroSession>;
 
    using Base = Session;
 
    using State = SessionState;
 
    using EventSlot = std::function<void (Event)>;
 
    using CallSlot = std::function<Outcome (Invocation)>;
 
    using InterruptSlot = std::function<Outcome (Interruption)>;
 
    template <typename TSpawnHandler>
    using YieldContext = boost::asio::basic_yield_context<TSpawnHandler>;
 
    static Ptr create(AnyIoExecutor exec, const Connector::Ptr& connector);
 
    static Ptr create(AnyIoExecutor exec, const ConnectorList& connectors);
 
    template <typename TExecutionContext>
    static CPPWAMP_ENABLED_TYPE(Ptr, isExecutionContext<TExecutionContext>())
    create(
        TExecutionContext& context, 
        const Connector::Ptr& connector 
        )
    {
        return create(context.get_executor(), connector);
    }
 
    template <typename TExecutionContext>
    static CPPWAMP_ENABLED_TYPE(Ptr, isExecutionContext<TExecutionContext>())
    create(
        TExecutionContext& context, 
        const ConnectorList& connectors) 
    {
        return create(context.get_executor(), connectors);
    }
 
 
    void connect(AsyncHandler<size_t> handler);
 
    template <typename H>
    size_t connect(YieldContext<H> yield, std::error_code* ec = nullptr);
 
    void join(Realm realm, AsyncHandler<SessionInfo> handler);
 
    template <typename H>
    SessionInfo join(Realm realm, YieldContext<H> yield,
                     std::error_code* ec = nullptr);
 
    void authenticate(Authentication auth);
 
    void leave(AsyncHandler<Reason> handler);
 
    template <typename H>
    Reason leave(YieldContext<H> yield, std::error_code* ec = nullptr);
 
    void leave(Reason reason, AsyncHandler<Reason> handler);
 
    template <typename H>
    Reason leave(Reason reason, YieldContext<H> yield,
                 std::error_code* ec = nullptr);
 
 
    void subscribe(Topic topic, EventSlot slot,
                   AsyncHandler<Subscription> handler);
 
    template <typename H>
    Subscription subscribe(Topic topic, EventSlot slot,
            YieldContext<H> yield, std::error_code* ec = nullptr);
 
    void unsubscribe(const Subscription& sub);
 
    void unsubscribe(const Subscription& sub, AsyncHandler<bool> handler);
 
    template <typename H>
    bool unsubscribe(const Subscription& sub, YieldContext<H> yield,
                     std::error_code* ec = nullptr);
 
    void publish(Pub pub);
 
    void publish(Pub pub, AsyncHandler<PublicationId> handler);
 
    template <typename H>
    PublicationId publish(Pub pub, YieldContext<H> yield,
                          std::error_code* ec = nullptr);
 
 
    void enroll(Procedure procedure, CallSlot slot,
                AsyncHandler<Registration> handler);
 
    template <typename H>
    Registration enroll(Procedure procedure, CallSlot slot,
            YieldContext<H> yield, std::error_code* ec = nullptr);
 
    void enroll(Procedure procedure, CallSlot callSlot,
                InterruptSlot interruptSlot,
                AsyncHandler<Registration> handler);
 
    template <typename H>
    Registration enroll(Procedure procedure, CallSlot slot,
                        InterruptSlot interruptSlot, YieldContext<H> yield,
                        std::error_code* ec = nullptr);
 
    void unregister(const Registration& reg);
 
    void unregister(const Registration& reg, AsyncHandler<bool> handler);
 
    template <typename H>
    bool unregister(const Registration& reg, YieldContext<H> yield,
                    std::error_code* ec = nullptr);
 
    RequestId call(Rpc procedure, AsyncHandler<Result> handler);
 
    template <typename H>
    Result call(Rpc rpc, YieldContext<H> yield, std::error_code* ec = nullptr);
 
 
    template <typename H>
    void suspend(YieldContext<H> yield);
 
 
    void reset() = delete;
 
protected:
    using Base::Base;
};
 
 
//******************************************************************************
// CoroSession implementation
//******************************************************************************
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
typename CoroSession<B>::Ptr CoroSession<B>::create(
    AnyIoExecutor exec,             
    const Connector::Ptr& connector 
    )
{
    return Ptr(new CoroSession(exec, {connector}));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
typename CoroSession<B>::Ptr CoroSession<B>::create(
    AnyIoExecutor exec,             
    const ConnectorList& connectors 
    )
{
    return Ptr(new CoroSession(exec, connectors));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::connect(
    AsyncHandler<size_t> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(state() == State::disconnected,
                        "Session is not disconnected");
    Base::connect(std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
size_t CoroSession<B>::connect(
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(state() == State::disconnected,
                        "Session is not disconnected");
    auto index = Base::connect(yield);
    if (!ec)
        return index.value();
    *ec = !index ? index.error() : std::error_code{};
    return index.value_or(0);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::join(
    Realm realm,                      
    AsyncHandler<SessionInfo> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::closed,
                        "Session is not closed");
    Base::join(std::move(realm), std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
SessionInfo CoroSession<B>::join(
    Realm realm,           
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::closed,
                        "Session is not closed");
    auto info = Base::join(std::move(realm), yield);
    if (!ec)
        return info.value();
    *ec = !info ? info.error() : std::error_code{};
    return info.value_or(SessionInfo{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::authenticate(Authentication auth)
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::authenticating,
                        "Session is not authenticating");
    Base::authenticate(std::move(auth));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::leave(
    AsyncHandler<Reason> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::leave(std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Reason CoroSession<B>::leave(
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto reason = Base::leave(yield);
    if (!ec)
        return reason.value();
    *ec = !reason ? reason.error() : std::error_code{};
    return reason.value_or(Reason{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::leave(
    Reason reason,               
    AsyncHandler<Reason> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::leave(std::move(reason), std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Reason CoroSession<B>::leave(
    Reason reason,         
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto result = Base::leave(std::move(reason), yield);
    if (!ec)
        return result.value();
    *ec = !result ? result.error() : std::error_code{};
    return result.value_or(Reason{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::subscribe(
    Topic topic,    
    EventSlot slot, 
    AsyncHandler<Subscription> handler 
)
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    using std::move;
    return Base::subscribe(move(topic), move(slot), move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Subscription CoroSession<B>::subscribe(
    Topic topic,           
    EventSlot slot,        
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto sub = Base::subscribe(std::move(topic), std::move(slot), yield);
    if (!ec)
        return sub.value();
    *ec = !sub ? sub.error() : std::error_code{};
    return sub.value_or(Subscription{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::unsubscribe(
    const Subscription& sub 
    )
{
    Base::unsubscribe(sub);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::unsubscribe(
    const Subscription& sub,   
    AsyncHandler<bool> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::unsubscribe(sub, std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
bool CoroSession<B>::unsubscribe(
    const Subscription& sub, 
    YieldContext<H> yield,   
    std::error_code* ec      
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto unsubscribed = Base::unsubscribe(sub, yield);
    if (!ec)
        return unsubscribed.value();
    *ec = !unsubscribed ? unsubscribed.error() : std::error_code{};
    return unsubscribed.value_or(false);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::publish(
    Pub pub 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::publish(pub);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::publish(
    Pub pub,                            
    AsyncHandler<PublicationId> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::publish(pub, std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
PublicationId CoroSession<B>::publish(
    Pub pub,                
    YieldContext<H> yield,  
    std::error_code* ec     
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto pubId = Base::publish(std::move(pub), yield);
    if (!ec)
        return pubId.value();
    *ec = !pubId ? pubId.error() : std::error_code{};
    return pubId.value_or(0);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::enroll(
    Procedure procedure, 
    CallSlot slot,       
    AsyncHandler<Registration> handler 
)
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::enroll(std::move(procedure), std::move(slot), std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Registration CoroSession<B>::enroll(
    Procedure procedure,   
    CallSlot slot,         
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto reg = Base::enroll(std::move(procedure), std::move(slot), yield);
    if (!ec)
        return reg.value();
    *ec = !reg ? reg.error() : std::error_code{};
    return reg.value_or(Registration{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::enroll(
    Procedure procedure, 
    CallSlot callSlot,   
    InterruptSlot interruptSlot, 
    AsyncHandler<Registration> handler 
)
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::enroll(std::move(procedure), std::move(callSlot),
                 std::move(interruptSlot), std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Registration CoroSession<B>::enroll(
    Procedure procedure,   
    CallSlot callSlot,     
    InterruptSlot interruptSlot, 
    YieldContext<H> yield, 
    std::error_code* ec    
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto reg = Base::enroll(std::move(procedure), std::move(callSlot),
                            std::move(interruptSlot), yield);
    if (!ec)
        return reg.value();
    *ec = !reg ? reg.error() : std::error_code{};
    return reg.value_or(Registration{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::unregister(
    const Registration& reg 
    )
{
    Base::unregister(reg);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
void CoroSession<B>::unregister(
    const Registration& reg,   
    AsyncHandler<bool> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    Base::unregister(reg, std::move(handler));
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
bool CoroSession<B>::unregister(
    const Registration& reg, 
    YieldContext<H> yield,   
    std::error_code* ec      
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto unregistered = Base::unregister(reg, yield);
    if (!ec)
        return unregistered.value();
    *ec = !unregistered ? unregistered.error() : std::error_code{};
    return unregistered.value_or(false);
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
RequestId CoroSession<B>::call(
    Rpc rpc,                     
    AsyncHandler<Result> handler 
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    CallChit chit;
    Base::call(std::move(rpc), chit, std::move(handler));
    return chit.requestId();
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
Result CoroSession<B>::call(
    Rpc rpc,                
    YieldContext<H> yield,  
    std::error_code* ec     
    )
{
    CPPWAMP_LOGIC_CHECK(this->state() == State::established,
                        "Session is not established");
    auto result = Base::call(std::move(rpc), yield);
    if (!ec)
        return result.value();
    *ec = !result ? result.error() : std::error_code{};
    return result.value_or(Result{});
}
 
//------------------------------------------------------------------------------
//------------------------------------------------------------------------------
template <typename B>
template <typename H>
void CoroSession<B>::suspend(YieldContext<H> yield)
{
    boost::asio::post(this->userIosvc(), yield);
}
 
} // namespace wamp
 
#endif // CPPWAMP_CORO_COROSESSION_HPP