From f2515b7ce288cea9a507752efbfda4621c5fe405 Mon Sep 17 00:00:00 2001 From: JohannesItten Date: Fri, 3 Jul 2026 10:28:30 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20MXL=5FERR=5FFLOW=5FINVALID=20reconnect,?= =?UTF-8?q?=20nullptr=E2=86=92""=20options,=20null=20guards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- nodes/fakesink/main.cpp | 14 ++++++++++++-- nodes/ndiin/main.cpp | 4 ++-- nodes/ndiout/main.cpp | 32 +++++++++++++++++++++++++++----- nodes/testpattern/main.cpp | 4 ++-- shared/NodeBase.hpp | 2 +- studio-manager/main.cpp | 2 +- 6 files changed, 45 insertions(+), 13 deletions(-) diff --git a/nodes/fakesink/main.cpp b/nodes/fakesink/main.cpp index 0990bd7..8cebc37 100644 --- a/nodes/fakesink/main.cpp +++ b/nodes/fakesink/main.cpp @@ -23,7 +23,7 @@ class FakeSinkNode : public dmf::NodeBase { log("flow active — starting read"); mxlFlowReader reader{}; - mxlStatus st = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &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; @@ -59,6 +59,16 @@ class FakeSinkNode : public dmf::NodeBase { 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; @@ -77,7 +87,7 @@ class FakeSinkNode : public dmf::NodeBase { log("stopped — total frames=%llu invalid=%llu late=%llu", frame_count, invalid_count, late_count); - mxlReleaseFlowReader(instance(), reader); + if (reader) mxlReleaseFlowReader(instance(), reader); } }; diff --git a/nodes/ndiin/main.cpp b/nodes/ndiin/main.cpp index bf35a8d..fe08d65 100644 --- a/nodes/ndiin/main.cpp +++ b/nodes/ndiin/main.cpp @@ -41,7 +41,7 @@ class NDIInNode : public dmf::NodeBase { mxlStatus st = mxlCreateFlowWriter( instance(), dmf::make_video_flow_def(video_flow_id, node_id(), width, height, fps_num, fps_den).c_str(), - nullptr, &video_writer, &video_cfg, &created); + "", &video_writer, &video_cfg, &created); if (st != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(st)); return; } const uint32_t video_stride = video_cfg.discrete.sliceSizes[0]; @@ -69,7 +69,7 @@ class NDIInNode : public dmf::NodeBase { instance(), dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, bit_depth, fps_num, fps_den).c_str(), - nullptr, &audio_writer, &audio_cfg, &created); + "", &audio_writer, &audio_cfg, &created); if (ast != MXL_STATUS_OK) { log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast)); has_audio = false; diff --git a/nodes/ndiout/main.cpp b/nodes/ndiout/main.cpp index 66f4a37..5699e55 100644 --- a/nodes/ndiout/main.cpp +++ b/nodes/ndiout/main.cpp @@ -64,7 +64,7 @@ class NDIOutNode : public dmf::NodeBase { log("flow active — starting read"); mxlFlowConfigInfo video_cfg{}; - mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), nullptr, &video_reader); + mxlStatus vst = mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader); if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowReader failed (%s)", dmf::mxl_status_str(vst)); return; } mxlFlowReaderGetConfigInfo(video_reader, &video_cfg); video_stride = video_cfg.discrete.sliceSizes[0]; @@ -77,10 +77,11 @@ class NDIOutNode : public dmf::NodeBase { int channels = 0; int samples_per_frame = 0; bool has_audio = config().contains("audio_flow_id"); + std::string audio_flow_id; if (has_audio) { const auto audio_flow_info = config().at("audio_flow_id"); - const auto audio_flow_id = audio_flow_info.at("id").get(); + audio_flow_id = audio_flow_info.at("id").get(); sample_rate = audio_flow_info.value("sample_rate", 48000); channels = audio_flow_info.value("channels", 2); samples_per_frame = sample_rate * fps_den / fps_num; @@ -88,7 +89,7 @@ class NDIOutNode : public dmf::NodeBase { log("audio flow=%s %d Hz %dch %d samples/frame", audio_flow_id.c_str(), sample_rate, channels, samples_per_frame); - mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), nullptr, &audio_reader); + mxlStatus ast = mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader); if (ast != MXL_STATUS_OK) { log("audio mxlCreateFlowReader failed (%s) — continuing without audio", dmf::mxl_status_str(ast)); has_audio = false; @@ -188,6 +189,16 @@ class NDIOutNode : public dmf::NodeBase { mxlFlowRuntimeInfo ari{}; mxlFlowReaderGetRuntimeInfo(audio_reader, &ari); audio_index = ari.headIndex; + } else if (ast == MXL_ERR_FLOW_INVALID) { + log("audio flow invalidated — reconnecting..."); + mxlReleaseFlowReader(instance(), audio_reader); + audio_reader = nullptr; + mxlSleepForNs(100'000'000); + if (mxlCreateFlowReader(instance(), audio_flow_id.c_str(), "", &audio_reader) == MXL_STATUS_OK) { + log("audio flow reconnected"); + const mxlRational r = {sample_rate, 1}; + audio_index = mxlGetCurrentIndex(&r); + } } } @@ -223,6 +234,17 @@ class NDIOutNode : public dmf::NodeBase { mxlFlowReaderGetRuntimeInfo(video_reader, &ri); video_index = ri.headIndex; + } else if (vst == MXL_ERR_FLOW_INVALID) { + log("video flow invalidated — reconnecting..."); + mxlReleaseFlowReader(instance(), video_reader); + video_reader = nullptr; + mxlSleepForNs(100'000'000); + if (mxlCreateFlowReader(instance(), flow_id.c_str(), "", &video_reader) == MXL_STATUS_OK) { + log("video flow reconnected"); + const mxlRational r = {fps_num, fps_den}; + video_index = mxlGetCurrentIndex(&r); + } + } else { log("unexpected video status (%s) on index=%llu", dmf::mxl_status_str(vst), video_index); break; @@ -248,8 +270,8 @@ class NDIOutNode : public dmf::NodeBase { else log("stopped"); - if (has_video) mxlReleaseFlowReader(instance(), video_reader); - if (has_audio) mxlReleaseFlowReader(instance(), audio_reader); + if (has_video && video_reader) mxlReleaseFlowReader(instance(), video_reader); + if (has_audio && audio_reader) mxlReleaseFlowReader(instance(), audio_reader); } }; diff --git a/nodes/testpattern/main.cpp b/nodes/testpattern/main.cpp index 93ca4fe..27e8974 100644 --- a/nodes/testpattern/main.cpp +++ b/nodes/testpattern/main.cpp @@ -27,7 +27,7 @@ class TestPatternNode : public dmf::NodeBase { mxlStatus vst = mxlCreateFlowWriter( instance(), dmf::make_video_flow_def(flow_id, node_id(), width, height, fps_num, fps_den).c_str(), - nullptr, &video_writer, &video_cfg, &created); + "", &video_writer, &video_cfg, &created); if (vst != MXL_STATUS_OK) { log("video mxlCreateFlowWriter failed (%s)", dmf::mxl_status_str(vst)); return; } const uint32_t video_stride = video_cfg.discrete.sliceSizes[0]; @@ -53,7 +53,7 @@ class TestPatternNode : public dmf::NodeBase { instance(), dmf::make_audio_flow_def(audio_flow_id, node_id(), sample_rate, channels, 32, fps_num, fps_den).c_str(), - nullptr, &audio_writer, &audio_cfg, &created); + "", &audio_writer, &audio_cfg, &created); if (ast != MXL_STATUS_OK) { log("audio mxlCreateFlowWriter failed (%s) — continuing without audio", dmf::mxl_status_str(ast)); has_audio = false; diff --git a/shared/NodeBase.hpp b/shared/NodeBase.hpp index 481fd0a..62e588a 100644 --- a/shared/NodeBase.hpp +++ b/shared/NodeBase.hpp @@ -80,7 +80,7 @@ public: log("domain=%s", domain_.c_str()); - inst_ = mxlCreateInstance(domain_.c_str(), nullptr); + inst_ = mxlCreateInstance(domain_.c_str(), ""); if (!inst_) { log("mxlCreateInstance failed at %s", domain_.c_str()); return 1; diff --git a/studio-manager/main.cpp b/studio-manager/main.cpp index 3e0c4b0..f50d4ea 100644 --- a/studio-manager/main.cpp +++ b/studio-manager/main.cpp @@ -155,7 +155,7 @@ int main(int argc, char* argv[]) { // Clean up stale flow directories left by any previous crashed run { - mxlInstance gc = mxlCreateInstance(domain.c_str(), nullptr); + mxlInstance gc = mxlCreateInstance(domain.c_str(), ""); if (gc) { mxlGarbageCollectFlows(gc); mxlDestroyInstance(gc);