#pragma once #include #include #include #include namespace zkutil { using ZooKeeperResponseCallback = std::function; class TestKeeperStorageDispatcher { private: Poco::Timespan operation_timeout{0, Coordination::DEFAULT_OPERATION_TIMEOUT_MS * 1000}; using clock = std::chrono::steady_clock; struct RequestInfo { Coordination::ZooKeeperRequestPtr request; clock::time_point time; int64_t session_id; }; std::mutex push_request_mutex; using RequestsQueue = ConcurrentBoundedQueue; RequestsQueue requests_queue{1}; std::atomic shutdown{false}; using SessionToResponseCallback = std::unordered_map; std::mutex session_to_response_callback_mutex; SessionToResponseCallback session_to_response_callback; ThreadFromGlobalPool processing_thread; TestKeeperStorage storage; private: void processingThread(); void finalize(); void setResponse(int64_t session_id, const Coordination::ZooKeeperResponsePtr & response); public: TestKeeperStorageDispatcher(); ~TestKeeperStorageDispatcher(); void putRequest(const Coordination::ZooKeeperRequestPtr & request, int64_t session_id); int64_t getSessionID() { return storage.getSessionID(); } void registerSession(int64_t session_id, ZooKeeperResponseCallback callback); }; }