fixed blocking queue
This commit is contained in:
@@ -10,6 +10,7 @@
|
|||||||
|
|
||||||
#include <mutex>
|
#include <mutex>
|
||||||
#include <condition_variable>
|
#include <condition_variable>
|
||||||
|
#include <future>
|
||||||
#include <boost/heap/priority_queue.hpp>
|
#include <boost/heap/priority_queue.hpp>
|
||||||
|
|
||||||
using namespace boost;
|
using namespace boost;
|
||||||
@@ -18,33 +19,56 @@ template <typename T>
|
|||||||
class BlockingQueue
|
class BlockingQueue
|
||||||
{
|
{
|
||||||
private:
|
private:
|
||||||
std::mutex d_mutex;
|
std::mutex mutex;
|
||||||
std::condition_variable d_condition;
|
heap::priority_queue<T, heap::stable<true>> p_queue;
|
||||||
heap::priority_queue<T, heap::stable<true>> p_queue;
|
|
||||||
|
std::promise<T> promise;
|
||||||
|
std::shared_future<T> future_pop;
|
||||||
|
|
||||||
public:
|
public:
|
||||||
void push(T const& value)
|
BlockingQueue()
|
||||||
|
{
|
||||||
|
this->future_pop = promise.get_future();
|
||||||
|
}
|
||||||
|
|
||||||
|
void push(const T& value)
|
||||||
|
{
|
||||||
|
std::lock_guard lock(this->mutex);
|
||||||
|
if(this->p_queue.empty())
|
||||||
{
|
{
|
||||||
std::unique_lock<std::mutex> lock(this->d_mutex);
|
this->promise.set_value(value);
|
||||||
p_queue.push(value);
|
return;
|
||||||
this->d_condition.notify_one();
|
}
|
||||||
|
p_queue.push(value);
|
||||||
|
}
|
||||||
|
|
||||||
|
T pop()
|
||||||
|
{
|
||||||
|
std::unique_lock lock(this->mutex);
|
||||||
|
|
||||||
|
auto status = future_pop.wait_for(std::chrono::seconds(0));
|
||||||
|
if(status == std::future_status::ready)
|
||||||
|
{
|
||||||
|
return future_pop.get();
|
||||||
}
|
}
|
||||||
|
|
||||||
T pop()
|
if (not this->p_queue.empty())
|
||||||
{
|
{
|
||||||
// TODO die queue funzt net weil wir nich pushen koennen wenn wir im pop warten
|
auto element = p_queue.top();
|
||||||
// TODO denk ma ueber future nach
|
p_queue.pop();
|
||||||
std::unique_lock<std::mutex> lock(this->d_mutex);
|
return element;
|
||||||
this->d_condition.wait(lock, [=]{ return !this->p_queue.empty(); });
|
|
||||||
T rc = *this->p_queue.end();
|
|
||||||
this->p_queue.pop();
|
|
||||||
return rc;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
void release()
|
lock.unlock();
|
||||||
{
|
|
||||||
this->d_condition.notify_one();
|
return future_pop.get();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private:
|
||||||
|
T get_first()
|
||||||
|
{
|
||||||
|
std::unique_lock lock(this->mutex);
|
||||||
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -6,7 +6,7 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
|
|
||||||
#include "../catch.hpp"
|
#include "catch.hpp"
|
||||||
#include "fakeit.hpp"
|
#include "fakeit.hpp"
|
||||||
|
|
||||||
#include <map>
|
#include <map>
|
||||||
@@ -60,11 +60,11 @@ SCENARIO("Creating a Detector object", "")
|
|||||||
|
|
||||||
AND_WHEN("an event can be found at gpio interface")
|
AND_WHEN("an event can be found at gpio interface")
|
||||||
{
|
{
|
||||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||||
THEN("only the fitting event should be distributed by event notifier")
|
THEN("after some time the only the fitting event should be distributed by event notifier")
|
||||||
{
|
{
|
||||||
REQUIRE((bool)Verify(Method(event_notifier_mock, distribute_event).Using(event2)));
|
REQUIRE((bool)Verify(Method(event_notifier_mock, distribute_event).Using(event2)));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user