client_internal.h

Go to the documentation of this file.

#pragma once
#include <vector>
#include <string>
#include <memory>
#include <chrono>
#include <atomic>
#include <unordered_map>
#include <cstdlib>
#include "webcface/common/encoding.h"
#include "webcface/common/internal/message/image.h"
#include "webcface/common/internal/mutex.h"
#include "webcface/field.h"
#include "webcface/log.h"
#include "queue.h"
#include "webcface/func_result.h"
#include "data_store1.h"
#include "data_store2.h"
#include "func_internal.h"
#include "webcface/image_frame.h"
#ifdef WEBCFACE_COMPILER_IS_GCC
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wabi"
#endif
#include <spdlog/logger.h>
#ifdef WEBCFACE_COMPILER_IS_GCC
#pragma GCC diagnostic pop
#endif
 
WEBCFACE_NS_BEGIN
 
class Log;
class Variant;
 
namespace message {
template <typename T>
std::string packSingle(const T &obj);
}
 
namespace internal {
 
void wsThreadMain(const std::shared_ptr<ClientData> &data);
// void syncThreadMain(const std::shared_ptr<ClientData> &data);
 
extern std::atomic<int> log_keep_lines;
 
struct ClientData : std::enable_shared_from_this<ClientData> {
    explicit ClientData(const SharedString &name,
                        const SharedString &host = nullptr, int port = -1);
    ClientData(const ClientData&) = delete;
    ClientData &operator=(const ClientData&) = delete;
    ~ClientData();
 
    void close();
 
    SharedString self_member_name;
    std::optional<unsigned int> self_member_id;
    bool isSelf(const FieldBase &base) const {
        return base.member_ == self_member_name;
    }
 
    // std::mutex curl_m;
    SharedString host;
    int port;
    bool current_curl_connected = false;
    void *current_curl_handle = nullptr;
    std::string current_curl_path;
    std::string current_ws_buf = "";
    std::vector<char> curl_err_buffer;
 
    std::thread ws_thread;
    // /*!
    //  * \brief recv_queueのメッセージを処理するスレッド
    //  *
    //  * auto_recv がtrueの場合のみ。
    //  *
    //  */
    // std::thread sync_thread;
 
    struct SyncDataSnapshot {
        std::chrono::system_clock::time_point time;
        StrMap1<std::shared_ptr<ValueData>> value_data;
        StrMap1<std::shared_ptr<TextData>> text_data;
        StrMap1<std::shared_ptr<RobotModelData>> robot_model_data;
        StrMap1<std::shared_ptr<message::ViewData>> view_prev, view_data;
        StrMap1<std::shared_ptr<message::Canvas3DData>> canvas3d_prev,
            canvas3d_data;
        StrMap1<std::shared_ptr<message::Canvas2DData>> canvas2d_prev,
            canvas2d_data;
        StrMap1<ImageData> image_data;
        StrMap1<std::vector<LogLineData>> log_data;
        StrMap1<std::shared_ptr<FuncData>> func_data;
    };
 
    struct SyncDataFirst {
        StrMap2<unsigned int> value_req, text_req, robot_model_req, view_req,
            canvas3d_req, canvas2d_req, image_req, log_req;
        StrMap2<message::ImageReq> image_req_info;
        bool ping_status_req;
        SyncDataSnapshot sync_data;
    };
 
    std::atomic<bool> closing = false;
 
    struct WsData {
        bool connected = false;
        bool sync_init_end = false;
        bool do_ws_init = false;
        bool do_ws_recv = false;
        bool recv_ready = false;
 
        std::queue<std::vector<std::pair<int, std::shared_ptr<void>>>>
            recv_queue;
        bool did_disconnect = false;
        std::queue<std::variant<std::string, SyncDataSnapshot>> sync_queue;
    };
 
    MutexProxy<WsData> ws_data;
 
    std::atomic<bool> auto_reconnect = true;
 
    struct SyncData {
        std::optional<SyncDataFirst> sync_first;
        SyncDataFirst syncDataFirst(internal::ClientData *this_);
        SyncDataSnapshot syncData(internal::ClientData *this_, bool is_first);
    };
 
    MutexProxy<SyncData> sync_data;
 
    template <typename T>
    bool messagePushOnline(const T &obj) {
        auto lock_ws = this->ws_data.lock();
        if (lock_ws->connected) {
            this->logger_internal->debug("-> queued to send: {}", obj);
            lock_ws->sync_queue.push(message::packSingle(obj));
            lock_ws.cond().notify_all();
            return true;
        } else {
            return false;
        }
    }
    template <typename T>
    bool messagePushReq(const T &obj) {
        if (this->sync_data.lock()->sync_first != std::nullopt) {
            this->logger_internal->debug("-> queued to send: {}", obj);
            auto lock_ws = this->ws_data.lock();
            lock_ws->sync_queue.push(message::packSingle(obj));
            lock_ws.cond().notify_all();
            return true;
        } else {
            return false;
        }
    }
    template <typename T>
    void messagePushAlways(const T &obj) {
        this->logger_internal->debug("-> queued to send: {}", obj);
        auto lock_ws = this->ws_data.lock();
        lock_ws->sync_queue.push(message::packSingle(obj));
        lock_ws.cond().notify_all();
    }
    void messagePushAlways(SyncDataSnapshot &&msg) {
        this->logger_internal->debug("-> sync data queued");
        auto lock_ws = this->ws_data.lock();
        lock_ws->sync_queue.push(std::move(msg));
        lock_ws.cond().notify_all();
    }
 
