fix: correct HTTP request in fetch_target_info, add socket timeouts and debug logging

- Content-Length was wrong (20 vs actual 18 bytes) - Removed running_ check from loop (was false during startup) - Added SO_RCVTIMEO/SO_SNDTIMEO to prevent infinite recv hangs - Added response size logging for debugging
fix: trigger startup on add_reader when target_host is set
2026-06-16 11:38:48 +03:00 · 2026-06-16 11:29:23 +03:00 · 2026-06-16 11:11:13 +03:00 · 2026-06-16 11:09:10 +03:00 · 2026-06-16 11:08:06 +03:00 · 2026-06-16 10:50:04 +03:00
23 changed files with 1845 additions and 24 deletions
@@ -12,7 +12,21 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")

 option(DMF_BUILD_TESTS "Build tests" ON)
-option(DMF_BUILD_MXL_TOOLS "Build MXL tools (testsrc, sink)" OFF)
+option(DMF_BUILD_MXL_TOOLS "Build MXL tools (testsrc, sink)" ON)
+option(DMF_BUILD_DECKLINK "Build DeckLink I/O nodes" ON)
+option(DMF_BUILD_FABRICS "Build Fabrics bridge nodes" OFF)
+
+set(DECKLINK_SDK_DIR "" CACHE PATH "Path to Blackmagic DeckLink SDK root")
+
+if(DMF_BUILD_DECKLINK AND DECKLINK_SDK_DIR)
+    add_subdirectory(nodes/decklink-in)
+    add_subdirectory(nodes/decklink-out)
+endif()
+
+if(DMF_BUILD_FABRICS)
+    set(MXL_ENABLE_FABRICS_OFI ON CACHE BOOL "" FORCE)
+    add_subdirectory(nodes/fabric-bridge)
+endif()

 find_package(fmt CONFIG REQUIRED)
 find_package(spdlog CONFIG REQUIRED)
@@ -20,6 +34,13 @@ find_package(nlohmann_json CONFIG REQUIRED)
 find_package(Libwebsockets CONFIG REQUIRED)
 find_package(Catch2 CONFIG QUIET)

+set(BUILD_TESTS OFF CACHE BOOL "" FORCE)
+
+if(DMF_BUILD_MXL_TOOLS)
+    set(BUILD_TOOLS ON CACHE BOOL "" FORCE)
+    set(BUILD_UTILS ON CACHE BOOL "" FORCE)
+endif()
+
 add_subdirectory(extern/mxl)
 add_subdirectory(libs/dmf-node)
 add_subdirectory(libs/dmf-engine)
