autobahn-cpp-dev 17.5.1+git7cc5d37-2 / usr / include / autobahn / wamp_session.ipp

///////////////////////////////////////////////////////////////////////////////
//
// Copyright (c) Crossbar.io Technologies GmbH and contributors
//
// Boost Software License - Version 1.0 - August 17th, 2003
//
// Permission is hereby granted, free of charge, to any person or organization
// obtaining a copy of the software and accompanying documentation covered by
// this license (the "Software") to use, reproduce, display, distribute,
// execute, and transmit the Software, and to prepare derivative works of the
// Software, and to permit third-parties to whom the Software is furnished to
// do so, all subject to the following:
//
// The copyright notices in the Software and this entire statement, including
// the above license grant, this restriction and the following disclaimer,
// must be included in all copies of the Software, in whole or in part, and
// all derivative works of the Software, unless such copies or derivative
// works are solely in the form of machine-executable object code generated by
// a source language processor.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
// SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
// FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//
///////////////////////////////////////////////////////////////////////////////

#include "exceptions.hpp"
#include "wamp_call.hpp"
#include "wamp_event.hpp"
#include "wamp_invocation.hpp"
#include "wamp_message.hpp"
#include "wamp_message_type.hpp"
#include "wamp_publication.hpp"
#include "wamp_registration.hpp"
#include "wamp_register_request.hpp"
#include "wamp_subscribe_request.hpp"
#include "wamp_subscription.hpp"
#include "wamp_transport.hpp"
#include "wamp_unregister_request.hpp"
#include "wamp_unsubscribe_request.hpp"
#include "wamp_authenticate.hpp"
#include "wamp_challenge.hpp"

#include "wamp_auth_utils.hpp"

#if !(defined(_WIN32) || defined(WIN32))
#include <arpa/inet.h>
#include <unistd.h>
#endif

#include <boost/system/error_code.hpp>
#include <cstdint>
#include <exception>
#include <iostream>
#include <sstream>
#include <stdlib.h>

namespace autobahn {

inline wamp_session::wamp_session(
        boost::asio::io_service& io_service,
        bool debug_enabled)
    : m_debug_enabled(debug_enabled)
    , m_io_service(io_service)
    , m_transport()
    , m_request_id(0)
    , m_session_id(0)
    , m_goodbye_sent(false)
    , m_running(false)
{
}

inline wamp_session::~wamp_session()
{
}

inline boost::future<void> wamp_session::start()
{
    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        if (m_running) {
            m_session_start.set_exception(boost::copy_exception(protocol_error("session already started")));
            return;
        }

        if (!m_transport) {
            m_session_start.set_exception(boost::copy_exception(no_transport_error()));
            return;
        }

        m_running = true;
        m_session_start.set_value();
    });

    return m_session_start.get_future();
}

inline boost::future<void> wamp_session::stop()
{
    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        if (!m_running) {
            m_session_stop.set_exception(boost::copy_exception(protocol_error("session already stopped")));
            return;
        }

        if (!m_transport) {
            m_session_start.set_exception(boost::copy_exception(no_transport_error()));
            return;
        }

        if (m_session_id) {
            m_session_stop.set_exception(boost::copy_exception(protocol_error("session still joined")));
            return;
        }

        m_running = false;
        m_session_stop.set_value();
    });

    return m_session_stop.get_future();
}

inline boost::future<uint64_t> wamp_session::join(
        const std::string& realm,
        const std::vector<std::string>& authentication_methods,
        const std::string& authentication_id)
{
    msgpack::zone zone;
    std::unordered_map<std::string, msgpack::object> roles;

    std::unordered_map<std::string, bool> caller_features;
    caller_features["call_timeout"] = true;
    std::unordered_map<std::string, msgpack::object> caller;
    caller["features"] = msgpack::object(caller_features, zone);
    roles["caller"] = msgpack::object(caller, zone);

    std::unordered_map<std::string, bool> callee_features;
    callee_features["call_timeout"] = true;
    std::unordered_map<std::string, msgpack::object> callee;
    callee["features"] = msgpack::object(callee_features, zone);
    roles["callee"] = msgpack::object(callee, zone);

    std::unordered_map<std::string, msgpack::object> publisher;
    roles["publisher"] = msgpack::object(publisher, zone);

    std::unordered_map<std::string, msgpack::object> subscriber;
    roles["subscriber"] = msgpack::object(subscriber, zone);

    std::unordered_map<std::string, msgpack::object> details;
    details["roles"] = msgpack::object(roles, zone);
    details["authmethods"] = msgpack::object(authentication_methods, zone);
    details["authid"] = msgpack::object(authentication_id, zone);

    auto message = std::make_shared<wamp_message>(3, std::move(zone));
    message->set_field(0, static_cast<int>(message_type::HELLO));
    message->set_field(1, realm);
    message->set_field(2, details);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        if (m_session_id) {
            m_session_join.set_exception(boost::copy_exception(protocol_error("session already joined")));
            return;
        }

        try {
            send_message(std::move(*message), false);
        } catch (const std::exception& e) {
            m_session_join.set_exception(boost::copy_exception(e));
        }
    });

    return m_session_join.get_future();
}

