HttpClient.hpp

Go to the documentation of this file.

#pragma once
#ifndef KURLYK_HEADER_KURLYK_HTTP_HTTP_CLIENT_HPP_INCLUDED
#define KURLYK_HEADER_KURLYK_HTTP_HTTP_CLIENT_HPP_INCLUDED

 
#if KURLYK_AUTH_SUPPORT
#   include "auth/IAuthProvider.hpp"
#endif
 
namespace kurlyk {

    class HttpClient {
    public:

        HttpClient() {
            ensure_initialized();
            m_request.group_id = HttpRequestManager::get_instance().generate_group_id();
        }

        HttpClient(const std::string& host) :
                m_host(host) {
            ensure_initialized();
            m_request.group_id = HttpRequestManager::get_instance().generate_group_id();
        }
 
        HttpClient(const HttpClient&) = delete;
        HttpClient& operator=(const HttpClient&) = delete;
        HttpClient(HttpClient&&) = delete;
        HttpClient& operator=(HttpClient&&) = delete;

        ~HttpClient() {
            cancel_requests();
            clear_rate_limit(RateLimitType::RL_GENERAL);
            clear_rate_limit(RateLimitType::RL_SPECIFIC);
        }

        void cancel_requests() {
            auto& worker = core::NetworkWorker::get_instance();
            auto& manager = HttpRequestManager::get_instance();
            const uint64_t group_id = m_request.group_id;
            if (worker.is_worker_thread()) {
                // On the worker thread: execute cancellation inline to avoid self-deadlock.
                manager.cancel_requests_by_group_id(group_id, nullptr);
                worker.process();
                return;
            }
            auto promise = std::make_shared<std::promise<void>>();
            auto future = promise->get_future();
            manager.cancel_requests_by_group_id(group_id, [promise](){
                try {
                    promise->set_value();
                } catch (const std::future_error& e) {
                    if (e.code() == std::make_error_condition(std::future_errc::promise_already_satisfied)) {
                        KURLYK_HANDLE_ERROR(e, "Promise already satisfied in HttpClient::request callback");
                    } else {
                        KURLYK_HANDLE_ERROR(e, "Future error in HttpClient::request callback");
                    }
                } catch (const std::exception& e) {
                    KURLYK_HANDLE_ERROR(e, "Unhandled exception in HttpClient::request callback");
                } catch (...) {
                    // Unknown fatal error in request callback
                }
            });
            worker.notify();
            try {
                future.get();
            } catch (const std::exception& e) {
                KURLYK_HANDLE_ERROR(e, "cancel_requests() future.get() failed");
            }
        }

        void wait_requests() {
            auto& worker = core::NetworkWorker::get_instance();
            if (worker.is_worker_thread()) {
                throw std::logic_error("HttpClient::wait_requests() must not be called from the network worker thread");
            }
            auto future = make_wait_requests_future(nullptr);
            worker.notify();
            try {
                future.get();
            } catch (const std::exception& e) {
                KURLYK_HANDLE_ERROR(e, "wait_requests() future.get() failed");
            }
        }

        bool wait_requests_for(std::chrono::milliseconds timeout) {
            auto& worker = core::NetworkWorker::get_instance();
            if (worker.is_worker_thread()) {
                return false;
            }
            uint64_t waiter_id = 0;
            auto future = make_wait_requests_future(&waiter_id);
            worker.notify();
            if (future.wait_for(timeout) == std::future_status::timeout) {
                HttpRequestManager::get_instance().cancel_wait_requests_by_group_id(
                    m_request.group_id,
                    waiter_id);
                return false;
            }
            try {
                future.get();
            } catch (const std::exception& e) {
                KURLYK_HANDLE_ERROR(e, "wait_requests_for() future.get() failed");
                return false;
            }
            return true;
        }

        void set_max_in_flight(std::size_t max_in_flight) {
            std::lock_guard<std::mutex> lock(m_submit_mutex);
            m_max_in_flight = max_in_flight;
        }

        std::size_t max_in_flight() const {
            std::lock_guard<std::mutex> lock(m_submit_mutex);
            return m_max_in_flight;
        }

        std::size_t in_flight_requests() const {
            return HttpRequestManager::get_instance().group_request_count(m_request.group_id);
        }

