client.hpp

Go to the documentation of this file.

// Copyright Takatoshi Kondo 2015
//
// Distributed under the Boost Software License, Version 1.0.
// (See accompanying file LICENSE_1_0.txt or copy at
// http://www.boost.org/LICENSE_1_0.txt)
 
#if !defined(MQTT_CLIENT_HPP)
#define MQTT_CLIENT_HPP
 
#include <mqtt/variant.hpp> // should be top to configure variant limit
 
#include <string>
#include <vector>
#include <functional>
#include <type_traits>
 
#include <mqtt/namespace.hpp>
#include <mqtt/optional.hpp>
 
#include <boost/lexical_cast.hpp>
#include <boost/asio.hpp>
 
#include <mqtt/endpoint.hpp>
#include <mqtt/null_strand.hpp>
#include <mqtt/move.hpp>
#include <mqtt/constant.hpp>
 
#include <mqtt/callable_overlay.hpp>
 
namespace MQTT_NS {
 
namespace as = boost::asio;
namespace mi = boost::multi_index;
 
template <typename Socket, std::size_t PacketIdBytes = 2>
class client : public endpoint<std::mutex, std::lock_guard, PacketIdBytes> {
    using this_type = client<Socket, PacketIdBytes>;
    using base = endpoint<std::mutex, std::lock_guard, PacketIdBytes>;
protected:
    struct constructor_access{};
public:
    using async_handler_t = typename base::async_handler_t;
 
