feat: add node HTTP control, engine-to-node communication, connect-input/output API
- Node control server now accepts HTTP POST /cmd for command dispatch (in addition to existing WebSocket control) - Added NodeControlClient: engine sends commands to nodes via HTTP POST - New REST endpoints: POST /api/graph/nodes/:id/connect-input - connect node input to MXL flow POST /api/graph/nodes/:id/connect-output - create new MXL flow + connect output POST /api/graph/nodes/:id/disconnect-port - remove reader/writer POST /api/graph/nodes/:id/command - send raw command to node - Flow IDs now use proper UUID v4 format (MXL requires standard UUIDs) - Flow definitions use NMOS format (urn:x-nmos:format:video) - Engine passes --mxl-domain to node processes - User-provided node IDs (via 'id' field in POST body) - End-to-end verified: testsrc → passthrough → new MXL output flow
This commit is contained in:
Johanness
@@ -2,6 +2,7 @@
|
||||
#include <dmf-engine/graph.hpp>
|
||||
#include <dmf-engine/flow_manager.hpp>
|
||||
#include <dmf-engine/process_manager.hpp>
|
||||
#include <dmf-engine/node_control_client.hpp>
|
||||
|
||||
#include <libwebsockets.h>
|
||||
|
||||
@@ -19,6 +20,8 @@ struct ApiServerImpl {
|
||||
Graph* graph = nullptr;
|
||||
FlowManager* flow_manager = nullptr;
|
||||
ProcessManager* process_manager = nullptr;
|
||||
NodeControlClient* control_client = nullptr;
|
||||
std::string mxl_domain = "/dev/shm/mxl";
|
||||
struct lws_context* context = nullptr;
|
||||
};
|
||||
|
||||
@@ -84,6 +87,7 @@ static void handle_request(const std::string& method, const std::string& path,
|
||||
auto& graph = *g_impl->graph;
|
||||
auto& fm = *g_impl->flow_manager;
|
||||
auto& pm = *g_impl->process_manager;
|
||||
auto& cc = *g_impl->control_client;
|
||||
|
||||
if (path == "/api/graph" && method == "GET") {
|
||||
ok(graph.serialize());
|
||||
@@ -93,11 +97,15 @@ static void handle_request(const std::string& method, const std::string& path,
|
||||
} else {
|
||||
auto type = req_body["type"].get<std::string>();
|
||||
auto config = req_body.value("config", nlohmann::json::object());
|
||||
if (req_body.contains("id") && req_body["id"].is_string()) {
|
||||
config["id"] = req_body["id"];
|
||||
}
|
||||
auto id = graph.add_node(type, config);
|
||||
created({{"id", id}});
|
||||
}
|
||||
} else if (path.find("/api/graph/nodes/") == 0 && method == "DELETE") {
|
||||
auto node_id = path.substr(std::string("/api/graph/nodes/").length());
|
||||
cc.unregister_node(node_id);
|
||||
if (graph.remove_node(node_id)) {
|
||||
ok({{"deleted", node_id}});
|
||||
} else {
|
||||
@@ -116,11 +124,54 @@ static void handle_request(const std::string& method, const std::string& path,
|
||||
auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
|
||||
|
||||
auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def);
|
||||
|
||||
auto from_port_num = cc.get_port(from_node);
|
||||
auto to_port_num = cc.get_port(to_node);
|
||||
|
||||
if (from_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "add_writer";
|
||||
cmd["port_id"] = from_port;
|
||||
cmd["flow_id"] = flow_id;
|
||||
cmd["flow_def"] = flow_def;
|
||||
cc.send_command(from_port_num, cmd.dump());
|
||||
}
|
||||
|
||||
if (to_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "add_reader";
|
||||
cmd["port_id"] = to_port;
|
||||
cmd["flow_id"] = flow_id;
|
||||
cc.send_command(to_port_num, cmd.dump());
|
||||
}
|
||||
|
||||
created({{"id", edge_id}, {"flow_id", flow_id}});
|
||||
}
|
||||
} else if (path.find("/api/graph/edges/") == 0 && method == "DELETE") {
|
||||
auto edge_id = path.substr(std::string("/api/graph/edges/").length());
|
||||
auto edges = graph.get_edges();
|
||||
const GraphEdge* edge = nullptr;
|
||||
for (auto& e : edges) {
|
||||
if (e.id == edge_id) { edge = &e; break; }
|
||||
}
|
||||
|
||||
if (graph.remove_edge(edge_id)) {
|
||||
if (edge) {
|
||||
auto from_port_num = cc.get_port(edge->from_node);
|
||||
auto to_port_num = cc.get_port(edge->to_node);
|
||||
if (from_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "remove_writer";
|
||||
cmd["port_id"] = edge->from_port;
|
||||
cc.send_command(from_port_num, cmd.dump());
|
||||
}
|
||||
if (to_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "remove_reader";
|
||||
cmd["port_id"] = edge->to_port;
|
||||
cc.send_command(to_port_num, cmd.dump());
|
||||
}
|
||||
}
|
||||
ok({{"deleted", edge_id}});
|
||||
} else {
|
||||
error_resp(404, "Edge not found: " + edge_id);
|
||||
@@ -129,15 +180,141 @@ static void handle_request(const std::string& method, const std::string& path,
|
||||
auto nodes = graph.get_nodes();
|
||||
uint16_t port = 9100;
|
||||
for (auto& node : nodes) {
|
||||
pm.start_node(const_cast<GraphNode&>(node), "/dev/shm/mxl", port++);
|
||||
pm.start_node(const_cast<GraphNode&>(node), g_impl->mxl_domain, port);
|
||||
cc.register_node(node.id, port);
|
||||
port++;
|
||||
}
|
||||
|
||||
for (auto& edge : graph.get_edges()) {
|
||||
auto from_port_num = cc.get_port(edge.from_node);
|
||||
auto to_port_num = cc.get_port(edge.to_node);
|
||||
|
||||
if (from_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "add_writer";
|
||||
cmd["port_id"] = edge.from_port;
|
||||
cmd["flow_id"] = edge.flow_id;
|
||||
cmd["flow_def"] = edge.flow_def;
|
||||
cc.send_command(from_port_num, cmd.dump());
|
||||
}
|
||||
|
||||
if (to_port_num > 0) {
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "add_reader";
|
||||
cmd["port_id"] = edge.to_port;
|
||||
cmd["flow_id"] = edge.flow_id;
|
||||
cc.send_command(to_port_num, cmd.dump());
|
||||
}
|
||||
}
|
||||
|
||||
ok({{"status", "started"}});
|
||||
} else if (path == "/api/graph/stop" && method == "POST") {
|
||||
auto nodes = graph.get_nodes();
|
||||
for (auto& node : nodes) {
|
||||
cc.unregister_node(node.id);
|
||||
pm.stop_node(const_cast<GraphNode&>(node));
|
||||
}
|
||||
ok({{"status", "stopped"}});
|
||||
} else if (path.find("/api/graph/nodes/") == 0 && path.find("/connect-input") != std::string::npos && method == "POST") {
|
||||
auto prefix = std::string("/api/graph/nodes/");
|
||||
auto suffix_start = path.find("/connect-input");
|
||||
auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
|
||||
|
||||
if (!req_body.contains("flow_id") || !req_body.contains("port_id")) {
|
||||
error_resp(400, "Missing flow_id/port_id");
|
||||
return;
|
||||
}
|
||||
|
||||
auto port_num = cc.get_port(node_id);
|
||||
if (port_num == 0) {
|
||||
error_resp(404, "Node not running or not found: " + node_id);
|
||||
return;
|
||||
}
|
||||
|
||||
nlohmann::json cmd;
|
||||
cmd["cmd"] = "add_reader";
|
||||
cmd["port_id"] = req_body["port_id"].get<std::string>();
|
||||
cmd["flow_id"] = req_body["flow_id"].get<std::string>();
|
||||
|
||||
if (cc.send_command(port_num, cmd.dump())) {
|
||||
ok({{"node_id", node_id}, {"connected_input", req_body["port_id"]}, {"flow_id", req_body["flow_id"]}});
|
||||
} else {
|
||||
error_resp(500, "Failed to send command to node");
|
||||
}
|
||||
} else if (path.find("/api/graph/nodes/") == 0 && path.find("/connect-output") != std::string::npos && method == "POST") {
|
||||
auto prefix = std::string("/api/graph/nodes/");
|
||||
auto suffix_start = path.find("/connect-output");
|
||||
auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
|
||||
|
||||
auto port_id = req_body.value("port_id", "video_out");
|
||||
|
||||
auto port_num = cc.get_port(node_id);
|
||||
if (port_num == 0) {
|
||||
error_resp(404, "Node not running or not found: " + node_id);
|
||||
return;
|
||||
}
|
||||
|
||||
auto flow_id = fm.create_flow_id();
|
||||
auto flow_def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
|
||||
|
||||
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())) {
|
||||
ok({{"node_id", node_id}, {"connected_output", port_id}, {"flow_id", flow_id}});
|
||||
} else {
|
||||
error_resp(500, "Failed to send command to node");
|
||||
}
|
||||
} else if (path.find("/api/graph/nodes/") == 0 && path.find("/disconnect-port") != std::string::npos && method == "POST") {
|
||||
auto prefix = std::string("/api/graph/nodes/");
|
||||
auto suffix_start = path.find("/disconnect-port");
|
||||
auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
|
||||
|
||||
if (!req_body.contains("port_id")) {
|
||||
error_resp(400, "Missing port_id");
|
||||
return;
|
||||
}
|
||||
|
||||
auto port_id = req_body["port_id"].get<std::string>();
|
||||
auto port_num = cc.get_port(node_id);
|
||||
if (port_num == 0) {
|
||||
error_resp(404, "Node not running or not found: " + node_id);
|
||||
return;
|
||||
}
|
||||
|
||||
auto direction = req_body.value("direction", "input");
|
||||
nlohmann::json cmd;
|
||||
if (direction == "output") {
|
||||
cmd["cmd"] = "remove_writer";
|
||||
} else {
|
||||
cmd["cmd"] = "remove_reader";
|
||||
}
|
||||
cmd["port_id"] = port_id;
|
||||
|
||||
if (cc.send_command(port_num, cmd.dump())) {
|
||||
ok({{"node_id", node_id}, {"disconnected", port_id}});
|
||||
} else {
|
||||
error_resp(500, "Failed to send command to node");
|
||||
}
|
||||
} 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");
|
||||
auto node_id = path.substr(prefix.length(), suffix_start - prefix.length());
|
||||
|
||||
auto port_num = cc.get_port(node_id);
|
||||
if (port_num == 0) {
|
||||
error_resp(404, "Node not running or not found: " + node_id);
|
||||
return;
|
||||
}
|
||||
|
||||
if (cc.send_command(port_num, req_body.dump())) {
|
||||
ok({{"node_id", node_id}, {"sent", true}});
|
||||
} else {
|
||||
error_resp(500, "Failed to send command to node");
|
||||
}
|
||||
} else {
|
||||
error_resp(404, "Not found: " + method + " " + path);
|
||||
}
|
||||
@@ -207,13 +384,15 @@ static const struct lws_protocols protocols[] = {
|
||||
{nullptr, nullptr, 0, 0},
|
||||
};
|
||||
|
||||
ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager)
|
||||
ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager, NodeControlClient& control_client, const std::string& mxl_domain)
|
||||
: impl_(std::make_unique<Impl>()) {
|
||||
impl_->data = std::make_unique<ApiServerImpl>();
|
||||
impl_->data->port = port;
|
||||
impl_->data->graph = &graph;
|
||||
impl_->data->flow_manager = &flow_manager;
|
||||
impl_->data->process_manager = &process_manager;
|
||||
impl_->data->control_client = &control_client;
|
||||
impl_->data->mxl_domain = mxl_domain;
|
||||
|
||||
g_impl = impl_->data.get();
|
||||
|
||||
|
||||
@@ -9,18 +9,29 @@
|
||||
|
||||
namespace dmf_engine {
|
||||
|
||||
FlowManager::FlowManager() {
|
||||
std::random_device rd;
|
||||
flow_counter_ = static_cast<int>(rd());
|
||||
}
|
||||
FlowManager::FlowManager() = default;
|
||||
|
||||
FlowId FlowManager::create_flow_id() {
|
||||
std::random_device rd;
|
||||
std::mt19937 gen(rd());
|
||||
std::uniform_int_distribution<uint32_t> dist(0, 0xFFFFFFFF);
|
||||
|
||||
uint32_t a = dist(gen);
|
||||
uint16_t b = dist(gen) & 0xFFFF;
|
||||
uint16_t c = (dist(gen) & 0x0FFF) | 0x4000;
|
||||
uint16_t d = (dist(gen) & 0x3FFF) | 0x8000;
|
||||
uint32_t e1 = dist(gen);
|
||||
uint16_t e2 = dist(gen) & 0xFFFF;
|
||||
|
||||
std::stringstream ss;
|
||||
ss << std::hex << (++flow_counter_);
|
||||
ss << std::hex << std::setfill('0');
|
||||
ss << std::setw(8) << a << "-";
|
||||
ss << std::setw(4) << b << "-";
|
||||
ss << std::setw(4) << c << "-";
|
||||
ss << std::setw(4) << d << "-";
|
||||
ss << std::setw(8) << e1 << std::setw(4) << e2;
|
||||
|
||||
auto id = ss.str();
|
||||
while (id.length() < 8) {
|
||||
id = "0" + id;
|
||||
}
|
||||
spdlog::info("FlowManager: created flow ID: {}", id);
|
||||
return id;
|
||||
}
|
||||
@@ -30,17 +41,22 @@ nlohmann::json FlowManager::create_v210_flow_def(const FlowId& flow_id, int widt
|
||||
|
||||
nlohmann::json flow_def;
|
||||
flow_def["id"] = flow_id;
|
||||
flow_def["version"] = "v1.0";
|
||||
flow_def["label"] = "DMF Studio Flow " + flow_id;
|
||||
flow_def["description"] = "Auto-generated V210 flow";
|
||||
flow_def["format"] = "urn:x-nmos:format:video";
|
||||
flow_def["grain_rate"] = {{"numerator", fps_numerator}, {"denominator", fps_denominator}};
|
||||
flow_def["format"] = "video/v210";
|
||||
flow_def["width"] = width;
|
||||
flow_def["height"] = height;
|
||||
flow_def["grain_size"] = grain_size;
|
||||
flow_def["components"] = nlohmann::json::array();
|
||||
flow_def["media_type"] = "video/v210";
|
||||
flow_def["frame_width"] = width;
|
||||
flow_def["frame_height"] = height;
|
||||
flow_def["interlace_mode"] = "progressive";
|
||||
flow_def["colorspace"] = "BT709";
|
||||
flow_def["components"] = nlohmann::json::array({
|
||||
{{"name", "Y"}, {"width", width}, {"height", height}, {"bit_depth", 10}},
|
||||
{{"name", "Cb"}, {"width", width / 2}, {"height", height}, {"bit_depth", 10}},
|
||||
{{"name", "Cr"}, {"width", width / 2}, {"height", height}, {"bit_depth", 10}},
|
||||
});
|
||||
flow_def["tags"] = nlohmann::json::object();
|
||||
flow_def["tags"]["grouphint"] = nlohmann::json::array({"DMF Studio:Video"});
|
||||
flow_def["tags"]["urn:x-nmos:tag:grouphint/v1.0"] = nlohmann::json::array({"DMF Studio:Video"});
|
||||
|
||||
return flow_def;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
#include <dmf-engine/node_control_client.hpp>
|
||||
|
||||
#include <spdlog/spdlog.h>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <cstring>
|
||||
#include <netinet/in.h>
|
||||
#include <sys/socket.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <sstream>
|
||||
|
||||
namespace dmf_engine {
|
||||
|
||||
void NodeControlClient::register_node(const std::string& node_id, uint16_t port) {
|
||||
node_ports_[node_id] = port;
|
||||
spdlog::info("NodeControlClient: registered node '{}' on port {}", node_id, port);
|
||||
}
|
||||
|
||||
void NodeControlClient::unregister_node(const std::string& node_id) {
|
||||
node_ports_.erase(node_id);
|
||||
}
|
||||
|
||||
uint16_t NodeControlClient::get_port(const std::string& node_id) const {
|
||||
auto it = node_ports_.find(node_id);
|
||||
return it != node_ports_.end() ? it->second : 0;
|
||||
}
|
||||
|
||||
bool NodeControlClient::send_command(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 false;
|
||||
}
|
||||
|
||||
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 false;
|
||||
}
|
||||
|
||||
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 false;
|
||||
}
|
||||
|
||||
char resp_buf[4096] = {};
|
||||
auto n = read(fd, resp_buf, sizeof(resp_buf) - 1);
|
||||
close(fd);
|
||||
|
||||
if (n <= 0) {
|
||||
spdlog::error("NodeControlClient: read failed on port {}", port);
|
||||
return false;
|
||||
}
|
||||
|
||||
std::string resp(resp_buf, n);
|
||||
bool ok = resp.find("200 OK") != std::string::npos || resp.find("201 Created") != std::string::npos;
|
||||
if (!ok) {
|
||||
spdlog::warn("NodeControlClient: command failed on port {}: {}", port, resp.substr(0, 100));
|
||||
}
|
||||
return ok;
|
||||
}
|
||||
|
||||
} // namespace dmf_engine
|
||||
Reference in New Issue
Block a user