WebSocketClient.hpp

Go to the documentation of this file.

#pragma once
#ifndef _KURLYK_WEBSOCKET_CLIENT_HPP_INCLUDED
#define _KURLYK_WEBSOCKET_CLIENT_HPP_INCLUDED

 
#include "WebSocketManager.hpp"
#include <mutex>
 
namespace kurlyk {

    class WebSocketClient {
    public:

        WebSocketClient() {
            ensure_initialized();
            m_client = WebSocketManager::get_instance().create_client();
            m_client->notify_handler() = []() {
                core::NetworkWorker::get_instance().notify();
            };
        }

        WebSocketClient(std::unique_ptr<WebSocketConfig> config, std::function<void(bool)> callback = nullptr) {
            ensure_initialized();
            m_client = WebSocketManager::get_instance().create_client();
            m_client->notify_handler() = []() {
                core::NetworkWorker::get_instance().notify();
            };
            m_client->set_config(std::move(config), std::move(callback));
        }

        WebSocketClient(
                const std::string& url,
                const Headers& headers = Headers(),
                const std::string& proxy_server = std::string(),
                const std::string& proxy_auth = std::string(),
                ProxyType proxy_type = ProxyType::PROXY_HTTP,
                long request_timeout = 20,
                bool reconnect = true,
                bool verify_cert = true,
                const std::string& ca_file = std::string(),
                int rpm = 200) {
            ensure_initialized();
            m_client = WebSocketManager::get_instance().create_client();
            m_client->notify_handler() = []() {
                core::NetworkWorker::get_instance().notify();
            };
            init_config();
 
            m_config->url = url;
            m_config->headers = headers;
            m_config->proxy_server = proxy_server;
            m_config->proxy_auth = proxy_auth;
            m_config->proxy_type = proxy_type;
            m_config->request_timeout = request_timeout;
            m_config->reconnect = reconnect;
            m_config->verify_cert = verify_cert;
            m_config->ca_file = ca_file;
            m_config->add_rate_limit(rpm, 60000);
        }
 
        WebSocketClient(const WebSocketClient&) = delete;
        WebSocketClient& operator=(const WebSocketClient&) = delete;

        virtual ~WebSocketClient() {
            auto client = m_client;
            core::NetworkWorker::get_instance().add_task([client](){
                client->shutdown();
            });
        }

        void on_event(std::function<void(std::unique_ptr<WebSocketEventData>)> callback) {
            m_client->event_handler() = std::move(callback);
        }

        std::function<void(std::unique_ptr<WebSocketEventData>)>& event_handler() {
            return m_client->event_handler();
        }

        std::future<bool> set_config(std::unique_ptr<WebSocketConfig> config) {
            auto promise = std::make_shared<std::promise<bool>>();
            auto future = promise->get_future();
            m_client->set_config(std::move(config), [promise](bool success) {
                promise->set_value(success);
            });
            core::NetworkWorker::get_instance().notify();
            return future;
        }

        void set_config(std::unique_ptr<WebSocketConfig> config, std::function<void(bool)> callback) {
            m_client->set_config(std::move(config), std::move(callback));
            core::NetworkWorker::get_instance().notify();
        }

        std::future<bool> connect() {
            if (m_config) {
#               if __cplusplus >= 201402L
                auto config = std::make_unique<WebSocketConfig>(*m_config.get());
#               else
                auto config = std::unique_ptr<WebSocketConfig>(new WebSocketConfig(*m_config.get()));
#               endif
                m_client->set_config(std::move(config), nullptr);
            }
            auto promise = std::make_shared<std::promise<bool>>();
            auto future = promise->get_future();
            m_client->connect([promise](bool success) {
                promise->set_value(success);
            });
            core::NetworkWorker::get_instance().notify();
            return future;
        }

        void connect(std::function<void(bool)> callback) {
            if (m_config) {
#               if __cplusplus >= 201402L
                auto config = std::make_unique<WebSocketConfig>(*m_config.get());
#               else
                auto config = std::unique_ptr<WebSocketConfig>(new WebSocketConfig(*m_config.get()));
#               endif
                m_client->set_config(std::move(config), nullptr);
            }
            m_client->connect(std::move(callback));
            core::NetworkWorker::get_instance().notify();
        }

        bool connect_and_wait() {
            auto connect_future = connect();
            return connect_future.get();
        }

        std::future<bool> disconnect() {
            auto promise = std::make_shared<std::promise<bool>>();
            auto future = promise->get_future();
            m_client->disconnect([promise](bool success) {
                promise->set_value(success);
            });
            core::NetworkWorker::get_instance().notify();
            return future;
        }

        bool disconnect_and_wait() {
            auto disconnect_future = disconnect();
            return disconnect_future.get();
        }

        void disconnect(std::function<void(bool success)> callback) {
            m_client->disconnect(std::move(callback));
            core::NetworkWorker::get_instance().notify();
        }

        const bool is_connected() const {
            return m_client->is_connected();
        }

        SubmitResult submit_message(
                const std::string &message,
                long rate_limit_id = 0,
                std::function<void(const std::error_code&)> callback = nullptr) {
            return m_client->submit_message(message, rate_limit_id, std::move(callback));
        }

