132 lines
4.0 KiB
C++
132 lines
4.0 KiB
C++
#include "MockDDS.h"
|
|
#include <map>
|
|
#include <mutex>
|
|
#include <thread>
|
|
#include <condition_variable>
|
|
|
|
namespace MockDDS {
|
|
|
|
struct TopicState {
|
|
std::vector<std::string> queue;
|
|
std::map<SubscriptionId, Subscriber> subs;
|
|
SubscriptionId next_sub_id{1};
|
|
unsigned long seq{0};
|
|
};
|
|
|
|
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);
|
|
// detach worker so test binary exit doesn't require join; worker will exit when g_async_running is false
|
|
g_async_thread.detach();
|
|
}
|
|
}
|
|
|
|
void publish(const std::string &topic, const std::string &msg) {
|
|
std::lock_guard<std::mutex> lk(g_mtx);
|
|
auto &ts = g_topics[topic];
|
|
ts.seq++;
|
|
// annotate payload with simple seq metadata for basic QoS checks
|
|
std::string annotated = std::to_string(ts.seq) + ":" + msg;
|
|
ts.queue.push_back(annotated);
|
|
// notify subscribers synchronously
|
|
for (auto &p : ts.subs) {
|
|
auto &cb = p.second;
|
|
if (cb) cb(annotated);
|
|
}
|
|
}
|
|
|
|
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];
|
|
SubscriptionId id = ts.next_sub_id++;
|
|
ts.subs.emplace(id, std::move(cb));
|
|
return id;
|
|
}
|
|
|
|
void unsubscribe(const std::string &topic, SubscriptionId id) {
|
|
std::lock_guard<std::mutex> lk(g_mtx);
|
|
auto it = g_topics.find(topic);
|
|
if (it == g_topics.end()) return;
|
|
it->second.subs.erase(id);
|
|
}
|
|
|
|
std::string take_last(const std::string &topic) {
|
|
std::lock_guard<std::mutex> lk(g_mtx);
|
|
auto it = g_topics.find(topic);
|
|
if (it == g_topics.end() || it->second.queue.empty()) return std::string();
|
|
return it->second.queue.back();
|
|
}
|
|
|
|
std::vector<std::string> get_queue(const std::string &topic) {
|
|
std::lock_guard<std::mutex> lk(g_mtx);
|
|
auto it = g_topics.find(topic);
|
|
if (it == g_topics.end()) return {};
|
|
return it->second.queue;
|
|
}
|
|
|
|
void clear_all() {
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_mtx);
|
|
g_topics.clear();
|
|
}
|
|
// clear pending async queue; do not join detached worker
|
|
{
|
|
std::lock_guard<std::mutex> lk(g_async_mtx);
|
|
g_async_queue.clear();
|
|
}
|
|
g_async_cv.notify_all();
|
|
}
|
|
|
|
} // namespace MockDDS
|