        void clear_rate_limit(RateLimitType type = RateLimitType::RL_GENERAL) {
            auto& instance = HttpRequestManager::get_instance();
 
            switch (type) {
            case RateLimitType::RL_GENERAL:
                if (m_owns_general_rate_limit) {
                    instance.remove_limit(m_request.general_rate_limit);
                }
                m_request.general_rate_limit.reset();
                m_owns_general_rate_limit = false;
                break;
 
            case RateLimitType::RL_SPECIFIC:
                if (m_owns_specific_rate_limit) {
                    instance.remove_limit(m_request.specific_rate_limit);
                }
                m_request.specific_rate_limit.reset();
                m_owns_specific_rate_limit = false;
                break;
            }
        }

        void set_host(const std::string& host) {
            m_host = host;
        }

        void set_headers(const kurlyk::Headers& headers) {
            m_request.headers = headers;
        }
 
#if KURLYK_AUTH_SUPPORT
        void set_auth_provider(std::shared_ptr<http::auth::IAuthProvider> provider) {
            m_auth_provider = provider;
        }

        void clear_auth_provider() {
            m_auth_provider.reset();
        }

        bool has_auth_provider() const {
            return m_auth_provider != nullptr;
        }

        std::shared_ptr<http::auth::IAuthProvider> auth_provider() const {
            return m_auth_provider;
        }
#endif

        bool assign_rate_limit_id(
                long limit_id,
                RateLimitType type = RateLimitType::RL_GENERAL) {
            return assign_rate_limit_handle(
                HttpRequestManager::get_instance().get_rate_limit(limit_id), type);
        }

        bool assign_rate_limit_handle(
                const HttpRateLimitHandlePtr& limit,
                RateLimitType type = RateLimitType::RL_GENERAL) {
            if (!limit) {
                return false;
            }
 
            clear_rate_limit(type);
 
            switch (type) {
            case RateLimitType::RL_GENERAL:
                m_request.general_rate_limit = limit;
                break;
 
            case RateLimitType::RL_SPECIFIC:
                m_request.specific_rate_limit = limit;
                break;
            }
            return true;
        }

        bool set_rate_limit_id(
                long limit_id,
                RateLimitType type = RateLimitType::RL_GENERAL) {
            return assign_rate_limit_id(limit_id, type);
        }

        bool set_rate_limit_handle(
                const HttpRateLimitHandlePtr& limit,
                RateLimitType type = RateLimitType::RL_GENERAL) {
            return assign_rate_limit_handle(limit, type);
        }

        void set_rate_limit(
                long requests_per_period,
                long period_ms,
                RateLimitType type = RateLimitType::RL_GENERAL,
                bool sequential = false) {
            auto& instance = HttpRequestManager::get_instance();
 
            clear_rate_limit(type);
 
            switch (type) {
            case RateLimitType::RL_GENERAL:
                m_request.general_rate_limit =
                    instance.create_rate_limit(requests_per_period, period_ms, sequential);
                m_owns_general_rate_limit = true;
                break;
 
            case RateLimitType::RL_SPECIFIC:
                m_request.specific_rate_limit =
                    instance.create_rate_limit(requests_per_period, period_ms, sequential);
                m_owns_specific_rate_limit = true;
                break;
            }
        }

        void set_rate_limit_rpm(
                long requests_per_minute,
                RateLimitType type = RateLimitType::RL_GENERAL,
                bool sequential = false) {
            long period_ms = 60000; // 1 minute in milliseconds
            set_rate_limit(requests_per_minute, period_ms, type, sequential);
        }

        void set_rate_limit_rps(
                long requests_per_second,
                RateLimitType type = RateLimitType::RL_GENERAL,
                bool sequential = false) {
            long period_ms = 1000; // 1 second in milliseconds
            set_rate_limit(requests_per_second, period_ms, type, sequential);
        }

        void set_rate_limit_key(const std::string& key) {
            m_request.general_rate_limit_key = key;
            m_request.specific_rate_limit_key = key;
        }

        void set_general_rate_limit_key(const std::string& key) {
            m_request.general_rate_limit_key = key;
        }

        void set_specific_rate_limit_key(const std::string& key) {
            m_request.specific_rate_limit_key = key;
        }

        void set_rate_limit_keys(
                const std::string& general_key,
                const std::string& specific_key) {
            m_request.general_rate_limit_key = general_key;
            m_request.specific_rate_limit_key = specific_key;
        }

