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

1 files changed, 250 insertions, 18 deletions

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