Added connect/transfer timeout options for Jupiter client.

1 files changed, 28 insertions, 19 deletions

diff --git a/zenserver/upstream/jupiter.cpp b/zenserver/upstream/jupiter.cpp
index 4caa5c8df..9223ea0f4 100644
--- a/zenserver/upstream/jupiter.cpp
+++ b/zenserver/upstream/jupiter.cpp
@@ -46,13 +46,20 @@ namespace detail {
 
 		void InvalidateAccessToken() { AccessToken = {}; }
 
-		void Reset()
+		void Reset(std::chrono::milliseconds ConnectTimeout, std::chrono::milliseconds Timeout)
 		{
 			Session.SetBody({});
 			Session.SetHeader({});
+			Session.SetConnectTimeout(ConnectTimeout);
+			Session.SetTimeout(Timeout);
 			AccessToken = GetAccessToken();
 		}
 
+		cpr::Session& GetSession() { return Session; }
+
+	private:
+		friend class CloudCacheClient;
+
 		CloudCacheClient&	  OwnerClient;
 		CloudCacheAccessToken AccessToken;
 		cpr::Session		  Session;
@@ -89,7 +96,7 @@ CloudCacheSession::GetDerivedData(std::string_view BucketId, std::string_view Ke
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -133,7 +140,7 @@ CloudCacheSession::GetRef(std::string_view BucketId, const IoHash& Key, ZenConte
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", ContentType}});
@@ -169,7 +176,7 @@ CloudCacheSession::GetBlob(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -206,7 +213,7 @@ CloudCacheSession::GetCompressedBlob(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-comp"}});
@@ -242,7 +249,7 @@ CloudCacheSession::GetObject(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -280,7 +287,7 @@ CloudCacheSession::PutDerivedData(std::string_view BucketId, std::string_view Ke
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/c/ddc/" << m_CacheClient->DdcNamespace() << "/" << BucketId << "/" << Key;
 
-	auto& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
@@ -331,7 +338,7 @@ CloudCacheSession::PutRef(std::string_view BucketId, const IoHash& Key, IoBuffer
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(
@@ -394,7 +401,7 @@ CloudCacheSession::FinalizeRef(std::string_view BucketId, const IoHash& Key, con
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString() << "/finalize/" << RefHash.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value},
@@ -454,7 +461,7 @@ CloudCacheSession::PutBlob(const IoHash& Key, IoBuffer Blob)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/octet-stream"}});
@@ -489,7 +496,7 @@ CloudCacheSession::PutCompressedBlob(const IoHash& Key, IoBuffer Blob)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compressed-blobs/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-comp"}});
@@ -524,7 +531,7 @@ CloudCacheSession::PutObject(const IoHash& Key, IoBuffer Object)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
@@ -560,7 +567,7 @@ CloudCacheSession::RefExists(std::string_view BucketId, const IoHash& Key)
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/refs/" << m_CacheClient->BlobStoreNamespace() << "/" << BucketId << "/"
 		<< Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
@@ -629,7 +636,7 @@ CloudCacheSession::PostComputeTasks(std::string_view ChannelId, IoBuffer TasksDa
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId;
 
-	auto& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Content-Type", "application/x-ue-cb"}});
@@ -662,7 +669,7 @@ CloudCacheSession::GetComputeUpdates(std::string_view ChannelId, const uint32_t
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/compute/" << ChannelId << "/updates?wait=" << WaitSeconds;
 
-	auto& Session = m_SessionState->Session;
+	auto& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -697,7 +704,7 @@ CloudCacheSession::GetObjectTree(const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/objects/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString() << "/tree";
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/octet-stream"}});
@@ -761,7 +768,7 @@ CloudCacheSession::CacheTypeExists(std::string_view TypeId, const IoHash& Key)
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/" << Key.ToHexString();
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}});
@@ -799,7 +806,7 @@ CloudCacheSession::CacheTypeExists(std::string_view TypeId, const std::set<IoHas
 	ExtendableStringBuilder<256> Uri;
 	Uri << m_CacheClient->ServiceUrl() << "/api/v1/" << TypeId << "/" << m_CacheClient->BlobStoreNamespace() << "/exists?" << Query;
 
-	cpr::Session& Session = m_SessionState->Session;
+	cpr::Session& Session = m_SessionState->GetSession();
 
 	Session.SetOption(cpr::Url{Uri.c_str()});
 	Session.SetOption(cpr::Header{{"Authorization", AccessToken.Value}, {"Accept", "application/x-ue-cb"}});
@@ -853,6 +860,8 @@ CloudCacheClient::CloudCacheClient(const CloudCacheClientOptions& Options)
 , m_OAuthClientId(Options.OAuthClientId)
 , m_OAuthSecret(Options.OAuthSecret)
 , m_AccessToken(Options.AccessToken)
+, m_ConnectTimeout(Options.ConnectTimeout)
+, m_Timeout(Options.Timeout)
 {
 	if (!Options.AccessToken.empty())
 	{
@@ -955,7 +964,7 @@ CloudCacheClient::AllocSessionState()
 		State = new detail::CloudCacheSessionState(*this);
 	}
 
-	State->Reset();
+	State->Reset(m_ConnectTimeout, m_Timeout);
 
 	return State;
 }