// Copyright Epic Games, Inc. All Rights Reserved. #define _SILENCE_CXX17_C_HEADER_DEPRECATION_WARNING #include #include #include #include #include #include #include #include #include #include #include #include #include #include #if ZEN_PLATFORM_WINDOWS # pragma comment(lib, "Crypt32.lib") # pragma comment(lib, "Wldap32.lib") #endif #include #include #include #include #include #include #include #include #include #include ////////////////////////////////////////////////////////////////////////// #include "projectclient.h" ////////////////////////////////////////////////////////////////////////// #define DOCTEST_CONFIG_IMPLEMENT #include #undef DOCTEST_CONFIG_IMPLEMENT using namespace fmt::literals; /* ___ ___ _________ _________ ________ ________ ___ ___ _______ ________ _________ |\ \|\ \|\___ ___\\___ ___\\ __ \ |\ ____\|\ \ |\ \|\ ___ \ |\ ___ \|\___ ___\ \ \ \\\ \|___ \ \_\|___ \ \_\ \ \|\ \ \ \ \___|\ \ \ \ \ \ \ __/|\ \ \\ \ \|___ \ \_| \ \ __ \ \ \ \ \ \ \ \ \ ____\ \ \ \ \ \ \ \ \ \ \ \_|/_\ \ \\ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ \___| \ \ \____\ \ \____\ \ \ \ \_|\ \ \ \\ \ \ \ \ \ \ \__\ \__\ \ \__\ \ \__\ \ \__\ \ \_______\ \_______\ \__\ \_______\ \__\\ \__\ \ \__\ \|__|\|__| \|__| \|__| \|__| \|_______|\|_______|\|__|\|_______|\|__| \|__| \|__| */ class HttpConnectionPool; /** * Http client connection * * Represents an established socket connection to a certain endpoint */ class HttpClientConnection { static HttpClientConnection* This(http_parser* Parser) { return (HttpClientConnection*)Parser->data; }; public: HttpClientConnection(asio::io_context& IoContext, HttpConnectionPool& Pool, asio::ip::tcp::socket&& InSocket) : m_IoContext(IoContext) , m_Pool(Pool) , m_Resolver(IoContext) , m_Socket(std::move(InSocket)) { } ~HttpClientConnection() {} HttpConnectionPool& ConnectionPool() { return m_Pool; } void SetKeepAlive(bool NewState) { m_KeepAlive = NewState; } void Get(const std::string_view Server, int Port, const std::string_view Path) { http_parser_init(&m_HttpParser, HTTP_RESPONSE); m_HttpParser.data = this; m_HttpParserSettings = http_parser_settings{ .on_message_begin = [](http_parser* p) -> int { return This(p)->OnMessageBegin(); }, .on_url = nullptr, .on_status = nullptr, .on_header_field = [](http_parser* p, const char* data, size_t size) { return This(p)->OnHeader(data, size); }, .on_header_value = [](http_parser* p, const char* data, size_t size) { return This(p)->OnHeaderValue(data, size); }, .on_headers_complete = [](http_parser* p) -> int { return This(p)->OnHeadersComplete(); }, .on_body = [](http_parser* p, const char* data, size_t size) { return This(p)->OnBody(data, size); }, .on_message_complete = [](http_parser* p) -> int { return This(p)->OnMessageComplete(); }, .on_chunk_header = nullptr, .on_chunk_complete = nullptr}; m_Headers.reserve(16); zen::ExtendableStringBuilder<256> RequestBody; RequestBody << "GET " << Path << " HTTP/1.1\r\n"; RequestBody << "Host: " << Server << "\r\n"; RequestBody << "Accept: */*\r\n"; RequestBody << "Connection: " << (m_KeepAlive ? "keep-alive" : "close") << "\r\n\r\n"; // TODO: support keep-alive m_RequestBody = RequestBody; OnConnected(); } private: void Reset() {} void OnError(const std::error_code& Error) { spdlog::error("HTTP client error! '{}'", Error.message()); } int OnHeader(const char* Data, size_t Bytes) { m_CurrentHeaderName = std::string_view(Data, Bytes); return 0; } int OnHeaderValue(const char* Data, size_t Bytes) { m_Headers.emplace_back(HeaderEntry{m_CurrentHeaderName, {Data, Bytes}}); return 0; } int OnHeadersComplete() { spdlog::debug("Headers complete"); return 0; } int OnMessageComplete() { if (http_should_keep_alive(&m_HttpParser)) { Reset(); } else { m_Socket.close(); m_RequestState = RequestState::Done; } return 0; } int OnMessageBegin() { return 0; } int OnBody(const char* Data, size_t Bytes) { return 0; } void OnConnected() { // Send initial request payload asio::async_write(m_Socket, asio::const_buffer(m_RequestBody.data(), m_RequestBody.size()), [this](const std::error_code& Error, size_t Bytes) { if (Error) { return OnError(Error); } OnRequestWritten(); }); } void OnRequestWritten() { asio::async_read(m_Socket, m_ResponseBuffer, asio::transfer_at_least(1), [this](const std::error_code& Error, size_t Bytes) { if (Error) { return OnError(Error); } OnStatusLineRead(Bytes); }); } void OnStatusLineRead(size_t Bytes) { // Parse size_t rv = http_parser_execute(&m_HttpParser, &m_HttpParserSettings, (const char*)m_ResponseBuffer.data(), Bytes); if (m_HttpParser.http_errno != 0) { // Something bad! spdlog::error("parse error {}", (uint32_t)m_HttpParser.http_errno); } switch (m_RequestState) { case RequestState::Init: asio::async_read(m_Socket, m_ResponseBuffer, asio::transfer_at_least(1), [this](const std::error_code& Error, size_t Bytes) { if (Error) { return OnError(Error); } OnStatusLineRead(Bytes); }); return; case RequestState::Done: break; } } private: asio::io_context& m_IoContext; HttpConnectionPool& m_Pool; asio::ip::tcp::resolver m_Resolver; asio::ip::tcp::socket m_Socket; std::string m_Uri; std::string m_RequestBody; // Initial request data http_parser m_HttpParser{}; http_parser_settings m_HttpParserSettings{}; uint8_t m_ResponseIoBuffer[4096]; asio::mutable_buffer m_ResponseBuffer{m_ResponseIoBuffer, sizeof m_ResponseIoBuffer}; enum class RequestState { Init, Done }; RequestState m_RequestState = RequestState::Init; struct HeaderEntry { std::string_view Name; std::string_view Value; }; std::string_view m_CurrentHeaderName; // Used while parsing headers std::vector m_Headers; bool m_KeepAlive = false; }; ////////////////////////////////////////////////////////////////////////// class HttpConnectionPool { public: HttpConnectionPool(asio::io_context& Context, std::string_view HostName, uint16_t Port); ~HttpConnectionPool(); std::unique_ptr GetConnection(); void ReturnConnection(std::unique_ptr&& Connection); private: zen::RwLock m_Lock; asio::io_context& m_Context; std::vector m_AvailableConnections; std::string m_HostName; uint16_t m_Port; }; HttpConnectionPool::HttpConnectionPool(asio::io_context& Context, std::string_view HostName, uint16_t Port) : m_Context(Context) , m_HostName(HostName) , m_Port(Port) { } HttpConnectionPool::~HttpConnectionPool() { zen::RwLock::ExclusiveLockScope ScopedLock(m_Lock); for (auto $ : m_AvailableConnections) { delete $; } } std::unique_ptr HttpConnectionPool::GetConnection() { zen::RwLock::ExclusiveLockScope ScopedLock(m_Lock); if (m_AvailableConnections.empty()) { zen::StringBuilder<16> Service; Service << int64_t(m_Port); asio::ip::tcp::resolver Resolver{m_Context}; std::error_code ErrCode; auto it = Resolver.resolve(m_HostName, Service, ErrCode); auto itEnd = asio::ip::tcp::resolver::iterator(); if (ErrCode) { return nullptr; } asio::ip::tcp::socket Socket{m_Context}; asio::connect(Socket, it, ErrCode); if (ErrCode) { return nullptr; } return std::make_unique(m_Context, *this, std::move(Socket)); } std::unique_ptr Connection{m_AvailableConnections.back()}; m_AvailableConnections.pop_back(); return std::move(Connection); } void HttpConnectionPool::ReturnConnection(std::unique_ptr&& Connection) { zen::RwLock::ExclusiveLockScope ScopedLock(m_Lock); m_AvailableConnections.emplace_back(Connection.release()); } ////////////////////////////////////////////////////////////////////////// class HttpContext { public: HttpContext(asio::io_context& Context) : m_Context(Context) {} ~HttpContext() = default; std::unique_ptr GetConnection(std::string_view HostName, uint16_t Port) { return ConnectionPool(HostName, Port).GetConnection(); } void ReturnConnection(std::unique_ptr Connection) { Connection->ConnectionPool().ReturnConnection(std::move(Connection)); } HttpConnectionPool& ConnectionPool(std::string_view HostName, uint16_t Port) { zen::RwLock::ExclusiveLockScope _(m_Lock); ConnectionId ConnId{std::string(HostName), Port}; if (auto It = m_ConnectionPools.find(ConnId); It == end(m_ConnectionPools)) { // Not found - create new entry auto In = m_ConnectionPools.insert({ConnId, std::move(HttpConnectionPool(m_Context, HostName, Port))}); return In.first->second; } else { return It->second; } } private: asio::io_context& m_Context; struct ConnectionId { inline bool operator<(const ConnectionId& Rhs) const { if (HostName != Rhs.HostName) { return HostName < Rhs.HostName; } return Port < Rhs.Port; } std::string HostName; uint16_t Port; }; zen::RwLock m_Lock; std::map m_ConnectionPools; }; ////////////////////////////////////////////////////////////////////////// class HttpClientRequest { public: HttpClientRequest(HttpContext& Context) : m_HttpContext(Context) {} ~HttpClientRequest() { if (m_Connection) { m_HttpContext.ReturnConnection(std::move(m_Connection)); } } void Get(const std::string_view Url) { http_parser_url ParsedUrl; int ErrCode = http_parser_parse_url(Url.data(), Url.size(), 0, &ParsedUrl); if (ErrCode) { ZEN_NOT_IMPLEMENTED(); } if ((ParsedUrl.field_set & (UF_HOST | UF_PORT | UF_PATH)) != (UF_HOST | UF_PORT | UF_PATH)) { // Bad URL } std::string_view HostName(Url.data() + ParsedUrl.field_data[UF_HOST].off, ParsedUrl.field_data[UF_HOST].len); std::string_view Path(Url.data() + ParsedUrl.field_data[UF_PATH].off); m_Connection = m_HttpContext.GetConnection(HostName, ParsedUrl.port); m_Connection->Get(HostName, ParsedUrl.port, Path); } private: HttpContext& m_HttpContext; std::unique_ptr m_Connection; }; ////////////////////////////////////////////////////////////////////////// // // Custom logging -- test code, this should be tweaked // namespace logging { using namespace spdlog; using namespace spdlog::details; using namespace std::literals; class full_formatter final : public spdlog::formatter { public: full_formatter(std::string_view LogId, std::chrono::time_point Epoch) : m_Epoch(Epoch), m_LogId(LogId) {} virtual std::unique_ptr clone() const override { return std::make_unique(m_LogId, m_Epoch); } static constexpr bool UseDate = false; virtual void format(const details::log_msg& msg, memory_buf_t& dest) override { using std::chrono::duration_cast; using std::chrono::milliseconds; using std::chrono::seconds; if constexpr (UseDate) { auto secs = std::chrono::duration_cast(msg.time.time_since_epoch()); if (secs != m_LastLogSecs) { m_CachedTm = os::localtime(log_clock::to_time_t(msg.time)); m_LastLogSecs = secs; } } const auto& tm_time = m_CachedTm; // cache the date/time part for the next second. auto duration = msg.time - m_Epoch; auto secs = duration_cast(duration); if (m_CacheTimestamp != secs || m_CachedDatetime.size() == 0) { m_CachedDatetime.clear(); m_CachedDatetime.push_back('['); if constexpr (UseDate) { fmt_helper::append_int(tm_time.tm_year + 1900, m_CachedDatetime); m_CachedDatetime.push_back('-'); fmt_helper::pad2(tm_time.tm_mon + 1, m_CachedDatetime); m_CachedDatetime.push_back('-'); fmt_helper::pad2(tm_time.tm_mday, m_CachedDatetime); m_CachedDatetime.push_back(' '); fmt_helper::pad2(tm_time.tm_hour, m_CachedDatetime); m_CachedDatetime.push_back(':'); fmt_helper::pad2(tm_time.tm_min, m_CachedDatetime); m_CachedDatetime.push_back(':'); fmt_helper::pad2(tm_time.tm_sec, m_CachedDatetime); } else { int Count = int(secs.count()); const int LogSecs = Count % 60; Count /= 60; const int LogMins = Count % 60; Count /= 60; const int LogHours = Count; fmt_helper::pad2(LogHours, m_CachedDatetime); m_CachedDatetime.push_back(':'); fmt_helper::pad2(LogMins, m_CachedDatetime); m_CachedDatetime.push_back(':'); fmt_helper::pad2(LogSecs, m_CachedDatetime); } m_CachedDatetime.push_back('.'); m_CacheTimestamp = secs; } dest.append(m_CachedDatetime.begin(), m_CachedDatetime.end()); auto millis = fmt_helper::time_fraction(msg.time); fmt_helper::pad3(static_cast(millis.count()), dest); dest.push_back(']'); dest.push_back(' '); if (!m_LogId.empty()) { dest.push_back('['); fmt_helper::append_string_view(m_LogId, dest); dest.push_back(']'); dest.push_back(' '); } // append logger name if exists if (msg.logger_name.size() > 0) { dest.push_back('['); fmt_helper::append_string_view(msg.logger_name, dest); dest.push_back(']'); dest.push_back(' '); } dest.push_back('['); // wrap the level name with color msg.color_range_start = dest.size(); fmt_helper::append_string_view(level::to_string_view(msg.level), dest); msg.color_range_end = dest.size(); dest.push_back(']'); dest.push_back(' '); // add source location if present if (!msg.source.empty()) { dest.push_back('['); const char* filename = details::short_filename_formatter::basename(msg.source.filename); fmt_helper::append_string_view(filename, dest); dest.push_back(':'); fmt_helper::append_int(msg.source.line, dest); dest.push_back(']'); dest.push_back(' '); } fmt_helper::append_string_view(msg.payload, dest); fmt_helper::append_string_view("\n"sv, dest); } private: std::chrono::time_point m_Epoch; std::tm m_CachedTm; std::chrono::seconds m_LastLogSecs; std::chrono::seconds m_CacheTimestamp{0}; memory_buf_t m_CachedDatetime; std::string m_LogId; }; } // namespace logging ////////////////////////////////////////////////////////////////////////// #if 0 # include # pragma comment(lib, "Crypt32.lib") # pragma comment(lib, "Wldap32.lib") int main() { mi_version(); zen::Sleep(1000); zen::Stopwatch timer; const int RequestCount = 100000; cpr::Session Sessions[10]; for (auto& Session : Sessions) { Session.SetUrl(cpr::Url{"http://localhost:1337/test/hello"}); //Session.SetUrl(cpr::Url{ "http://arn-wd-l0182:1337/test/hello" }); } auto Run = [](cpr::Session& Session) { for (int i = 0; i < 10000; ++i) { cpr::Response Result = Session.Get(); if (Result.status_code != 200) { spdlog::warn("request response: {}", Result.status_code); } } }; Concurrency::parallel_invoke([&] { Run(Sessions[0]); }, [&] { Run(Sessions[1]); }, [&] { Run(Sessions[2]); }, [&] { Run(Sessions[3]); }, [&] { Run(Sessions[4]); }, [&] { Run(Sessions[5]); }, [&] { Run(Sessions[6]); }, [&] { Run(Sessions[7]); }, [&] { Run(Sessions[8]); }, [&] { Run(Sessions[9]); }); // cpr::Response r = cpr::Get(cpr::Url{ "http://localhost:1337/test/hello" }); spdlog::info("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(timer.getElapsedTimeMs()), zen::NiceRate(RequestCount, (uint32_t)timer.getElapsedTimeMs(), "req")); return 0; } #elif 0 //#include int main() { mi_version(); restinio::run(restinio::on_thread_pool(32).port(8080).request_handler( [](auto req) { return req->create_response().set_body("Hello, World!").done(); })); return 0; } #else ZenTestEnvironment TestEnv; int main(int argc, char** argv) { mi_version(); zencore_forcelinktests(); spdlog::set_level(spdlog::level::debug); spdlog::set_formatter(std::make_unique("test", std::chrono::system_clock::now())); std::filesystem::path ProgramBaseDir = std::filesystem::path(argv[0]).parent_path(); std::filesystem::path TestBaseDir = ProgramBaseDir.parent_path().parent_path() / ".test"; TestEnv.Initialize(ProgramBaseDir, TestBaseDir); spdlog::info("Running tests..."); return doctest::Context(argc, argv).run(); } # if 1 TEST_CASE("asio.http") { std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); ZenServerInstance Instance(TestEnv); Instance.SetTestDir(TestDir); Instance.SpawnServer(13337); spdlog::info("Waiting..."); Instance.WaitUntilReady(); // asio test asio::io_context IoContext; HttpContext HttpCtx(IoContext); HttpClientRequest Request(HttpCtx); Request.Get("http://localhost:13337/test/hello"); IoContext.run(); } # endif TEST_CASE("default.single") { std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); ZenServerInstance Instance(TestEnv); Instance.SetTestDir(TestDir); Instance.SpawnServer(13337); spdlog::info("Waiting..."); Instance.WaitUntilReady(); std::atomic RequestCount{0}; std::atomic BatchCounter{0}; spdlog::info("Running single server test..."); auto IssueTestRequests = [&] { const uint64_t BatchNo = BatchCounter.fetch_add(1); const DWORD ThreadId = GetCurrentThreadId(); spdlog::info("query batch {} started (thread {})", BatchNo, ThreadId); cpr::Session cli; cli.SetUrl(cpr::Url{"http://localhost:13337/test/hello"}); for (int i = 0; i < 10000; ++i) { auto res = cli.Get(); ++RequestCount; } spdlog::info("query batch {} ended (thread {})", BatchNo, ThreadId); }; auto fun10 = [&] { Concurrency::parallel_invoke(IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests); }; zen::Stopwatch timer; // Concurrency::parallel_invoke(fun10, fun10, fun, fun, fun, fun, fun, fun, fun, fun); Concurrency::parallel_invoke(IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests, IssueTestRequests); uint64_t Elapsed = timer.getElapsedTimeMs(); spdlog::info("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req")); } TEST_CASE("multi.basic") { ZenServerInstance Instance1(TestEnv); std::filesystem::path TestDir1 = TestEnv.CreateNewTestDir(); Instance1.SetTestDir(TestDir1); Instance1.SpawnServer(13337); ZenServerInstance Instance2(TestEnv); std::filesystem::path TestDir2 = TestEnv.CreateNewTestDir(); Instance2.SetTestDir(TestDir2); Instance2.SpawnServer(13338); spdlog::info("Waiting..."); Instance1.WaitUntilReady(); Instance2.WaitUntilReady(); std::atomic RequestCount{0}; std::atomic BatchCounter{0}; auto IssueTestRequests = [&](int PortNumber) { const uint64_t BatchNo = BatchCounter.fetch_add(1); const DWORD ThreadId = GetCurrentThreadId(); spdlog::info("query batch {} started (thread {}) for port {}", BatchNo, ThreadId, PortNumber); cpr::Session cli; cli.SetUrl(cpr::Url{"http://localhost:{}/test/hello"_format(PortNumber)}); for (int i = 0; i < 10000; ++i) { auto res = cli.Get(); ++RequestCount; } spdlog::info("query batch {} ended (thread {})", BatchNo, ThreadId); }; zen::Stopwatch timer; spdlog::info("Running multi-server test..."); Concurrency::parallel_invoke([&] { IssueTestRequests(13337); }, [&] { IssueTestRequests(13338); }, [&] { IssueTestRequests(13337); }, [&] { IssueTestRequests(13338); }); uint64_t Elapsed = timer.getElapsedTimeMs(); spdlog::info("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req")); } TEST_CASE("cas.basic") { std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); const uint16_t PortNumber = 13337; const int IterationCount = 1000; std::vector ChunkSizes(IterationCount); std::vector ChunkHashes(IterationCount); { ZenServerInstance Instance1(TestEnv); Instance1.SetTestDir(TestDir); Instance1.SpawnServer(PortNumber); Instance1.WaitUntilReady(); std::atomic RequestCount{0}; std::atomic BatchCounter{0}; zen::Stopwatch timer; std::mt19937_64 mt; auto BaseUri = "http://localhost:{}/cas"_format(PortNumber); cpr::Session cli; cli.SetUrl(cpr::Url{BaseUri}); // Populate CAS with some generated data for (int i = 0; i < IterationCount; ++i) { const int ChunkSize = mt() % 10000 + 5; std::string body = fmt::format("{}", i); body.resize(ChunkSize, ' '); ChunkSizes[i] = ChunkSize; ChunkHashes[i] = zen::IoHash::HashMemory(body.data(), body.size()); cli.SetBody(body); auto res = cli.Post(); CHECK(!res.error); ++RequestCount; } // Verify that the chunks persisted for (int i = 0; i < IterationCount; ++i) { zen::ExtendableStringBuilder<128> Uri; Uri << BaseUri << "/"; ChunkHashes[i].ToHexString(Uri); auto res = cpr::Get(cpr::Url{Uri.c_str()}); CHECK(!res.error); CHECK(res.status_code == 200); CHECK(res.text.size() == ChunkSizes[i]); zen::IoHash Hash = zen::IoHash::HashMemory(res.text.data(), res.text.size()); CHECK(ChunkHashes[i] == Hash); ++RequestCount; } uint64_t Elapsed = timer.getElapsedTimeMs(); spdlog::info("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req")); } // Verify that the data persists between process runs (the previous server has exited at this point) { ZenServerInstance Instance2(TestEnv); Instance2.SetTestDir(TestDir); Instance2.SpawnServer(PortNumber); Instance2.WaitUntilReady(); for (int i = 0; i < IterationCount; ++i) { zen::ExtendableStringBuilder<128> Uri; Uri << "http://localhost:{}/cas/"_format(PortNumber); ChunkHashes[i].ToHexString(Uri); auto res = cpr::Get(cpr::Url{Uri.c_str()}); CHECK(res.status_code == 200); CHECK(res.text.size() == ChunkSizes[i]); zen::IoHash Hash = zen::IoHash::HashMemory(res.text.data(), res.text.size()); CHECK(ChunkHashes[i] == Hash); } } } TEST_CASE("project.basic") { using namespace std::literals; std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); const uint16_t PortNumber = 13337; ZenServerInstance Instance1(TestEnv); Instance1.SetTestDir(TestDir); Instance1.SpawnServer(PortNumber); Instance1.WaitUntilReady(); std::atomic RequestCount{0}; zen::Stopwatch timer; std::mt19937_64 mt; zen::StringBuilder<64> BaseUri; BaseUri << "http://localhost:{}/prj/test"_format(PortNumber); SUBCASE("build store init") { { { zen::CbObjectWriter Body; Body << "id" << "test"; Body << "root" << "/zooom"; Body << "project" << "/zooom"; Body << "engine" << "/zooom"; zen::MemoryOutStream MemOut; zen::BinaryWriter Writer{MemOut}; Body.Save(Writer); auto Response = cpr::Post(cpr::Url{BaseUri.c_str()}, cpr::Body{(const char*)MemOut.Data(), MemOut.Size()}); CHECK(Response.status_code == 201); } { auto Response = cpr::Get(cpr::Url{BaseUri.c_str()}); CHECK(Response.status_code == 200); zen::CbObjectView ResponseObject = zen::CbFieldView(Response.text.data()).AsObjectView(); CHECK(ResponseObject["id"].AsString() == "test"sv); CHECK(ResponseObject["root"].AsString() == "/zooom"sv); } } BaseUri << "/oplog/ps5"; { { zen::StringBuilder<64> PostUri; PostUri << BaseUri; auto Response = cpr::Post(cpr::Url{PostUri.c_str()}); CHECK(Response.status_code == 201); } { auto Response = cpr::Get(cpr::Url{BaseUri.c_str()}); CHECK(Response.status_code == 200); zen::CbObjectView ResponseObject = zen::CbFieldView(Response.text.data()).AsObjectView(); CHECK(ResponseObject["id"].AsString() == "ps5"sv); CHECK(ResponseObject["project"].AsString() == "test"sv); } } SUBCASE("build store persistence") { uint8_t AttachData[] = {1, 2, 3}; zen::CbAttachment Attach{zen::SharedBuffer::Clone(zen::MemoryView{AttachData, 3})}; zen::CbObjectWriter OpWriter; OpWriter << "key" << "foo" << "attachment" << Attach; const std::string_view ChunkId{ "00000000" "00000000" "00010000"}; auto FileOid = zen::Oid::FromHexString(ChunkId); OpWriter.BeginArray("files"); OpWriter.BeginObject(); OpWriter << "id" << FileOid; OpWriter << "path" << __FILE__; OpWriter.EndObject(); OpWriter.EndArray(); OpWriter.BeginArray("serverfiles"); OpWriter.BeginObject(); OpWriter << "id" << FileOid; OpWriter << "path" << __FILE__; OpWriter.EndObject(); OpWriter.EndArray(); zen::CbObject Op = OpWriter.Save(); zen::MemoryOutStream MemOut; zen::BinaryWriter Writer(MemOut); zen::CbPackage OpPackage(Op); OpPackage.AddAttachment(Attach); OpPackage.Save(Writer); { zen::StringBuilder<64> PostUri; PostUri << BaseUri << "/new"; auto Response = cpr::Post(cpr::Url{PostUri.c_str()}, cpr::Body{(const char*)MemOut.Data(), MemOut.Size()}); REQUIRE(!Response.error); CHECK(Response.status_code == 201); } // Read file data { zen::StringBuilder<128> ChunkGetUri; ChunkGetUri << BaseUri << "/" << ChunkId; auto Response = cpr::Get(cpr::Url{ChunkGetUri.c_str()}); REQUIRE(!Response.error); CHECK(Response.status_code == 200); } { zen::StringBuilder<128> ChunkGetUri; ChunkGetUri << BaseUri << "/" << ChunkId << "?offset=1&size=10"; auto Response = cpr::Get(cpr::Url{ChunkGetUri.c_str()}); REQUIRE(!Response.error); CHECK(Response.status_code == 200); CHECK(Response.text.size() == 10); } spdlog::info("+++++++"); } SUBCASE("build store op commit") { spdlog::info("-------"); } } const uint64_t Elapsed = timer.getElapsedTimeMs(); spdlog::info("{} requests in {} ({})", RequestCount, zen::NiceTimeSpanMs(Elapsed), zen::NiceRate(RequestCount, (uint32_t)Elapsed, "req")); } # if 0 // this is extremely WIP TEST_CASE("project.pipe") { using namespace std::literals; std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); const uint16_t PortNumber = 13337; ZenServerInstance Instance1(TestEnv); Instance1.SetTestDir(TestDir); Instance1.SpawnServer(PortNumber); Instance1.WaitUntilReady(); zen::LocalProjectClient LocalClient(PortNumber); zen::CbObjectWriter Cbow; Cbow << "hey" << 42; zen::CbObject Response = LocalClient.MessageTransaction(Cbow.Save()); } # endif TEST_CASE("z$.basic") { using namespace std::literals; std::filesystem::path TestDir = TestEnv.CreateNewTestDir(); const uint16_t PortNumber = 13337; const int kIterationCount = 100; const auto BaseUri = "http://localhost:{}/z$"_format(PortNumber); { ZenServerInstance Instance1(TestEnv); Instance1.SetTestDir(TestDir); Instance1.SpawnServer(PortNumber); Instance1.WaitUntilReady(); // Populate with some simple data for (int i = 0; i < kIterationCount; ++i) { zen::CbObjectWriter Cbo; Cbo << "index" << i; zen::MemoryOutStream MemOut; zen::BinaryWriter Writer{MemOut}; Cbo.Save(Writer); zen::IoHash Key = zen::IoHash::HashMemory(&i, sizeof i); cpr::Response Result = cpr::Put(cpr::Url{"{}/{}/{}"_format(BaseUri, "test", Key)}, cpr::Body{(const char*)MemOut.Data(), MemOut.Size()}, cpr::Header{{"Content-Type", "application/x-ue-cb"}}); CHECK(Result.status_code == 201); } // Retrieve data for (int i = 0; i < kIterationCount; ++i) { zen::IoHash Key = zen::IoHash::HashMemory(&i, sizeof i); cpr::Response Result = cpr::Get(cpr::Url{"{}/{}/{}"_format(BaseUri, "test", Key)}); CHECK(Result.status_code == 200); } } // Verify that the data persists between process runs (the previous server has exited at this point) { ZenServerInstance Instance1(TestEnv); Instance1.SetTestDir(TestDir); Instance1.SpawnServer(PortNumber); Instance1.WaitUntilReady(); // Retrieve data again for (int i = 0; i < kIterationCount; ++i) { zen::IoHash Key = zen::IoHash::HashMemory(&i, sizeof i); cpr::Response Result = cpr::Get(cpr::Url{"{}/{}/{}"_format(BaseUri, "test", Key)}); CHECK(Result.status_code == 200); } } } #endif