        void set_accept_encoding(
                bool identity = false,
                bool deflate = false,
                bool gzip = false,
                bool brotli = false) {
            m_request.set_accept_encoding(identity, deflate, gzip, brotli);
        }

        void set_accept_encoding(const std::string& value) {
            m_request.accept_encoding = value;
        }

        void set_accept_language(const std::string& value) {
            set_header("Accept-Language", value);
        }

        void set_content_type(const std::string& value) {
            set_header("Content-Type", value);
        }

        void set_origin(const std::string& value) {
            set_header("Origin", value);
        }

        void set_referer(const std::string& value) {
            set_header("Referer", value);
        }

        void set_dnt(const bool value) {
            m_request.headers.erase("dnt");
            if (value) m_request.headers.emplace("dnt", "1");
        }

        void set_follow_location(bool value) {
            m_request.follow_location = value;
        }

        void set_auto_referer(bool value) {
            m_request.auto_referer = value;
        }

        void set_proxy_tunnel(bool value) {
            m_request.proxy_tunnel = value;
        }

        void set_head_only(bool value) {
            m_request.head_only = value;
        }

        void set_streaming(bool streaming) {
            m_request.streaming = streaming;
        }

        void set_proxy(
                    const std::string& ip,
                    int port,
                    ProxyType type = ProxyType::PROXY_HTTP) {
            m_request.set_proxy(ip, port, type);
        }

        void set_proxy(
                const std::string& ip,
                const int port,
                const std::string& username,
                const std::string& password,
                ProxyType type = ProxyType::PROXY_HTTP) {
            m_request.set_proxy(ip, port, username, password, type);
        }

        void set_proxy_auth(
                const std::string& username,
                const std::string& password) {
            m_request.set_proxy_auth(username, password);
        }

        void set_proxy_server(const std::string& server) {
            m_request.set_proxy_server(server);
        }

        void set_proxy_auth(const std::string& auth) {
            m_request.set_proxy_auth(auth);
        }

        void set_proxy_type(ProxyType type) {
            m_request.set_proxy_type(type);
        }

        void set_retry_attempts(long retry_attempts, long retry_delay_ms) {
            m_request.set_retry_attempts(retry_attempts, retry_delay_ms);
        }

        void add_valid_status(long status) {
            m_request.add_valid_status(status);
        }

        void set_valid_statuses(const std::set<long>& statuses) {
            m_request.set_valid_statuses(statuses);
        }

        void clear_valid_statuses() {
            m_request.clear_valid_statuses();
        }

        void set_user_agent(const std::string& user_agent) {
            m_request.set_user_agent(user_agent);
        }

        void set_cookie(const std::string& cookie) {
            m_request.set_cookie(cookie);
        }

        void set_cert_file(const std::string& cert_file) {
            m_request.set_cert_file(cert_file);
        }

        void set_ca_file(const std::string& ca_file) {
            m_request.set_ca_file(ca_file);
        }

        void set_timeout(long timeout) {
            m_request.set_timeout(timeout);
        }

        void set_connect_timeout(long connect_timeout) {
            m_request.set_connect_timeout(connect_timeout);
        }

        void set_verbose(bool verbose) {
            m_request.verbose = verbose;
        }

        void set_debug_header(bool debug_header) {
            m_request.debug_header = debug_header;
        }

        void set_max_redirects(long max_redirects) {
            m_request.max_redirects = max_redirects;
        }

        SubmitResult submit_request(
                std::unique_ptr<HttpRequest> request_ptr,
                HttpResponseCallback callback) {
            SubmitResult submit_result;
            {
                std::lock_guard<std::mutex> lock(m_submit_mutex);
 
                if (m_max_in_flight != 0 &&
                    HttpRequestManager::get_instance().group_request_count(m_request.group_id) >= m_max_in_flight) {
                    return SubmitResult{false, utils::make_error_code(utils::ClientError::QueueLimitExceeded)};
                }
 
                submit_result = HttpRequestManager::get_instance().submit_request(
                    std::move(request_ptr), std::move(callback));
            }
 
            if (submit_result) {
                core::NetworkWorker::get_instance().notify();
            }
            return submit_result;
        }

        bool request(
                const std::string &method,
                const std::string& path,
                const QueryParams &query,
                const Headers &headers,
                const std::string &content,
                HttpResponseCallback callback) {
            return request(make_request(method, path, query, headers, content), std::move(callback));
        }

