247 complete httpclient implementation (#269)

* implemented HttpClient connection pooling
* implemented missing verbs
* added response helpers (CbObject/CbPackage/text)
* added RwLock::WithSharedLock and RwLock::WithExclusiveLock
* added some noexcept annotations on RwLock
* removed CPR dependency in httpclient.h

5 files changed, 310 insertions, 44 deletions

diff --git a/src/zencore/include/zencore/thread.h b/src/zencore/include/zencore/thread.h
index a9c96d422..09c25996f 100644
--- a/src/zencore/include/zencore/thread.h
+++ b/src/zencore/include/zencore/thread.h
@@ -25,18 +25,18 @@ void SetCurrentThreadName(std::string_view ThreadName);
 class RwLock
 {
 public:
-	ZENCORE_API void AcquireShared();
-	ZENCORE_API void ReleaseShared();
+	ZENCORE_API void AcquireShared() noexcept;
+	ZENCORE_API void ReleaseShared() noexcept;
 
-	ZENCORE_API void AcquireExclusive();
-	ZENCORE_API void ReleaseExclusive();
+	ZENCORE_API void AcquireExclusive() noexcept;
+	ZENCORE_API void ReleaseExclusive() noexcept;
 
 	struct SharedLockScope
 	{
 		SharedLockScope(RwLock& Lock) : m_Lock(&Lock) { Lock.AcquireShared(); }
 		~SharedLockScope() { ReleaseNow(); }
 
-		void ReleaseNow()
+		void ReleaseNow() noexcept
 		{
 			if (m_Lock)
 			{
@@ -49,12 +49,18 @@ public:
 		RwLock* m_Lock;
 	};
 
+	inline void WithSharedLock(auto&& Fun)
+	{
+		SharedLockScope $(*this);
+		Fun();
+	}
+
 	struct ExclusiveLockScope
 	{
 		ExclusiveLockScope(RwLock& Lock) : m_Lock(&Lock) { Lock.AcquireExclusive(); }
 		~ExclusiveLockScope() { ReleaseNow(); }
 
-		void ReleaseNow()
+		void ReleaseNow() noexcept
 		{
 			if (m_Lock)
 			{
@@ -67,6 +73,12 @@ public:
 		RwLock* m_Lock;
 	};
 
+	inline void WithExclusiveLock(auto&& Fun)
+	{
+		ExclusiveLockScope $(*this);
+		Fun();
+	}
+
 private:
 	std::shared_mutex m_Mutex;
 };
diff --git a/src/zencore/thread.cpp b/src/zencore/thread.cpp
index 1597a7dd9..86609c210 100644
--- a/src/zencore/thread.cpp
+++ b/src/zencore/thread.cpp
@@ -121,25 +121,25 @@ SetCurrentThreadName([[maybe_unused]] std::string_view ThreadName)
 }  // namespace zen
 
 void
-RwLock::AcquireShared()
+RwLock::AcquireShared() noexcept
 {
 	m_Mutex.lock_shared();
 }
 
 void
-RwLock::ReleaseShared()
+RwLock::ReleaseShared() noexcept
 {
 	m_Mutex.unlock_shared();
 }
 
 void
-RwLock::AcquireExclusive()
+RwLock::AcquireExclusive() noexcept
 {
 	m_Mutex.lock();
 }
 
 void
-RwLock::ReleaseExclusive()
+RwLock::ReleaseExclusive() noexcept
 {
 	m_Mutex.unlock();
 }
diff --git a/src/zenhttp/httpclient.cpp b/src/zenhttp/httpclient.cpp
index e6813d407..891ada83e 100644
--- a/src/zenhttp/httpclient.cpp
+++ b/src/zenhttp/httpclient.cpp
@@ -13,6 +13,10 @@
 #include <zencore/testing.h>
 #include <zenhttp/httpshared.h>
 
+ZEN_THIRD_PARTY_INCLUDES_START
+#include <cpr/cpr.h>
+ZEN_THIRD_PARTY_INCLUDES_END
+
 static std::atomic<uint32_t> HttpClientRequestIdCounter{0};
 
 namespace zen {
@@ -22,12 +26,154 @@ using namespace std::literals;
 HttpClient::Response
 FromCprResponse(cpr::Response& InResponse)
 {
-	return {.StatusCode = int(InResponse.status_code)};
+	return {.StatusCode = HttpResponseCode(InResponse.status_code)};
+}
+
+cpr::Body
+AsCprBody(const CbObject& Obj)
+{
+	return cpr::Body((const char*)Obj.GetBuffer().GetData(), Obj.GetBuffer().GetSize());
+}
+
+cpr::Body
+AsCprBody(const IoBuffer& Obj)
+{
+	return cpr::Body((const char*)Obj.GetData(), Obj.GetSize());
+}
+
+cpr::Body
+AsCprBody(const CompositeBuffer& Buffers)
+{
+	SharedBuffer Buffer = Buffers.Flatten();
+
+	// This is super inefficient, should be fixed
+	std::string String{(const char*)Buffer.GetData(), Buffer.GetSize()};
+	return cpr::Body{std::move(String)};
+}
+
+CbObject
+AsCbObject(cpr::Response& Response)
+{
+	const std::string& Data = Response.text;
+
+	return LoadCompactBinaryObject(IoBufferBuilder::MakeCloneFromMemory(Data.data(), Data.size()));
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+HttpClient::Response
+ResponseWithPayload(cpr::Response& HttpResponse, const HttpResponseCode WorkResponseCode)
+{
+	// This ends up doing a memcpy, would be good to get rid of it by streaming results
+	// into buffer directly
+	IoBuffer ResponseBuffer = IoBuffer(IoBuffer::Clone, HttpResponse.text.data(), HttpResponse.text.size());
+
+	if (auto It = HttpResponse.header.find("Content-Type"); It != HttpResponse.header.end())
+	{
+		const HttpContentType ContentType = ParseContentType(It->second);
+
+		ResponseBuffer.SetContentType(ContentType);
+	}
+
+	return HttpClient::Response{.StatusCode = WorkResponseCode, .ResponsePayload = std::move(ResponseBuffer)};
+}
+
+HttpClient::Response
+CommonResponse(cpr::Response&& HttpResponse)
+{
+	const HttpResponseCode WorkResponseCode = HttpResponseCode(HttpResponse.status_code);
+
+	if (WorkResponseCode == HttpResponseCode::NoContent || HttpResponse.text.empty())
+	{
+		return HttpClient::Response{.StatusCode = WorkResponseCode};
+	}
+	else
+	{
+		return ResponseWithPayload(HttpResponse, WorkResponseCode);
+	}
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+struct HttpClient::Impl : public RefCounted
+{
+	Impl();
+	~Impl();
+
+	// Session allocation
+
+	struct Session
+	{
+		Session(Impl* InOuter, cpr::Session* InSession) : Outer(InOuter), CprSession(InSession) {}
+		~Session() { Outer->ReleaseSession(CprSession); }
+
+		inline cpr::Session* operator->() const { return CprSession; }
+
+	private:
+		Impl*		  Outer;
+		cpr::Session* CprSession;
+
+		Session(Session&&) = delete;
+		Session& operator=(Session&&) = delete;
+	};
+
+	Session AllocSession(const std::string_view BaseUrl, const std::string_view Url);
+
+private:
+	RwLock					   m_SessionLock;
+	std::vector<cpr::Session*> m_Sessions;
+
+	void ReleaseSession(cpr::Session*);
+};
+
+HttpClient::Impl::Impl()
+{
+}
+
+HttpClient::Impl::~Impl()
+{
+	m_SessionLock.WithExclusiveLock([&] {
+		for (auto CprSession : m_Sessions)
+		{
+			delete CprSession;
+		}
+		m_Sessions.clear();
+	});
+}
+
+HttpClient::Impl::Session
+HttpClient::Impl::AllocSession(const std::string_view BaseUrl, const std::string_view ResourcePath)
+{
+	RwLock::ExclusiveLockScope _(m_SessionLock);
+
+	ExtendableStringBuilder<128> UrlBuffer;
+	UrlBuffer << BaseUrl << ResourcePath;
+
+	if (m_Sessions.empty())
+	{
+		cpr::Session* NewSession = new cpr::Session();
+		NewSession->SetUrl(UrlBuffer.c_str());
+		return Session(this, NewSession);
+	}
+	else
+	{
+		cpr::Session* NewSession = m_Sessions.back();
+		m_Sessions.pop_back();
+
+		NewSession->SetUrl(UrlBuffer.c_str());
+		return Session(this, NewSession);
+	}
+}
+
+void
+HttpClient::Impl::ReleaseSession(cpr::Session* CprSession)
+{
+	m_SessionLock.WithExclusiveLock([&] { m_Sessions.push_back(CprSession); });
 }
 
 //////////////////////////////////////////////////////////////////////////
 
-HttpClient::HttpClient(std::string_view BaseUri) : m_BaseUri(BaseUri)
+HttpClient::HttpClient(std::string_view BaseUri) : m_BaseUri(BaseUri), m_Impl(new Impl)
 {
 	StringBuilder<32> SessionId;
 	GetSessionId().ToString(SessionId);
@@ -41,8 +187,7 @@ HttpClient::~HttpClient()
 HttpClient::Response
 HttpClient::TransactPackage(std::string_view Url, CbPackage Package)
 {
-	cpr::Session Sess;
-	Sess.SetUrl(m_BaseUri + std::string(Url));
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
 
 	// First, list of offered chunks for filtering on the server end
 
@@ -69,10 +214,10 @@ HttpClient::TransactPackage(std::string_view Url, CbPackage Package)
 		BinaryWriter MemWriter;
 		Writer.Save(MemWriter);
 
-		Sess.SetHeader({{"Content-Type", "application/x-ue-offer"}, {"UE-Session", m_SessionId}, {"UE-Request", RequestIdString}});
-		Sess.SetBody(cpr::Body{(const char*)MemWriter.Data(), MemWriter.Size()});
+		Sess->SetHeader({{"Content-Type", "application/x-ue-offer"}, {"UE-Session", m_SessionId}, {"UE-Request", RequestIdString}});
+		Sess->SetBody(cpr::Body{(const char*)MemWriter.Data(), MemWriter.Size()});
 
-		cpr::Response FilterResponse = Sess.Post();
+		cpr::Response FilterResponse = Sess->Post();
 
 		if (FilterResponse.status_code == 200)
 		{
@@ -111,10 +256,10 @@ HttpClient::TransactPackage(std::string_view Url, CbPackage Package)
 	CompositeBuffer Message		= FormatPackageMessageBuffer(SendPackage);
 	SharedBuffer	FlatMessage = Message.Flatten();
 
-	Sess.SetHeader({{"Content-Type", "application/x-ue-cbpkg"}, {"UE-Session", m_SessionId}, {"UE-Request", RequestIdString}});
-	Sess.SetBody(cpr::Body{(const char*)FlatMessage.GetData(), FlatMessage.GetSize()});
+	Sess->SetHeader({{"Content-Type", "application/x-ue-cbpkg"}, {"UE-Session", m_SessionId}, {"UE-Request", RequestIdString}});
+	Sess->SetBody(cpr::Body{(const char*)FlatMessage.GetData(), FlatMessage.GetSize()});
 
-	cpr::Response FilterResponse = Sess.Post();
+	cpr::Response FilterResponse = Sess->Post();
 
 	if (!IsHttpSuccessCode(FilterResponse.status_code))
 	{
@@ -130,31 +275,118 @@ HttpClient::TransactPackage(std::string_view Url, CbPackage Package)
 		ResponseBuffer.SetContentType(ContentType);
 	}
 
-	return {.StatusCode = int(FilterResponse.status_code), .ResponsePayload = ResponseBuffer};
+	return {.StatusCode = HttpResponseCode(FilterResponse.status_code), .ResponsePayload = ResponseBuffer};
 }
 
+//////////////////////////////////////////////////////////////////////////
+//
+// Standard HTTP verbs
+//
+
 HttpClient::Response
-HttpClient::Put(std::string_view Url, IoBuffer Payload)
+HttpClient::Put(std::string_view Url, const IoBuffer& Payload)
 {
-	ZEN_UNUSED(Url);
-	ZEN_UNUSED(Payload);
-	return {};
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+	Sess->SetBody(AsCprBody(Payload));
+	Sess->SetHeader(cpr::Header{{"Content-Type", std::string(MapContentTypeToString(Payload.GetContentType()))}});
+
+	return CommonResponse(Sess->Put());
 }
 
 HttpClient::Response
 HttpClient::Get(std::string_view Url)
 {
-	ZEN_UNUSED(Url);
-	return {};
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+
+	return CommonResponse(Sess->Get());
 }
 
 HttpClient::Response
 HttpClient::Delete(std::string_view Url)
 {
-	ZEN_UNUSED(Url);
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+
+	return CommonResponse(Sess->Delete());
+}
+
+HttpClient::Response
+HttpClient::Post(std::string_view Url, const IoBuffer& Payload)
+{
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+
+	Sess->SetBody(AsCprBody(Payload));
+	Sess->SetHeader(cpr::Header{{"Content-Type", std::string(MapContentTypeToString(Payload.GetContentType()))}});
+
+	return CommonResponse(Sess->Post());
+}
+
+HttpClient::Response
+HttpClient::Post(std::string_view Url, CbObject Payload)
+{
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+
+	Sess->SetBody(AsCprBody(Payload));
+	Sess->SetHeader(cpr::Header{{"Content-Type", std::string(MapContentTypeToString(ZenContentType::kCbObject))}});
+
+	return CommonResponse(Sess->Post());
+}
+
+HttpClient::Response
+HttpClient::Post(std::string_view Url, CbPackage Pkg)
+{
+	CompositeBuffer Message = zen::FormatPackageMessageBuffer(Pkg);
+
+	Impl::Session Sess = m_Impl->AllocSession(m_BaseUri, Url);
+	Sess->SetBody(AsCprBody(Message));
+	Sess->SetHeader(cpr::Header{{"Content-Type", std::string(MapContentTypeToString(ZenContentType::kCbPackage))}});
+
+	return CommonResponse(Sess->Post());
+}
+
+//////////////////////////////////////////////////////////////////////////
+
+CbObject
+HttpClient::Response::AsObject()
+{
+	// TODO: sanity check the payload format etc
+
+	if (ResponsePayload)
+	{
+		return LoadCompactBinaryObject(ResponsePayload);
+	}
+
+	return {};
+}
+
+CbPackage
+HttpClient::Response::AsPackage()
+{
+	// TODO: sanity checks and error handling
+	if (ResponsePayload)
+	{
+		return ParsePackageMessage(ResponsePayload);
+	}
+
 	return {};
 }
 
+std::string_view
+HttpClient::Response::AsText()
+{
+	if (ResponsePayload)
+	{
+		return std::string_view(reinterpret_cast<const char*>(ResponsePayload.GetData()), ResponsePayload.GetSize());
+	}
+
+	return {};
+}
+
+bool
+HttpClient::Response::IsSuccess() const noexcept
+{
+	return IsHttpSuccessCode(StatusCode);
+}
+
 //////////////////////////////////////////////////////////////////////////
 
 #if ZEN_WITH_TESTS
diff --git a/src/zenhttp/include/zenhttp/httpclient.h b/src/zenhttp/include/zenhttp/httpclient.h
index 8316a9b9f..8ec29d548 100644
--- a/src/zenhttp/include/zenhttp/httpclient.h
+++ b/src/zenhttp/include/zenhttp/httpclient.h
@@ -8,10 +8,6 @@
 #include <zencore/uid.h>
 #include <zenhttp/httpcommon.h>
 
-ZEN_THIRD_PARTY_INCLUDES_START
-#include <cpr/cpr.h>
-ZEN_THIRD_PARTY_INCLUDES_END
-
 namespace zen {
 
 class CbPackage;
@@ -20,6 +16,7 @@ class CbPackage;
 
 	Currently simple and synchronous, should become lean and asynchronous
  */
+
 class HttpClient
 {
 public:
@@ -28,20 +25,39 @@ public:
 
 	struct Response
 	{
-		int		 StatusCode = 0;
-		IoBuffer ResponsePayload;  // Note: this also includes the content type
+		HttpResponseCode StatusCode = HttpResponseCode::ImATeapot;
+		IoBuffer		 ResponsePayload;  // Note: this also includes the content type
+
+		CbObject  AsObject();
+		CbPackage AsPackage();
+
+		// Return the response payload as a string. Note that this does not attempt to
+		// validate that the content type or content itself makes sense as a string.
+		std::string_view AsText();
+
+		bool   IsSuccess() const noexcept;
+		inline operator bool() const noexcept { return IsSuccess(); }
 	};
 
-	[[nodiscard]] Response Put(std::string_view Url, IoBuffer Payload);
+	[[nodiscard]] Response Put(std::string_view Url, const IoBuffer& Payload);
 	[[nodiscard]] Response Get(std::string_view Url);
-	[[nodiscard]] Response TransactPackage(std::string_view Url, CbPackage Package);
 	[[nodiscard]] Response Delete(std::string_view Url);
+	[[nodiscard]] Response Post(std::string_view Url, const IoBuffer& Payload);
+	[[nodiscard]] Response Post(std::string_view Url, CbObject Payload);
+	[[nodiscard]] Response Post(std::string_view Url, CbPackage Payload);
+
+	[[nodiscard]] Response TransactPackage(std::string_view Url, CbPackage Package);
+
+	inline std::string GetBaseUri() const { return m_BaseUri; }
 
 private:
+	struct Impl;
+
 	std::string m_BaseUri;
 	std::string m_SessionId;
+	Ref<Impl>	m_Impl;
 };
 
-}  // namespace zen
-
 void httpclient_forcelink();  // internal
+
+}  // namespace zen
diff --git a/src/zenhttp/include/zenhttp/httpcommon.h b/src/zenhttp/include/zenhttp/httpcommon.h
index 19fda8db4..fb0d128a2 100644
--- a/src/zenhttp/include/zenhttp/httpcommon.h
+++ b/src/zenhttp/include/zenhttp/httpcommon.h
@@ -30,12 +30,6 @@ extern HttpContentType (*ParseContentType)(const std::string_view& ContentTypeSt
 std::string_view ReasonStringForHttpResultCode(int HttpCode);
 bool			 TryParseHttpRangeHeader(std::string_view RangeHeader, HttpRanges& Ranges);
 
-[[nodiscard]] inline bool
-IsHttpSuccessCode(int HttpCode)
-{
-	return (HttpCode >= 200) && (HttpCode < 300);
-}
-
 enum class HttpVerb : uint8_t
 {
 	kGet	 = 1 << 0,
@@ -178,4 +172,16 @@ enum class HttpResponseCode
 	NetworkAuthenticationRequired = 511,  //!< Indicates that the client needs to authenticate to gain network access.
 };
 
+[[nodiscard]] inline bool
+IsHttpSuccessCode(int HttpCode) noexcept
+{
+	return (HttpCode >= 200) && (HttpCode < 300);
+}
+
+[[nodiscard]] inline bool
+IsHttpSuccessCode(HttpResponseCode HttpCode)
+{
+	return IsHttpSuccessCode(int(HttpCode));
+}
+
 }  // namespace zen