add ability to abort http requests (#586)

* add abort-check function to httpclient

1 files changed, 61 insertions, 20 deletions

diff --git a/src/zenhttp/clients/httpclientcpr.cpp b/src/zenhttp/clients/httpclientcpr.cpp
index 568106887..27cffd912 100644
--- a/src/zenhttp/clients/httpclientcpr.cpp
+++ b/src/zenhttp/clients/httpclientcpr.cpp
@@ -16,9 +16,9 @@
 namespace zen {
 
 HttpClientBase*
-CreateCprHttpClient(std::string_view BaseUri, const HttpClientSettings& ConnectionSettings)
+CreateCprHttpClient(std::string_view BaseUri, const HttpClientSettings& ConnectionSettings, std::function<bool()>&& CheckIfAbortFunction)
 {
-	return new CprHttpClient(BaseUri, ConnectionSettings);
+	return new CprHttpClient(BaseUri, ConnectionSettings, std::move(CheckIfAbortFunction));
 }
 
 static std::atomic<uint32_t> HttpClientRequestIdCounter{0};
@@ -291,12 +291,17 @@ DoWithRetry(
 	std::string_view							 SessionId,
 	std::function<cpr::Response()>&&			 Func,
 	uint8_t										 RetryCount,
+	std::function<bool()>&						 CheckIfAbortFunction,
 	std::function<bool(cpr::Response& Result)>&& Validate = [](cpr::Response&) { return true; })
 {
 	uint8_t		  Attempt = 0;
 	cpr::Response Result  = Func();
 	while (Attempt < RetryCount)
 	{
+		if (CheckIfAbortFunction && CheckIfAbortFunction())
+		{
+			return Result;
+		}
 		if (!ShouldRetry(Result))
 		{
 			if (Result.error || !IsHttpSuccessCode(Result.status_code))
@@ -320,12 +325,17 @@ static cpr::Response
 DoWithRetry(std::string_view						  SessionId,
 			std::function<cpr::Response()>&&		  Func,
 			std::unique_ptr<detail::TempPayloadFile>& PayloadFile,
-			uint8_t									  RetryCount)
+			uint8_t									  RetryCount,
+			std::function<bool()>&					  CheckIfAbortFunction)
 {
 	uint8_t		  Attempt = 0;
 	cpr::Response Result  = Func();
 	while (Attempt < RetryCount)
 	{
+		if (CheckIfAbortFunction && CheckIfAbortFunction())
+		{
+			return Result;
+		}
 		if (!ShouldRetry(Result))
 		{
 			if (Result.error || !IsHttpSuccessCode(Result.status_code))
@@ -353,8 +363,10 @@ HeaderContentType(ZenContentType ContentType)
 
 //////////////////////////////////////////////////////////////////////////
 
-CprHttpClient::CprHttpClient(std::string_view BaseUri, const HttpClientSettings& Connectionsettings)
-: HttpClientBase(BaseUri, Connectionsettings)
+CprHttpClient::CprHttpClient(std::string_view		   BaseUri,
+							 const HttpClientSettings& Connectionsettings,
+							 std::function<bool()>&&   CheckIfAbortFunction)
+: HttpClientBase(BaseUri, Connectionsettings, std::move(CheckIfAbortFunction))
 {
 }
 
@@ -562,7 +574,8 @@ CprHttpClient::Put(std::string_view Url, const IoBuffer& Payload, const KeyValue
 				Sess->UpdateHeader({HeaderContentType(Payload.GetContentType())});
 				return Sess.Put();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -583,7 +596,8 @@ CprHttpClient::Put(std::string_view Url, const KeyValueMap& Parameters)
 															  GetAccessToken());
 								  return Sess.Put();
 							  },
-							  m_ConnectionSettings.RetryCount));
+							  m_ConnectionSettings.RetryCount,
+							  m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -600,6 +614,7 @@ CprHttpClient::Get(std::string_view Url, const KeyValueMap& AdditionalHeader, co
 				return Sess.Get();
 			},
 			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction,
 			[](cpr::Response& Result) {
 				std::unique_ptr<detail::TempPayloadFile> NoTempFile;
 				return ValidatePayload(Result, NoTempFile);
@@ -619,7 +634,8 @@ CprHttpClient::Head(std::string_view Url, const KeyValueMap& AdditionalHeader)
 				Session Sess = AllocSession(m_BaseUri, Url, m_ConnectionSettings, AdditionalHeader, {}, m_SessionId, GetAccessToken());
 				return Sess.Head();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -635,7 +651,8 @@ CprHttpClient::Delete(std::string_view Url, const KeyValueMap& AdditionalHeader)
 				Session Sess = AllocSession(m_BaseUri, Url, m_ConnectionSettings, AdditionalHeader, {}, m_SessionId, GetAccessToken());
 				return Sess.Delete();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -652,7 +669,8 @@ CprHttpClient::Post(std::string_view Url, const KeyValueMap& AdditionalHeader, c
 					AllocSession(m_BaseUri, Url, m_ConnectionSettings, AdditionalHeader, Parameters, m_SessionId, GetAccessToken());
 				return Sess.Post();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -690,7 +708,8 @@ CprHttpClient::Post(std::string_view Url, const IoBuffer& Payload, ZenContentTyp
 				Sess->SetBody(AsCprBody(Payload));
 				return Sess.Post();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -709,7 +728,8 @@ CprHttpClient::Post(std::string_view Url, CbObject Payload, const KeyValueMap& A
 				Sess->UpdateHeader({HeaderContentType(ZenContentType::kCbObject)});
 				return Sess.Post();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -732,13 +752,18 @@ CprHttpClient::Post(std::string_view Url, const CompositeBuffer& Payload, ZenCon
 				Sess->UpdateHeader({HeaderContentType(ContentType)});
 
 				detail::CompositeBufferReadStream Reader(Payload, 512u * 1024u);
-				auto							  ReadCallback = [&Reader](char* buffer, size_t& size, intptr_t) {
+				auto							  ReadCallback = [this, &Reader](char* buffer, size_t& size, intptr_t) {
+					 if (m_CheckIfAbortFunction && m_CheckIfAbortFunction())
+					 {
+						 return false;
+					 }
 					 size = Reader.Read(buffer, size);
 					 return true;
 				};
 				return Sess.Post(cpr::ReadCallback(gsl::narrow<cpr::cpr_off_t>(Payload.GetSize()), ReadCallback));
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -759,7 +784,12 @@ CprHttpClient::Upload(std::string_view Url, const IoBuffer& Payload, const KeyVa
 				{
 					uint64_t					   Offset = 0;
 					detail::BufferedReadFileStream Buffer(FileRef.FileHandle, FileRef.FileChunkOffset, FileRef.FileChunkSize, 512u * 1024u);
-					auto						   ReadCallback = [&Payload, &Offset, &Buffer](char* buffer, size_t& size, intptr_t) {
+					auto						   ReadCallback = [this, &Payload, &Offset, &Buffer](char* buffer, size_t& size, intptr_t) {
+						  if (m_CheckIfAbortFunction && m_CheckIfAbortFunction())
+						  {
+							  return false;
+						  }
+
 						  size = Min<size_t>(size, Payload.GetSize() - Offset);
 						  Buffer.Read(buffer, size);
 						  Offset += size;
@@ -770,7 +800,8 @@ CprHttpClient::Upload(std::string_view Url, const IoBuffer& Payload, const KeyVa
 				Sess->SetBody(AsCprBody(Payload));
 				return Sess.Put();
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -787,13 +818,18 @@ CprHttpClient::Upload(std::string_view Url, const CompositeBuffer& Payload, ZenC
 				Sess->UpdateHeader({HeaderContentType(ContentType)});
 
 				detail::CompositeBufferReadStream Reader(Payload, 512u * 1024u);
-				auto							  ReadCallback = [&Reader](char* buffer, size_t& size, intptr_t) {
+				auto							  ReadCallback = [this, &Reader](char* buffer, size_t& size, intptr_t) {
+					 if (m_CheckIfAbortFunction && m_CheckIfAbortFunction())
+					 {
+						 return false;
+					 }
 					 size = Reader.Read(buffer, size);
 					 return true;
 				};
 				return Sess.Put(cpr::ReadCallback(gsl::narrow<cpr::cpr_off_t>(Payload.GetSize()), ReadCallback));
 			},
-			m_ConnectionSettings.RetryCount));
+			m_ConnectionSettings.RetryCount,
+			m_CheckIfAbortFunction));
 }
 
 CprHttpClient::Response
@@ -825,6 +861,10 @@ CprHttpClient::Download(std::string_view Url, const std::filesystem::path& TempF
 			   };
 
 			   auto DownloadCallback = [&](std::string data, intptr_t) {
+				   if (m_CheckIfAbortFunction && m_CheckIfAbortFunction())
+				   {
+					   return false;
+				   }
 				   if (PayloadFile)
 				   {
 					   ZEN_ASSERT(PayloadString.empty());
@@ -1027,9 +1067,10 @@ CprHttpClient::Download(std::string_view Url, const std::filesystem::path& TempF
 			   return Response;
 		   },
 		   PayloadFile,
-		   m_ConnectionSettings.RetryCount);
+		   m_ConnectionSettings.RetryCount,
+		   m_CheckIfAbortFunction);
 
 	return CommonResponse(m_SessionId, std::move(Response), PayloadFile ? PayloadFile->DetachToIoBuffer() : IoBuffer{});
 }
 
-}  // namespace zen
\ No newline at end of file
+}  // namespace zen