From c6df63ef7f970f2faec0b4f84cdb846f96dace48 Mon Sep 17 00:00:00 2001
From: Stefan Boberg <stefan@boberg.org>
Date: Tue, 5 Oct 2021 16:35:57 +0200
Subject: zencore: Added BlockingQueue (moved from upstreamcache.cpp)

---
 zenserver/upstream/upstreamcache.cpp | 67 ++----------------------------------
 1 file changed, 2 insertions(+), 65 deletions(-)

(limited to 'zenserver/upstream/upstreamcache.cpp')

diff --git a/zenserver/upstream/upstreamcache.cpp b/zenserver/upstream/upstreamcache.cpp
index a183e7ebf..15effb08f 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>
@@ -33,70 +34,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 +966,7 @@ private:
 
 	spdlog::logger& Log() { return m_Log; }
 
-	using UpstreamQueue = detail::BlockingQueue<UpstreamCacheRecord>;
+	using UpstreamQueue = BlockingQueue<UpstreamCacheRecord>;
 
 	struct RunState
 	{
-- 
cgit v1.2.3