inline boost::future<std::string> wamp_session::leave(const std::string& reason)
{
    auto message = std::make_shared<wamp_message>(3);
    message->set_field(0, static_cast<int>(message_type::GOODBYE));
    message->set_field(1, std::unordered_map<int, int>() /* No Details */);
    message->set_field(2, reason);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        if (m_goodbye_sent) {
            m_session_leave.set_exception(boost::copy_exception(protocol_error("goodbye already sent")));
        }

        try {
            send_message(std::move(*message), false);
            m_goodbye_sent = true;
        } catch (const std::exception& e) {
            m_session_leave.set_exception(boost::copy_exception(e));
        }

        m_session_id = 0;
    });

    return m_session_leave.get_future();
}

inline boost::future<void> wamp_session::publish(const std::string& topic,const wamp_publish_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(4);
    message->set_field(0, static_cast<int>(message_type::PUBLISH));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, topic);

    auto result = std::make_shared<boost::promise<void>>();
    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            result->set_value();
        } catch (const std::exception& e) {
            result->set_exception(boost::copy_exception(e));
        }
    });

    return result->get_future();
}

template <typename List>
inline boost::future<void> wamp_session::publish(const std::string& topic, const List& arguments,const wamp_publish_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(5);
    message->set_field(0, static_cast<int>(message_type::PUBLISH));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, topic);
    message->set_field(4, arguments);

    auto result = std::make_shared<boost::promise<void>>();
    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            result->set_value();
        } catch (const std::exception& e) {
            result->set_exception(boost::copy_exception(e));
        }
    });

    return result->get_future();
}

template <typename List, typename Map>
inline boost::future<void> wamp_session::publish(
        const std::string& topic, const List& arguments, const Map& kw_arguments,const wamp_publish_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(6);
    message->set_field(0, static_cast<int>(message_type::PUBLISH));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, topic);
    message->set_field(4, arguments);
    message->set_field(5, kw_arguments);

    auto result = std::make_shared<boost::promise<void>>();
    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            result->set_value();
        } catch (const std::exception& e) {
            result->set_exception(boost::copy_exception(e));
        }
    });

    return result->get_future();
}

inline boost::future<wamp_subscription> wamp_session::subscribe(
        const std::string& topic,
        const wamp_event_handler& handler,
        const wamp_subscribe_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(4);
    message->set_field(0, static_cast<int>(message_type::SUBSCRIBE));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, topic);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto subscribe_request = std::make_shared<wamp_subscribe_request>(handler);

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_subscribe_requests.emplace(request_id, subscribe_request);
        } catch (const std::exception& e) {
            subscribe_request->response().set_exception(boost::copy_exception(e));
        }
    });

    return subscribe_request->response().get_future();
}

inline boost::future<void> wamp_session::unsubscribe(const wamp_subscription& subscription)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(3);
    message->set_field(0, static_cast<int>(message_type::UNSUBSCRIBE));
    message->set_field(1, request_id);
    message->set_field(2, subscription.id());

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto unsubscribe_request = std::make_shared<wamp_unsubscribe_request>(subscription);

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_unsubscribe_requests.emplace(request_id, unsubscribe_request);
        } catch (const std::exception& e) {
            unsubscribe_request->response().set_exception(boost::copy_exception(e));
        }
    });

    return unsubscribe_request->response().get_future();
}

inline boost::future<wamp_call_result> wamp_session::call(
        const std::string& procedure,
        const wamp_call_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(4);
    message->set_field(0, static_cast<int>(message_type::CALL));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, procedure);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto call = std::make_shared<wamp_call>();

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_calls.emplace(request_id, call);
        } catch (const std::exception& e) {
            call->result().set_exception(boost::copy_exception(e));
        }
    });

    return call->result().get_future();
}

template<typename List>
inline boost::future<wamp_call_result> wamp_session::call(
        const std::string& procedure,
        const List& arguments,
        const wamp_call_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(5);
    message->set_field(0, static_cast<int>(message_type::CALL));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, procedure);
    message->set_field(4, arguments);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto call = std::make_shared<wamp_call>();

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_calls.emplace(request_id, call);
        } catch (const std::exception& e) {
            call->result().set_exception(boost::copy_exception(e));
        }
    });

    return call->result().get_future();
}

