Merged from upstream

3 files changed, 10 insertions, 69 deletions

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index a183e7ebf..ba5991b3f 100644
--- a/zenserver/upstream/upstreamcache.cpp
+++ b/zenserver/upstream/upstreamcache.cpp
@@ -8,6 +8,7 @@
 #include <zencore/compactbinarybuilder.h>
 #include <zencore/compactbinarypackage.h>
 #include <zencore/compactbinaryvalidation.h>
+#include <zencore/blockingqueue.h>
 #include <zencore/fmtutils.h>
 #include <zencore/stats.h>
 #include <zencore/stream.h>
@@ -23,7 +24,6 @@
 
 #include <algorithm>
 #include <atomic>
-#include <deque>
 #include <thread>
 #include <unordered_map>
 
@@ -33,70 +33,6 @@ using namespace std::literals;
 
 namespace detail {
 
-	template<typename T>
-	class BlockingQueue
-	{
-	public:
-		BlockingQueue() = default;
-
-		~BlockingQueue() { CompleteAdding(); }
-
-		void Enqueue(T&& Item)
-		{
-			{
-				std::lock_guard Lock(m_Lock);
-				m_Queue.emplace_back(std::move(Item));
-				m_Size++;
-			}
-
-			m_NewItemSignal.notify_one();
-		}
-
-		bool WaitAndDequeue(T& Item)
-		{
-			if (m_CompleteAdding.load())
-			{
-				return false;
-			}
-
-			std::unique_lock Lock(m_Lock);
-			m_NewItemSignal.wait(Lock, [this]() { return !m_Queue.empty() || m_CompleteAdding.load(); });
-
-			if (!m_Queue.empty())
-			{
-				Item = std::move(m_Queue.front());
-				m_Queue.pop_front();
-				m_Size--;
-
-				return true;
-			}
-
-			return false;
-		}
-
-		void CompleteAdding()
-		{
-			if (!m_CompleteAdding.load())
-			{
-				m_CompleteAdding.store(true);
-				m_NewItemSignal.notify_all();
-			}
-		}
-
-		std::size_t Size() const
-		{
-			std::unique_lock Lock(m_Lock);
-			return m_Queue.size();
-		}
-
-	private:
-		mutable std::mutex		m_Lock;
-		std::condition_variable m_NewItemSignal;
-		std::deque<T>			m_Queue;
-		std::atomic_bool		m_CompleteAdding{false};
-		std::atomic_uint32_t	m_Size;
-	};
-
 	class JupiterUpstreamEndpoint final : public UpstreamEndpoint
 	{
 	public:
@@ -1029,7 +965,7 @@ private:
 
 	spdlog::logger& Log() { return m_Log; }
 
-	using UpstreamQueue = detail::BlockingQueue<UpstreamCacheRecord>;
+	using UpstreamQueue = BlockingQueue<UpstreamCacheRecord>;
 
 	struct RunState
 	{
diff --git a/zenserver/upstream/zen.cpp b/zenserver/upstream/zen.cpp
index c988a6b0b..6141fd397 100644
--- a/zenserver/upstream/zen.cpp
+++ b/zenserver/upstream/zen.cpp
@@ -328,7 +328,9 @@ namespace detail {
 
 //////////////////////////////////////////////////////////////////////////
 
-ZenStructuredCacheClient::ZenStructuredCacheClient(std::string_view ServiceUrl) : m_ServiceUrl(ServiceUrl)
+ZenStructuredCacheClient::ZenStructuredCacheClient(std::string_view ServiceUrl)
+: m_Log(logging::Get(std::string_view("zenclient")))
+, m_ServiceUrl(ServiceUrl)
 {
 }
 
@@ -369,7 +371,7 @@ ZenStructuredCacheClient::FreeSessionState(detail::ZenCacheSessionState* State)
 using namespace std::literals;
 
 ZenStructuredCacheSession::ZenStructuredCacheSession(ZenStructuredCacheClient& OuterClient)
-: m_Log(logging::Get("zenclient"sv))
+: m_Log(OuterClient.Log())
 , m_Client(OuterClient)
 {
 	m_SessionState = m_Client.AllocSessionState();
diff --git a/zenserver/upstream/zen.h b/zenserver/upstream/zen.h
index 158be668a..12e46bd8d 100644
--- a/zenserver/upstream/zen.h
+++ b/zenserver/upstream/zen.h
@@ -138,8 +138,11 @@ public:
 
 	std::string_view ServiceUrl() const { return m_ServiceUrl; }
 
+	inline spdlog::logger& Log() { return m_Log; }
+
 private:
-	std::string m_ServiceUrl;
+	spdlog::logger& m_Log;
+	std::string		m_ServiceUrl;
 
 	RwLock									 m_SessionStateLock;
 	std::list<detail::ZenCacheSessionState*> m_SessionStateCache;