        bool send_message(
                const std::string &message,
                long rate_limit_id = 0,
                std::function<void(const std::error_code&)> callback = nullptr) {
            return submit_message(message, rate_limit_id, std::move(callback)).accepted;
        }

        SubmitResult submit_close(
                const int status = 1000,
                const std::string &reason = std::string(),
                std::function<void(const std::error_code&)> callback = nullptr) {
            return m_client->submit_close(status, reason, std::move(callback));
        }

        bool send_close(
                const int status = 1000,
                const std::string &reason = std::string(),
                std::function<void(const std::error_code&)> callback = nullptr) {
            return submit_close(status, reason, std::move(callback)).accepted;
        }

        std::list<std::unique_ptr<WebSocketEventData>> receive_events() {
            return m_client->receive_events();
        }

        std::unique_ptr<WebSocketEventData> receive_event() {
            return m_client->receive_event();
        }

        std::string get_http_version() const {
            return m_client->get_http_version();
        }

        Headers get_headers() const {
            return m_client->get_headers();
        }

        std::string get_remote_endpoint() const {
            return m_client->get_remote_endpoint();
        }

        void set_url(const std::string& host, const std::string& path, const std::string& query = "") {
            init_config();
            return m_config->set_url(host, path, query);
        }

        void set_url(const std::string& url, const QueryParams& query) {
            init_config();
            return m_config->set_url(url, query);
        }

        void set_accept_encoding(bool identity = false, bool deflate = false, bool gzip = false, bool brotli = false) {
            init_config();
            return m_config->set_accept_encoding(identity, deflate, gzip, brotli);
        }

        void set_proxy(
                const std::string& ip,
                int port,
                ProxyType type = ProxyType::PROXY_HTTP) {
            init_config();
            return m_config->set_proxy(ip, port, type);
        }

        void set_proxy(
                const std::string& ip,
                int port,
                const std::string& username,
                const std::string& password,
                ProxyType type = ProxyType::PROXY_HTTP) {
            init_config();
            return m_config->set_proxy(ip, port, username, password, type);
        }

        void set_proxy_server(const std::string& server) {
            init_config();
            return m_config->set_proxy_server(server);
        }

        void set_proxy_auth(const std::string& auth) {
            init_config();
            return m_config->set_proxy_auth(auth);
        }

        void set_proxy_type(ProxyType type) {
            init_config();
            return m_config->set_proxy_type(type);
        }

        void set_proxy_auth(const std::string& username, const std::string& password) {
            init_config();
            return m_config->set_proxy_auth(username, password);
        }

        void set_reconnect(bool reconnect, long reconnect_attempts = 0, long reconnect_delay = 0) {
            init_config();
            return m_config->set_reconnect(reconnect, reconnect_attempts, reconnect_delay);
        }

        void set_user_agent(const std::string& user_agent) {
            init_config();
            return m_config->set_user_agent(user_agent);
        }

        void set_accept_language(const std::string& accept_language) {
            init_config();
            return m_config->set_accept_language(accept_language);
        }

        void set_cookie(const std::string& cookie) {
            init_config();
            return m_config->set_cookie(cookie);
        }

        void set_idle_timeout(long idle_timeout) {
            init_config();
            return m_config->set_idle_timeout(idle_timeout);
        }

        void set_request_timeout(long request_timeout) {
            init_config();
            return m_config->set_request_timeout(request_timeout);
        }

        void set_max_send_queue_size(std::size_t max_send_queue_size) {
            init_config();
            return m_config->set_max_send_queue_size(max_send_queue_size);
        }

        void set_ca_file(const std::string& ca_file) {
            init_config();
            return m_config->set_ca_file(ca_file);
        }

        void set_ca_file(bool verify_cert, const std::string& ca_file) {
            init_config();
            return m_config->set_ca_file(verify_cert, ca_file);
        }

        void set_verify_cert(bool verify_cert) {
            init_config();
            return m_config->set_verify_cert(verify_cert);
        }

        long add_rate_limit(long requests_per_period, long period_ms) {
            init_config();
            return m_config->add_rate_limit(requests_per_period, period_ms);
        }

        long add_rate_limit_rpm(long requests_per_minute) {
            long period_ms = 60000; // 1 minute in milliseconds
            return add_rate_limit(requests_per_minute, period_ms);
        }

        long add_rate_limit_rps(long requests_per_second) {
            long period_ms = 1000; // 1 second in milliseconds
            return add_rate_limit(requests_per_second, period_ms);
        }
 
    private:
        selected_backend_client_ptr m_client; 
        std::unique_ptr<WebSocketConfig> m_config; 

        void init_config() {
            if (!m_config) {
#               if __cplusplus >= 201402L
                m_config = std::make_unique<WebSocketConfig>();
#               else
                m_config = std::unique_ptr<WebSocketConfig>(new WebSocketConfig());
#               endif
            }
        }

        static void ensure_initialized() {
            static std::once_flag once;
            std::call_once(once, []() {
                HttpRequestManager::get_instance();
                WebSocketManager::get_instance();
                core::NetworkWorker::get_instance().start(KURLYK_AUTO_INIT_USE_ASYNC);
            });
        }
 
    }; // WebSocketClient
 
} // namespace kurlyk
 
#endif // _KURLYK_WEBSOCKET_CLIENT_HPP_INCLUDED