        bool request(
                const std::string &method,
                const std::string& path,
                const QueryParams &query,
                const Headers &headers,
                const std::string &content,
                long specific_rate_limit_id,
                HttpResponseCallback callback) {
            return request(
                method,
                path,
                query,
                headers,
                content,
                HttpRequestManager::get_instance().get_rate_limit(specific_rate_limit_id),
                std::move(callback));
        }

        bool request(
                const std::string &method,
                const std::string& path,
                const QueryParams &query,
                const Headers &headers,
                const std::string &content,
                const HttpRateLimitHandlePtr& specific_rate_limit,
                HttpResponseCallback callback) {
            auto request_ptr = make_request(method, path, query, headers, content);
            request_ptr->specific_rate_limit = specific_rate_limit;
            return request(std::move(request_ptr), std::move(callback));
        }

        bool get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                HttpResponseCallback callback) {
            return request("GET", path, query, headers, std::string(), std::move(callback));
        }

        bool post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                HttpResponseCallback callback) {
            return request("POST", path, query, headers, content, std::move(callback));
        }

        bool get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                long specific_rate_limit_id,
                HttpResponseCallback callback) {
            return request("GET", path, query, headers, std::string(), specific_rate_limit_id, std::move(callback));
        }

        bool get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const HttpRateLimitHandlePtr& specific_rate_limit,
                HttpResponseCallback callback) {
            return request("GET", path, query, headers, std::string(), specific_rate_limit, std::move(callback));
        }

        bool post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                long specific_rate_limit_id,
                HttpResponseCallback callback) {
            return request("POST", path, query, headers, content, specific_rate_limit_id, std::move(callback));
        }

        bool post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                const HttpRateLimitHandlePtr& specific_rate_limit,
                HttpResponseCallback callback) {
            return request("POST", path, query, headers, content, specific_rate_limit, std::move(callback));
        }

        std::future<HttpResponsePtr> request(
                const std::string& method,
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content) {
            return submit_future_request(make_request(method, path, query, headers, content));
        }

        std::future<HttpResponsePtr> request(
                const std::string& method,
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                long specific_rate_limit_id) {
            return request(
                method,
                path,
                query,
                headers,
                content,
                HttpRequestManager::get_instance().get_rate_limit(specific_rate_limit_id));
        }

        std::future<HttpResponsePtr> request(
                const std::string& method,
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                const HttpRateLimitHandlePtr& specific_rate_limit) {
            auto request_ptr = make_request(method, path, query, headers, content);
            request_ptr->specific_rate_limit = specific_rate_limit;
            return submit_future_request(std::move(request_ptr));
        }

        std::future<HttpResponsePtr> get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers) {
            return request("GET", path, query, headers, std::string());
        }

        std::future<HttpResponsePtr> post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content) {
            return request("POST", path, query, headers, content);
        }

        std::future<HttpResponsePtr> get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                long specific_rate_limit_id) {
            return request("GET", path, query, headers, std::string(), specific_rate_limit_id);
        }

        std::future<HttpResponsePtr> post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                long specific_rate_limit_id) {
            return request("POST", path, query, headers, content, specific_rate_limit_id);
        }

        std::future<HttpResponsePtr> get(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const HttpRateLimitHandlePtr& specific_rate_limit) {
            return request("GET", path, query, headers, std::string(), specific_rate_limit);
        }

        std::future<HttpResponsePtr> post(
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content,
                const HttpRateLimitHandlePtr& specific_rate_limit) {
            return request("POST", path, query, headers, content, specific_rate_limit);
        }
 
    private:
        HttpRequest m_request;  
        std::string m_host;     
        bool m_owns_general_rate_limit = false; 
        bool m_owns_specific_rate_limit = false; 
        mutable std::mutex m_submit_mutex; 
        std::size_t m_max_in_flight = 0; 
#       if KURLYK_AUTH_SUPPORT
        std::shared_ptr<http::auth::IAuthProvider> m_auth_provider; 