@@ -0,0 +1,66 @@
+#!/bin/bash
+set -e
+
+ENGINE_PORT=${ENGINE_PORT:-9000}
+BASE_URL="http://127.0.0.1:${ENGINE_PORT}"
+DEVICE_INDEX=${DEVICE_INDEX:-0}
+MODE=${MODE:-1080i50}
+
+log() { echo "=== $1 ==="; }
+
+log "Adding decklink-in node (device=${DEVICE_INDEX}, mode=${MODE})"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d "{\"type\":\"decklink-in\",\"id\":\"sdi_in\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Adding passthrough node"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d '{"type":"passthrough","id":"pass1"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Adding decklink-out node (device=${DEVICE_INDEX}, mode=${MODE})"
+curl -s -X POST "${BASE_URL}/api/graph/nodes" \
+  -H "Content-Type: application/json" \
+  -d "{\"type\":\"decklink-out\",\"id\":\"sdi_out\",\"config\":{\"device_index\":${DEVICE_INDEX},\"mode\":\"${MODE}\"}}" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Connecting sdi_in → pass1"
+curl -s -X POST "${BASE_URL}/api/graph/edges" \
+  -H "Content-Type: application/json" \
+  -d '{"from_node":"sdi_in","from_port":"video_out","to_node":"pass1","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Connecting pass1 → sdi_out"
+curl -s -X POST "${BASE_URL}/api/graph/edges" \
+  -H "Content-Type: application/json" \
+  -d '{"from_node":"pass1","from_port":"video_out","to_node":"sdi_out","to_port":"video_in"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Starting graph"
+curl -s -X POST "${BASE_URL}/api/graph/start" | python3 -m json.tool 2>/dev/null || echo ""
+
+sleep 3
+
+log "Status: sdi_in"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_in/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Status: pass1"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/pass1/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Status: sdi_out"
+curl -s -X POST "${BASE_URL}/api/graph/nodes/sdi_out/command" \
+  -H "Content-Type: application/json" \
+  -d '{"cmd":"status"}' | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Graph state"
+curl -s "${BASE_URL}/api/graph" | python3 -m json.tool 2>/dev/null || echo ""
+
+echo ""
+echo "=== SDI pipeline running. Press Enter to stop. ==="
+read
+
+log "Stopping graph"
+curl -s -X POST "${BASE_URL}/api/graph/stop" | python3 -m json.tool 2>/dev/null || echo ""
+
+log "Done"
@@ -1 +1 @@
-/home/itten/DMF/mxl
+../mxl
@@ -9,6 +9,7 @@ namespace dmf_engine {
 class NodeControlClient {
 public:
    bool send_command(uint16_t port, const std::string& json_cmd);
+    std::string send_command_with_response(uint16_t port, const std::string& json_cmd);

    void register_node(const std::string& node_id, uint16_t port);
    void unregister_node(const std::string& node_id);
@@ -12,6 +12,7 @@

 #include <cstring>
 #include <string>
+#include <thread>

 namespace dmf_engine {

@@ -178,7 +179,7 @@ static void handle_request(const std::string& method, const std::string& path,
            }
        } else if (path == "/api/graph/start" && method == "POST") {
            auto nodes = graph.get_nodes();
-            uint16_t port = 9100;
+            uint16_t port = g_impl->port + 100;
            for (auto& node : nodes) {
                pm.start_node(const_cast<GraphNode&>(node), g_impl->mxl_domain, port);
                cc.register_node(node.id, port);
@@ -254,8 +255,29 @@ static void handle_request(const std::string& method, const std::string& path,
                return;
            }

+            int fps_num = 50, fps_den = 1;
+            int width = 1920, height = 1080;
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump());
+            if (!status_resp.empty()) {
+                try {
+                    auto sr = nlohmann::json::parse(status_resp);
+                    if (sr.contains("data")) {
+                        auto& d = sr["data"];
+                        if (d.contains("grain_rate")) {
+                            fps_num = d["grain_rate"].value("numerator", fps_num);
+                            fps_den = d["grain_rate"].value("denominator", fps_den);
+                        }
+                        width = d.value("width", width);
+                        height = d.value("height", height);
+                    }
+                } catch (...) {}
+            }
+
            auto flow_id = fm.create_flow_id();
-            auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
+            auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den);

            nlohmann::json cmd;
            cmd["cmd"] = "add_writer";
@@ -299,6 +321,172 @@ static void handle_request(const std::string& method, const std::string& path,
            } else {
                error_resp(500, "Failed to send command to node");
            }
+        } else if (path.find("/api/graph/nodes/") == 0 && path.find("/register") != std::string::npos && method == "POST") {
+            auto prefix = std::string("/api/graph/nodes/");
+            auto suffix_start = path.find("/register");
+            auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
+
+            if (!req_body.contains("port")) {
+                error_resp(400, "Missing port");
+                return;
+            }
+
+            auto control_port = static_cast<uint16_t>(req_body["port"].get<int>());
+            cc.register_node(node_id, control_port);
+
+            auto* node = graph.get_node_mut(node_id);
+            if (node) {
+                node->control_port = control_port;
+                node->state = NodeState::Running;
+            }
+
+            ok({{"node_id", node_id}, {"control_port", control_port}});
+        } else if (path == "/api/fabric/create-target" && method == "POST") {
+            if (!req_body.contains("node_id")) {
+                error_resp(400, "Missing node_id");
+                return;
+            }
+
+            auto node_id = req_body["node_id"].get<std::string>();
+            auto port_id = req_body.value("port_id", "video_out");
+            auto flow_id = req_body.value("flow_id", fm.create_flow_id());
+
+            auto port_num = cc.get_port(node_id);
+            if (port_num == 0) {
+                error_resp(404, "Node not running: " + node_id);
+                return;
+            }
+
+            int fps_num = 25, fps_den = 1;
+            int width = 1920, height = 1080;
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            auto status_resp = cc.send_command_with_response(port_num, status_cmd.dump());
+            if (!status_resp.empty()) {
+                try {
+                    auto sr = nlohmann::json::parse(status_resp);
+                    if (sr.contains("data")) {
+                        auto& d = sr["data"];
+                        if (d.contains("grain_rate")) {
+                            fps_num = d["grain_rate"].value("numerator", fps_num);
+                            fps_den = d["grain_rate"].value("denominator", fps_den);
+                        }
+                        width = d.value("width", width);
+                        height = d.value("height", height);
+                    }
+                } catch (...) {}
+            }
+
+            auto flow_def = fm.create_v210_flow_def(flow_id, width, height, fps_num, fps_den);
+
+            nlohmann::json cmd;
+            cmd["cmd"] = "add_writer";
+            cmd["port_id"] = port_id;
+            cmd["flow_id"] = flow_id;
+            cmd["flow_def"] = flow_def;
+            if (!cc.send_command(port_num, cmd.dump())) {
+                error_resp(500, "Failed to add writer to target node");
+                return;
+            }
+
+            std::string target_info;
+            constexpr int max_poll_attempts = 30;
+            for (int i = 0; i < max_poll_attempts; ++i) {
+                auto resp = cc.send_command_with_response(port_num, status_cmd.dump());
+                if (!resp.empty()) {
+                    try {
+                        auto sr = nlohmann::json::parse(resp);
+                        if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
+                            target_info = sr["data"]["target_info"].get<std::string>();
+                            break;
+                        }
+                    } catch (...) {}
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+
+            if (target_info.empty()) {
+                error_resp(504, "Target node did not produce target_info in time");
+                return;
+            }
+
+            ok({
+                {"node_id", node_id},
+                {"flow_id", flow_id},
+                {"target_info", target_info}
+            });
+        } else if (path == "/api/fabric/connect" && method == "POST") {
+            if (!req_body.contains("target_node_id") || !req_body.contains("initiator_node_id")) {
+                error_resp(400, "Missing target_node_id/initiator_node_id");
+                return;
+            }
+
+            auto target_node_id = req_body["target_node_id"].get<std::string>();
+            auto initiator_node_id = req_body["initiator_node_id"].get<std::string>();
+
+            auto target_port = cc.get_port(target_node_id);
+            auto initiator_port = cc.get_port(initiator_node_id);
+
+            if (target_port == 0) {
+                error_resp(404, "Target node not running: " + target_node_id);
+                return;
+            }
+            if (initiator_port == 0) {
+                error_resp(404, "Initiator node not running: " + initiator_node_id);
+                return;
+            }
+
+            nlohmann::json status_cmd;
+            status_cmd["cmd"] = "status";
+            std::string target_info;
+            constexpr int max_poll_attempts = 30;
+            for (int i = 0; i < max_poll_attempts; ++i) {
+                auto resp = cc.send_command_with_response(target_port, status_cmd.dump());
+                if (!resp.empty()) {
+                    try {
+                        auto sr = nlohmann::json::parse(resp);
+                        if (sr.contains("data") && sr["data"].contains("target_info") && !sr["data"]["target_info"].get<std::string>().empty()) {
+                            target_info = sr["data"]["target_info"].get<std::string>();
+                            break;
+                        }
+                    } catch (...) {}
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+
+            if (target_info.empty()) {
+                error_resp(504, "Target node did not produce target_info in time");
+                return;
+            }
+
+            nlohmann::json configure_cmd;
+            configure_cmd["cmd"] = "configure";
+            configure_cmd["params"]["target_info"] = target_info;
+            if (!cc.send_command(initiator_port, configure_cmd.dump())) {
+                error_resp(500, "Failed to send target_info to initiator node");
+                return;
+            }
+
+            std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+            bool initiator_running = false;
+            auto init_resp = cc.send_command_with_response(initiator_port, status_cmd.dump());
+            if (!init_resp.empty()) {
+                try {
+                    auto ir = nlohmann::json::parse(init_resp);
+                    if (ir.contains("data") && ir["data"].contains("running")) {
+                        initiator_running = ir["data"]["running"].get<bool>();
+                    }
+                } catch (...) {}
+            }
+
+            ok({
+                {"target_node_id", target_node_id},
+                {"initiator_node_id", initiator_node_id},
+                {"target_info_length", target_info.size()},
+                {"initiator_running", initiator_running}
+            });
        } else if (path.find("/api/graph/nodes/") == 0 && path.find("/command") != std::string::npos && method == "POST") {
            auto prefix = std::string("/api/graph/nodes/");
            auto suffix_start = path.find("/command");
@@ -85,4 +85,60 @@ bool NodeControlClient::send_command(uint16_t port, const std::string& json_cmd)
    return ok;
 }

+std::string NodeControlClient::send_command_with_response(uint16_t port, const std::string& json_cmd) {
+    int fd = socket(AF_INET, SOCK_STREAM, 0);
+    if (fd < 0) {
+        spdlog::error("NodeControlClient: socket() failed: {}", strerror(errno));
+        return "";
+    }
+
+    struct timeval tv;
+    tv.tv_sec = 2;
+    tv.tv_usec = 0;
+    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+    setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+
+    struct sockaddr_in addr;
+    std::memset(&addr, 0, sizeof(addr));
+    addr.sin_family = AF_INET;
+    addr.sin_port = htons(port);
+    inet_pton(AF_INET, "127.0.0.1", &addr.sin_addr);
+
+    if (connect(fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
+        spdlog::error("NodeControlClient: connect to port {} failed: {}", port, strerror(errno));
+        close(fd);
+        return "";
+    }
+
+    std::ostringstream req;
+    req << "POST /cmd HTTP/1.1\r\n"
+        << "Host: 127.0.0.1:" << port << "\r\n"
+        << "Content-Type: application/json\r\n"
+        << "Content-Length: " << json_cmd.size() << "\r\n"
+        << "Connection: close\r\n"
+        << "\r\n"
+        << json_cmd;
+
+    auto request = req.str();
+    auto sent = write(fd, request.data(), request.size());
+    if (sent != static_cast<ssize_t>(request.size())) {
+        spdlog::error("NodeControlClient: write failed on port {}", port);
+        close(fd);
+        return "";
+    }
+
+    std::string full_resp;
+    char resp_buf[4096];
+    while (true) {
+        auto n = read(fd, resp_buf, sizeof(resp_buf));
+        if (n <= 0) break;
+        full_resp.append(resp_buf, n);
+    }
+    close(fd);
+
+    auto body_start = full_resp.find("\r\n\r\n");
+    if (body_start == std::string::npos) return "";
+    return full_resp.substr(body_start + 4);
+}
+
 } // namespace dmf_engine
@@ -11,7 +11,7 @@ struct lws;

 namespace dmf_node {

-using CommandHandler = std::function<void(const nlohmann::json& payload)>;
+using CommandHandler = std::function<nlohmann::json(const nlohmann::json& payload)>;
 using StatusCallback = std::function<void(const nlohmann::json& event)>;

 class ControlServer {
@@ -9,6 +9,7 @@
 #include <nlohmann/json.hpp>

 #include <memory>
+#include <unordered_map>
 #include <vector>

 namespace dmf_node {
@@ -21,12 +22,25 @@ public:
    virtual std::vector<PortDef> input_ports() const = 0;
    virtual std::vector<PortDef> output_ports() const = 0;
    virtual void configure(const nlohmann::json& params) = 0;
-    virtual void on_add_writer(const std::string& port_id, mxlFlowWriter writer) = 0;
-    virtual void on_add_reader(const std::string& port_id, mxlFlowReader reader) = 0;
+    virtual void on_add_writer(const std::string& port_id, mxlFlowWriter writer) {}
+    virtual void on_add_reader(const std::string& port_id, mxlFlowReader reader) {}
    virtual void on_remove_writer(const std::string& port_id) = 0;
    virtual void on_remove_reader(const std::string& port_id) = 0;
    virtual void process() = 0;
    virtual nlohmann::json status() const = 0;
+
+    void set_mxl_instance(mxlInstance instance) { mxl_instance_ = instance; }
+    mxlInstance get_mxl_instance() const { return mxl_instance_; }
+
+    void set_writer_config(const std::string& port_id, const mxlFlowConfigInfo& config) { writer_configs_[port_id] = config; }
+    void set_reader_config(const std::string& port_id, const mxlFlowConfigInfo& config) { reader_configs_[port_id] = config; }
+    const mxlFlowConfigInfo* get_writer_config(const std::string& port_id) const { auto it = writer_configs_.find(port_id); return it != writer_configs_.end() ? &it->second : nullptr; }
+    const mxlFlowConfigInfo* get_reader_config(const std::string& port_id) const { auto it = reader_configs_.find(port_id); return it != reader_configs_.end() ? &it->second : nullptr; }
+
+private:
+    mxlInstance mxl_instance_ = nullptr;
+    std::unordered_map<std::string, mxlFlowConfigInfo> writer_configs_;
+    std::unordered_map<std::string, mxlFlowConfigInfo> reader_configs_;
 };

 } // namespace dmf_node
@@ -29,17 +29,18 @@ struct ControlServer::Impl {
    struct lws_context* context = nullptr;
 };

-static void dispatch_command(ControlServerData* data, const nlohmann::json& msg) {
+static nlohmann::json dispatch_command(ControlServerData* data, const nlohmann::json& msg) {
    if (!msg.contains("cmd")) {
        spdlog::warn("Control: message missing 'cmd' field");
-        return;
+        return {{"error", "missing 'cmd' field"}};
    }
    auto cmd = msg["cmd"].get<std::string>();
    auto it = data->commands.find(cmd);
    if (it != data->commands.end()) {
-        it->second(msg);
+        return it->second(msg);
    } else {
        spdlog::warn("Control: unknown command '{}'", cmd);
+        return {{"error", "unknown command: " + cmd}};
    }
 }

@@ -102,8 +103,8 @@ static int callback_all(struct lws* wsi, enum lws_callback_reasons reason,
        }
        try {
            auto msg = nlohmann::json::parse(ps->http_body);
-            dispatch_command(ps->data, msg);
-            return send_http_json(wsi, "200 OK", R"({"ok":true})");
+            auto result = dispatch_command(ps->data, msg);
+            return send_http_json(wsi, "200 OK", result.dump());
        } catch (const nlohmann::json::parse_error& e) {
            return send_http_json(wsi, "400 Bad Request",
                                 nlohmann::json({{"error", e.what()}}).dump());
@@ -57,6 +57,7 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
    std::signal(SIGINT, signal_handler);
    std::signal(SIGTERM, signal_handler);

+    spdlog::flush_on(spdlog::level::info);
    spdlog::info("Starting node '{}' type='{}'", node_id_, node->type());

    if (!std::filesystem::exists(mxl_domain_)) {
@@ -71,6 +72,8 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
    }
    spdlog::info("MXL instance created on domain: {}", mxl_domain_);

+    node->set_mxl_instance(mxl_instance_);
+
    mxlGarbageCollectFlows(mxl_instance_);

    if (!config_str_.empty()) {
@@ -101,7 +104,7 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
    };
    std::vector<FlowResource> flow_resources;

-    control_server->register_command("add_writer", [&](const nlohmann::json& msg) {
+    control_server->register_command("add_writer", [&](const nlohmann::json& msg) -> nlohmann::json {
        auto flow_id = msg["flow_id"].get<std::string>();
        auto port_id = msg["port_id"].get<std::string>();
        auto flow_def = msg["flow_def"].dump();
@@ -112,14 +115,16 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
        auto status = mxlCreateFlowWriter(mxl_instance_, flow_def.c_str(), nullptr, &writer, &config_info, &created);
        if (status != MXL_STATUS_OK || !writer) {
            spdlog::error("Failed to create flow writer for flow {}: status={}", flow_id, static_cast<int>(status));
-            return;
+            return {{"ok", false}, {"error", "failed to create flow writer"}};
        }
 spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created);
        flow_resources.push_back({port_id, writer, nullptr});
+        node->set_writer_config(port_id, config_info);
        node->on_add_writer(port_id, writer);
+        return {{"ok", true}, {"created", created}};
    });

-    control_server->register_command("add_reader", [&](const nlohmann::json& msg) {
+    control_server->register_command("add_reader", [&](const nlohmann::json& msg) -> nlohmann::json {
        auto flow_id = msg["flow_id"].get<std::string>();
        auto port_id = msg["port_id"].get<std::string>();

@@ -127,14 +132,18 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
        auto status = mxlCreateFlowReader(mxl_instance_, flow_id.c_str(), nullptr, &reader);
        if (status != MXL_STATUS_OK || !reader) {
            spdlog::error("Failed to create flow reader for flow {}: status={}", flow_id, static_cast<int>(status));
-            return;
+            return {{"ok", false}, {"error", "failed to create flow reader"}};
        }
 spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id);
        flow_resources.push_back({port_id, nullptr, reader});
+        mxlFlowConfigInfo config{};
+        mxlFlowReaderGetConfigInfo(reader, &config);
+        node->set_reader_config(port_id, config);
        node->on_add_reader(port_id, reader);
+        return {{"ok", true}};
    });

-    control_server->register_command("remove_writer", [&](const nlohmann::json& msg) {
+    control_server->register_command("remove_writer", [&](const nlohmann::json& msg) -> nlohmann::json {
        auto port_id = msg["port_id"].get<std::string>();
        node->on_remove_writer(port_id);
        for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) {
@@ -145,9 +154,10 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
            }
        }
        spdlog::info("Removed writer on port '{}'", port_id);
+        return {{"ok", true}};
    });

-    control_server->register_command("remove_reader", [&](const nlohmann::json& msg) {
+    control_server->register_command("remove_reader", [&](const nlohmann::json& msg) -> nlohmann::json {
        auto port_id = msg["port_id"].get<std::string>();
        node->on_remove_reader(port_id);
        for (auto it = flow_resources.begin(); it != flow_resources.end(); ++it) {
@@ -158,26 +168,30 @@ int NodeRunner::exec(std::unique_ptr<Node> node) {
            }
        }
        spdlog::info("Removed reader on port '{}'", port_id);
+        return {{"ok", true}};
    });

-    control_server->register_command("configure", [&](const nlohmann::json& msg) {
+    control_server->register_command("configure", [&](const nlohmann::json& msg) -> nlohmann::json {
        if (msg.contains("params")) {
            node->configure(msg["params"]);
            spdlog::info("Reconfigured node '{}'", node_id_);
        }
+        return {{"ok", true}};
    });

-    control_server->register_command("status", [&](const nlohmann::json& /*msg*/) {
+    control_server->register_command("status", [&](const nlohmann::json& /*msg*/) -> nlohmann::json {
        nlohmann::json resp;
        resp["event"] = "status";
        resp["node_id"] = node_id_;
        resp["data"] = node->status();
        control_server->send_event(resp);
+        return resp;
    });

-    control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) {
+    control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) -> nlohmann::json {
        spdlog::info("Shutdown command received");
        g_running = false;
+        return {{"ok", true}};
    });

    nlohmann::json ready_event;
@@ -0,0 +1,21 @@
+cmake_minimum_required(VERSION 3.24)
+
+project(dmf-node-decklink-in LANGUAGES CXX)
+
+set(DECKLINK_INCLUDE "${DECKLINK_SDK_DIR}/Linux/include")
+
+add_executable(dmf-node-decklink-in
+    src/main.cpp
+    src/decklink_in_node.cpp
+    "${DECKLINK_SDK_DIR}/Linux/include/DeckLinkAPIDispatch.cpp"
+)
+
+target_include_directories(dmf-node-decklink-in PRIVATE
+    "${DECKLINK_INCLUDE}"
+)
+
+target_link_libraries(dmf-node-decklink-in PRIVATE
+    dmf-node
+)
+
+target_compile_options(dmf-node-decklink-in PRIVATE -Wno-unused-parameter)
@@ -0,0 +1,292 @@
+#include "decklink_in_node.hpp"
+
+#include <mxl/flow.h>
+#include <mxl/mxl.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPIConfiguration.h>
+
+#include <spdlog/spdlog.h>
+
+#include <cstring>
+
+namespace dmf_node {
+
+DeckLinkInNode::~DeckLinkInNode() {
+    close_device();
+}
+
+void DeckLinkInNode::configure(const nlohmann::json& params) {
+    if (params.contains("device_index")) {
+        device_index_ = params["device_index"].get<int>();
+    }
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "1080i50") display_mode_ = bmdModeHD1080i50;
+        else if (mode_str == "1080p50") display_mode_ = bmdModeHD1080p50;
+        else if (mode_str == "1080p25") display_mode_ = bmdModeHD1080p25;
+        else if (mode_str == "1080i5994") display_mode_ = bmdModeHD1080i5994;
+        else if (mode_str == "1080p5994") display_mode_ = bmdModeHD1080p5994;
+        else if (mode_str == "1080p2997") display_mode_ = bmdModeHD1080p2997;
+        else if (mode_str == "720p50") display_mode_ = bmdModeHD720p50;
+        else if (mode_str == "720p5994") display_mode_ = bmdModeHD720p5994;
+        else {
+            spdlog::warn("DeckLink-in: unknown mode '{}', defaulting to 1080i50", mode_str);
+        }
+    }
+    if (params.contains("input_connection")) {
+        auto conn_str = params["input_connection"].get<std::string>();
+        if (conn_str == "sdi") input_connection_ = bmdVideoConnectionSDI;
+        else if (conn_str == "hdmi") input_connection_ = bmdVideoConnectionHDMI;
+        else if (conn_str == "optical_sdi") input_connection_ = bmdVideoConnectionOpticalSDI;
+        else if (conn_str == "component") input_connection_ = bmdVideoConnectionComponent;
+        else if (conn_str == "composite") input_connection_ = bmdVideoConnectionComposite;
+        else if (conn_str == "svideo") input_connection_ = bmdVideoConnectionSVideo;
+        else {
+            spdlog::warn("DeckLink-in: unknown input_connection '{}', defaulting to auto", conn_str);
+        }
+    }
+
+    if (!open_device()) {
+        spdlog::error("DeckLink-in: failed to open device during configure");
+    }
+}
+
+void DeckLinkInNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
+    if (port_id == "video_out") {
+        writer_ = writer;
+
+        auto now = mxlGetTime();
+        write_index_ = mxlTimestampToIndex(&grain_rate_, now);
+
+        spdlog::info("DeckLink-in: writer added, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+
+        if (input_ && !capturing_) {
+            if (input_->StartStreams() != S_OK) {
+                spdlog::error("DeckLink-in: failed to start streams");
+                return;
+            }
+            capturing_ = true;
+            spdlog::info("DeckLink-in: capture started");
+        }
+    }
+}
+
+void DeckLinkInNode::on_remove_writer(const std::string& port_id) {
+    if (port_id == "video_out") {
+        close_device();
+        writer_.reset();
+        spdlog::info("DeckLink-in: writer removed");
+    }
+}
+
+bool DeckLinkInNode::open_device() {
+    auto* iter = CreateDeckLinkIteratorInstance();
+    if (!iter) {
+        spdlog::error("DeckLink-in: DeckLink drivers not found");
+        return false;
+    }
+
+    IDeckLink* device = nullptr;
+    for (int i = 0; i <= device_index_; ++i) {
+        if (iter->Next(&device) != S_OK) {
+            spdlog::error("DeckLink-in: device index {} not found", device_index_);
+            iter->Release();
+            return false;
+        }
+        if (i < device_index_) {
+            device->Release();
+        }
+    }
+    iter->Release();
+
+    const char* model_name = nullptr;
+    device->GetModelName(&model_name);
+    spdlog::info("DeckLink-in: opened device '{}'", model_name ? model_name : "unknown");
+
+    if (input_connection_ != bmdVideoConnectionUnspecified) {
+        IDeckLinkConfiguration* config = nullptr;
+        if (device->QueryInterface(IID_IDeckLinkConfiguration, (void**)&config) == S_OK && config) {
+            if (config->SetInt(bmdDeckLinkConfigVideoInputConnection, input_connection_) != S_OK) {
+                spdlog::warn("DeckLink-in: failed to set input connection");
+            } else {
+                spdlog::info("DeckLink-in: input connection configured");
+            }
+            config->Release();
+        }
+    }
+
+    if (device->QueryInterface(IID_IDeckLinkInput, (void**)&input_) != S_OK) {
+        spdlog::error("DeckLink-in: device has no input interface");
+        device->Release();
+        return false;
+    }
+
+    decklink_ = device;
+
+    callback_ = std::make_unique<CaptureCallback>(*this);
+    input_->SetCallback(callback_.get());
+
+    auto flags = bmdVideoInputEnableFormatDetection;
+    if (input_->EnableVideoInput(display_mode_, bmdFormat10BitYUV, flags) != S_OK) {
+        spdlog::error("DeckLink-in: failed to enable video input");
+        return false;
+    }
+
+    IDeckLinkDisplayMode* mode = nullptr;
+    if (input_->GetDisplayMode(display_mode_, &mode) == S_OK) {
+        frame_width_ = mode->GetWidth();
+        frame_height_ = mode->GetHeight();
+        BMDTimeValue duration = 0;
+        BMDTimeScale scale = 0;
+        mode->GetFrameRate(&duration, &scale);
+        mode->Release();
+
+        is_interlaced_ = (display_mode_ == bmdModeHD1080i50 ||
+                          display_mode_ == bmdModeHD1080i5994);
+
+        grain_rate_.numerator = static_cast<int32_t>(scale);
+        grain_rate_.denominator = static_cast<int32_t>(duration);
+
+        auto g = std::__gcd(grain_rate_.numerator, grain_rate_.denominator);
+        grain_rate_.numerator /= g;
+        grain_rate_.denominator /= g;
+
+        spdlog::info("DeckLink-in: {}x{} @ {}/{} fps, interlaced={}",
+                     frame_width_, frame_height_,
+                     grain_rate_.numerator, grain_rate_.denominator,
+                     is_interlaced_);
+    }
+
+    spdlog::info("DeckLink-in: device opened, waiting for writer to start capture");
+    return true;
+}
+
+void DeckLinkInNode::close_device() {
+    if (input_) {
+        if (capturing_) {
+            input_->StopStreams();
+        }
+        input_->DisableVideoInput();
+        input_->Release();
+        input_ = nullptr;
+    }
+    if (decklink_) {
+        decklink_->Release();
+        decklink_ = nullptr;
+    }
+    capturing_ = false;
+    callback_.reset();
+    spdlog::info("DeckLink-in: device closed");
+}
+
+void DeckLinkInNode::process() {
+    if (!writer_ || !capturing_) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        return;
+    }
+
+    auto deadline = mxlIndexToTimestamp(&grain_rate_, write_index_ + 1);
+    mxlSleepUntil(deadline);
+
+    void* src_data = nullptr;
+    long src_row_bytes = 0;
+    long src_width = 0;
+    long src_height = 0;
+
+    {
+        std::lock_guard<std::mutex> lock(frame_mutex_);
+        if (frame_data_) {
+            src_data = frame_data_;
+            src_row_bytes = frame_row_bytes_;
+            src_width = frame_width_;
+            src_height = frame_height_;
+        }
+        frame_ready_ = false;
+    }
+
+    if (!src_data) {
+        write_index_++;
+        return;
+    }
+
+    mxlGrainInfo out_grain{};
+    uint8_t* out_payload = nullptr;
+    auto status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload);
+    if (status != MXL_STATUS_OK) {
+        auto now = mxlGetTime();
+        auto current = mxlTimestampToIndex(&grain_rate_, now);
+        write_index_ = current + 1;
+        return;
+    }
+
+    auto dst_row_bytes = out_grain.grainSize / src_height;
+    auto copy_row_bytes = std::min(static_cast<long>(dst_row_bytes), src_row_bytes);
+
+    for (long y = 0; y < src_height && y < static_cast<long>(out_grain.grainSize / dst_row_bytes); ++y) {
+        std::memcpy(out_payload + y * dst_row_bytes,
+                    static_cast<uint8_t*>(src_data) + y * src_row_bytes,
+                    copy_row_bytes);
+    }
+
+    out_grain.validSlices = out_grain.totalSlices;
+    mxlFlowWriterCommitGrain(*writer_, &out_grain);
+
+    write_index_++;
+    grains_written_++;
+
+    if (grains_written_ == 1) {
+        spdlog::info("DeckLink-in: first grain written, index={}", write_index_ - 1);
+    }
+}
+
+nlohmann::json DeckLinkInNode::status() const {
+    return {
+        {"type", "decklink-in"},
+        {"grains_written", grains_written_},
+        {"write_index", write_index_},
+        {"capturing", capturing_.load()},
+        {"has_writer", writer_.has_value()},
+        {"grain_rate", {{"numerator", grain_rate_.numerator}, {"denominator", grain_rate_.denominator}}},
+        {"width", frame_width_},
+        {"height", frame_height_},
+        {"interlaced", is_interlaced_},
+    };
+}
+
+HRESULT DeckLinkInNode::CaptureCallback::VideoInputFormatChanged(
+    BMDVideoInputFormatChangedEvents, IDeckLinkDisplayMode* newMode, BMDDetectedVideoInputFormatFlags) {
+    spdlog::info("DeckLink-in: input format changed");
+    return S_OK;
+}
+
+HRESULT DeckLinkInNode::CaptureCallback::VideoInputFrameArrived(
+    IDeckLinkVideoInputFrame* videoFrame, IDeckLinkAudioInputPacket*) {
+    if (!videoFrame || (videoFrame->GetFlags() & bmdFrameHasNoInputSource)) {
+        return S_OK;
+    }
+
+    void* bytes = nullptr;
+    IDeckLinkVideoBuffer* buf = nullptr;
+    if (videoFrame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+        buf->StartAccess(bmdBufferAccessRead);
+        buf->GetBytes(&bytes);
+        buf->EndAccess(bmdBufferAccessRead);
+        buf->Release();
+    }
+
+    if (!bytes) {
+        return S_OK;
+    }
+
+    std::lock_guard<std::mutex> lock(owner_.frame_mutex_);
+    owner_.frame_data_ = bytes;
+    owner_.frame_row_bytes_ = videoFrame->GetRowBytes();
+    owner_.frame_width_ = videoFrame->GetWidth();
+    owner_.frame_height_ = videoFrame->GetHeight();
+    owner_.frame_ready_ = true;
+
+    return S_OK;
+}
+
+} // namespace dmf_node
@@ -0,0 +1,84 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPI.h>
+
+#include <atomic>
+#include <mutex>
+#include <optional>
+#include <thread>
+
+namespace dmf_node {
+
+class DeckLinkInNode : public Node {
+public:
+    DeckLinkInNode() = default;
+    ~DeckLinkInNode();
+
+    std::string type() const override { return "decklink-in"; }
+
+    std::vector<PortDef> input_ports() const override { return {}; }
+
+    std::vector<PortDef> output_ports() const override {
+        return {{"video_out", PortDirection::Output, MediaType::VideoV210}};
+    }
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override;
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override {}
+    void on_remove_writer(const std::string& port_id) override;
+    void on_remove_reader(const std::string& port_id) override {}
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    bool open_device();
+    void close_device();
+
+    std::optional<mxlFlowWriter> writer_;
+    mxlRational grain_rate_{25, 1};
+    uint64_t write_index_ = 0;
+    uint64_t grains_written_ = 0;
+
+    int device_index_ = 0;
+    BMDDisplayMode display_mode_ = bmdModeHD1080i50;
+    BMDVideoConnection input_connection_ = bmdVideoConnectionUnspecified;
+
+    IDeckLink* decklink_ = nullptr;
+    IDeckLinkInput* input_ = nullptr;
+
+    std::atomic<bool> capturing_{false};
+    bool is_interlaced_ = false;
+    std::mutex frame_mutex_;
+    void* frame_data_ = nullptr;
+    long frame_row_bytes_ = 0;
+    long frame_width_ = 1920;
+    long frame_height_ = 1080;
+    bool frame_ready_ = false;
+
+    class CaptureCallback : public IDeckLinkInputCallback {
+    public:
+        CaptureCallback(DeckLinkInNode& owner) : owner_(owner) {}
+
+        HRESULT STDMETHODCALLTYPE QueryInterface(REFIID, void**) override { return E_NOINTERFACE; }
+        ULONG STDMETHODCALLTYPE AddRef() override { return 1; }
+        ULONG STDMETHODCALLTYPE Release() override { return 1; }
+
+        HRESULT STDMETHODCALLTYPE VideoInputFormatChanged(
+            BMDVideoInputFormatChangedEvents, IDeckLinkDisplayMode*, BMDDetectedVideoInputFormatFlags) override;
+
+        HRESULT STDMETHODCALLTYPE VideoInputFrameArrived(
+            IDeckLinkVideoInputFrame* videoFrame, IDeckLinkAudioInputPacket*) override;
+
+    private:
+        DeckLinkInNode& owner_;
+    };
+
+    std::unique_ptr<CaptureCallback> callback_;
+};
+
+} // namespace dmf_node
@@ -0,0 +1,6 @@
+#include <dmf-node/node_runner.hpp>
+#include "decklink_in_node.hpp"
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::DeckLinkInNode>(argc, argv);
+}
@@ -0,0 +1,21 @@
+cmake_minimum_required(VERSION 3.24)
+
+project(dmf-node-decklink-out LANGUAGES CXX)
+
+set(DECKLINK_INCLUDE "${DECKLINK_SDK_DIR}/Linux/include")
+
+add_executable(dmf-node-decklink-out
+    src/main.cpp
+    src/decklink_out_node.cpp
+    "${DECKLINK_SDK_DIR}/Linux/include/DeckLinkAPIDispatch.cpp"
+)
+
+target_include_directories(dmf-node-decklink-out PRIVATE
+    "${DECKLINK_INCLUDE}"
+)
+
+target_link_libraries(dmf-node-decklink-out PRIVATE
+    dmf-node
+)
+
+target_compile_options(dmf-node-decklink-out PRIVATE -Wno-unused-parameter)
@@ -0,0 +1,365 @@
+#include "decklink_out_node.hpp"
+
+#include <mxl/flow.h>
+#include <mxl/mxl.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPIConfiguration.h>
+
+#include <spdlog/spdlog.h>
+
+#include <cstring>
+
+namespace dmf_node {
+
+DeckLinkOutNode::~DeckLinkOutNode() {
+    close_device();
+}
+
+void DeckLinkOutNode::configure(const nlohmann::json& params) {
+    if (params.contains("device_index")) {
+        device_index_ = params["device_index"].get<int>();
+    }
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "1080i50") display_mode_ = bmdModeHD1080i50;
+        else if (mode_str == "1080p50") display_mode_ = bmdModeHD1080p50;
+        else if (mode_str == "1080p25") display_mode_ = bmdModeHD1080p25;
+        else if (mode_str == "1080i5994") display_mode_ = bmdModeHD1080i5994;
+        else if (mode_str == "1080p5994") display_mode_ = bmdModeHD1080p5994;
+        else if (mode_str == "1080p2997") display_mode_ = bmdModeHD1080p2997;
+        else if (mode_str == "720p50") display_mode_ = bmdModeHD720p50;
+        else if (mode_str == "720p5994") display_mode_ = bmdModeHD720p5994;
+        else {
+            spdlog::warn("DeckLink-out: unknown mode '{}', defaulting to 1080i50", mode_str);
+        }
+    }
+    if (params.contains("output_connection")) {
+        auto conn_str = params["output_connection"].get<std::string>();
+        if (conn_str == "sdi") output_connection_ = bmdVideoConnectionSDI;
+        else if (conn_str == "hdmi") output_connection_ = bmdVideoConnectionHDMI;
+        else if (conn_str == "optical_sdi") output_connection_ = bmdVideoConnectionOpticalSDI;
+        else if (conn_str == "component") output_connection_ = bmdVideoConnectionComponent;
+        else if (conn_str == "composite") output_connection_ = bmdVideoConnectionComposite;
+        else if (conn_str == "svideo") output_connection_ = bmdVideoConnectionSVideo;
+        else {
+            spdlog::warn("DeckLink-out: unknown output_connection '{}', defaulting to auto", conn_str);
+        }
+    }
+}
+
+void DeckLinkOutNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
+    if (port_id == "video_in") {
+        reader_ = reader;
+
+        mxlFlowConfigInfo config{};
+        mxlFlowReaderGetConfigInfo(*reader_, &config);
+        flow_rate_ = config.common.grainRate;
+
+        auto now = mxlGetTime();
+        auto current_index = mxlTimestampToIndex(&flow_rate_, now);
+        read_index_ = current_index - 2;
+
+        spdlog::info("DeckLink-out: reader added, flow_rate={}/{}", flow_rate_.numerator, flow_rate_.denominator);
+
+        if (!open_device()) {
+            spdlog::error("DeckLink-out: failed to open device");
+            return;
+        }
+    }
+}
+
+void DeckLinkOutNode::on_remove_reader(const std::string& port_id) {
+    if (port_id == "video_in") {
+        close_device();
+        reader_.reset();
+        spdlog::info("DeckLink-out: reader removed");
+    }
+}
+
+bool DeckLinkOutNode::open_device() {
+    auto* iter = CreateDeckLinkIteratorInstance();
+    if (!iter) {
+        spdlog::error("DeckLink-out: DeckLink drivers not found");
+        return false;
+    }
+
+    IDeckLink* device = nullptr;
+    for (int i = 0; i <= device_index_; ++i) {
+        if (iter->Next(&device) != S_OK) {
+            spdlog::error("DeckLink-out: device index {} not found", device_index_);
+            iter->Release();
+            return false;
+        }
+        if (i < device_index_) {
+            device->Release();
+        }
+    }
+    iter->Release();
+
+    const char* model_name = nullptr;
+    device->GetModelName(&model_name);
+    spdlog::info("DeckLink-out: opened device '{}'", model_name ? model_name : "unknown");
+
+    if (device->QueryInterface(IID_IDeckLinkOutput, (void**)&output_) != S_OK) {
+        spdlog::error("DeckLink-out: device has no output interface");
+        device->Release();
+        return false;
+    }
+
+    decklink_ = device;
+
+    if (output_connection_ != bmdVideoConnectionUnspecified) {
+        IDeckLinkConfiguration* config = nullptr;
+        if (decklink_->QueryInterface(IID_IDeckLinkConfiguration, (void**)&config) == S_OK && config) {
+            if (config->SetInt(bmdDeckLinkConfigVideoOutputConnection, output_connection_) != S_OK) {
+                spdlog::warn("DeckLink-out: failed to set output connection");
+            } else {
+                spdlog::info("DeckLink-out: set output connection configured");
+            }
+            config->Release();
+        } else {
+            spdlog::warn("DeckLink-out: IDeckLinkConfiguration not available");
+        }
+    }
+
+    callback_ = std::make_unique<OutputCallback>(*this);
+    output_->SetScheduledFrameCompletionCallback(callback_.get());
+
+    if (output_->EnableVideoOutput(display_mode_, bmdVideoOutputFlagDefault) != S_OK) {
+        spdlog::error("DeckLink-out: failed to enable video output");
+        return false;
+    }
+
+    IDeckLinkDisplayMode* mode = nullptr;
+    if (output_->GetDisplayMode(display_mode_, &mode) == S_OK) {
+        out_width_ = mode->GetWidth();
+        out_height_ = mode->GetHeight();
+        BMDTimeValue duration = 0;
+        BMDTimeScale scale = 0;
+        mode->GetFrameRate(&duration, &scale);
+        frame_duration_ = duration;
+        time_scale_ = scale;
+        mode->Release();
+
+        output_rate_.numerator = static_cast<int32_t>(scale);
+        output_rate_.denominator = static_cast<int32_t>(duration);
+        auto g = std::__gcd(output_rate_.numerator, output_rate_.denominator);
+        output_rate_.numerator /= g;
+        output_rate_.denominator /= g;
+
+        is_interlaced_ = (display_mode_ == bmdModeHD1080i50 ||
+                          display_mode_ == bmdModeHD1080i5994);
+
+        spdlog::info("DeckLink-out: {}x{} @ {}/{} fps, interlaced={}",
+                     out_width_, out_height_,
+                     output_rate_.numerator, output_rate_.denominator,
+                     is_interlaced_);
+    }
+
+    preroll_frames_ = 5;
+    for (int i = 0; i < preroll_frames_; ++i) {
+        IDeckLinkMutableVideoFrame* frame = nullptr;
+        int32_t out_row_bytes = out_width_ * 16 / 6;
+        if (output_->CreateVideoFrame(out_width_, out_height_, out_row_bytes,
+                                       bmdFormat10BitYUV, bmdFrameFlagDefault, &frame) != S_OK || !frame) {
+            out_row_bytes = out_width_ * 2;
+            if (output_->CreateVideoFrame(out_width_, out_height_, out_row_bytes,
+                                           bmdFormat8BitYUV, bmdFrameFlagDefault, &frame) != S_OK || !frame) {
+                spdlog::error("DeckLink-out: failed to create preroll frame");
+                return false;
+            }
+            spdlog::info("DeckLink-out: using 8BitYUV output format");
+            output_format_ = bmdFormat8BitYUV;
+        } else {
+            output_format_ = bmdFormat10BitYUV;
+        }
+        IDeckLinkVideoBuffer* buf = nullptr;
+        if (frame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+            buf->StartAccess(bmdBufferAccessWrite);
+            void* dst = nullptr;
+            buf->GetBytes(&dst);
+            if (dst) {
+                std::memset(dst, 0, out_row_bytes * out_height_);
+            }
+            buf->EndAccess(bmdBufferAccessWrite);
+            buf->Release();
+        }
+        auto stream_time = i * frame_duration_;
+        output_->ScheduleVideoFrame(frame, stream_time, frame_duration_, time_scale_);
+        frame->Release();
+    }
+
+    stream_frame_ = preroll_frames_;
+    if (output_->StartScheduledPlayback(0, time_scale_, 1.0) != S_OK) {
+        spdlog::error("DeckLink-out: failed to start scheduled playback");
+        return false;
+    }
+
+    playing_ = true;
+    spdlog::info("DeckLink-out: playback started");
+    return true;
+}
+
+void DeckLinkOutNode::close_device() {
+    if (output_) {
+        if (playing_) {
+            output_->StopScheduledPlayback(0, nullptr, time_scale_);
+        }
+        output_->DisableVideoOutput();
+        output_->Release();
+        output_ = nullptr;
+    }
+    if (decklink_) {
+        decklink_->Release();
+        decklink_ = nullptr;
+    }
+    playing_ = false;
+    callback_.reset();
+    {
+        std::lock_guard<std::mutex> lock(frame_pool_mutex_);
+        for (auto* f : frame_pool_) {
+            f->Release();
+        }
+        frame_pool_.clear();
+    }
+    spdlog::info("DeckLink-out: device closed");
+}
+
+void DeckLinkOutNode::schedule_frame(void* mxl_payload, long width, long height, long row_bytes) {
+    if (!output_) return;
+
+    int32_t out_row_bytes = 0;
+    if (output_format_ == bmdFormat10BitYUV) {
+        if (output_->RowBytesForPixelFormat(bmdFormat10BitYUV, width, &out_row_bytes) != S_OK || out_row_bytes <= 0) {
+            out_row_bytes = ((width + 5) / 6) * 16;
+        }
+    } else {
+        out_row_bytes = width * 2;
+    }
+
+    IDeckLinkMutableVideoFrame* frame = nullptr;
+    {
+        std::lock_guard<std::mutex> lock(frame_pool_mutex_);
+        if (!frame_pool_.empty()) {
+            frame = frame_pool_.back();
+            frame_pool_.pop_back();
+        }
+    }
+
+    if (!frame) {
+        return;
+    }
+
+    IDeckLinkVideoBuffer* buf = nullptr;
+    if (frame->QueryInterface(IID_IDeckLinkVideoBuffer, (void**)&buf) == S_OK && buf) {
+        buf->StartAccess(bmdBufferAccessWrite);
+        void* dst = nullptr;
+        buf->GetBytes(&dst);
+        if (dst) {
+            auto copy_row_bytes = std::min(static_cast<long>(out_row_bytes), row_bytes);
+            for (long y = 0; y < height; ++y) {
+                std::memcpy(static_cast<uint8_t*>(dst) + y * out_row_bytes,
+                            static_cast<uint8_t*>(mxl_payload) + y * row_bytes,
+                            copy_row_bytes);
+            }
+        }
+        buf->EndAccess(bmdBufferAccessWrite);
+        buf->Release();
+    }
+
+    auto stream_time = stream_frame_ * frame_duration_;
+    output_->ScheduleVideoFrame(frame, stream_time, frame_duration_, time_scale_);
+    frame->Release();
+    stream_frame_++;
+}
+
+void DeckLinkOutNode::process() {
+    if (!reader_ || !playing_) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        return;
+    }
+
+    if (first_frame_) {
+        auto now = mxlGetTime();
+        grains_read_ = mxlTimestampToIndex(&output_rate_, now);
+        first_frame_ = false;
+        spdlog::info("DeckLink-out: starting at output index {}", grains_read_);
+    }
+
+    auto deadline = mxlIndexToTimestamp(&output_rate_, grains_read_ + 1);
+    mxlSleepUntil(deadline);
+
+    auto out_timestamp = mxlIndexToTimestamp(&output_rate_, grains_read_);
+    auto source_index = mxlTimestampToIndex(&flow_rate_, out_timestamp);
+
+    mxlGrainInfo grain_info{};
+    uint8_t* payload = nullptr;
+    auto status = mxlFlowReaderGetGrain(*reader_, source_index, 5000000ULL, &grain_info, &payload);
+    if (status != MXL_STATUS_OK) {
+        if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
+            auto now = mxlGetTime();
+            auto current_index = mxlTimestampToIndex(&flow_rate_, now);
+            spdlog::warn("DeckLink-out: grain too late, skipping to index {}", current_index - 2);
+            source_index = current_index - 2;
+            status = mxlFlowReaderGetGrain(*reader_, source_index, 5000000ULL, &grain_info, &payload);
+            if (status != MXL_STATUS_OK) {
+                grains_read_++;
+                return;
+            }
+        } else if (status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+            return;
+        } else {
+            return;
+        }
+    }
+
+    mxlFlowConfigInfo config{};
+    mxlFlowReaderGetConfigInfo(*reader_, &config);
+    auto grain_size = grain_info.grainSize;
+
+    long src_row_bytes = (config.discrete.sliceSizes[0] > 0) ? static_cast<long>(config.discrete.sliceSizes[0]) : (grain_size / out_height_);
+    long src_height = (src_row_bytes > 0) ? (grain_size / src_row_bytes) : out_height_;
+    long src_width = (src_row_bytes * 3) / 8;
+
+    schedule_frame(payload, src_width, src_height, src_row_bytes);
+
+    grains_read_++;
+
+    if (grains_read_ == 1) {
+        spdlog::info("DeckLink-out: first grain output, source_index={}", source_index);
+    }
+}
+
+nlohmann::json DeckLinkOutNode::status() const {
+    return {
+        {"type", "decklink-out"},
+        {"grains_read", grains_read_},
+        {"read_index", read_index_},
+        {"playing", playing_.load()},
+        {"has_reader", reader_.has_value()},
+        {"flow_rate", {{"numerator", flow_rate_.numerator}, {"denominator", flow_rate_.denominator}}},
+        {"output_rate", {{"numerator", output_rate_.numerator}, {"denominator", output_rate_.denominator}}},
+        {"width", out_width_},
+        {"height", out_height_},
+    };
+}
+
+HRESULT DeckLinkOutNode::OutputCallback::ScheduledFrameCompleted(
+    IDeckLinkVideoFrame* completedFrame, BMDOutputFrameCompletionResult result) {
+    if (completedFrame) {
+        completedFrame->AddRef();
+        std::lock_guard<std::mutex> lock(owner_.frame_pool_mutex_);
+        owner_.frame_pool_.push_back(static_cast<IDeckLinkMutableVideoFrame*>(completedFrame));
+        owner_.frame_pool_cv_.notify_one();
+    }
+    return S_OK;
+}
+
+HRESULT DeckLinkOutNode::OutputCallback::ScheduledPlaybackHasStopped() {
+    owner_.playing_ = false;
+    spdlog::info("DeckLink-out: scheduled playback stopped");
+    return S_OK;
+}
+
+} // namespace dmf_node
@@ -0,0 +1,93 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/time.h>
+
+#include <DeckLinkAPI.h>
+
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+namespace dmf_node {
+
+class DeckLinkOutNode : public Node {
+public:
+    DeckLinkOutNode() = default;
+    ~DeckLinkOutNode();
+
+    std::string type() const override { return "decklink-out"; }
+
+    std::vector<PortDef> input_ports() const override {
+        return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
+    }
+
+    std::vector<PortDef> output_ports() const override { return {}; }
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override {}
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override;
+    void on_remove_writer(const std::string& port_id) override {}
+    void on_remove_reader(const std::string& port_id) override;
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    bool open_device();
+    void close_device();
+    void schedule_frame(void* mxl_payload, long width, long height, long row_bytes);
+
+    std::optional<mxlFlowReader> reader_;
+    mxlRational flow_rate_{50, 1};
+    mxlRational output_rate_{50, 1};
+    uint64_t read_index_ = 0;
+    uint64_t grains_read_ = 0;
+    bool first_frame_ = true;
+
+    int device_index_ = 0;
+    BMDDisplayMode display_mode_ = bmdModeHD1080i50;
+    BMDVideoConnection output_connection_ = bmdVideoConnectionUnspecified;
+
+    IDeckLink* decklink_ = nullptr;
+    IDeckLinkOutput* output_ = nullptr;
+    BMDTimeValue frame_duration_ = 1000;
+    BMDTimeScale time_scale_ = 50000;
+    BMDPixelFormat output_format_ = bmdFormat10BitYUV;
+    int preroll_frames_ = 3;
+
+    std::atomic<bool> playing_{false};
+    int create_fail_count_ = 0;
+    uint64_t stream_frame_ = 0;
+    bool is_interlaced_ = false;
+    long out_width_ = 1920;
+    long out_height_ = 1080;
+
+    class OutputCallback : public IDeckLinkVideoOutputCallback {
+    public:
+        OutputCallback(DeckLinkOutNode& owner) : owner_(owner) {}
+
+        HRESULT STDMETHODCALLTYPE QueryInterface(REFIID, void**) override { return E_NOINTERFACE; }
+        ULONG STDMETHODCALLTYPE AddRef() override { return 1; }
+        ULONG STDMETHODCALLTYPE Release() override { return 1; }
+
+        HRESULT STDMETHODCALLTYPE ScheduledFrameCompleted(
+            IDeckLinkVideoFrame* completedFrame, BMDOutputFrameCompletionResult result) override;
+        HRESULT STDMETHODCALLTYPE ScheduledPlaybackHasStopped() override;
+
+    private:
+        DeckLinkOutNode& owner_;
+    };
+
+    std::unique_ptr<OutputCallback> callback_;
+    std::vector<IDeckLinkMutableVideoFrame*> frame_pool_;
+    std::mutex frame_pool_mutex_;
+    std::condition_variable frame_pool_cv_;
+    IDeckLinkMutableVideoFrame* scheduled_frame_ = nullptr;
+};
+
+} // namespace dmf_node
@@ -0,0 +1,6 @@
+#include <dmf-node/node_runner.hpp>
+#include "decklink_out_node.hpp"
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::DeckLinkOutNode>(argc, argv);
+}
@@ -0,0 +1,28 @@
+cmake_minimum_required(VERSION 3.24 FATAL_ERROR)
+
+project(dmf-node-fabric-bridge VERSION 0.1.0 LANGUAGES CXX)
+
+find_package(fmt CONFIG REQUIRED)
+find_package(spdlog CONFIG REQUIRED)
+find_package(nlohmann_json CONFIG REQUIRED)
+find_package(PkgConfig REQUIRED)
+pkg_check_modules(libfabric REQUIRED IMPORTED_TARGET libfabric)
+
+add_executable(dmf-node-fabric-bridge
+    src/fabric_bridge_node.cpp
+    src/main.cpp
+)
+
+target_link_libraries(dmf-node-fabric-bridge PRIVATE
+    dmf-node
+    mxl-fabrics-objects
+    mxl-fabrics-headers
+    PkgConfig::libfabric
+    spdlog::spdlog
+    fmt::fmt
+    nlohmann_json::nlohmann_json
+)
+
+target_include_directories(dmf-node-fabric-bridge PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}/src
+)
@@ -0,0 +1,470 @@
+#include "fabric_bridge_node.hpp"
+
+#include <mxl/mxl.h>
+
+#include <spdlog/spdlog.h>
+
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <unistd.h>
+#include <netdb.h>
+#include <cerrno>
+
+namespace dmf_node {
+
+FabricBridgeNode::~FabricBridgeNode() {
+    running_ = false;
+    if (worker_thread_.has_value() && worker_thread_->joinable()) {
+        worker_thread_->join();
+    }
+}
+
+std::vector<PortDef> FabricBridgeNode::input_ports() const {
+    if (mode_ == FabricMode::Target) {
+        return {};
+    }
+    return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
+}
+
+std::vector<PortDef> FabricBridgeNode::output_ports() const {
+    if (mode_ == FabricMode::Initiator) {
+        return {};
+    }
+    return {{"video_out", PortDirection::Output, MediaType::VideoV210}};
+}
+
+void FabricBridgeNode::configure(const nlohmann::json& params) {
+    if (params.contains("mode")) {
+        auto mode_str = params["mode"].get<std::string>();
+        if (mode_str == "initiator") mode_ = FabricMode::Initiator;
+        else if (mode_str == "target") mode_ = FabricMode::Target;
+        else spdlog::warn("Fabric-bridge: unknown mode '{}', defaulting to initiator", mode_str);
+    }
+
+    if (params.contains("bind_node")) bind_node_ = params["bind_node"].get<std::string>();
+    if (params.contains("bind_service")) bind_service_ = params["bind_service"].get<std::string>();
+    if (params.contains("provider")) {
+        auto prov_str = params["provider"].get<std::string>();
+        mxlFabricsProviderFromString(prov_str.c_str(), &provider_);
+    }
+
+    if (params.contains("target_info")) {
+        target_info_str_ = params["target_info"].get<std::string>();
+        if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_info_str_.empty()) {
+            start_requested_.store(true);
+            spdlog::info("Fabric-bridge: target_info set, signaling startup");
+        }
+    }
+
+    if (params.contains("target_host")) {
+        target_host_ = params["target_host"].get<std::string>();
+        target_port_ = params.value("target_port", static_cast<uint16_t>(9100));
+        spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_);
+    }
+}
+
+void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
+    if (port_id == "video_out") {
+        writer_ = writer;
+        start_requested_.store(true);
+        spdlog::info("Fabric-bridge: writer added (target mode), signaling startup");
+    }
+}
+
+void FabricBridgeNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
+    if (port_id == "video_in") {
+        reader_ = reader;
+        if (mode_ == FabricMode::Initiator) {
+            if (!target_info_str_.empty() || !target_host_.empty()) {
+                start_requested_.store(true);
+                spdlog::info("Fabric-bridge: reader added (initiator mode), signaling startup");
+            } else {
+                spdlog::info("Fabric-bridge: reader added (initiator mode), waiting for target_info");
+            }
+        }
+    }
+}
+
+void FabricBridgeNode::on_remove_writer(const std::string& port_id) {
+    if (port_id == "video_out") {
+        writer_.reset();
+        spdlog::info("Fabric-bridge: writer removed");
+    }
+}
+
+void FabricBridgeNode::on_remove_reader(const std::string& port_id) {
+    if (port_id == "video_in") {
+        reader_.reset();
+        spdlog::info("Fabric-bridge: reader removed");
+    }
+}
+
+void FabricBridgeNode::fetch_target_info() {
+    if (target_host_.empty()) return;
+
+    spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_);
+
+    std::string body = R"({"cmd":"status"})";
+    std::string request = "POST /cmd HTTP/1.1\r\n"
+                           "Host: " + target_host_ + "\r\n"
+                           "Content-Type: application/json\r\n"
+                           "Connection: close\r\n"
+                           "Content-Length: " + std::to_string(body.size()) + "\r\n"
+                           "\r\n" +
+                           body;
+
+    for (int attempt = 0; attempt < 30 && target_info_str_.empty(); ++attempt) {
+        int sock = socket(AF_INET, SOCK_STREAM, 0);
+        if (sock < 0) {
+            spdlog::warn("Fabric-bridge: socket() failed, retry {}...", attempt + 1);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        struct timeval tv{};
+        tv.tv_sec = 2;
+        tv.tv_usec = 0;
+        setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+        setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+
+        struct hostent* he = gethostbyname(target_host_.c_str());
+        if (!he) {
+            spdlog::warn("Fabric-bridge: cannot resolve {}, retry {}...", target_host_, attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        struct sockaddr_in addr{};
+        addr.sin_family = AF_INET;
+        addr.sin_port = htons(target_port_);
+        std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length);
+
+        if (connect(sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
+            spdlog::warn("Fabric-bridge: connect to {}:{} failed (errno={}), retry {}...", target_host_, target_port_, errno, attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+            continue;
+        }
+
+        if (send(sock, request.c_str(), request.size(), 0) < 0) {
+            spdlog::warn("Fabric-bridge: send failed, retry {}...", attempt + 1);
+            close(sock);
+            std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            continue;
+        }
+
+        std::string response;
+        char buf[4096];
+        ssize_t n;
+        while ((n = recv(sock, buf, sizeof(buf), 0)) > 0) {
+            response.append(buf, n);
+        }
+        close(sock);
+
+        spdlog::info("Fabric-bridge: received {} bytes from target", response.size());
+
+        auto body_pos = response.find("\r\n\r\n");
+        if (body_pos == std::string::npos) {
+            spdlog::warn("Fabric-bridge: no HTTP body in response, retry {}...", attempt + 1);
+            std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            continue;
+        }
+
+        std::string response_body = response.substr(body_pos + 4);
+
+        try {
+            auto json = nlohmann::json::parse(response_body);
+            if (json.contains("data") && json["data"].contains("target_info")) {
+                auto ti = json["data"]["target_info"].get<std::string>();
+                if (!ti.empty()) {
+                    target_info_str_ = ti;
+                    spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size());
+                    return;
+                }
+            }
+            spdlog::debug("Fabric-bridge: response data: {}", json.contains("data") ? json["data"].dump() : "(no data)");
+        } catch (const nlohmann::json::exception& e) {
+            spdlog::warn("Fabric-bridge: JSON parse error: {}", e.what());
+        }
+
+        spdlog::info("Fabric-bridge: target_info not ready yet, retry {}...", attempt + 1);
+        std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    }
+
+    if (target_info_str_.empty()) {
+        spdlog::error("Fabric-bridge: failed to fetch target_info after 30 attempts");
+    }
+}
+
+void FabricBridgeNode::process() {
+    if (!worker_thread_.has_value() && start_requested_.load()) {
+        if (mode_ == FabricMode::Initiator && reader_.has_value() && !target_host_.empty() && target_info_str_.empty()) {
+            fetch_target_info();
+        }
+        if (start_requested_.load()) {
+            running_ = true;
+            spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
+            if (mode_ == FabricMode::Initiator) {
+                worker_thread_ = std::thread(&FabricBridgeNode::run_initiator, this);
+            } else {
+                worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
+            }
+        }
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+}
+
+void FabricBridgeNode::run_initiator() {
+    spdlog::info("Fabric-bridge: starting initiator mode");
+
+    auto instance = get_mxl_instance();
+    if (!instance) {
+        spdlog::error("Fabric-bridge: no MXL instance");
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: creating fabrics instance...");
+    mxlFabricsInstance fab_inst = nullptr;
+    auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !fab_inst) {
+        spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
+        running_ = false;
+        return;
+    }
+    fabrics_instance_ = fab_inst;
+
+    spdlog::info("Fabric-bridge: creating initiator...");
+    mxlFabricsInitiator initiator = nullptr;
+    status = mxlFabricsCreateInitiator(fab_inst, &initiator);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInitiator returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !initiator) {
+        spdlog::error("Fabric-bridge: failed to create initiator: {}", static_cast<int>(status));
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+    initiator_ = initiator;
+
+    mxlFlowConfigInfo config{};
+    mxlFlowReaderGetConfigInfo(*reader_, &config);
+    grain_rate_ = config.common.grainRate;
+
+    mxlFabricsInitiatorConfig init_cfg{};
+    init_cfg.version = MXL_FABRICS_API_VERSION;
+    init_cfg.interface.version = MXL_FABRICS_API_VERSION;
+    init_cfg.interface.provider = provider_;
+    init_cfg.interface.address.node = bind_node_.c_str();
+    init_cfg.interface.address.service = bind_service_.c_str();
+    init_cfg.reader = *reader_;
+
+    spdlog::info("Fabric-bridge: calling mxlFabricsInitiatorSetup on {}:{}", bind_node_, bind_service_);
+    status = mxlFabricsInitiatorSetup(initiator, &init_cfg, nullptr);
+    if (status != MXL_STATUS_OK) {
+        spdlog::error("Fabric-bridge: initiator setup failed: {}", static_cast<int>(status));
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    if (target_info_str_.empty()) {
+        spdlog::error("Fabric-bridge: no target_info available, cannot add target");
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    mxlFabricsTargetInfo target_info = nullptr;
+    spdlog::info("Fabric-bridge: parsing target_info ({} bytes)", target_info_str_.size());
+    status = mxlFabricsTargetInfoFromString(target_info_str_.c_str(), &target_info);
+    if (status != MXL_STATUS_OK || !target_info) {
+        spdlog::error("Fabric-bridge: failed to parse target_info: {}", static_cast<int>(status));
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    status = mxlFabricsInitiatorAddTarget(initiator, target_info);
+    if (status != MXL_STATUS_OK) {
+        spdlog::error("Fabric-bridge: failed to add target: {}", static_cast<int>(status));
+        mxlFabricsFreeTargetInfo(target_info);
+        mxlFabricsDestroyInitiator(fab_inst, initiator);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: waiting for connection...");
+    do {
+        status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 250);
+    } while (status == MXL_ERR_NOT_READY && running_);
+
+    auto now = mxlGetTime();
+    read_index_ = mxlTimestampToIndex(&grain_rate_, now) - 2;
+
+    spdlog::info("Fabric-bridge: initiator started, grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+
+    while (running_) {
+        auto deadline = mxlIndexToTimestamp(&grain_rate_, read_index_ + 1);
+        mxlSleepUntil(deadline);
+
+        mxlGrainInfo grain_info{};
+        uint8_t* payload = nullptr;
+        auto grain_status = mxlFlowReaderGetGrain(*reader_, read_index_, 5000000ULL, &grain_info, &payload);
+        if (grain_status != MXL_STATUS_OK) {
+            if (grain_status == MXL_ERR_OUT_OF_RANGE_TOO_LATE || grain_status == MXL_ERR_OUT_OF_RANGE_TOO_EARLY) {
+                auto ts = mxlGetTime();
+                read_index_ = mxlTimestampToIndex(&grain_rate_, ts) - 2;
+                continue;
+            }
+            read_index_++;
+            continue;
+        }
+
+        auto transfer_status = mxlFabricsInitiatorTransferGrain(initiator, read_index_, 0, grain_info.validSlices);
+        if (transfer_status != MXL_STATUS_OK) {
+            spdlog::warn("Fabric-bridge: transfer grain {} failed: {}", read_index_, static_cast<int>(transfer_status));
+            read_index_++;
+            continue;
+        }
+
+        do {
+            status = mxlFabricsInitiatorMakeProgressBlocking(initiator, 10);
+        } while (status == MXL_ERR_NOT_READY && running_);
+
+        read_index_++;
+        grains_transferred_++;
+
+        if (grains_transferred_ == 1) {
+            spdlog::info("Fabric-bridge: first grain transferred, index={}", read_index_ - 1);
+        }
+    }
+
+    mxlFabricsInitiatorRemoveTarget(initiator, target_info);
+    mxlFabricsFreeTargetInfo(target_info);
+    mxlFabricsDestroyInitiator(fab_inst, initiator);
+    mxlFabricsDestroyInstance(fab_inst);
+    spdlog::info("Fabric-bridge: initiator stopped");
+}
+
+void FabricBridgeNode::run_target() {
+    spdlog::info("Fabric-bridge: starting target mode");
+
+    auto instance = get_mxl_instance();
+    if (!instance) {
+        spdlog::error("Fabric-bridge: no MXL instance");
+        running_ = false;
+        return;
+    }
+
+    spdlog::info("Fabric-bridge: creating fabrics instance...");
+    mxlFabricsInstance fab_inst = nullptr;
+    auto status = mxlFabricsCreateInstance(instance, nullptr, &fab_inst);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateInstance returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !fab_inst) {
+        spdlog::error("Fabric-bridge: failed to create fabrics instance: {}", static_cast<int>(status));
+        running_ = false;
+        return;
+    }
+    fabrics_instance_ = fab_inst;
+
+    spdlog::info("Fabric-bridge: creating target...");
+    mxlFabricsTarget tgt = nullptr;
+    status = mxlFabricsCreateTarget(fab_inst, &tgt);
+    spdlog::info("Fabric-bridge: mxlFabricsCreateTarget returned {}", static_cast<int>(status));
+    if (status != MXL_STATUS_OK || !tgt) {
+        spdlog::error("Fabric-bridge: failed to create target: {}", static_cast<int>(status));
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+    target_ = tgt;
+
+    auto* config = get_writer_config("video_out");
+    if (config) {
+        grain_rate_ = config->common.grainRate;
+        spdlog::info("Fabric-bridge: target grain_rate={}/{}", grain_rate_.numerator, grain_rate_.denominator);
+    } else {
+        spdlog::warn("Fabric-bridge: no writer config, using default grain_rate");
+    }
+
+    mxlFabricsTargetConfig target_cfg{};
+    target_cfg.version = MXL_FABRICS_API_VERSION;
+    target_cfg.interface.version = MXL_FABRICS_API_VERSION;
+    target_cfg.interface.provider = provider_;
+    target_cfg.interface.address.node = bind_node_.c_str();
+    target_cfg.interface.address.service = bind_service_.c_str();
+    target_cfg.writer = *writer_;
+
+    spdlog::info("Fabric-bridge: calling mxlFabricsTargetSetup on {}:{}", bind_node_, bind_service_);
+    mxlFabricsTargetInfo target_info = nullptr;
+    status = mxlFabricsTargetSetup(tgt, &target_cfg, nullptr, &target_info);
+    if (status != MXL_STATUS_OK || !target_info) {
+        spdlog::error("Fabric-bridge: target setup failed: {}", static_cast<int>(status));
+        mxlFabricsDestroyTarget(fab_inst, tgt);
+        mxlFabricsDestroyInstance(fab_inst);
+        running_ = false;
+        return;
+    }
+
+    size_t info_size = 0;
+    mxlFabricsTargetInfoToString(target_info, nullptr, &info_size);
+    std::vector<char> info_buf(info_size);
+    mxlFabricsTargetInfoToString(target_info, info_buf.data(), &info_size);
+    target_info_str_ = std::string(info_buf.data(), info_size);
+
+    spdlog::info("Fabric-bridge: target ready, target_info length={}", target_info_str_.size());
+    spdlog::info("Fabric-bridge: TARGET_INFO={}", target_info_str_);
+
+    while (running_) {
+        uint64_t grain_index = 0;
+        status = mxlFabricsTargetReadGrain(tgt, 200, &grain_index);
+        if (status == MXL_ERR_TIMEOUT || status == MXL_ERR_NOT_READY) {
+            continue;
+        }
+        if (status != MXL_STATUS_OK) {
+            spdlog::warn("Fabric-bridge: target read grain failed: {}", static_cast<int>(status));
+            continue;
+        }
+
+        mxlGrainInfo grain_info{};
+        uint8_t* dummy_payload = nullptr;
+        auto open_status = mxlFlowWriterOpenGrain(*writer_, grain_index, &grain_info, &dummy_payload);
+        if (open_status == MXL_STATUS_OK) {
+            grain_info.validSlices = grain_info.totalSlices;
+            mxlFlowWriterCommitGrain(*writer_, &grain_info);
+            grains_transferred_++;
+            if (grains_transferred_ == 1) {
+                spdlog::info("Fabric-bridge: first grain committed, index={}", grain_index);
+            }
+        } else {
+            spdlog::warn("Fabric-bridge: open grain {} failed: {}", grain_index, static_cast<int>(open_status));
+        }
+    }
+
+    mxlFabricsFreeTargetInfo(target_info);
+    mxlFabricsDestroyTarget(fab_inst, tgt);
+    mxlFabricsDestroyInstance(fab_inst);
+    spdlog::info("Fabric-bridge: target stopped");
+}
+
+nlohmann::json FabricBridgeNode::status() const {
+    return {
+        {"type", "fabric-bridge"},
+        {"mode", mode_ == FabricMode::Initiator ? "initiator" : "target"},
+        {"grains_transferred", grains_transferred_},
+        {"running", running_.load()},
+        {"has_reader", reader_.has_value()},
+        {"has_writer", writer_.has_value()},
+        {"target_info_length", target_info_str_.size()},
+        {"target_info", target_info_str_},
+    };
+}
+
+} // namespace dmf_node
@@ -0,0 +1,68 @@
+#pragma once
+
+#include <dmf-node/node.hpp>
+#include <mxl/flow.h>
+#include <mxl/fabrics.h>
+#include <mxl/time.h>
+
+#include <atomic>
+#include <optional>
+#include <string>
+#include <thread>
+
+namespace dmf_node {
+
+enum class FabricMode { Initiator, Target };
+
+class FabricBridgeNode : public Node {
+public:
+    FabricBridgeNode() = default;
+    ~FabricBridgeNode();
+
+    std::string type() const override { return "fabric-bridge"; }
+
+    std::vector<PortDef> input_ports() const override;
+    std::vector<PortDef> output_ports() const override;
+
+    void configure(const nlohmann::json& params) override;
+    void on_add_writer(const std::string& port_id, mxlFlowWriter writer) override;
+    void on_add_reader(const std::string& port_id, mxlFlowReader reader) override;
+    void on_remove_writer(const std::string& port_id) override;
+    void on_remove_reader(const std::string& port_id) override;
+
+    void process() override;
+    nlohmann::json status() const override;
+
+private:
+    void run_initiator();
+    void run_target();
+
+    FabricMode mode_ = FabricMode::Initiator;
+
+    std::string bind_node_ = "0.0.0.0";
+    std::string bind_service_ = "0";
+    mxlFabricsProvider provider_ = MXL_FABRICS_PROVIDER_ANY;
+
+    std::optional<mxlFlowReader> reader_;
+    std::optional<mxlFlowWriter> writer_;
+    mxlRational grain_rate_{50, 1};
+    uint64_t read_index_ = 0;
+    uint64_t grains_transferred_ = 0;
+
+    std::string target_info_str_;
+
+    std::string target_host_;
+    uint16_t target_port_ = 9100;
+
+    void fetch_target_info();
+
+    mxlFabricsInstance fabrics_instance_ = nullptr;
+    mxlFabricsInitiator initiator_ = nullptr;
+    mxlFabricsTarget target_ = nullptr;
+
+    std::optional<std::thread> worker_thread_;
+    std::atomic<bool> running_{false};
+    std::atomic<bool> start_requested_{false};
+};
+
+} // namespace dmf_node
@@ -0,0 +1,6 @@
+#include "fabric_bridge_node.hpp"
+#include <dmf-node/node_runner.hpp>
+
+int main(int argc, char* argv[]) {
+    return dmf_node::NodeRunner::run<dmf_node::FabricBridgeNode>(argc, argv);
+}
Author	SHA1	Message	Date
Johanness	12896182c9	fix: correct HTTP request in fetch_target_info, add socket timeouts and debug logging - Content-Length was wrong (20 vs actual 18 bytes) - Removed running_ check from loop (was false during startup) - Added SO_RCVTIMEO/SO_SNDTIMEO to prevent infinite recv hangs - Added response size logging for debugging	2026-06-16 11:38:48 +03:00
Johanness	d093d64ad1	fix: trigger startup on add_reader when target_host is set The initiator couldn't start because start_requested_ was never set when target_host was provided without target_info. Now on_add_reader signals startup when either target_info_str_ or target_host_ is available.	2026-06-16 11:29:23 +03:00
Johanness	e6e439b16e	feat: add node register endpoint and fabric create-target/connect APIs - POST /api/graph/nodes/{id}/register: register manually-started node with control port - POST /api/fabric/create-target: add_writer + poll for target_info - POST /api/fabric/connect: auto-propagate target_info to initiator - fabric-bridge: auto-fetch target_info from remote target via HTTP	2026-06-16 11:11:13 +03:00
Johanness	0944f574ad	fabric-bridge: remove unnecessary win32 includes	2026-06-16 11:09:10 +03:00
Johanness	cca0cd810a	feat: fabric-bridge auto-fetch target_info from remote target node Initiator can now auto-discover target_info by polling the target's HTTP control server. Just set target_host and target_port in config: --config '{"mode":"initiator",...,"target_host":"10.10.100.1","target_port":9100}' No more manual copy-paste of target_info JSON between hosts.	2026-06-16 11:08:06 +03:00
Johanness	ac3a506a5c	feat: add engine API for fabric-bridge target_info exchange - POST /api/fabric/create-target: add_writer to target node, poll for target_info - POST /api/fabric/connect: get target_info from target, configure initiator node	2026-06-16 10:50:04 +03:00
Johanness	991620092e	Merge branch 'develop'	2026-06-16 10:42:28 +03:00
Johanness	da836785df	fix: set running_=true before starting fabric-bridge worker thread	2026-06-14 21:16:23 +03:00
Johanness	0a49b31094	fabric-bridge: flush logs on info, log worker thread start	2026-06-14 21:09:50 +03:00
Johanness	62e2dc49bf	cmake: enable MXL tools build by default (testsrc, sink)	2026-06-14 17:03:50 +03:00
Johanness	8fa058f37f	fix: add missing return values in add_writer/add_reader handlers (UB causing SIGILL)	2026-06-14 16:57:22 +03:00
Johanness	a6ac2bf1a1	fabric-bridge: thread-safe startup, detailed Fabrics API logging	2026-06-14 16:54:37 +03:00
Johanness	17a8b23d03	fabric-bridge: add detailed logging before each Fabrics API call	2026-06-14 16:49:02 +03:00
Johanness	a9d4994504	fix: base control port on engine port to avoid conflicts (port+100)	2026-06-14 16:41:52 +03:00
Johanness	919f61fc72	fix: fabric-bridge build - use NodeRunner::run, get writer config from node base class	2026-06-14 15:27:47 +03:00
Johanness	1ab41b6740	feat: add fabric-bridge node for RDMA/inter-host MXL flow sharing - New fabric-bridge node: initiator mode reads local MXL flow and transfers via Fabrics (RDMA/TCP); target mode receives remote data and commits to local MXL flow writer - Add set_mxl_instance/get_mxl_instance to Node base class - CMake option DMF_BUILD_FABRICS=OFF by default (needs libfabric) - Target publishes target_info in status for out-of-band exchange - Initiator receives target_info via config parameter	2026-06-14 11:26:00 +03:00
Johanness	7568c8472d	Merge develop: Phase 3 DeckLink I/O + MXL update	2026-06-14 11:20:57 +03:00
Johanness	4402fb2e12	fix decklink-out: use stream_frame_ counter for DeckLink scheduling instead of TAI-based grains_read_	2026-05-30 23:26:51 +03:00
Johanness	b434c61b86	decklink-out: non-blocking frame pool, skip late frames instead of realign, 5 preroll frames	2026-05-30 23:24:54 +03:00
Johanness	bd1f060c73	decklink-out: reuse frames via ScheduledFrameCompleted callback pool	2026-05-30 23:23:15 +03:00
Johanness	26febc41d6	decklink-out: preroll frames before StartScheduledPlayback, detect output format	2026-05-30 23:21:13 +03:00
Johanness	a3cd45cc49	decklink-out: add mode support check, limit CreateVideoFrame error spam	2026-05-30 23:18:02 +03:00
Johanness	b4f766fb9d	add HRESULT logging to CreateVideoFrame for debug	2026-05-30 23:15:00 +03:00
Johanness	b5d9c7cd3f	fix decklink-out: init grains_read from current time to avoid epoch realignment	2026-05-30 23:13:24 +03:00
Johanness	c2e823beda	fix realignment	2026-05-30 23:10:35 +03:00
Johanness	003279669e	fix output res&frame rate	2026-05-30 23:07:15 +03:00
Johanness	2206fd71b3	some fixes	2026-05-30 22:12:47 +03:00
Johanness	8b5fb3dc36	fix skipped write_index	2026-05-28 23:55:16 +03:00
Johanness	34f72c22f0	hardcoded decklink mode fix	2026-05-28 23:43:26 +03:00
Johanness	96bef6cd32	fix: use relative symlink for extern/mxl (works on any checkout path)	2026-05-28 22:51:45 +03:00
Johanness	d02223224f	feat: add DeckLink input and output nodes decklink-in: - IDeckLinkInputCallback::VideoInputFrameArrived captures V210 frames - Frame data stored with mutex, copied to MXL grain in process thread - Uses mxlSleepUntil for TAI-time-based grain pacing - Configurable: device_index (int), mode (1080i50/1080p50/etc) - Auto-detect input format via bmdVideoInputEnableFormatDetection - 1 output port: video_out (V210) decklink-out: - Reads V210 grains from MXL, schedules playback via DeckLink output - Creates IDeckLinkMutableVideoFrame, copies grain data, schedules - Uses ScheduledFrameCompleted callback for frame completion - Configurable: device_index (int), mode (1080i50/1080p50/etc) - 1 input port: video_in (V210) Both nodes: - Use DeckLinkAPIDispatch.cpp for CreateDeckLinkIteratorInstance - Access pixel data via IDeckLinkVideoBuffer (latest SDK API) - Graceful device open/close on writer/reader add/remove - Build conditionally via DMF_BUILD_DECKLINK + DECKLINK_SDK_DIR	2026-05-28 22:23:00 +03:00
Johanness	e4f3498617	Merge develop: Phase 1 complete — Framework + Passthrough	2026-05-26 23:04:51 +03:00
Johanness	db41a91967	Merge feature/phase1-framework: Phase 1 complete Phase 1: Framework + Passthrough — Two processes exchanging video via MXL, controlled by engine. Implemented: - Project scaffold: CMake + vcpkg, MXL SDK submodule - libdmf-node: Node interface, PortDef, ControlServer (WS + HTTP /cmd), NodeRunner (CLI, MXL lifecycle, separate process thread) - libdmf-engine: Graph model, FlowManager (UUID v4, NMOS V210 flow defs), ProcessManager (fork/exec), ApiServer (REST), NodeControlClient - dmf-node-passthrough: V210 grain read→memcpy→write, TAI-time-based grain pacing with mxlSleepUntil, 2-grain read delay - dmf-studio-engine: REST API with full graph CRUD + start/stop + connect-input/connect-output/disconnect-port/command endpoints - Proper shutdown: release MXL readers/writers before destroy, engine kills child processes on exit - Unit tests: 6 cases, 22 assertions, all passing	2026-05-26 23:04:45 +03:00