diff --git a/libs/dmf-node/src/control_server.cpp b/libs/dmf-node/src/control_server.cpp index cfb1d1d..acdd6b5 100644 --- a/libs/dmf-node/src/control_server.cpp +++ b/libs/dmf-node/src/control_server.cpp @@ -47,12 +47,14 @@ static int send_http_json(struct lws* wsi, const std::string& status, const std: auto hdr = "HTTP/1.1 " + status + "\r\n" "Content-Type: application/json\r\n" "Content-Length: " + std::to_string(body.size()) + "\r\n" + "Connection: close\r\n" "\r\n"; std::vector buf(LWS_PRE + hdr.size() + body.size()); std::memcpy(buf.data() + LWS_PRE, hdr.data(), hdr.size()); std::memcpy(buf.data() + LWS_PRE + hdr.size(), body.data(), body.size()); lws_write(wsi, buf.data() + LWS_PRE, hdr.size() + body.size(), LWS_WRITE_HTTP); - return lws_http_transaction_completed(wsi) ? -1 : 0; + lws_http_transaction_completed(wsi); + return -1; } struct PerSession { diff --git a/libs/dmf-node/src/node_runner.cpp b/libs/dmf-node/src/node_runner.cpp index 266e04c..974eda6 100644 --- a/libs/dmf-node/src/node_runner.cpp +++ b/libs/dmf-node/src/node_runner.cpp @@ -144,6 +144,14 @@ int NodeRunner::exec(std::unique_ptr node) { } }); + control_server->register_command("status", [&](const nlohmann::json& /*msg*/) { + nlohmann::json resp; + resp["event"] = "status"; + resp["node_id"] = node_id_; + resp["data"] = node->status(); + control_server->send_event(resp); + }); + control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) { spdlog::info("Shutdown command received"); g_running = false; diff --git a/nodes/passthrough/src/passthrough_node.cpp b/nodes/passthrough/src/passthrough_node.cpp index 1d52bba..ad9407d 100644 --- a/nodes/passthrough/src/passthrough_node.cpp +++ b/nodes/passthrough/src/passthrough_node.cpp @@ -19,7 +19,10 @@ void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter wr void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) { if (port_id == "video_in") { reader_ = reader; - spdlog::info("Passthrough: reader added on video_in"); + mxlFlowRuntimeInfo runtime{}; + mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + read_index_ = runtime.headIndex; + spdlog::info("Passthrough: reader added on video_in, starting at index {}", read_index_); } } @@ -45,26 +48,23 @@ void PassthroughNode::process() { mxlGrainInfo grain_info{}; uint8_t* payload = nullptr; - auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload); + auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 20000000ULL, &grain_info, &payload); if (status != MXL_STATUS_OK) { - if (status == MXL_ERR_TIMEOUT) { - return; - } - if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) { + if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) { mxlFlowRuntimeInfo runtime{}; mxlFlowReaderGetRuntimeInfo(*reader_, &runtime); + spdlog::warn("Passthrough: index {} out of range, jumping to head {}", read_index_, runtime.headIndex); read_index_ = runtime.headIndex; - return; } return; } mxlGrainInfo out_grain{}; uint8_t* out_payload = nullptr; - status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload); + status = mxlFlowWriterOpenGrain(*writer_, grain_info.index, &out_grain, &out_payload); if (status != MXL_STATUS_OK) { - spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast(status)); - read_index_++; + spdlog::warn("Passthrough: failed to open output grain at index {}: {}", grain_info.index, static_cast(status)); + read_index_ = grain_info.index + 1; return; } @@ -75,9 +75,12 @@ void PassthroughNode::process() { out_grain.flags = grain_info.flags; mxlFlowWriterCommitGrain(*writer_, &out_grain); - read_index_++; - write_index_++; + read_index_ = grain_info.index + 1; grains_processed_++; + + if (grains_processed_ <= 5 || grains_processed_ % 50 == 0) { + spdlog::info("Passthrough: grain #{}, index={}, size={}", grains_processed_, grain_info.index, grain_info.grainSize); + } } nlohmann::json PassthroughNode::status() const { @@ -85,7 +88,6 @@ nlohmann::json PassthroughNode::status() const { {"type", "passthrough"}, {"grains_processed", grains_processed_}, {"read_index", read_index_}, - {"write_index", write_index_}, {"has_reader", reader_.has_value()}, {"has_writer", writer_.has_value()}, };