template<typename List, typename Map>
inline boost::future<wamp_call_result> wamp_session::call(
        const std::string& procedure,
        const List& arguments,
        const Map& kw_arguments,
        const wamp_call_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(6);
    message->set_field(0, static_cast<int>(message_type::CALL));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, procedure);
    message->set_field(4, arguments);
    message->set_field(5, kw_arguments);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto call = std::make_shared<wamp_call>();

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_calls.emplace(request_id, call);
        } catch (const std::exception& e) {
            call->result().set_exception(boost::copy_exception(e));
        }
    });

    return call->result().get_future();
}

inline boost::future<wamp_registration> wamp_session::provide(
        const std::string& name,
        const wamp_procedure& procedure,
        const provide_options& options)
{
    uint64_t request_id = ++m_request_id;

    auto message = std::make_shared<wamp_message>(4);
    message->set_field(0, static_cast<int>(message_type::REGISTER));
    message->set_field(1, request_id);
    message->set_field(2, options);
    message->set_field(3, name);

    auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
    auto register_request = std::make_shared<wamp_register_request>(procedure);

    m_io_service.dispatch([=]() {
        auto shared_self = weak_self.lock();
        if (!shared_self) {
            return;
        }

        try {
            send_message(std::move(*message));
            m_register_requests.emplace(request_id, register_request);
        } catch (const std::exception& e) {
            register_request->response().set_exception(boost::copy_exception(e));
        }
    });

    return register_request->response().get_future();
}

inline boost::future<void> wamp_session::unprovide(const wamp_registration& registration)
{
    uint64_t request_id = ++m_request_id;

	auto message = std::make_shared<wamp_message>(3);
	message->set_field(0, static_cast<int>(message_type::UNREGISTER));
	message->set_field(1, request_id);
	message->set_field(2, registration.id());

	auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
	auto unregister_request = std::make_shared<wamp_unregister_request>(registration);

	m_io_service.dispatch([=]() {
		auto shared_self = weak_self.lock();
		if (!shared_self) {
			return;
		}

		try {
			send_message(std::move(*message));
			m_unregister_requests.emplace(request_id, unregister_request);
		}
		catch (const std::exception& e) {
			unregister_request->response().set_exception(boost::copy_exception(e));
		}
	});

	return unregister_request->response().get_future();
}

inline boost::future<wamp_authenticate> wamp_session::on_challenge(const wamp_challenge& challenge)
{
    // a dummy implementation
    boost::promise<wamp_authenticate> dummy;
    dummy.set_value( wamp_authenticate( "" ) );
    return dummy.get_future();
}

inline void wamp_session::on_attach(const std::shared_ptr<wamp_transport>& transport)
{
    // FIXME: We should be deferring this operation to the io service. This
    //        will almost certainly require us to return a future here to
    //        all the caller to sync up with the actual attaching of the
    //        transport.

    if (m_transport) {
        throw protocol_error("Transport already attached to session");
    }

    // This should never be possible as you cannot start a session without
    // having a transport already attached.
    assert(!m_running);

    m_transport = transport;
}

inline void wamp_session::on_detach(bool was_clean, const std::string& reason)
{
    // FIXME: We should be deferring this operation to the io service. This
    //        will almost certainly require us to return a future here to
    //        all the caller to sync up with the actual detaching of the
    //        transport.

    if (!m_transport) {
        throw protocol_error("Transport already detached from session");
    }

    // FIXME: Figure out what to do if we are detaching a transport
    //        from a session that is still running. Ideally we would
    //        not detach the transport until m_session_stop is satisfied.
    //        Perhaps we could use the same promise/future discussed above.
    //        One side effect here will be if the transport is re-used for
    //        another session as it may still receive messages for the old
    //        session.
    assert(!m_running);

    m_transport.reset();
}

