samiandClaude Sonnet 5
ddf36e848a
util/ carries no wire surface, so it lands before the contract freeze. - error: enum class Error, the engine-wide failure taxonomy; is_retryable enumerates every value (no default:) so -Wswitch forces the retry decision on each future addition. ErrorInfo carries context/http_status. - result: Result<T> over std::expected<T, ErrorInfo>, Result<void>, VDM_TRY / VDM_TRY_ASSIGN. Errors returned, never thrown, on the transfer path. - bytes: span aliases, LE load_le/store_le (debug-asserted precondition, not input validation), and a bounds-checked latching ByteReader for the .veloxpart.meta reader. - event_bus: typed thread-safe pub/sub; header states plainly that unsubscribe is not a quiesce point and download_task will need its own drain. - thread_pool: std::jthread pool; dtor joins in the body before members die (fixed a use-after-destruction on cv_/mu_). Header notes shutdown is drain-only and DAEMON will need a cancel mode. - log: sink interface (core does no I/O); DAEMON installs one. Tested: -Werror clean, 6 binaries green under plain / ASan+UBSan / TSan. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS
107 lines
2.8 KiB
C++
107 lines
2.8 KiB
C++
#include "vdm/util/event_bus.hpp"
|
|
|
|
#include <atomic>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
#include "vtest.hpp"
|
|
|
|
using vdm::EventBus;
|
|
|
|
namespace {
|
|
struct Progress {
|
|
int task;
|
|
long downloaded;
|
|
};
|
|
struct StateChange {
|
|
int task;
|
|
std::string state;
|
|
};
|
|
} // namespace
|
|
|
|
VT_TEST(bus_delivers_to_matching_type_only) {
|
|
EventBus bus;
|
|
int progress_hits = 0;
|
|
int state_hits = 0;
|
|
|
|
bus.subscribe<Progress>([&](const Progress &p) {
|
|
++progress_hits;
|
|
VT_CHECK_EQ(p.task, 7);
|
|
});
|
|
bus.subscribe<StateChange>([&](const StateChange &) { ++state_hits; });
|
|
|
|
bus.publish(Progress{7, 1024});
|
|
VT_CHECK_EQ(progress_hits, 1);
|
|
VT_CHECK_EQ(state_hits, 0);
|
|
|
|
bus.publish(StateChange{7, "downloading"});
|
|
VT_CHECK_EQ(progress_hits, 1);
|
|
VT_CHECK_EQ(state_hits, 1);
|
|
}
|
|
|
|
VT_TEST(bus_invokes_in_registration_order) {
|
|
EventBus bus;
|
|
std::vector<int> order;
|
|
bus.subscribe<Progress>([&](const Progress &) { order.push_back(1); });
|
|
bus.subscribe<Progress>([&](const Progress &) { order.push_back(2); });
|
|
bus.subscribe<Progress>([&](const Progress &) { order.push_back(3); });
|
|
bus.publish(Progress{0, 0});
|
|
VT_REQUIRE(order.size() == 3);
|
|
VT_CHECK_EQ(order[0], 1);
|
|
VT_CHECK_EQ(order[1], 2);
|
|
VT_CHECK_EQ(order[2], 3);
|
|
}
|
|
|
|
VT_TEST(bus_unsubscribe_stops_delivery) {
|
|
EventBus bus;
|
|
int hits = 0;
|
|
auto tok = bus.subscribe<Progress>([&](const Progress &) { ++hits; });
|
|
bus.publish(Progress{0, 0});
|
|
bus.unsubscribe(tok);
|
|
bus.publish(Progress{0, 0});
|
|
VT_CHECK_EQ(hits, 1);
|
|
}
|
|
|
|
VT_TEST(bus_scoped_subscription_auto_unsubscribes) {
|
|
EventBus bus;
|
|
int hits = 0;
|
|
{
|
|
auto sub = bus.subscribe_scoped<Progress>([&](const Progress &) { ++hits; });
|
|
bus.publish(Progress{0, 0});
|
|
}
|
|
bus.publish(Progress{0, 0});
|
|
VT_CHECK_EQ(hits, 1);
|
|
}
|
|
|
|
VT_TEST(bus_handler_may_unsubscribe_itself_during_dispatch) {
|
|
EventBus bus;
|
|
int hits = 0;
|
|
EventBus::Token tok = EventBus::kInvalid;
|
|
tok = bus.subscribe<Progress>([&](const Progress &) {
|
|
++hits;
|
|
bus.unsubscribe(tok); // must not deadlock or invalidate the dispatch loop
|
|
});
|
|
bus.publish(Progress{0, 0});
|
|
bus.publish(Progress{0, 0});
|
|
VT_CHECK_EQ(hits, 1);
|
|
}
|
|
|
|
VT_TEST(bus_concurrent_publish_is_safe) {
|
|
EventBus bus;
|
|
std::atomic<long> total{0};
|
|
bus.subscribe<Progress>([&](const Progress &p) { total += p.downloaded; });
|
|
|
|
constexpr int kThreads = 8;
|
|
constexpr int kPerThread = 2000;
|
|
std::vector<std::jthread> ts;
|
|
for (int i = 0; i < kThreads; ++i)
|
|
ts.emplace_back([&, i] {
|
|
for (int j = 0; j < kPerThread; ++j)
|
|
bus.publish(Progress{i, 1});
|
|
});
|
|
ts.clear(); // join
|
|
|
|
VT_CHECK_EQ(total.load(), static_cast<long>(kThreads) * kPerThread);
|
|
}
|