66 lines
1.8 KiB
C++
66 lines
1.8 KiB
C++
#include "MockDDS.h"
|
|
#include <map>
|
|
#include <mutex>
|
|
|
|
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;
|
|
|
|
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);
|
|
}
|
|
}
|
|
|
|
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();
|
|
}
|
|
|
|
} // namespace MockDDS
|