    std::string packSyncDataFirst(const SyncDataFirst &data);
    std::string packSyncData(std::stringstream &buffer, int &len,
                             const SyncDataSnapshot &data);
 
    void start();
    void join();
 
    void syncImpl(bool sync, bool forever,
                  std::optional<std::chrono::microseconds> timeout);
 
    void
    onRecv(const std::vector<std::pair<int, std::shared_ptr<void>>> &messages);
 
    SharedMutexProxy<StrMap1<bool>> member_entry;
 
    SyncDataStore2<std::shared_ptr<ValueData>> value_store;
    SyncDataStore2<std::shared_ptr<TextData>> text_store;
    SyncDataStore2<std::shared_ptr<FuncData>> func_store;
    SyncDataStore2<std::shared_ptr<message::ViewData>> view_store;
    SyncDataStore2<ImageData, message::ImageReq> image_store;
    SyncDataStore2<std::shared_ptr<RobotModelData>> robot_model_store;
    SyncDataStore2<std::shared_ptr<message::Canvas3DData>> canvas3d_store;
    SyncDataStore2<std::shared_ptr<message::Canvas2DData>> canvas2d_store;
    SyncDataStore2<std::shared_ptr<LogData>> log_store;
    SyncDataStore1<std::chrono::system_clock::time_point> sync_time_store;
    FuncResultStore func_result_store;
 
    MutexProxy<StrMap1<Queue<FuncCallHandle>>> func_listener_handlers;
 
    SharedMutexProxy<StrMap1<unsigned int>> member_ids;
    SharedMutexProxy<std::unordered_map<unsigned int, std::string>>
        member_lib_name, member_lib_ver, member_addr;
    const SharedString &getMemberNameFromId(unsigned int id) const {
        if (self_member_id && *self_member_id == id) {
            return self_member_name;
        }
        auto lock_member_ids = member_ids.shared_lock();
        for (const auto &it : lock_member_ids.get()) {
            if (it.second == id) {
                return it.first;
            }
        }
        static SharedString empty;
        return empty;
    }
    unsigned int getMemberIdFromName(const SharedString &name) const {
        if (name == self_member_name && self_member_id) {
            return *self_member_id;
        }
        auto lock_member_ids = member_ids.shared_lock();
        auto it = lock_member_ids->find(name);
        if (it != lock_member_ids->end()) {
            return it->second;
        }
        return 0;
    }
 
    std::shared_ptr<std::function<void(Member)>> member_entry_event;
 
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Member)>>>>
        member_connected_event, member_disconnected_event;
 
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Value)>>>>
        value_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Variant)>>>>
        text_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Image)>>>>
        image_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(RobotModel)>>>>
        robot_model_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(View)>>>>
        view_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Canvas3D)>>>>
        canvas3d_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Canvas2D)>>>>
        canvas2d_change_event;
    SharedMutexProxy<StrMap2<std::shared_ptr<std::function<void(Log)>>>>
        log_append_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Member)>>>>
        sync_event, ping_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Value)>>>>
        value_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Text)>>>>
        text_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Func)>>>>
        func_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(View)>>>>
        view_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Image)>>>>
        image_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(RobotModel)>>>>
        robot_model_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Canvas3D)>>>>
        canvas3d_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Canvas2D)>>>>
        canvas2d_entry_event;
    SharedMutexProxy<StrMap1<std::shared_ptr<std::function<void(Log)>>>>
        log_entry_event;
 
    std::shared_ptr<spdlog::logger> logger_internal;
    SharedMutexProxy<StrMap1<std::unique_ptr<std::streambuf>>> logger_buf;
    SharedMutexProxy<StrMap1<std::unique_ptr<std::ostream>>> logger_os;
    SharedMutexProxy<StrMap1<std::unique_ptr<std::wstreambuf>>> logger_buf_w;
    SharedMutexProxy<StrMap1<std::unique_ptr<std::wostream>>> logger_os_w;
 
    SharedMutexProxy<std::string> svr_name, svr_version, svr_hostname;
 
    std::shared_ptr<std::unordered_map<unsigned int, int>> ping_status =
        nullptr;
    bool ping_status_req = false;
    void pingStatusReq();
};
} // namespace internal
WEBCFACE_NS_END