    template<typename ... Args>
    client(constructor_access, Args && ... args)
     : client(std::forward<Args>(args)...)
    { }
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
    make_client(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>>>>
    make_client_no_strand(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
#if defined(MQTT_USE_WS)
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
    make_client_ws(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>>>>
    make_client_no_strand_ws(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
    make_tls_client(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
    make_tls_client_no_strand(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
#if defined(MQTT_USE_WS)
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
    make_tls_client_ws(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
    make_tls_client_no_strand_ws(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
#endif // defined(MQTT_USE_WS)
#endif // defined(MQTT_USE_TLS)
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
    make_client_32(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
    make_client_no_strand_32(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
#if defined(MQTT_USE_WS)
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
    make_client_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
    make_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
    make_tls_client_32(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
    make_tls_client_no_strand_32(as::io_context& ioc, std::string host, std::string port, protocol_version version);
 
#if defined(MQTT_USE_WS)
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
    make_tls_client_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
 
    friend std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
    make_tls_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path, protocol_version version);
#endif // defined(MQTT_USE_WS)
#endif // defined(MQTT_USE_TLS)
 
    void set_host(std::string host) {
        host_ = force_move(host);
    }
 
    void set_port(std::string port) {
        port_ = force_move(port);
    }
 
    void set_client_id(std::string id) {
        client_id_ = force_move(id);
    }
 
    std::string const& get_client_id() const {
        return client_id_;
    }
 
    void set_clean_session(bool cs) {
        set_clean_start(cs);
    }
 
    void set_clean_start(bool cs) {
        base::clean_start_ = cs;
    }
 
    void set_user_name(std::string name) {
        user_name_ = force_move(name);
    }
 
    void set_password(std::string password) {
        password_ = force_move(password);
    }
 
    void set_will(will w) {
        will_ = force_move(w);
    }
 
    template<typename ... Args>
    void set_will(Args && ... args) {
        will_.emplace(std::forward<Args>(args)...);
    }
 
#if defined(MQTT_USE_TLS)
 
    tls::context& get_ssl_context() {
        static_assert(has_tls<std::decay_t<decltype(*this)>>::value, "Client is required to support TLS.");
        return ctx_;
    }
 
    tls::context const& get_ssl_context() const {
        static_assert(has_tls<std::decay_t<decltype(*this)>>::value, "Client is required to support TLS.");
        return ctx_;
    }
 
#endif // defined(MQTT_USE_TLS)
 
 
    void set_keep_alive_sec(std::uint16_t keep_alive_sec, std::chrono::steady_clock::duration ping) {
        if ((ping_duration_ != std::chrono::steady_clock::duration::zero()) && base::connected() && (ping == std::chrono::steady_clock::duration::zero())) {
            tim_ping_.cancel();
        }
        keep_alive_sec_ = keep_alive_sec;
        ping_duration_ = force_move(ping);
    }
 
    void set_keep_alive_sec_ping_ms(std::uint16_t keep_alive_sec, std::size_t ping_ms) {
            set_keep_alive_sec(keep_alive_sec, std::chrono::milliseconds(ping_ms));
    }
 
    void set_keep_alive_sec(std::uint16_t keep_alive_sec) {
        set_keep_alive_sec(keep_alive_sec, std::chrono::seconds(keep_alive_sec / 2));
    }
 
 
    void connect(any session_life_keeper = any()) {
        connect(v5::properties{}, force_move(session_life_keeper));
    }
 
    void connect(boost::system::error_code& ec, any session_life_keeper = any()) {
        connect(v5::properties{}, ec, force_move(session_life_keeper));
    }
 
    void connect(v5::properties props, any session_life_keeper = any()) {
        setup_socket(socket_);
        connect_impl(force_move(props), force_move(session_life_keeper));
    }
 
    void connect(
        v5::properties props,
        boost::system::error_code& ec,
        any session_life_keeper = any()) {
        setup_socket(socket_);
        connect_impl(force_move(props), force_move(session_life_keeper), ec);
    }
 
    void connect(std::shared_ptr<Socket>&& socket, any session_life_keeper = any()) {
        connect(force_move(socket), v5::properties{}, force_move(session_life_keeper));
    }
 
    void connect(
        std::shared_ptr<Socket>&& socket,
        boost::system::error_code& ec,
        any session_life_keeper = any()) {
        connect(force_move(socket), v5::properties{}, ec, force_move(session_life_keeper));
    }
 
    void connect(
        std::shared_ptr<Socket>&& socket,
        v5::properties props,
        any session_life_keeper = any()) {
        socket_ = force_move(socket);
        base::socket_sp_ref() = socket_;
        connect_impl(force_move(props), force_move(session_life_keeper));
    }
 
    void connect(
        std::shared_ptr<Socket>&& socket,
        v5::properties props,
        boost::system::error_code& ec,
        any session_life_keeper = any()) {
        socket_ = force_move(socket);
        base::socket_sp_ref() = socket_;
        connect_impl(force_move(props), force_move(session_life_keeper), ec);
    }
 
    void async_connect() {
        async_connect(any(), async_handler_t());
    }
 
    template <typename T>
    // to avoid ambiguousness between any and async_handler_t
    std::enable_if_t<
        !std::is_convertible<T, async_handler_t>::value
    >
    async_connect(T session_life_keeper) {
        async_connect(force_move(session_life_keeper), async_handler_t());
    }
 
    void async_connect(async_handler_t func) {
        async_connect(any(), force_move(func));
    }
 
    void async_connect(any session_life_keeper, async_handler_t func) {
        async_connect(v5::properties{}, force_move(session_life_keeper), force_move(func));
    }
 
    template <typename T>
    // to avoid ambiguousness between any and async_handler_t
    std::enable_if_t<
        !std::is_convertible<T, async_handler_t>::value
    >
    async_connect(v5::properties props, T session_life_keeper) {
        async_connect(force_move(props), force_move(session_life_keeper), async_handler_t());
    }
 
    void async_connect(v5::properties props, async_handler_t func) {
        async_connect(force_move(props), any(), force_move(func));
    }
 
    void async_connect(v5::properties props, any session_life_keeper, async_handler_t func) {
        setup_socket(socket_);
        async_connect_impl(force_move(props), force_move(session_life_keeper), force_move(func));
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket) {
        async_connect(force_move(socket), v5::properties{}, any(), async_handler_t());
    }
 
    template <typename T>
    // to avoid ambiguousness between any and async_handler_t
    std::enable_if_t<
        !std::is_convertible<T, async_handler_t>::value
    >
    async_connect(std::shared_ptr<Socket>&& socket, T session_life_keeper) {
        async_connect(force_move(socket), v5::properties{}, force_move(session_life_keeper), async_handler_t());
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket, async_handler_t func) {
        async_connect(force_move(socket), v5::properties{}, any(), force_move(func));
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket, any session_life_keeper, async_handler_t func) {
        async_connect(force_move(socket), v5::properties{}, force_move(session_life_keeper), force_move(func));
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket, v5::properties props) {
        async_connect(force_move(socket), force_move(props), any(), async_handler_t());
    }
 
    template <typename T>
    // to avoid ambiguousness between any and async_handler_t
    std::enable_if_t<
        !std::is_convertible<T, async_handler_t>::value
    >
    async_connect(std::shared_ptr<Socket>&& socket, v5::properties props, T session_life_keeper) {
        async_connect(force_move(socket), force_move(props), force_move(session_life_keeper), async_handler_t());
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket, v5::properties props, async_handler_t func) {
        async_connect(force_move(socket), force_move(props), any(), force_move(func));
    }
 
    void async_connect(std::shared_ptr<Socket>&& socket, v5::properties props, any session_life_keeper, async_handler_t func) {
        socket_ = force_move(socket);
        base::socket_sp_ref() = socket_;
        async_connect_impl(force_move(props), force_move(session_life_keeper), force_move(func));
    }
 
    void disconnect(
        std::chrono::steady_clock::duration timeout,
        v5::disconnect_reason_code reason_code = v5::disconnect_reason_code::normal_disconnection,
        v5::properties props = {}
    ) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
            tim_close_.expires_after(force_move(timeout));
            tim_close_.async_wait(
                [wp = force_move(wp)](error_code ec) {
                    if (auto sp = wp.lock()) {
                        if (!ec) {
                            sp->socket()->post(
                                [sp] {
                                    sp->force_disconnect();
                                }
                            );
                        }
                    }
                }
            );
            base::disconnect(reason_code, force_move(props));
        }
    }
 
    void disconnect(
        v5::disconnect_reason_code reason_code = v5::disconnect_reason_code::normal_disconnection,
        v5::properties props = {}
    ) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            base::disconnect(reason_code, force_move(props));
        }
    }
 
    void async_disconnect(
        std::chrono::steady_clock::duration timeout,
        async_handler_t func = async_handler_t()) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
            tim_close_.expires_after(force_move(timeout));
            tim_close_.async_wait(
                [wp = force_move(wp)](error_code ec) {
                    if (auto sp = wp.lock()) {
                        if (!ec) {
                            sp->socket()->post(
                                [sp] {
                                    sp->force_disconnect();
                                }
                            );
                        }
                    }
                }
            );
            base::async_disconnect(force_move(func));
        }
    }
 
    void async_disconnect(
        std::chrono::steady_clock::duration timeout,
        v5::disconnect_reason_code reason_code,
        v5::properties props,
        async_handler_t func = async_handler_t()) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
            tim_close_.expires_after(force_move(timeout));
            tim_close_.async_wait(
                [wp = force_move(wp)](error_code ec) {
                    if (auto sp = wp.lock()) {
                        if (!ec) {
                            sp->socket()->post(
                                [sp] {
                                    sp->force_disconnect();
                                }
                            );
                        }
                    }
                }
            );
            base::async_disconnect(reason_code, force_move(props), force_move(func));
        }
    }
 
    void async_disconnect(
        async_handler_t func = async_handler_t()) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            base::async_disconnect(force_move(func));
        }
    }
 
