Zen upstream support (#7)

1 files changed, 109 insertions, 21 deletions

diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index e9102ad45..3d4999e5d 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -8,6 +8,7 @@
 #include <zencore/stream.h>
 
 #include "cache/structuredcachestore.h"
+#include "diag/logging.h"
 
 // cpr ////////////////////////////////////////////////////////////////////
 //
@@ -20,7 +21,6 @@
 #include <cpr/cpr.h>
 #pragma warning(pop)
 
-#include <spdlog/spdlog.h>
 #include <xxhash.h>
 #include <gsl/gsl-lite.hpp>
 
@@ -322,6 +322,45 @@ namespace detail {
 		ZenStructuredCacheClient& OwnerClient;
 		cpr::Session			  Session;
 	};
+
+	static void LogResponse(spdlog::logger& Log, std::string_view Verb, const cpr::Response& Response)
+	{
+		using namespace std::literals;
+
+		std::string_view ContentType = "unknown"sv;
+		if (auto It = Response.header.find("Content-Type"); It != Response.header.end())
+		{
+			ContentType = It->second;
+		}
+
+		const uint64_t Bytes = Verb == "GET"sv ? Response.downloaded_bytes : Response.uploaded_bytes;
+
+		const bool IsBinary =
+			ContentType == "application/x-ue-cb"sv || ContentType == "application/x-ue-comp"sv || ContentType == "application/octet-stream";
+
+		if (IsBinary)
+		{
+			Log.debug("{} '{}', Status: '{}', Elapsed: '{}', Content-Type: '{}' '{}' Bytes, Reason: '{}'",
+					  Verb,
+					  Response.url.str(),
+					  Response.status_code,
+					  Response.elapsed,
+					  ContentType,
+					  Bytes,
+					  Response.reason);
+		}
+		else
+		{
+			Log.debug("{} '{}', Status: '{}', Elapsed: '{}', Content-Type: '{}': '{}', Reason: '{}'",
+					  Verb,
+					  Response.url.str(),
+					  Response.status_code,
+					  Response.elapsed,
+					  ContentType,
+					  Response.text,
+					  Response.reason);
+		}
+	}
 }  // namespace detail
 
 //////////////////////////////////////////////////////////////////////////
@@ -364,50 +403,99 @@ ZenStructuredCacheClient::FreeSessionState(detail::ZenCacheSessionState* State)
 
 //////////////////////////////////////////////////////////////////////////
 
-ZenStructuredCacheSession::ZenStructuredCacheSession(ZenStructuredCacheClient& OuterClient) : m_Client(OuterClient)
+using namespace std::literals;
+
+ZenStructuredCacheSession::ZenStructuredCacheSession(ZenStructuredCacheClient& OuterClient)
+: m_Log(zen::logging::Get("zenclient"sv))
+, m_Client(OuterClient)
 {
+	m_SessionState = m_Client.AllocSessionState();
 }
 
 ZenStructuredCacheSession::~ZenStructuredCacheSession()
 {
+	m_Client.FreeSessionState(m_SessionState);
 }
 
-IoBuffer
-ZenStructuredCacheSession::Get(std::string_view BucketId, std::string_view Key)
+ZenCacheResult
+ZenStructuredCacheSession::GetCacheRecord(std::string_view BucketId, const IoHash& Key, ZenContentType Type)
 {
-	ZEN_UNUSED(BucketId, Key);
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString();
+
+	cpr::Session& Session = m_SessionState->Session;
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(cpr::Header{{"Accept", Type == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream"}});
+
+	cpr::Response Response = Session.Get();
+	detail::LogResponse(m_Log, "GET"sv, Response);
+
+	if (Response.status_code == 200)
+	{
+		return {.Response = IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()), .Success = true};
+	}
 
 	return {};
 }
 
-void
-ZenStructuredCacheSession::Put(std::string_view BucketId, std::string_view Key, IoBuffer Data)
+ZenCacheResult
+ZenStructuredCacheSession::GetCachePayload(std::string_view BucketId, const IoHash& Key, const IoHash& PayloadId)
 {
-	ZEN_UNUSED(BucketId, Key, Data);
-}
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString() << "/" << PayloadId.ToHexString();
 
-// Structured cache operations
+	cpr::Session& Session = m_SessionState->Session;
 
-IoBuffer
-ZenStructuredCacheSession::Get(std::string_view BucketId, const IoHash& Key)
-{
-	ZEN_UNUSED(BucketId, Key);
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(cpr::Header{{"Accept", "application/x-ue-comp"}});
+
+	cpr::Response Response = Session.Get();
+	detail::LogResponse(m_Log, "GET"sv, Response);
+
+	if (Response.status_code == 200)
+	{
+		return {.Response = IoBufferBuilder::MakeCloneFromMemory(Response.text.data(), Response.text.size()), .Success = true};
+	}
 
 	return {};
 }
 
-IoBuffer
-ZenStructuredCacheSession::Get(std::string_view BucketId, const IoHash& Key, const IoHash& ContentId)
+ZenCacheResult
+ZenStructuredCacheSession::PutCacheRecord(std::string_view BucketId, const IoHash& Key, IoBuffer Value, ZenContentType Type)
 {
-	ZEN_UNUSED(BucketId, Key, ContentId);
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString();
 
-	return {};
+	cpr::Session& Session = m_SessionState->Session;
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(
+		cpr::Header{{"Content-Type", Type == ZenContentType::kCbObject ? "application/x-ue-cb" : "application/octet-stream"}});
+	Session.SetBody(cpr::Body{static_cast<const char*>(Value.Data()), Value.Size()});
+
+	cpr::Response Response = Session.Put();
+	detail::LogResponse(m_Log, "PUT"sv, Response);
+
+	return {.Success = Response.status_code == 200};
 }
 
-void
-ZenStructuredCacheSession::Put(std::string_view BucketId, const IoHash& Key, ZenCacheValue Data)
+ZenCacheResult
+ZenStructuredCacheSession::PutCachePayload(std::string_view BucketId, const IoHash& Key, const IoHash& PayloadId, IoBuffer Payload)
 {
-	ZEN_UNUSED(BucketId, Key, Data);
+	ExtendableStringBuilder<256> Uri;
+	Uri << m_Client.ServiceUrl() << "/z$/" << BucketId << "/" << Key.ToHexString() << "/" << PayloadId.ToHexString();
+
+	cpr::Session& Session = m_SessionState->Session;
+
+	Session.SetOption(cpr::Url{Uri.c_str()});
+	Session.SetHeader(cpr::Header{{"Content-Type", "application/x-ue-comp"}});
+	Session.SetBody(cpr::Body{static_cast<const char*>(Payload.Data()), Payload.Size()});
+
+	cpr::Response Response = Session.Put();
+	detail::LogResponse(m_Log, "PUT"sv, Response);
+
+	return {.Success = Response.status_code == 200};
 }
 
 }  // namespace zen