#       endif

        bool request(
                std::unique_ptr<HttpRequest> request_ptr,
                HttpResponseCallback callback) {
            return submit_request(std::move(request_ptr), std::move(callback)).accepted;
        }
 
        void set_header(const std::string& name, const std::string& value) {
            m_request.headers.erase(name);
            m_request.headers.emplace(name, value);
        }
 
        std::unique_ptr<HttpRequest> make_request(
                const std::string& method,
                const std::string& path,
                const QueryParams& query,
                const Headers& headers,
                const std::string& content) const {
#           if __cplusplus >= 201402L
            auto request_ptr = std::make_unique<HttpRequest>(m_request);
#           else
            auto request_ptr = std::unique_ptr<HttpRequest>(new HttpRequest(m_request));
#           endif
 
            request_ptr->request_id = HttpRequestManager::get_instance().generate_request_id();
            request_ptr->method = method;
            request_ptr->set_url(m_host, path, query);
            request_ptr->headers.insert(headers.begin(), headers.end());
            request_ptr->content = content;
#           if KURLYK_AUTH_SUPPORT
            if (m_auth_provider) {
                m_auth_provider->authorize(*request_ptr);
            }
#           endif
            return request_ptr;
        }
 
        std::future<HttpResponsePtr> submit_future_request(std::unique_ptr<HttpRequest> request_ptr) {
            auto promise = std::make_shared<std::promise<HttpResponsePtr>>();
            auto future = promise->get_future();
 
            HttpResponseCallback callback = [promise](HttpResponsePtr response) {
                safe_set_response(promise, std::move(response));
            };
 
            safe_submit_request(promise, std::move(request_ptr), std::move(callback));
            return future;
        }

        std::future<void> make_wait_requests_future(uint64_t* waiter_id) {
            auto promise = std::make_shared<std::promise<void>>();
            auto future = promise->get_future();
 
            const uint64_t registered_waiter_id =
                HttpRequestManager::get_instance().wait_requests_by_group_id(
                m_request.group_id,
                [promise]() {
                    try {
                        promise->set_value();
                    } catch (const std::future_error& e) {
                        if (e.code() == std::make_error_condition(std::future_errc::promise_already_satisfied)) {
                            KURLYK_HANDLE_ERROR(e, "Promise already satisfied in HttpClient::wait_requests callback");
                        } else {
                            KURLYK_HANDLE_ERROR(e, "Future error in HttpClient::wait_requests callback");
                        }
                    } catch (const std::exception& e) {
                        KURLYK_HANDLE_ERROR(e, "Unhandled exception in HttpClient::wait_requests callback");
                    } catch (...) {
                        // Unknown fatal error in wait callback
                    }
                });
            if (waiter_id) {
                *waiter_id = registered_waiter_id;
            }
 
            return future;
        }

        static void safe_set_response(
                std::shared_ptr<std::promise<HttpResponsePtr>> promise,
                HttpResponsePtr response) {
            if (!response || !response->ready) return;
            try {
                promise->set_value(std::move(response));
            } catch (const std::future_error& e) {
                if (e.code() == std::make_error_condition(std::future_errc::promise_already_satisfied)) {
                    KURLYK_HANDLE_ERROR(e, "Promise already satisfied in HttpClient::request callback");
                } else {
                    KURLYK_HANDLE_ERROR(e, "Future error in HttpClient::request callback");
                }
            } catch (const std::exception& e) {
                KURLYK_HANDLE_ERROR(e, "Unhandled exception in HttpClient::request callback");
            } catch (...) {
                // Unknown fatal error in request callback
            }
        }

        static HttpResponsePtr make_submit_error_response(const SubmitResult& submit_result) {
#           if __cplusplus >= 201402L
            auto response = std::make_unique<HttpResponse>();
#           else
            auto response = std::unique_ptr<HttpResponse>(new HttpResponse());
#           endif
            response->ready = true;
            response->status_code = 0;
            response->error_code = submit_result.error_code;
            response->error_message = submit_result.error_code.message();
            return response;
        }

        void safe_submit_request(
                std::shared_ptr<std::promise<HttpResponsePtr>> promise,
                std::unique_ptr<HttpRequest> request_ptr,
                HttpResponseCallback callback) {
            try {
                const SubmitResult submit_result = submit_request(std::move(request_ptr), std::move(callback));
                if (!submit_result) {
                    safe_set_response(promise, make_submit_error_response(submit_result));
                }
            } catch (const std::exception& e) {
                KURLYK_HANDLE_ERROR(e, "Exception while submitting request");
                try {
                    promise->set_exception(std::current_exception());
                } catch (...) {} // fallback
            } catch (...) {
                // Unknown fatal error while submitting request
                try {
                    promise->set_exception(std::current_exception());
                } catch (...) {} // fallback
            }
        }

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