JohannesItten
f2515b7ce2
Readers (ndiout, fakesink) now handle MXL_ERR_FLOW_INVALID by releasing and recreating the flow reader, then realigning the index to current time. This lets consumer nodes survive a producer restart without exiting. All mxlCreateInstance/FlowWriter/FlowReader options args changed from nullptr to "" to match MXL reference implementation style. mxlReleaseFlowReader calls guarded with null checks so cleanup is safe when a mid-run reconnect attempt fails. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
98 lines
3.6 KiB
C++
98 lines
3.6 KiB
C++
#include <chrono>
|
|
#include <string>
|
|
#include <mxl/flow.h>
|
|
#include <mxl/time.h>
|
|
#include "NodeBase.hpp"
|
|
|
|
class FakeSinkNode : public dmf::NodeBase {
|
|
void run() override {
|
|
const auto flow_info = config().at("flow_id");
|
|
const auto flow_id = flow_info.at("id").get<std::string>();
|
|
const int fps_num = flow_info.value("fps_num", 25);
|
|
const int fps_den = flow_info.value("fps_den", 1);
|
|
|
|
log("flow=%s", flow_id.c_str());
|
|
|
|
log("waiting for flow to become active...");
|
|
bool active = false;
|
|
while (!active && dmf::g_running.load(std::memory_order_relaxed)) {
|
|
mxlIsFlowActive(instance(), flow_id.c_str(), &active);
|
|
if (!active) mxlSleepForNs(100'000'000);
|
|
}
|
|
if (!dmf::g_running) return;
|
|
log("flow active — starting read");
|
|
|
|
mxlFlowReader reader{};
|
|
mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader);
|
|
if (st != MXL_STATUS_OK) {
|
|
log("mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(st));
|
|
return;
|
|
}
|
|
|
|
const mxlRational rate = {fps_num, fps_den};
|
|
|
|
uint64_t index = mxlGetCurrentIndex(&rate);
|
|
uint64_t frame_count = 0;
|
|
uint64_t invalid_count = 0;
|
|
uint64_t late_count = 0;
|
|
auto wall_start = std::chrono::steady_clock::now();
|
|
auto last_log_time = wall_start;
|
|
|
|
while (dmf::g_running.load(std::memory_order_relaxed)) {
|
|
mxlGrainInfo grain{};
|
|
uint8_t* buf = nullptr;
|
|
|
|
st = mxlFlowReaderGetGrainNonBlocking(reader, index, &grain, &buf);
|
|
|
|
if (st == MXL_STATUS_OK) {
|
|
frame_count++;
|
|
if (grain.flags & MXL_GRAIN_FLAG_INVALID) invalid_count++;
|
|
index++;
|
|
|
|
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
|
|
mxlSleepForNs(1'000'000); // 1 ms poll
|
|
|
|
} else if (st == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
|
|
late_count++;
|
|
// Jump to the most recent frame in the ring buffer
|
|
mxlFlowRuntimeInfo ri{};
|
|
mxlFlowReaderGetRuntimeInfo(reader, &ri);
|
|
index = ri.headIndex;
|
|
|
|
} else if (st == MXL_ERR_FLOW_INVALID) {
|
|
log("flow invalidated — reconnecting...");
|
|
if (reader) mxlReleaseFlowReader(instance(), reader);
|
|
reader = nullptr;
|
|
mxlSleepForNs(100'000'000);
|
|
if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &reader) == MXL_STATUS_OK) {
|
|
log("flow reconnected");
|
|
index = mxlGetCurrentIndex(&rate);
|
|
}
|
|
|
|
} else {
|
|
log("unexpected status (%s) on index=%llu", dmf::mxl_status_str(st), index);
|
|
break;
|
|
}
|
|
|
|
// Log stats every second (wall clock) — avoid per-frame fprintf
|
|
auto now = std::chrono::steady_clock::now();
|
|
if (std::chrono::duration<double>(now - last_log_time).count() >= 1.0) {
|
|
const double elapsed = std::chrono::duration<double>(now - wall_start).count();
|
|
log("frames=%llu invalid=%llu late=%llu avg=%.2f fps",
|
|
frame_count, invalid_count, late_count,
|
|
static_cast<double>(frame_count) / elapsed);
|
|
last_log_time = now;
|
|
}
|
|
}
|
|
|
|
log("stopped — total frames=%llu invalid=%llu late=%llu",
|
|
frame_count, invalid_count, late_count);
|
|
if (reader) mxlReleaseFlowReader(instance(), reader);
|
|
}
|
|
};
|
|
|
|
int main() {
|
|
FakeSinkNode node;
|
|
return node.execute();
|
|
}
|