|
/********************************************************************
|
|
* CONFIDENTIAL - All source code is the propriety and confidential *
|
|
* information of David Sorber. *
|
|
* *
|
|
* Copyright 2024 David Sorber *
|
|
* Unpublished -- all rights reserved under the copyright laws *
|
|
* of the United States. *
|
|
********************************************************************/
|
|
|
|
#ifndef __DBS_PROTECTEDQUEUE_H__
|
|
#define __DBS_PROTECTEDQUEUE_H__
|
|
|
|
#include <condition_variable>
|
|
#include <functional>
|
|
#include <iostream>
|
|
#include <queue>
|
|
#include <mutex>
|
|
|
|
|
|
namespace DBS
|
|
{
|
|
|
|
template<class T>
|
|
class ProtectedQueue
|
|
{
|
|
public:
|
|
|
|
ProtectedQueue()
|
|
: m_terminateFlag(false),
|
|
p_itemAddedCallback(nullptr),
|
|
p_itemRemovedCallback(nullptr)
|
|
{}
|
|
|
|
virtual ~ProtectedQueue() {}
|
|
|
|
/**
|
|
* Add a new element to the back of the queue
|
|
*/
|
|
inline void push_back(T newElement)
|
|
{
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
m_deque.push_back(newElement);
|
|
|
|
// Call item added callback if one exists
|
|
if (p_itemAddedCallback != nullptr)
|
|
{
|
|
(*p_itemAddedCallback)(m_deque.size());
|
|
}
|
|
}
|
|
m_CV.notify_one();
|
|
}
|
|
|
|
/**
|
|
* Retrieve the first element of the queue and remove it from the queue
|
|
*/
|
|
inline T pop_front()
|
|
{
|
|
// Wait on queue blocking
|
|
std::unique_lock<std::mutex> muxLock(m_mutex);
|
|
while (m_deque.empty())
|
|
{
|
|
m_CV.wait_for(muxLock, std::chrono::milliseconds(100));
|
|
|
|
// If we've been asked to terminate, break out and return
|
|
// NOTE: the return value is garbage if terminate has been
|
|
// called prior this call returning
|
|
if (m_terminateFlag)
|
|
{
|
|
T elementToReturn{};
|
|
return elementToReturn;
|
|
}
|
|
}
|
|
|
|
T elementToReturn = m_deque.front();
|
|
m_deque.pop_front();
|
|
|
|
// Call item removed callback if one exists
|
|
if (p_itemRemovedCallback != nullptr)
|
|
{
|
|
(*p_itemRemovedCallback)(m_deque.size());
|
|
}
|
|
|
|
return elementToReturn;
|
|
}
|
|
|
|
/**
|
|
* Retrieve the first element of the queue and remove it from the queue
|
|
*/
|
|
inline T pop_front(bool waitOnEmptyQueue, bool& elementReturned)
|
|
{
|
|
// Wait on queue blocking
|
|
std::unique_lock<std::mutex> muxLock(m_mutex);
|
|
if (waitOnEmptyQueue)
|
|
{
|
|
while (m_deque.empty())
|
|
{
|
|
m_CV.wait_for(muxLock, std::chrono::milliseconds(100));
|
|
|
|
// If we've been asked to terminate, break out and return
|
|
// NOTE: the return value is garbage if terminate has been
|
|
// called prior this call returning
|
|
if (m_terminateFlag)
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
T elementToReturn{};
|
|
if (m_deque.size() > 0)
|
|
{
|
|
elementReturned = true;
|
|
elementToReturn = m_deque.front();
|
|
m_deque.pop_front();
|
|
|
|
// Call item removed callback if one exists
|
|
if (p_itemRemovedCallback != nullptr)
|
|
{
|
|
(*p_itemRemovedCallback)(m_deque.size());
|
|
}
|
|
}
|
|
else
|
|
{
|
|
elementReturned = false;
|
|
}
|
|
|
|
return elementToReturn;
|
|
}
|
|
|
|
/**
|
|
* Terminate any waiting pop_front() requests
|
|
*/
|
|
void terminate()
|
|
{
|
|
m_terminateFlag = true;
|
|
m_CV.notify_all();
|
|
}
|
|
|
|
/*
|
|
* Resets the terminate flag to allow the queue to function again after
|
|
* being terminated previously
|
|
*/
|
|
void unTerminate()
|
|
{
|
|
m_terminateFlag = false;
|
|
}
|
|
|
|
void cleanup(std::function<void(T)>& callback)
|
|
{
|
|
// Call the callback on all remaining items in the queue purging
|
|
// each afterwards
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
while (! m_deque.empty())
|
|
{
|
|
callback(m_deque.front());
|
|
m_deque.pop_front();
|
|
}
|
|
}
|
|
m_CV.notify_all();
|
|
}
|
|
|
|
inline bool empty()
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
return m_deque.empty();
|
|
}
|
|
|
|
inline uint32_t size()
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
return m_deque.size();
|
|
}
|
|
|
|
inline void setItemAddedCallback(std::function<void(uint32_t)>& callback)
|
|
{
|
|
p_itemAddedCallback = &callback;
|
|
}
|
|
|
|
inline void setItemRemovedCallback(std::function<void(uint32_t)>& callback)
|
|
{
|
|
p_itemRemovedCallback = &callback;
|
|
}
|
|
|
|
/**
|
|
* Get the position of item in the queue
|
|
*
|
|
* @return position of item in queue if it exists, else UINT32_MAX
|
|
*/
|
|
uint32_t getPositionOf(T item)
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
uint32_t position = 0;
|
|
for (auto queueItem : m_deque)
|
|
{
|
|
if (item == queueItem)
|
|
{
|
|
return position;
|
|
}
|
|
position ++;
|
|
}
|
|
|
|
return UINT32_MAX;
|
|
}
|
|
|
|
/**
|
|
* Removes item from queue. If item does not exist in queue, nothing
|
|
* will happen
|
|
*
|
|
* @param item Item to remove
|
|
*/
|
|
void removeItem(T item)
|
|
{
|
|
std::lock_guard<std::mutex> lock(m_mutex);
|
|
|
|
// Erase item from deque
|
|
auto iter = m_deque.begin();
|
|
while (iter != m_deque.end())
|
|
{
|
|
if (*iter == item)
|
|
{
|
|
m_deque.erase(iter);
|
|
return;
|
|
}
|
|
++iter;
|
|
}
|
|
}
|
|
|
|
private:
|
|
|
|
bool m_terminateFlag;
|
|
|
|
std::deque<T> m_deque;
|
|
std::mutex m_mutex;
|
|
std::condition_variable m_CV;
|
|
|
|
std::function<void(uint32_t)>* p_itemAddedCallback;
|
|
std::function<void(uint32_t)>* p_itemRemovedCallback;
|
|
};
|
|
|
|
}
|
|
|
|
#endif
|