    void async_disconnect(
        v5::disconnect_reason_code reason_code,
        v5::properties props,
        async_handler_t func = async_handler_t()) {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        if (base::connected()) {
            base::async_disconnect(reason_code, force_move(props), force_move(func));
        }
    }
 
    void force_disconnect() {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
        tim_close_.cancel();
        base::force_disconnect();
    }
 
    void set_async_pingreq(bool b) {
        async_pingreq_ = b;
    }
 
    std::shared_ptr<Socket> const& socket() const {
        return socket_;
    }
 
    std::shared_ptr<Socket>& socket() {
        return socket_;
    }
 
protected:
    client(as::io_context& ioc,
           std::string host,
           std::string port
#if defined(MQTT_USE_WS)
           ,
           std::string path = "/"
#endif // defined(MQTT_USE_WS)
           ,
           protocol_version version = protocol_version::v3_1_1,
           bool async_store_send = false
    )
        :base(ioc, version, async_store_send),
         ioc_(ioc),
         tim_ping_(ioc_),
         tim_close_(ioc_),
         host_(force_move(host)),
         port_(force_move(port))
#if defined(MQTT_USE_WS)
         ,
         path_(force_move(path))
#endif // defined(MQTT_USE_WS)
    {
#if defined(MQTT_USE_TLS)
        ctx_.set_verify_mode(tls::verify_peer);
#endif // defined(MQTT_USE_TLS)
    }
 
private:
    template <typename Strand>
    void setup_socket(std::shared_ptr<tcp_endpoint<as::ip::tcp::socket, Strand>>& socket) {
        socket = std::make_shared<Socket>(ioc_);
        base::socket_sp_ref() = socket;
    }
 
#if defined(MQTT_USE_WS)
    template <typename Strand>
    void setup_socket(std::shared_ptr<ws_endpoint<as::ip::tcp::socket, Strand>>& socket) {
        socket = std::make_shared<Socket>(ioc_);
        base::socket_sp_ref() = socket;
    }
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
    template <typename Strand>
    void setup_socket(std::shared_ptr<tcp_endpoint<tls::stream<as::ip::tcp::socket>, Strand>>& socket) {
        socket = std::make_shared<Socket>(ioc_, ctx_);
        base::socket_sp_ref() = socket;
    }
 
#if defined(MQTT_USE_WS)
    template <typename Strand>
    void setup_socket(std::shared_ptr<ws_endpoint<tls::stream<as::ip::tcp::socket>, Strand>>& socket) {
        socket = std::make_shared<Socket>(ioc_, ctx_);
        base::socket_sp_ref() = socket;
    }
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
    void start_session(v5::properties props, any session_life_keeper) {
        base::async_read_control_packet_type(force_move(session_life_keeper));
        // sync base::connect() refer to parameters only in the function.
        // So they can be passed as view.
        base::connect(
            buffer(string_view(client_id_)),
            ( user_name_ ? buffer(string_view(user_name_.value()))
                         : optional<buffer>() ),
            ( password_  ? buffer(string_view(password_.value()))
                         : optional<buffer>() ),
            will_,
            keep_alive_sec_,
            force_move(props)
        );
    }
 
