client_internal.h

Go to the documentation of this file.

#pragma once
#include <mutex>
#include <condition_variable>
#include <vector>
#include <string>
#include <memory>
#include <chrono>
#include <atomic>
#include <unordered_map>
#include <cstdlib>
#include <spdlog/logger.h>
#include "webcface/common/encoding.h"
#include "webcface/common/internal/message/image.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"
 
WEBCFACE_NS_BEGIN
 
class Log;
class Variant;
 
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);
 
    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::shared_ptr<void> curl_initializer;
 
    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 WsMutexedData {
        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;
        std::queue<std::variant<std::string, SyncDataSnapshot>> sync_queue;
    };
 
  private:
    WsMutexedData ws_data;
    std::mutex ws_m;
 
  public:
    std::condition_variable ws_cond;
 
    class ScopedWsLock : public std::unique_lock<std::mutex> {
        ClientData *data;
 
      public:
        explicit ScopedWsLock(ClientData *data)
            : std::unique_lock<std::mutex>(data->ws_m), data(data) {}
        explicit ScopedWsLock(const std::shared_ptr<ClientData> &data)
            : ScopedWsLock(data.get()) {}
        auto &getData() {
            assert(this->owns_lock());
            return data->ws_data;
        }
    };
 
    std::atomic<bool> auto_reconnect = true;
 
    struct SyncMutexedData {
        std::optional<SyncDataFirst> sync_first;
        SyncDataFirst syncDataFirst(internal::ClientData *this_);
        SyncDataSnapshot syncData(internal::ClientData *this_, bool is_first);
    };
 
  private:
    SyncMutexedData sync_data;
    std::mutex sync_m;
 
  public:
    class ScopedSyncLock : public std::unique_lock<std::mutex> {
        ClientData *data;
 
      public:
        explicit ScopedSyncLock(ClientData *data)
            : std::unique_lock<std::mutex>(data->sync_m), data(data) {}
        explicit ScopedSyncLock(const std::shared_ptr<ClientData> &data)
            : ScopedSyncLock(data.get()) {}
        auto &getData() {
            assert(this->owns_lock());
            return data->sync_data;
        }
    };
 
    bool messagePushOnline(std::string &&msg) {
        ScopedWsLock lock_ws(this);
        if (lock_ws.getData().connected) {
            lock_ws.getData().sync_queue.push(std::move(msg));
            this->ws_cond.notify_all();
            return true;
        } else {
            return false;
        }
    }
    bool messagePushReq(std::string &&msg) {
        bool has_sync_first;
        {
            ScopedSyncLock lock_s(this);
            has_sync_first = (lock_s.getData().sync_first != std::nullopt);
        }
        if (has_sync_first) {
            ScopedWsLock lock_ws(this);
            lock_ws.getData().sync_queue.push(std::move(msg));
            this->ws_cond.notify_all();
            return true;
        } else {
            return false;
        }
    }
    void messagePushAlways(std::string &&msg) {
        ScopedWsLock lock_ws(this);
        lock_ws.getData().sync_queue.push(std::move(msg));
        this->ws_cond.notify_all();
    }
    void messagePushAlways(SyncDataSnapshot &&msg) {
        ScopedWsLock lock_ws(this);
        lock_ws.getData().sync_queue.push(std::move(msg));
        this->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);
 
    std::mutex entry_m;
    StrSet1 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;
 
    StrMap1<Queue<FuncCallHandle>> func_listener_handlers;
 
    StrMap1<unsigned int> member_ids;
    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;
        }
        for (const auto &it : member_ids) {
            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 it = member_ids.find(name);
        if (it != member_ids.end()) {
            return it->second;
        }
        return 0;
    }
 
    std::mutex event_m;
    std::shared_ptr<std::function<void(Member)>> member_entry_event;
 
    StrMap2<std::shared_ptr<std::function<void(Value)>>> value_change_event;
    StrMap2<std::shared_ptr<std::function<void(Variant)>>> text_change_event;
    StrMap2<std::shared_ptr<std::function<void(Image)>>> image_change_event;
    StrMap2<std::shared_ptr<std::function<void(RobotModel)>>>
        robot_model_change_event;
    StrMap2<std::shared_ptr<std::function<void(View)>>> view_change_event;
    StrMap2<std::shared_ptr<std::function<void(Canvas3D)>>>
        canvas3d_change_event;
    StrMap2<std::shared_ptr<std::function<void(Canvas2D)>>>
        canvas2d_change_event;
    StrMap2<std::shared_ptr<std::function<void(Log)>>> log_append_event;
    StrMap1<std::shared_ptr<std::function<void(Member)>>> sync_event,
        ping_event;
    StrMap1<std::shared_ptr<std::function<void(Value)>>> value_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Text)>>> text_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Func)>>> func_entry_event;
    StrMap1<std::shared_ptr<std::function<void(View)>>> view_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Image)>>> image_entry_event;
    StrMap1<std::shared_ptr<std::function<void(RobotModel)>>>
        robot_model_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Canvas3D)>>>
        canvas3d_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Canvas2D)>>>
        canvas2d_entry_event;
    StrMap1<std::shared_ptr<std::function<void(Log)>>>
        log_entry_event;
 
    std::shared_ptr<spdlog::logger> logger_internal;
    std::mutex logger_m;
    StrMap1<std::unique_ptr<std::streambuf>> logger_buf;
    StrMap1<std::unique_ptr<std::ostream>> logger_os;
    StrMap1<std::unique_ptr<std::wstreambuf>> logger_buf_w;
    StrMap1<std::unique_ptr<std::wostream>> logger_os_w;
 
    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