NetworkWorker.hpp

Go to the documentation of this file.

#pragma once
#ifndef _KURLYK_NETWORK_WORKER_HPP_INCLUDED
#define _KURLYK_NETWORK_WORKER_HPP_INCLUDED

 
#define KURLYK_HANDLE_ERROR(e, msg) \
    ::kurlyk::core::NetworkWorker::get_instance().handle_error((e), (msg), __FILE__, __LINE__, __FUNCTION__)
 
#include <thread>
 
namespace kurlyk::core {

    class NetworkWorker {
    public:
        using ErrorHandler = std::function<void(const std::exception&, const char*, const char*, int, const char*)>;

        static NetworkWorker& get_instance() {
            static NetworkWorker* instance = new NetworkWorker();
            return *instance;
        }

        void add_error_handler(ErrorHandler handler) {
            std::lock_guard<std::mutex> lock(m_error_handlers_mutex);
            m_error_handlers.push_back(std::move(handler));
        }

        void handle_error(
                const std::exception& e,
                const char* msg,
                const char* file,
                int line,
                const char* func) {
 
            std::unique_lock<std::mutex> lock(m_error_handlers_mutex);
            if (m_error_handlers.empty()) return;
            std::vector<ErrorHandler> handlers = m_error_handlers;
            lock.unlock();
 
            for (const auto& handler : handlers) {
                try {
                    handler(e, msg, file, line, func);
                } catch (...) {
                    // Never let handler crash the system
                }
            }
        }

        void handle_error(
                std::exception_ptr eptr,
                const char* msg,
                const char* file,
                int line,
                const char* func) {
 
            if (!eptr) return;
 
            try {
                std::rethrow_exception(eptr);
            } catch (const std::exception& e) {
                handle_error(e, msg, file, line, func);
            } catch (...) {
                const std::runtime_error unknown("Unknown non-std::exception");
                handle_error(unknown, msg, file, line, func);
            }
        }

        bool is_worker_thread() const {
            return std::this_thread::get_id() == m_worker_thread_id;
        }

        void add_task(std::function<void()> task) {
            std::unique_lock<std::mutex> lock(m_tasks_list_mutex);
            m_tasks_list.push_back(std::move(task));
            lock.unlock();
            notify();
        }

        void register_manager(INetworkTaskManager* manager) {
            std::lock_guard<std::mutex> lock(m_managers_mutex);
            if (std::find(m_managers.begin(), m_managers.end(), manager) == m_managers.end()) {
                m_managers.push_back(manager);
            }
        }

        void process() {
            std::unique_lock<std::mutex> lock(m_managers_mutex);
            for (auto* m : m_managers) m->process();
            lock.unlock();
            process_tasks();
        }

        void notify() {
            std::lock_guard<std::mutex> locker(m_notify_mutex);
            m_notify = true;
            m_notify_condition.notify_one();
        }

        void start(const bool use_async) {
            std::unique_lock<std::mutex> locker(m_is_worker_started_mutex);
            if (m_is_worker_started) return;
            m_is_worker_started = true;
            if (!use_async) return;
            locker.unlock();
 
            m_future = std::async(
                    std::launch::async,
                    [this] {
                m_worker_thread_id = std::this_thread::get_id();
                for (;;) {
                    std::unique_lock<std::mutex> locker(m_notify_mutex);
                    m_notify_condition.wait(locker, [this]() { return m_notify; });
                    m_notify = false;
                    locker.unlock();
 
                    if (m_shutdown) {
                        shutdown();
                        return;
                    }
 
                    while (is_loaded()) {
                        process();
                        if (m_shutdown) {
                            shutdown();
                            return;
                        }
 
                        std::unique_lock<std::mutex> locker(m_notify_mutex);
                        m_notify_condition.wait_for(locker, std::chrono::milliseconds(1), [this] {
                            return m_notify || m_shutdown;
                        });
                        m_notify = false;
                        locker.unlock();
 
                        if (m_shutdown) {
                            shutdown();
                            return;
                        }
                    }
 
                    if (m_shutdown) {
                        shutdown();
                        return;
                    }
                }
            }).share();
        }

        void stop() {
            {
                std::lock_guard<std::mutex> locker(m_is_worker_started_mutex);
                if (!m_is_worker_started && !m_future.valid()) return;
            }
 
            const bool was_shutdown = m_shutdown.exchange(true);
            if (!m_future.valid()) {
                if (!was_shutdown) shutdown();
                return;
            }
 
            notify();
            try {
                m_future.wait();
                m_future.get();
            } catch(...) {
                KURLYK_HANDLE_ERROR(std::current_exception(), "Exception during NetworkWorker shutdown");
            };
        }

        void shutdown() {
            std::unique_lock<std::mutex> lock(m_managers_mutex);
            for (auto* m : m_managers) m->shutdown();
            lock.unlock();
            process_tasks();
        }
 
    private:
        std::shared_future<void>    m_future;                           
        std::atomic<bool>           m_shutdown = ATOMIC_VAR_INIT(false);
        std::thread::id             m_worker_thread_id;                 
        std::mutex                  m_notify_mutex;                     
        std::condition_variable     m_notify_condition;                 
        bool                        m_notify = false;                   
        std::mutex                  m_is_worker_started_mutex;          
        bool                        m_is_worker_started = false;        
        mutable std::mutex          m_tasks_list_mutex;                 
        std::list<std::function<void()>> m_tasks_list;                  
        mutable std::mutex          m_managers_mutex;                   
        std::vector<INetworkTaskManager*> m_managers;                   
        std::mutex                  m_error_handlers_mutex;             
        std::vector<ErrorHandler>   m_error_handlers;                   
 

        NetworkWorker() {
            utils::convert_user_agent_to_sec_ch_ua("");
            utils::is_valid_email_id("");
        }

        ~NetworkWorker() {
            stop();
        }

        NetworkWorker(const NetworkWorker&) = delete;

        NetworkWorker& operator=(const NetworkWorker&) = delete;

        void process_tasks() {
            std::unique_lock<std::mutex> lock(m_tasks_list_mutex);
            if (m_tasks_list.empty()) return;
            auto tasks_list = std::move(m_tasks_list);
            m_tasks_list.clear();
            lock.unlock();
            for (auto &item : tasks_list) {
                item();
            }
            tasks_list.clear();
        }

        const bool has_pending_tasks() const {
            std::lock_guard<std::mutex> lock(m_tasks_list_mutex);
            return !m_tasks_list.empty();
        }

        const bool is_loaded() const {
            std::unique_lock<std::mutex> lock(m_managers_mutex);
            for (auto* m : m_managers) {
                if (m->is_loaded()) return true;
            }
            lock.unlock();
            return has_pending_tasks();
        }
 
    }; // NetworkWorker
 
}; // namespace kurlyk
 
#endif // _KURLYK_NETWORK_WORKER_HPP_INCLUDED