    void async_start_session(v5::properties props, any session_life_keeper, async_handler_t func) {
        base::async_read_control_packet_type(force_move(session_life_keeper));
        // sync base::connect() refer to parameters only in the function.
        // So they can be passed as view.
        base::async_connect(
            buffer(string_view(client_id_)),
            ( user_name_ ? buffer(string_view(user_name_.value()))
                         : optional<buffer>() ),
            ( password_  ? buffer(string_view(password_.value()))
                         : optional<buffer>() ),
            will_,
            keep_alive_sec_,
            force_move(props),
            force_move(func)
        );
    }
 
    template <typename Strand>
    void handshake_socket(
        tcp_endpoint<as::ip::tcp::socket, Strand>&,
        v5::properties props,
        any session_life_keeper) {
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
    template <typename Strand>
    void handshake_socket(
        tcp_endpoint<as::ip::tcp::socket, Strand>&,
        v5::properties props,
        any session_life_keeper,
        boost::system::error_code& ec) {
        start_session(force_move(props), force_move(session_life_keeper));
        ec = boost::system::errc::make_error_code(boost::system::errc::success);
    }
 
#if defined(MQTT_USE_WS)
 
    template <typename Strand>
    void handshake_socket(
        ws_endpoint<as::ip::tcp::socket, Strand>& socket,
        v5::properties props,
        any session_life_keeper) {
        socket.handshake(host_, path_);
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
    template <typename Strand>
    void handshake_socket(
        ws_endpoint<as::ip::tcp::socket, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        boost::system::error_code& ec) {
        socket.handshake(host_, path_, ec);
        if (ec) return;
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
 
    template <typename Strand>
    void handshake_socket(
        tcp_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper) {
        socket.handshake(tls::stream_base::client);
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
    template <typename Strand>
    void handshake_socket(
        tcp_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        boost::system::error_code& ec) {
        socket.handshake(tls::stream_base::client, ec);
        if (ec) return;
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
#if defined(MQTT_USE_WS)
 
    template <typename Strand>
    void handshake_socket(
        ws_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper) {
        socket.next_layer().handshake(tls::stream_base::client);
        socket.handshake(host_, path_);
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
    template <typename Strand>
    void handshake_socket(
        ws_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        boost::system::error_code& ec) {
        socket.next_layer().handshake(tls::stream_base::client, ec);
        if (ec) return;
        socket.handshake(host_, path_, ec);
        if (ec) return;
        start_session(force_move(props), force_move(session_life_keeper));
    }
 
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
    template <typename Strand>
    void async_handshake_socket(
        tcp_endpoint<as::ip::tcp::socket, Strand>&,
        v5::properties props,
        any session_life_keeper,
        async_handler_t func) {
        async_start_session(force_move(props), force_move(session_life_keeper), force_move(func));
    }
 
#if defined(MQTT_USE_WS)
    template <typename Strand>
    void async_handshake_socket(
        ws_endpoint<as::ip::tcp::socket, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        async_handler_t func) {
        socket.async_handshake(
            host_,
            path_,
            [
                this,
                self = this->shared_from_this(),
                session_life_keeper = force_move(session_life_keeper),
                props = force_move(props),
                func = force_move(func)
            ]
            (error_code ec) mutable {
                if (ec) {
                  if (func) func(ec);
                  return;
                }
                async_start_session(force_move(props), force_move(session_life_keeper), force_move(func));
            });
    }
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
 
    template <typename Strand>
    void async_handshake_socket(
        tcp_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        async_handler_t func) {
        socket.async_handshake(
            tls::stream_base::client,
            [
                this,
                self = this->shared_from_this(),
                session_life_keeper = force_move(session_life_keeper),
                props = force_move(props),
                func = force_move(func)
            ]
            (error_code ec) mutable {
                if (ec) {
                  if (func) func(ec);
                  return;
                }
                async_start_session(force_move(props), force_move(session_life_keeper), force_move(func));
            });
    }
 
#if defined(MQTT_USE_WS)
    template <typename Strand>
    void async_handshake_socket(
        ws_endpoint<tls::stream<as::ip::tcp::socket>, Strand>& socket,
        v5::properties props,
        any session_life_keeper,
        async_handler_t func) {
        socket.next_layer().async_handshake(
            tls::stream_base::client,
            [
                this,
                self = this->shared_from_this(),
                session_life_keeper = force_move(session_life_keeper),
                &socket,
                props = force_move(props),
                func = force_move(func)
            ]
            (error_code ec) mutable {
                if (ec) {
                    if (func) func(ec);
                    return;
                }
                socket.async_handshake(
                    host_,
                    path_,
                    [
                        this,
                        self,
                        session_life_keeper = force_move(session_life_keeper),
                        props = force_move(props),
                        func = force_move(func)
                    ]
                    (error_code ec) mutable {
                        if (ec) {
                            if (func) func(ec);
                            return;
                        }
                        async_start_session(force_move(props), force_move(session_life_keeper), force_move(func));
                    });
            });
    }
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
    void connect_impl(
        v5::properties props,
        any session_life_keeper) {
        as::ip::tcp::resolver r(ioc_);
        auto eps = r.resolve(host_, port_);
        as::connect(socket_->lowest_layer(), eps.begin(), eps.end());
        base::set_connect();
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) {
            set_timer();
        }
        handshake_socket(*socket_, force_move(props), force_move(session_life_keeper));
    }
 
    void connect_impl(
        v5::properties props,
        any session_life_keeper,
        boost::system::error_code& ec) {
        as::ip::tcp::resolver r(ioc_);
        auto eps = r.resolve(host_, port_, ec);
        if (ec) return;
        as::connect(socket_->lowest_layer(), eps.begin(), eps.end(), ec);
        if (ec) return;
        base::set_connect();
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) {
            set_timer();
        }
        handshake_socket(*socket_, force_move(props), force_move(session_life_keeper), ec);
    }
 
    void async_connect_impl(
        v5::properties props,
        any session_life_keeper,
        async_handler_t func) {
        auto r = std::make_shared<as::ip::tcp::resolver>(ioc_);
        auto p = r.get();
        p->async_resolve(
            host_,
            port_,
            [
                this,
                self = this->shared_from_this(),
                props = force_move(props),
                session_life_keeper = force_move(session_life_keeper),
                func = force_move(func),
                r = force_move(r)
            ]
            (
                error_code ec,
                as::ip::tcp::resolver::results_type eps
            ) mutable {
                if (ec) {
                    if (func) func(ec);
                    return;
                }
                as::async_connect(
                    socket_->lowest_layer(), eps.begin(), eps.end(),
                    [
                        this,
                        self = force_move(self),
                        props = force_move(props),
                        session_life_keeper = force_move(session_life_keeper),
                        func = force_move(func)
                    ]
                    (error_code ec, auto /* unused */) mutable {
                        if (ec) {
                            if (func) func(ec);
                            return;
                        }
                        base::set_connect();
                        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) {
                            set_timer();
                        }
                        async_handshake_socket(*socket_, force_move(props), force_move(session_life_keeper), force_move(func));
                    }
                );
            }
        );
    }
 
protected:
    void on_pre_send() noexcept override {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) {
            reset_timer();
        }
    }
 
private:
    void handle_timer(error_code ec) {
        if (!ec) {
            if (async_pingreq_) {
                base::async_pingreq();
            }
            else {
                base::pingreq();
            }
        }
    }
 
    void set_timer() {
        tim_ping_.expires_after(ping_duration_);
        std::weak_ptr<this_type> wp(std::static_pointer_cast<this_type>(this->shared_from_this()));
        tim_ping_.async_wait(
            [wp = force_move(wp)](error_code ec) {
                if (auto sp = wp.lock()) {
                    sp->handle_timer(ec);
                }
            }
        );
    }
 
    void reset_timer() {
        tim_ping_.cancel();
        set_timer();
    }
 
protected:
    void on_close() noexcept override {
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
    }
 
    void on_error(error_code ec) noexcept override {
        (void)ec;
        if (ping_duration_ != std::chrono::steady_clock::duration::zero()) tim_ping_.cancel();
    }
 
    // Ensure that only code that knows the *exact* type of an object
    // inheriting from this abstract base class can destruct it.
    // This avoids issues of the destructor not triggering destruction
    // of derived classes, and any member variables contained in them.
    // Note: Not virtual to avoid need for a vtable when possible.
    ~client() = default;
 
private:
 
#if defined(MQTT_USE_TLS)
 
    template <typename T>
    struct has_tls : std::false_type {
    };
 
    template <typename U>
    struct has_tls<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, U>>> : std::true_type {
    };
 
#if defined(MQTT_USE_WS)
 
    template <typename U>
    struct has_tls<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, U>>> : std::true_type {
    };
 
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
    std::shared_ptr<Socket> socket_;
    as::io_context& ioc_;
    as::steady_timer tim_ping_;
    as::steady_timer tim_close_;
    std::string host_;
    std::string port_;
    std::uint16_t keep_alive_sec_{0};
    std::chrono::steady_clock::duration ping_duration_{std::chrono::steady_clock::duration::zero()};
    std::string client_id_;
    optional<will> will_;
    optional<std::string> user_name_;
    optional<std::string> password_;
    bool async_pingreq_ = false;
#if defined(MQTT_USE_TLS)
    tls::context ctx_{tls::context::tlsv12};
#endif // defined(MQTT_USE_TLS)
#if defined(MQTT_USE_WS)
    std::string path_;
#endif // defined(MQTT_USE_WS)
};
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
make_client(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
make_client(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_client(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>>>>
make_client_no_strand(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<as::ip::tcp::socket, null_strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>>>>
make_client_no_strand(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_client_no_strand(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
#if defined(MQTT_USE_WS)
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
make_client_ws(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>>>>
make_client_ws(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_client_ws(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>>>>
make_client_no_strand_ws(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<as::ip::tcp::socket, null_strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>>>>
make_client_no_strand_ws(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_client_no_strand_ws(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
make_tls_client(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
make_tls_client(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
make_tls_client_no_strand(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
make_tls_client_no_strand(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_no_strand(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
#if defined(MQTT_USE_WS)
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
make_tls_client_ws(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>>>>
make_tls_client_ws(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_ws(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
make_tls_client_no_strand_ws(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>>>>
make_tls_client_no_strand_ws(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_no_strand_ws(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
 
// 32bit Packet Id (experimental)
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
make_client_32(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
make_client_32(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_client_32(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
make_client_no_strand_32(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<as::ip::tcp::socket, null_strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
make_client_no_strand_32(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_client_no_strand_32(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
#if defined(MQTT_USE_WS)
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
make_client_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, as::io_context::strand>, 4>>>
make_client_ws_32(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_client_ws_32(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
make_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<as::ip::tcp::socket, null_strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<as::ip::tcp::socket, null_strand>, 4>>>
make_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_client_no_strand_ws_32(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
#endif // defined(MQTT_USE_WS)
 
#if defined(MQTT_USE_TLS)
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
make_tls_client_32(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
make_tls_client_32(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_32(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
make_tls_client_no_strand_32(as::io_context& ioc, std::string host, std::string port, protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
#if defined(MQTT_USE_WS)
        "/",
#endif // defined(MQTT_USE_WS)
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<tcp_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
make_tls_client_no_strand_32(as::io_context& ioc, std::string host, std::uint16_t port, protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_no_strand_32(
        ioc,
        force_move(host),
        std::to_string(port),
        version
    );
}
 
#if defined(MQTT_USE_WS)
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
make_tls_client_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, as::io_context::strand>, 4>>>
make_tls_client_ws_32(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_ws_32(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
make_tls_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::string port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    using client_t = client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>;
    return std::make_shared<callable_overlay<client_t>>(
        client_t::constructor_access(),
        ioc,
        force_move(host),
        force_move(port),
        force_move(path),
        version
    );
}
 
inline std::shared_ptr<callable_overlay<client<ws_endpoint<tls::stream<as::ip::tcp::socket>, null_strand>, 4>>>
make_tls_client_no_strand_ws_32(as::io_context& ioc, std::string host, std::uint16_t port, std::string path = "/", protocol_version version = protocol_version::v3_1_1) {
    return make_tls_client_no_strand_ws_32(
        ioc,
        force_move(host),
        std::to_string(port),
        force_move(path),
        version
    );
}
 
#endif // defined(MQTT_USE_WS)
 
#endif // defined(MQTT_USE_TLS)
 
} // namespace MQTT_NS
 
#endif // MQTT_CLIENT_HPP