Project

General

Profile

Download (6.01 KB) Statistics
| Branch: | Revision:

root/src/ProtectedQueue.h @ eee13fa3

2c54c4c3 david.sorber
/********************************************************************
* 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