tests(dds): enable GTest smoke-tests and add async-capable realistic MockDDS with GTest async subscription test
This commit is contained in:
1 parent
14988660cd
commit
6d8d843ee0
4 files changed
+115
-8
No files matched your search
@@ -1,6 +1,8 @@
|
||||
#include "MockDDS.h"
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <condition_variable>
|
||||
|
||||
namespace MockDDS {
|
||||
|
||||
@@ -14,6 +16,51 @@ struct TopicState {
|
||||
static std::map<std::string, TopicState> g_topics;
|
||||
static std::mutex g_mtx;
|
||||
|
||||
// Background dispatcher for async publishes
|
||||
static std::vector<std::tuple<std::string, std::string, unsigned>> g_async_queue;
|
||||
static std::mutex g_async_mtx;
|
||||
static std::condition_variable g_async_cv;
|
||||
static bool g_async_running = false;
|
||||
static std::thread g_async_thread;
|
||||
|
||||
static void async_worker() {
|
||||
std::unique_lock<std::mutex> lk(g_async_mtx);
|
||||
while (g_async_running) {
|
||||
if (g_async_queue.empty()) {
|
||||
g_async_cv.wait(lk, [] { return !g_async_queue.empty() || !g_async_running; });
|
||||
if (!g_async_running) break;
|
||||
}
|
||||
auto task = g_async_queue.front();
|
||||
g_async_queue.erase(g_async_queue.begin());
|
||||
lk.unlock();
|
||||
auto &topic = std::get<0>(task);
|
||||
auto &msg = std::get<1>(task);
|
||||
unsigned delay_ms = std::get<2>(task);
|
||||
if (delay_ms) std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms));
|
||||
// perform synchronous publish semantics for delivered message
|
||||
{
|
||||
std::lock_guard<std::mutex> lk2(g_mtx);
|
||||
auto &ts = g_topics[topic];
|
||||
ts.seq++;
|
||||
std::string annotated = std::to_string(ts.seq) + ":" + msg;
|
||||
ts.queue.push_back(annotated);
|
||||
for (auto &p : ts.subs) {
|
||||
auto &cb = p.second;
|
||||
if (cb) cb(annotated);
|
||||
}
|
||||
}
|
||||
lk.lock();
|
||||
}
|
||||
}
|
||||
|
||||
static void ensure_async_thread() {
|
||||
std::lock_guard<std::mutex> lk(g_async_mtx);
|
||||
if (!g_async_running) {
|
||||
g_async_running = true;
|
||||
g_async_thread = std::thread(async_worker);
|
||||
}
|
||||
}
|
||||
|
||||
void publish(const std::string &topic, const std::string &msg) {
|
||||
std::lock_guard<std::mutex> lk(g_mtx);
|
||||
auto &ts = g_topics[topic];
|
||||
@@ -28,6 +75,15 @@ void publish(const std::string &topic, const std::string &msg) {
|
||||
}
|
||||
}
|
||||
|
||||
void publish_async(const std::string &topic, const std::string &msg, unsigned delay_ms) {
|
||||
ensure_async_thread();
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(g_async_mtx);
|
||||
g_async_queue.emplace_back(topic, msg, delay_ms);
|
||||
}
|
||||
g_async_cv.notify_one();
|
||||
}
|
||||
|
||||
MockDDS::SubscriptionId subscribe(const std::string &topic, Subscriber cb) {
|
||||
std::lock_guard<std::mutex> lk(g_mtx);
|
||||
auto &ts = g_topics[topic];
|
||||
@@ -58,8 +114,21 @@ std::vector<std::string> get_queue(const std::string &topic) {
|
||||
}
|
||||
|
||||
void clear_all() {
|
||||
std::lock_guard<std::mutex> lk(g_mtx);
|
||||
g_topics.clear();
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(g_mtx);
|
||||
g_topics.clear();
|
||||
}
|
||||
// shutdown async worker gracefully
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(g_async_mtx);
|
||||
g_async_running = false;
|
||||
}
|
||||
g_async_cv.notify_all();
|
||||
if (g_async_thread.joinable()) g_async_thread.join();
|
||||
{
|
||||
std::lock_guard<std::mutex> lk(g_async_mtx);
|
||||
g_async_queue.clear();
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace MockDDS
|
||||
Reference in new issue
Block a user