inline void wamp_session::on_message(wamp_message&& message)
{
    // FIXME: Move this check into the transport
    //if (obj.type != msgpack::type::ARRAY) {
    //    throw protocol_error("invalid message structure - message is not an array");
    //}

    if (message.size() < 1) {
        throw protocol_error("invalid message structure - missing message code");
    }

    if (!message.is_field_type(0, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("invalid message code type - not an integer");
    }

    message_type code = static_cast<message_type>(message.field<int>(0));

    switch (code) {
        case message_type::HELLO:
            throw protocol_error("received HELLO message unexpected for WAMP client roles");
        case message_type::WELCOME:
            process_welcome(std::move(message));
            break;
        case message_type::ABORT:
            process_abort(std::move(message));
            break;
        case message_type::CHALLENGE:
            process_challenge(std::move(message));
            break;
        case message_type::AUTHENTICATE:
            throw protocol_error("received AUTHENTICATE message unexpected for WAMP client roles");
        case message_type::GOODBYE:
            process_goodbye(std::move(message));
            break;
        case message_type::HEARTBEAT:
            // FIXME
            break;
        case message_type::ERROR:
            process_error(std::move(message));
            break;
        case message_type::PUBLISH:
            throw protocol_error("received PUBLISH message unexpected for WAMP client roles");
        case message_type::PUBLISHED:
            // FIXME
            break;
        case message_type::SUBSCRIBE:
            throw protocol_error("received SUBSCRIBE message unexpected for WAMP client roles");
        case message_type::SUBSCRIBED:
            process_subscribed(std::move(message));
            break;
        case message_type::UNSUBSCRIBE:
            throw protocol_error("received UNSUBSCRIBE message unexpected for WAMP client roles");
        case message_type::UNSUBSCRIBED:
            process_unsubscribed(std::move(message));
            break;
        case message_type::EVENT:
            process_event(std::move(message));
            break;
        case message_type::CALL:
            throw protocol_error("received CALL message unexpected for WAMP client roles");
        case message_type::CANCEL:
            throw protocol_error("received CANCEL message unexpected for WAMP client roles");
        case message_type::RESULT:
            process_call_result(std::move(message));
            break;
        case message_type::REGISTER:
            throw protocol_error("received REGISTER message unexpected for WAMP client roles");
        case message_type::REGISTERED:
            process_registered(std::move(message));
            break;
        case message_type::UNREGISTER:
            throw protocol_error("received UNREGISTER message unexpected for WAMP client roles");
        case message_type::UNREGISTERED:
            // FIXME
            break;
        case message_type::INVOCATION:
            process_invocation(std::move(message));
            break;
        case message_type::INTERRUPT:
            throw protocol_error("received INTERRUPT message - not implemented");
        case message_type::YIELD:
            throw protocol_error("received YIELD message unexpected for WAMP client roles");
    }
}

inline void wamp_session::process_challenge(wamp_message&& message)
{
    // kind of authentication
    std::string whatAuth = message.field<std::string>(1);

    /////////////////////////////////////////
    // wampcra authentication
    /////////////////////////////////////////

    wamp_challenge challenge_object("");

    if (whatAuth == "wampcra") {
        if (!message.is_field_type(2, msgpack::type::MAP)) {
            throw protocol_error("CHALLENGE - Details must be a dictionary");
        }

        std::string challenge, salt;
        int iterations = 0 , keylen = 0;

        // parse the details, and fill variables above
        try {
            std::unordered_map<std::string, msgpack::object> details;
            message.field(2).convert(details);
            auto itr = details.find("challenge");
            if (itr != details.end()) {
                challenge = itr->second.as<std::string>();
            } else {
                throw protocol_error("wampcra must always introduce a challenge ( in details )");
            }

            itr = details.find("salt");
            if (itr != details.end()) {
                // ok we must salt our secret
                salt = itr->second.as<std::string>();

                itr = details.find("iterations");
                if (itr == details.end()) {
                    throw protocol_error("wampcra must always tell a number of iterations when introducing salting ( in details )");
                }
                iterations = itr->second.as<int>();


                itr = details.find("keylen");
                if (itr == details.end()) {
                    throw protocol_error("wampcra must always tell a key length (keylen) when introducing salting ( in details )");
                }
                keylen = itr->second.as<int>();
            }

            // make the challenge object
            challenge_object = wamp_challenge("wampcra",challenge,salt,iterations,keylen);

        } catch (const std::exception&) {
            if (m_debug_enabled) {
                std::cerr << "failed to parse challenge details" << std::endl;
            }
            throw protocol_error("wampcra authentication: Failed parse challange details");
        };
    /////////////////////////////////////////
    // ticket authentication
    /////////////////////////////////////////
    } else if ( whatAuth == "ticket" ) {
        // make the challenge object
        challenge_object = wamp_challenge("ticket");
    }
    else {
        throw protocol_error("not supported challenge type - can now only handle 'wampcra' and 'ticket'");
    }

    // I am not sure if this is neccesary. Looking at other
    // comments in this code and the examples, it seems like the
    // context_response should live at least until the end of
    // the lambda callback - below
    std::shared_ptr< boost::future< void > > context_response = std::make_shared< boost::future<void> >();

    // call the context, to get a signature...
    (*context_response) = on_challenge(challenge_object).then([=]( boost::future<wamp_authenticate> fu_auth) {
        try {
            const wamp_authenticate sig = fu_auth.get();

            auto message = std::make_shared<wamp_message>(3);
            message->set_field(0, static_cast<int>(message_type::AUTHENTICATE));
            message->set_field(1, sig.signature());
            message->set_field(2, std::unordered_map<int, int>() /* No Extra/Dict */);

            auto weak_self = std::weak_ptr<wamp_session>(this->shared_from_this());
            m_io_service.dispatch([=]() {
                auto shared_self = weak_self.lock();
                if (!shared_self) {
                    return;
                }

                try {
                    send_message(std::move(*message), false);
                } catch (const std::exception&) {
                    if (m_debug_enabled) {
                        std::cerr << "failed to handle authentication" << std::endl;
                    }
                    throw protocol_error("authentication error: failed send signature");
                }
            });

            // make sure the context_response is copied into this lambda...
            context_response.get();
        } catch (const std::exception&) {
            if (m_debug_enabled) {
                std::cerr << "failed to handle authentication" << std::endl;
            }
            throw protocol_error("authentication error: failed send signature");
        }
    });
}

inline void wamp_session::process_welcome(wamp_message&& message)
{
    m_session_id = message.field<uint64_t>(1);
    message.field(2).convert(m_welcome_details);
    m_session_join.set_value(m_session_id);
}

inline void wamp_session::process_abort(wamp_message&& message)
{
    // [ABORT, Details|dict, Reason|uri]

    if (message.size() != 3 ) {
        throw protocol_error("ABORT - length must be 3");
    }

    // Details|dict
    if (!message.is_field_type(1, msgpack::type::MAP)) {
        throw protocol_error("ABORT - Details must be a dictionary");
    }

    // Reason|uri
    if (!message.is_field_type(2, msgpack::type::STR)) {
        throw protocol_error("ABORT - REASON must be a string (URI)");
    }

    std::string uri = message.field<std::string>(2);
    m_session_join.set_exception(boost::copy_exception(abort_error(uri)));
}

inline void wamp_session::process_goodbye(wamp_message&& message)
{
    m_session_id = 0;

    // if we did not initiate closing, reply ..
    if (!m_goodbye_sent) {
        // [GOODBYE, Details|dict, Reason|uri]
        wamp_message goodbye(3);
        goodbye.set_field(0, static_cast<int>(message_type::GOODBYE));
        goodbye.set_field(1, std::unordered_map<int,int>() /* No Details */);
        goodbye.set_field(2, std::string("wamp.error.goodbye_and_out"));

        send_message(std::move(goodbye));
    } else {
        // we previously initiated closing, so this
        // is the peer reply.
    }

    std::string reason = message.field<std::string>(2);
    m_session_leave.set_value(reason);
}

inline void wamp_session::process_error(wamp_message&& message)
{
    // [ERROR, REQUEST.Type|int, REQUEST.Request|id, Details|dict, Error|uri]
    // [ERROR, REQUEST.Type|int, REQUEST.Request|id, Details|dict, Error|uri, Arguments|list]
    // [ERROR, REQUEST.Type|int, REQUEST.Request|id, Details|dict, Error|uri, Arguments|list, ArgumentsKw|dict]

    if (message.size() < 5 || message.size() > 7) {
        throw protocol_error("invalid ERROR message structure - length must be 5, 6 or 7");
    }

    // REQUEST.Type|int
    //
    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("invalid ERROR message structure - REQUEST.Type must be an integer");
    }
    auto request_type = static_cast<message_type>(message.field<int>(1));

    if (request_type != message_type::CALL &&
         request_type != message_type::REGISTER &&
         request_type != message_type::UNREGISTER &&
         request_type != message_type::PUBLISH &&
         request_type != message_type::SUBSCRIBE &&
         request_type != message_type::UNSUBSCRIBE) {
        throw protocol_error("invalid ERROR message - ERROR.Type must one of CALL, REGISTER, UNREGISTER, SUBSCRIBE, UNSUBSCRIBE");
    }

    // REQUEST.Request|id
    if (!message.is_field_type(2, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("invalid ERROR message structure - REQUEST.Request must be an integer");
    }
    auto request_id = message.field<uint64_t>(2);

    // Details
    if (!message.is_field_type(3, msgpack::type::MAP)) {
        throw protocol_error("invalid ERROR message structure - Details must be a dictionary");
    }

    // Error|uri
    if (!message.is_field_type(4, msgpack::type::STR)) {
        throw protocol_error("invalid ERROR message - Error must be a string (URI)");
    }
    auto error = message.field<std::string>(4);

    // Arguments|list
    if (message.size() > 5) {
        if (!message.is_field_type(5, msgpack::type::ARRAY)) {
            throw protocol_error("invalid ERROR message structure - Arguments must be a list");
        }
    }

    // ArgumentsKw|list
    if (message.size() > 6) {
        if (!message.is_field_type(6, msgpack::type::MAP)) {
            throw protocol_error("invalid ERROR message structure - ArgumentsKw must be a dictionary");
        }
        try {
            auto kw_args = message.field<std::unordered_map<std::string, std::string>>(6);
            const auto itr = kw_args.find("what");
            if (itr != kw_args.end()) {
                error += ": ";
                error += itr->second;
            }
        } catch (const std::exception&) {
            if (m_debug_enabled) {
                std::cerr << "failed to parse error message keyword arguments" << std::endl;
            }

            error += ": unknown exception";
        }
    }

    switch (request_type) {

        case message_type::CALL:
            {
                //
                // process CALL ERROR
                //
                auto call_itr = m_calls.find(request_id);

                if (call_itr != m_calls.end()) {
                    // FIXME: Forward all error info.
                    call_itr->second->result().set_exception(boost::copy_exception(std::runtime_error(error)));
                    m_calls.erase(call_itr);
                } else {
                    throw protocol_error("bogus ERROR message for non-pending CALL request ID: " + error);
                }
            }
            break;
        case message_type::REGISTER:
            {
                auto reg_itr = m_register_requests.find(request_id);
                if (reg_itr != m_register_requests.end())
                {
                    reg_itr->second->response().set_exception(boost::copy_exception(std::runtime_error(error)));
                    m_register_requests.erase(reg_itr);
                } else {
                    throw protocol_error("bogus ERROR message for non-pending REGISTER request ID: " + error);
                }
            }
            break;
        case message_type::UNREGISTER:
            {
                auto unreg_itr = m_unregister_requests.find(request_id);
                if (unreg_itr != m_unregister_requests.end())
                {
                   unreg_itr->second->response().set_exception(boost::copy_exception(std::runtime_error(error)));
                   m_unregister_requests.erase(unreg_itr);
                } else {
                    throw protocol_error("bogus ERROR message for non-pending UNREGISTER request ID: " + error);
                }
            }
            break;
        case message_type::PUBLISH:
            {
                //TODO: there's currently no way to get to the future returned by the publish function
                // but there needs to be something more sensibly then just propagating the error to the
                // function running the io_service
                throw protocol_error("Received ERROR for a PUBLISH request: " + error);
            }
            break;
        case message_type::SUBSCRIBE:
            {
                auto sub_itr = m_subscribe_requests.find(request_id);
                if (sub_itr != m_subscribe_requests.end())
                {
                    sub_itr->second->response().set_exception(boost::copy_exception(std::runtime_error(error)));
                    m_subscribe_requests.erase(sub_itr);
                } else {
                    throw protocol_error("bogus ERROR message for non-pending SUBSCRIBE request ID: " + error);
                }
            }
            break;
        case message_type::UNSUBSCRIBE:
            {
                auto unsub_itr = m_unsubscribe_requests.find(request_id);
                if (unsub_itr != m_unsubscribe_requests.end())
                {
                    unsub_itr->second->response().set_exception(boost::copy_exception(std::runtime_error(error)));
                    m_unsubscribe_requests.erase(unsub_itr);
                } else {
                    throw protocol_error("bogus ERROR message for non-pending UNSUBSCRIBE request ID: " + error);
                }
            }
            break;
        default:
            throw protocol_error("unhandled ERROR message: " + error);
            break;
    }
}


inline void wamp_session::process_invocation(wamp_message&& message)
{
    // [INVOCATION, Request|id, REGISTERED.Registration|id, Details|dict]
    // [INVOCATION, Request|id, REGISTERED.Registration|id, Details|dict, CALL.Arguments|list]
    // [INVOCATION, Request|id, REGISTERED.Registration|id, Details|dict, CALL.Arguments|list, CALL.ArgumentsKw|dict]

    if (message.size() < 4 || message.size() > 6) {
        throw protocol_error("INVOCATION message length must be 4, 5 or 6");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("INVOCATION.Request must be an integer");
    }
    uint64_t request_id = message.field<uint64_t>(1);

    if (!message.is_field_type(2, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("INVOCATION.Registration must be an integer");
    }
    uint64_t registration_id = message.field<uint64_t>(2);

    auto procedure_itr = m_procedures.find(registration_id);
    if (procedure_itr != m_procedures.end()) {
        if (!message.is_field_type(3, msgpack::type::MAP)) {
            throw protocol_error("INVOCATION.Details must be a map");
        }

        wamp_invocation invocation = std::make_shared<wamp_invocation_impl>();
        invocation->set_request_id(request_id);
        invocation->set_details(message.field(3));
        if (message.size() > 4) {
            if (!message.is_field_type(4, msgpack::type::ARRAY)) {
                throw protocol_error("INVOCATION.Arguments must be an array/vector");
            }
            invocation->set_arguments(message.field(4));

            if (message.size() > 5) {
                if (!message.is_field_type(5, msgpack::type::MAP)) {
                    throw protocol_error("INVOCATION.KwArguments must be a map");
                }
                invocation->set_kw_arguments(message.field(5));
            }
        }

        invocation->set_zone(std::move(message.zone()));

        auto weak_this = std::weak_ptr<wamp_session>(this->shared_from_this());

        auto send_result_fn = [weak_this] (const std::shared_ptr<wamp_message>& message) {
            // Make sure the session still exists, since the invocation could run
            // on a different thread.
            auto shared_this = weak_this.lock();
            if (!shared_this) {
                return; // FIXME: or throw exception?
            }

            // Send to the io_service thread, and make sure the session still exists (again).
            shared_this->m_io_service.dispatch([weak_this, message] {
                auto shared_this = weak_this.lock();
                if (!shared_this) {
                    return; // FIXME: or throw exception?
                }
                shared_this->send_message(std::move(*message));
            });
        };

        invocation->set_send_result_fn(std::move(send_result_fn));

        try {
            if (m_debug_enabled) {
                std::cerr << "Invoking procedure registered under " << registration_id << std::endl;
            }
            procedure_itr->second(invocation);
        }

        // FIXME: implement Autobahn-specific exception with error URI
        catch (const std::exception& e) {
            // we can at least describe the error with e.what()
            //
            if (invocation->sendable()) {
                std::map<std::string, std::string> error_kw_arguments;
                error_kw_arguments["what"] = e.what();
                invocation->error("wamp.error.runtime_error", EMPTY_ARGUMENTS, error_kw_arguments);
            }
        }
        catch (...) {
            // no information available on actual error
            //
            if (invocation->sendable()) {
                invocation->error("wamp.error.runtime_error");
            }
        }
    } else {
        throw protocol_error("bogus INVOCATION message for non-registered registration ID");
    }
}

inline void wamp_session::process_call_result(wamp_message&& message)
{
    // [RESULT, CALL.Request|id, Details|dict]
    // [RESULT, CALL.Request|id, Details|dict, YIELD.Arguments|list]
    // [RESULT, CALL.Request|id, Details|dict, YIELD.Arguments|list, YIELD.ArgumentsKw|dict]

    if (message.size() < 3 || message.size() > 5) {
        throw protocol_error("RESULT - length must be 3, 4 or 5");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("RESULT - CALL.Request must be an id");
    }
    uint64_t request_id = message.field<uint64_t>(1);

    auto call_itr = m_calls.find(request_id);
    if (call_itr != m_calls.end()) {
        if (!message.is_field_type(2, msgpack::type::MAP)) {
            throw protocol_error("RESULT - Details must be a dictionary");
        }

        wamp_call_result result(std::move(message.zone()));
        if (message.size() > 3) {
            if (!message.is_field_type(3, msgpack::type::ARRAY)) {
                throw protocol_error("RESULT - YIELD.Arguments must be a list");
            }
            result.set_arguments(message.field(3));

            if (message.size() > 4) {
                if (!message.is_field_type(4, msgpack::type::MAP)) {
                    throw protocol_error("RESULT - YIELD.ArgumentsKw must be a dictionary");
                }
                result.set_kw_arguments(message.field(4));
            }
        }
        call_itr->second->set_result(std::move(result));
        m_calls.erase(call_itr);
    } else {
        throw protocol_error("bogus RESULT message for non-pending request ID");
    }
}

inline void wamp_session::process_subscribed(wamp_message&& message)
{
    // [SUBSCRIBED, SUBSCRIBE.Request|id, Subscription|id]
    if (message.size() != 3) {
        throw protocol_error("SUBSCRIBED - length must be 3");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("SUBSCRIBED - SUBSCRIBED.Request must be an integer");
    }
    uint64_t request_id = message.field<uint64_t>(1);

    auto subscribe_request_itr = m_subscribe_requests.find(request_id);
    if (subscribe_request_itr != m_subscribe_requests.end()) {
        if (!message.is_field_type(2, msgpack::type::POSITIVE_INTEGER)) {
            throw protocol_error("SUBSCRIBED - SUBSCRIBED.Subscription must be an integer");
        }

        uint64_t subscription_id = message.field<uint64_t>(2);
        m_subscription_handlers.insert(
                std::make_pair(subscription_id, subscribe_request_itr->second->handler()));
        subscribe_request_itr->second->set_response(wamp_subscription(subscription_id));
        m_subscribe_requests.erase(request_id);
    } else {
        throw protocol_error("SUBSCRIBED - no pending request ID");
    }
}

inline void wamp_session::process_unsubscribed(wamp_message&& message)
{
    // [UNSUBSCRIBED, UNSUBSCRIBE.Request|id]
    if (message.size() != 2) {
        throw protocol_error("UNSUBSCRIBED - length must be 2");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("UNSUBSCRIBED - UNSUBSCRIBED.Request must be an integer");
    }
    uint64_t request_id = message.field<uint64_t>(1);

    auto unsubscribe_request_itr = m_unsubscribe_requests.find(request_id);
    if (unsubscribe_request_itr != m_unsubscribe_requests.end()) {
        uint64_t subscription_id = unsubscribe_request_itr->second->subscription().id();
        m_subscription_handlers.erase(subscription_id);
        unsubscribe_request_itr->second->set_response();
        m_unsubscribe_requests.erase(request_id);
    } else {
        throw protocol_error("UNSUBSCRIBED - no pending request ID");
    }
}

inline void wamp_session::process_event(wamp_message&& message)
{
    // [EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id, Details|dict]
    // [EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id, Details|dict, PUBLISH.Arguments|list]
    // [EVENT, SUBSCRIBED.Subscription|id, PUBLISHED.Publication|id, Details|dict, PUBLISH.Arguments|list, PUBLISH.ArgumentsKw|dict]

    if (message.size() < 4 || message.size() > 6) {
        throw protocol_error("EVENT - length must be 4, 5 or 6");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("EVENT - SUBSCRIBED.Subscription must be an integer");
    }
    uint64_t subscription_id = message.field<uint64_t>(1);

    auto subscription_handlers_itr = m_subscription_handlers.lower_bound(subscription_id);
    auto subscription_handlers_end = m_subscription_handlers.upper_bound(subscription_id);

    if (subscription_handlers_itr != m_subscription_handlers.end() &&
            subscription_handlers_itr != subscription_handlers_end) {

        if (!message.is_field_type(2, msgpack::type::POSITIVE_INTEGER)) {
            throw protocol_error("EVENT - PUBLISHED.Publication must be an id");
        }
        //uint64_t publication_id = message[2].as<uint64_t>();

        if (!message.is_field_type(3, msgpack::type::MAP)) {
            throw protocol_error("EVENT - Details must be a dictionary");
        }

        wamp_event event(std::move(message.zone()));

        event.set_details(message.field(3));

        if (message.size() > 4) {
            if (!message.is_field_type(4, msgpack::type::ARRAY)) {
                throw protocol_error("EVENT - EVENT.Arguments must be a list");
            }
            event.set_arguments(message.field(4));

            if (message.size() > 5) {
                if (!message.is_field_type(5, msgpack::type::MAP)) {
                    throw protocol_error("EVENT - EVENT.ArgumentsKw must be a dictionary");
                }
                event.set_kw_arguments(message.field(5));
            }
        }

        try {
            // now trigger the user supplied event handler ..
            //
            while (subscription_handlers_itr != subscription_handlers_end) {
                 (subscription_handlers_itr->second)(event);
                 ++subscription_handlers_itr;
            }
        } catch (...) {
            if (m_debug_enabled) {
                std::cerr << "Warning: event handler threw exception" << std::endl;
            }
        }

    } else {
        // silently swallow EVENT for non-existent subscription IDs.
        // We may have just unsubscribed, this EVENT might be have
        // already been in-flight.
        if (m_debug_enabled) {
            std::cerr << "EVENT - non-existent subscription ID " << subscription_id << std::endl;
        }
    }
}

inline void wamp_session::process_registered(wamp_message&& message)
{
    // [REGISTERED, REGISTER.Request|id, Registration|id]

    if (message.size() != 3) {
        throw protocol_error("REGISTERED - length must be 3");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("REGISTERED - REGISTERED.Request must be an integer");
    }
    uint64_t request_id = message.field<uint64_t>(1);

    auto register_request_itr = m_register_requests.find(request_id);
    if (register_request_itr != m_register_requests.end()) {
        if (!message.is_field_type(2, msgpack::type::POSITIVE_INTEGER)) {
            throw protocol_error("REGISTERED - REGISTERED.Registration must be an integer");
        }
        uint64_t registration_id = message.field<uint64_t>(2);

        m_procedures[registration_id] = register_request_itr->second->procedure();
        register_request_itr->second->set_response(wamp_registration(registration_id));
        m_register_requests.erase(register_request_itr);
    } else {
        throw protocol_error("REGISTERED - no pending request ID");
    }
}

inline void wamp_session::process_unregistered(wamp_message&& message)
{
    // [UNREGISTERED, UNREGISTER.Request|id]
    if (message.size() != 2) {
        throw protocol_error("UNREGISTERED - length must be 2");
    }

    if (!message.is_field_type(1, msgpack::type::POSITIVE_INTEGER)) {
        throw protocol_error("UNREGISTERED - UNREGISTERED.Request must be an integer");
    }

    uint64_t request_id = message.field<uint64_t>(1);
	auto unregister_request_itr = m_unregister_requests.find(request_id);
    if (unregister_request_itr != m_unregister_requests.end()) {
        uint64_t registration_id = unregister_request_itr->second->registration().id();
        m_procedures.erase(registration_id);
        unregister_request_itr->second->set_response();
        m_unregister_requests.erase(request_id);
    } else {
        throw protocol_error("UNREGISTERED - no pending request ID");
    }
}

inline void wamp_session::send_message(wamp_message&& message, bool session_established)
{
    if (!m_running) {
        throw protocol_error("session not running");
    }

	if (!m_transport) {
        throw no_transport_error();
    }

	if (!m_transport->is_connected()) {
		throw no_transport_error();
	}

    if (session_established && !m_session_id) {
        throw no_session_error();
    }

    m_transport->send_message(std::move(message));
}

inline const std::unordered_map<std::string, msgpack::object>&  wamp_session::welcome_details()
{
    return m_welcome_details;
}

} // namespace autobahn