feat: scaffold project framework with engine, node skeleton, and passthrough node

- CMake build system with vcpkg dependency management
- libdmf-node: Node interface, port definitions, WebSocket control
  server, node runner with MXL lifecycle management
- libdmf-engine: Graph model (nodes/edges CRUD, serialization),
  FlowManager (UUID + V210 NMOS flow definitions), ProcessManager
  (fork/exec node processes), ApiServer (REST API for graph control)
- Passthrough node: reads V210 grains from MXL, copies, writes to MXL
- Unit tests for Graph and FlowManager (6 cases, 21 assertions)
- MXL SDK as git submodule (symlinked from extern/mxl)
This commit is contained in:
Johanness
2026-05-26 01:32:50 +03:00
parent ea0eaf8d48
commit 5b5ffa1308
30 changed files with 1547 additions and 0 deletions
+12
View File
@@ -1 +1,13 @@
ref_arch.pdf ref_arch.pdf
build/
.cache/
CMakeUserPresets.json
compile_commands.json
.vcpkg/
vcpkg_installed/
node_modules/
web/dist/
*.o
*.a
*.so
*.d
+31
View File
@@ -0,0 +1,31 @@
cmake_minimum_required(VERSION 3.24 FATAL_ERROR)
project(dmf-studio
VERSION 0.1.0
LANGUAGES CXX C
)
set(CMAKE_CXX_STANDARD 20)
set(CMAKE_CXX_STANDARD_REQUIRED ON)
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)
find_package(fmt CONFIG REQUIRED)
find_package(spdlog CONFIG REQUIRED)
find_package(nlohmann_json CONFIG REQUIRED)
find_package(Libwebsockets CONFIG REQUIRED)
find_package(Catch2 CONFIG QUIET)
add_subdirectory(extern/mxl)
add_subdirectory(libs/dmf-node)
add_subdirectory(libs/dmf-engine)
add_subdirectory(nodes/passthrough)
add_subdirectory(engine)
if(DMF_BUILD_TESTS AND Catch2_FOUND)
add_subdirectory(tests)
endif()
+5
View File
@@ -0,0 +1,5 @@
add_executable(dmf-studio-engine
src/main.cpp
)
target_link_libraries(dmf-studio-engine PRIVATE dmf-engine)
+46
View File
@@ -0,0 +1,46 @@
#include <dmf-engine/api_server.hpp>
#include <dmf-engine/graph.hpp>
#include <dmf-engine/flow_manager.hpp>
#include <dmf-engine/process_manager.hpp>
#include <spdlog/spdlog.h>
#include <csignal>
#include <atomic>
static std::atomic<bool> g_running{true};
static void signal_handler(int /*signum*/) {
g_running = false;
}
int main(int argc, char* argv[]) {
uint16_t port = 8080;
for (int i = 1; i < argc; ++i) {
std::string arg = argv[i];
if ((arg == "--port" || arg == "-p") && i + 1 < argc) {
port = static_cast<uint16_t>(std::stoi(argv[++i]));
}
}
std::signal(SIGINT, signal_handler);
std::signal(SIGTERM, signal_handler);
spdlog::info("DMF Studio Engine starting on port {}", port);
dmf_engine::Graph graph;
dmf_engine::FlowManager flow_manager;
dmf_engine::ProcessManager process_manager;
dmf_engine::ApiServer api_server(port, graph, flow_manager, process_manager);
spdlog::info("DMF Studio Engine ready");
while (g_running) {
api_server.poll(100);
}
spdlog::info("DMF Studio Engine shutting down");
return 0;
}
Vendored Symlink
+1
View File
@@ -0,0 +1 @@
/home/itten/DMF/mxl
+18
View File
@@ -0,0 +1,18 @@
add_library(dmf-engine STATIC
src/graph.cpp
src/flow_manager.cpp
src/process_manager.cpp
src/api_server.cpp
)
target_include_directories(dmf-engine PUBLIC
include
)
target_link_libraries(dmf-engine PUBLIC
dmf-node
nlohmann_json::nlohmann_json
spdlog::spdlog
fmt::fmt
websockets
)
@@ -0,0 +1,33 @@
#pragma once
#include <dmf-engine/graph.hpp>
#include <functional>
#include <memory>
#include <string>
struct lws_context;
namespace dmf_engine {
using RequestHandler = std::function<std::string(const std::string& method, const std::string& path, const std::string& body)>;
class ApiServer {
public:
ApiServer(uint16_t port, Graph& graph, class FlowManager& flow_manager, class ProcessManager& process_manager);
~ApiServer();
ApiServer(const ApiServer&) = delete;
ApiServer& operator=(const ApiServer&) = delete;
void poll(int timeout_ms);
private:
void handle_rest(const std::string& method, const std::string& path, const std::string& body,
std::string& response_status, std::string& response_content_type, std::string& response_body);
struct Impl;
std::unique_ptr<Impl> impl_;
};
} // namespace dmf_engine
@@ -0,0 +1,22 @@
#pragma once
#include <dmf-engine/types.hpp>
#include <nlohmann/json.hpp>
#include <string>
namespace dmf_engine {
class FlowManager {
public:
FlowManager();
FlowId create_flow_id();
nlohmann::json create_v210_flow_def(const FlowId& flow_id, int width, int height, int fps_numerator, int fps_denominator) const;
private:
int flow_counter_ = 0;
};
} // namespace dmf_engine
@@ -0,0 +1,62 @@
#pragma once
#include <dmf-engine/types.hpp>
#include <nlohmann/json.hpp>
#include <optional>
#include <string>
#include <unordered_map>
#include <vector>
namespace dmf_engine {
enum class NodeState {
Stopped,
Starting,
Running,
Stopping,
};
struct GraphNode {
NodeId id;
std::string type;
nlohmann::json config;
NodeState state = NodeState::Stopped;
uint16_t control_port = 0;
int pid = 0;
};
struct GraphEdge {
EdgeId id;
NodeId from_node;
PortId from_port;
NodeId to_node;
PortId to_port;
FlowId flow_id;
nlohmann::json flow_def;
};
class Graph {
public:
NodeId add_node(const std::string& type, const nlohmann::json& config = {});
bool remove_node(const NodeId& node_id);
const GraphNode* get_node(const NodeId& node_id) const;
std::vector<GraphNode> get_nodes() const;
EdgeId add_edge(const NodeId& from_node, const PortId& from_port,
const NodeId& to_node, const PortId& to_port,
const FlowId& flow_id, const nlohmann::json& flow_def);
bool remove_edge(const EdgeId& edge_id);
std::vector<GraphEdge> get_edges() const;
std::vector<GraphEdge> get_edges_for_node(const NodeId& node_id) const;
nlohmann::json serialize() const;
private:
std::unordered_map<NodeId, GraphNode> nodes_;
std::unordered_map<EdgeId, GraphEdge> edges_;
int next_node_num_ = 0;
};
} // namespace dmf_engine
@@ -0,0 +1,22 @@
#pragma once
#include <dmf-engine/graph.hpp>
#include <string>
#include <unordered_map>
namespace dmf_engine {
class ProcessManager {
public:
bool start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port);
bool stop_node(GraphNode& node);
void stop_all();
bool is_running(const NodeId& node_id) const;
private:
std::string find_node_binary(const std::string& node_type) const;
};
} // namespace dmf_engine
@@ -0,0 +1,13 @@
#pragma once
#include <string>
#include <cstdint>
namespace dmf_engine {
using NodeId = std::string;
using EdgeId = std::string;
using FlowId = std::string;
using PortId = std::string;
} // namespace dmf_engine
+223
View File
@@ -0,0 +1,223 @@
#include <dmf-engine/api_server.hpp>
#include <dmf-engine/graph.hpp>
#include <dmf-engine/flow_manager.hpp>
#include <dmf-engine/process_manager.hpp>
#include <libwebsockets.h>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
#include <cstring>
#include <string>
namespace dmf_engine {
struct ApiServerImpl {
uint16_t port = 0;
Graph* graph = nullptr;
FlowManager* flow_manager = nullptr;
ProcessManager* process_manager = nullptr;
struct lws_context* context = nullptr;
};
struct ApiServer::Impl {
std::unique_ptr<ApiServerImpl> data;
};
struct HttpRequest {
std::string method;
std::string path;
std::string body;
bool body_done = false;
};
static ApiServerImpl* g_impl = nullptr;
static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
void* user, void* in, size_t len) {
auto* req = static_cast<HttpRequest*>(user);
switch (reason) {
case LWS_CALLBACK_HTTP: {
new (req) HttpRequest();
if (lws_hdr_total_length(wsi, WSI_TOKEN_POST_URI) > 0) {
req->method = "POST";
char buf[256] = {};
lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_POST_URI);
req->path = buf;
} else {
req->method = "GET";
char buf[256] = {};
lws_hdr_copy(wsi, buf, sizeof(buf), WSI_TOKEN_GET_URI);
req->path = buf;
}
break;
}
case LWS_CALLBACK_HTTP_BODY: {
req->body.append(static_cast<char*>(in), len);
break;
}
case LWS_CALLBACK_HTTP_BODY_COMPLETION: {
if (!g_impl || !g_impl->graph) {
lws_return_http_status(wsi, HTTP_STATUS_INTERNAL_SERVER_ERROR, "Server error");
return -1;
}
std::string status_str, content_type, response_body;
auto ok = [&](const nlohmann::json& j) {
status_str = "200 OK";
response_body = j.dump();
};
auto created = [&](const nlohmann::json& j) {
status_str = "201 Created";
response_body = j.dump();
};
auto error_resp = [&](int code, const std::string& msg) {
status_str = std::to_string(code) + " Error";
response_body = nlohmann::json({{"error", msg}}).dump();
};
content_type = "application/json";
try {
nlohmann::json req_body = req->body.empty() ? nlohmann::json::object() : nlohmann::json::parse(req->body);
auto& graph = *g_impl->graph;
auto& fm = *g_impl->flow_manager;
auto& pm = *g_impl->process_manager;
if (req->path == "/api/graph" && req->method == "GET") {
ok(graph.serialize());
} else if (req->path == "/api/graph/nodes" && req->method == "POST") {
if (!req_body.contains("type")) {
error_resp(400, "Missing 'type' field");
} else {
auto type = req_body["type"].get<std::string>();
auto config = req_body.value("config", nlohmann::json::object());
auto id = graph.add_node(type, config);
created({{"id", id}});
}
} else if (req->path.find("/api/graph/nodes/") == 0 && req->method == "DELETE") {
auto node_id = req->path.substr(std::string("/api/graph/nodes/").length());
if (graph.remove_node(node_id)) {
ok({{"deleted", node_id}});
} else {
error_resp(404, "Node not found: " + node_id);
}
} else if (req->path == "/api/graph/edges" && req->method == "POST") {
if (!req_body.contains("from_node") || !req_body.contains("to_node")) {
error_resp(400, "Missing from_node/to_node");
} else {
auto from_node = req_body["from_node"].get<std::string>();
auto from_port = req_body.value("from_port", "video_out");
auto to_node = req_body["to_node"].get<std::string>();
auto to_port = req_body.value("to_port", "video_in");
auto flow_id = fm.create_flow_id();
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);
created({{"id", edge_id}, {"flow_id", flow_id}});
}
} else if (req->path.find("/api/graph/edges/") == 0 && req->method == "DELETE") {
auto edge_id = req->path.substr(std::string("/api/graph/edges/").length());
if (graph.remove_edge(edge_id)) {
ok({{"deleted", edge_id}});
} else {
error_resp(404, "Edge not found: " + edge_id);
}
} else if (req->path == "/api/graph/start" && req->method == "POST") {
auto nodes = graph.get_nodes();
uint16_t port = 9100;
for (auto& node : nodes) {
pm.start_node(const_cast<GraphNode&>(node), "/dev/shm/mxl", port++);
}
ok({{"status", "started"}});
} else if (req->path == "/api/graph/stop" && req->method == "POST") {
auto nodes = graph.get_nodes();
for (auto& node : nodes) {
pm.stop_node(const_cast<GraphNode&>(node));
}
ok({{"status", "stopped"}});
} else {
error_resp(404, "Not found: " + req->method + " " + req->path);
}
} catch (const nlohmann::json::exception& e) {
error_resp(400, std::string("JSON error: ") + e.what());
} catch (const std::exception& e) {
error_resp(500, e.what());
}
auto headers = "HTTP/1.1 " + status_str + "\r\n"
"Content-Type: " + content_type + "\r\n"
"Access-Control-Allow-Origin: *\r\n"
"Access-Control-Allow-Methods: GET, POST, DELETE, PUT, OPTIONS\r\n"
"Access-Control-Allow-Headers: Content-Type\r\n"
"Content-Length: " + std::to_string(response_body.size()) + "\r\n"
"\r\n";
std::vector<uint8_t> buf(LWS_PRE + headers.size() + response_body.size());
std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size());
std::memcpy(buf.data() + LWS_PRE + headers.size(), response_body.data(), response_body.size());
lws_write(wsi, buf.data() + LWS_PRE, headers.size() + response_body.size(), LWS_WRITE_HTTP);
if (lws_http_transaction_completed(wsi)) {
return -1;
}
return 0;
}
default:
break;
}
return 0;
}
static const struct lws_protocols protocols[] = {
{"http-api", callback_http, sizeof(HttpRequest), 0},
{nullptr, nullptr, 0, 0},
};
ApiServer::ApiServer(uint16_t port, Graph& graph, FlowManager& flow_manager, ProcessManager& process_manager)
: 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;
g_impl = impl_->data.get();
struct lws_context_creation_info info;
std::memset(&info, 0, sizeof(info));
info.port = port;
info.protocols = protocols;
info.gid = -1;
info.uid = -1;
impl_->data->context = lws_create_context(&info);
if (!impl_->data->context) {
spdlog::error("Failed to create HTTP context on port {}", port);
throw std::runtime_error("Failed to create HTTP context");
}
spdlog::info("API server: listening on port {}", port);
}
ApiServer::~ApiServer() {
if (impl_->data && impl_->data->context) {
lws_context_destroy(impl_->data->context);
}
if (g_impl == impl_->data.get()) {
g_impl = nullptr;
}
}
void ApiServer::poll(int timeout_ms) {
lws_service(impl_->data->context, timeout_ms);
}
} // namespace dmf_engine
+48
View File
@@ -0,0 +1,48 @@
#include <dmf-engine/flow_manager.hpp>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
#include <random>
#include <sstream>
namespace dmf_engine {
FlowManager::FlowManager() {
std::random_device rd;
flow_counter_ = static_cast<int>(rd());
}
FlowId FlowManager::create_flow_id() {
std::stringstream ss;
ss << std::hex << (++flow_counter_);
auto id = ss.str();
while (id.length() < 8) {
id = "0" + id;
}
spdlog::info("FlowManager: created flow ID: {}", id);
return id;
}
nlohmann::json FlowManager::create_v210_flow_def(const FlowId& flow_id, int width, int height, int fps_numerator, int fps_denominator) const {
auto grain_size = (width * 2) * height;
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["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["tags"] = nlohmann::json::object();
flow_def["tags"]["grouphint"] = nlohmann::json::array({"DMF Studio:Video"});
return flow_def;
}
} // namespace dmf_engine
+135
View File
@@ -0,0 +1,135 @@
#include <dmf-engine/graph.hpp>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
namespace dmf_engine {
NodeId Graph::add_node(const std::string& type, const nlohmann::json& config) {
auto id = type + "_" + std::to_string(next_node_num_++);
GraphNode node;
node.id = id;
node.type = type;
node.config = config;
node.state = NodeState::Stopped;
nodes_[id] = std::move(node);
spdlog::info("Graph: added node '{}' type='{}'", id, type);
return id;
}
bool Graph::remove_node(const NodeId& node_id) {
auto it = nodes_.find(node_id);
if (it == nodes_.end()) {
spdlog::warn("Graph: node '{}' not found", node_id);
return false;
}
std::vector<EdgeId> edges_to_remove;
for (const auto& [eid, edge] : edges_) {
if (edge.from_node == node_id || edge.to_node == node_id) {
edges_to_remove.push_back(eid);
}
}
for (const auto& eid : edges_to_remove) {
edges_.erase(eid);
spdlog::info("Graph: removed edge '{}' (connected to removed node '{}')", eid, node_id);
}
nodes_.erase(it);
spdlog::info("Graph: removed node '{}'", node_id);
return true;
}
const GraphNode* Graph::get_node(const NodeId& node_id) const {
auto it = nodes_.find(node_id);
return it != nodes_.end() ? &it->second : nullptr;
}
std::vector<GraphNode> Graph::get_nodes() const {
std::vector<GraphNode> result;
for (const auto& [_, node] : nodes_) {
result.push_back(node);
}
return result;
}
EdgeId Graph::add_edge(const NodeId& from_node, const PortId& from_port,
const NodeId& to_node, const PortId& to_port,
const FlowId& flow_id, const nlohmann::json& flow_def) {
auto id = from_node + ":" + from_port + "->" + to_node + ":" + to_port;
GraphEdge edge;
edge.id = id;
edge.from_node = from_node;
edge.from_port = from_port;
edge.to_node = to_node;
edge.to_port = to_port;
edge.flow_id = flow_id;
edge.flow_def = flow_def;
edges_[id] = std::move(edge);
spdlog::info("Graph: added edge '{}' flow_id={}", id, flow_id);
return id;
}
bool Graph::remove_edge(const EdgeId& edge_id) {
auto it = edges_.find(edge_id);
if (it == edges_.end()) {
spdlog::warn("Graph: edge '{}' not found", edge_id);
return false;
}
edges_.erase(it);
spdlog::info("Graph: removed edge '{}'", edge_id);
return true;
}
std::vector<GraphEdge> Graph::get_edges() const {
std::vector<GraphEdge> result;
for (const auto& [_, edge] : edges_) {
result.push_back(edge);
}
return result;
}
std::vector<GraphEdge> Graph::get_edges_for_node(const NodeId& node_id) const {
std::vector<GraphEdge> result;
for (const auto& [_, edge] : edges_) {
if (edge.from_node == node_id || edge.to_node == node_id) {
result.push_back(edge);
}
}
return result;
}
nlohmann::json Graph::serialize() const {
auto j = nlohmann::json::object();
auto nodes_arr = nlohmann::json::array();
for (const auto& [id, node] : nodes_) {
nodes_arr.push_back({
{"id", node.id},
{"type", node.type},
{"config", node.config},
{"state", static_cast<int>(node.state)},
{"control_port", node.control_port},
{"pid", node.pid},
});
}
j["nodes"] = nodes_arr;
auto edges_arr = nlohmann::json::array();
for (const auto& [id, edge] : edges_) {
edges_arr.push_back({
{"id", edge.id},
{"from_node", edge.from_node},
{"from_port", edge.from_port},
{"to_node", edge.to_node},
{"to_port", edge.to_port},
{"flow_id", edge.flow_id},
});
}
j["edges"] = edges_arr;
return j;
}
} // namespace dmf_engine
+97
View File
@@ -0,0 +1,97 @@
#include <dmf-engine/process_manager.hpp>
#include <spdlog/spdlog.h>
#include <cstdlib>
#include <filesystem>
#include <signal.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
namespace dmf_engine {
std::string ProcessManager::find_node_binary(const std::string& node_type) const {
std::string binary_name = "dmf-node-" + node_type;
if (const char* env_path = std::getenv("DMF_NODE_PATH")) {
auto candidate = std::filesystem::path(env_path) / binary_name;
if (std::filesystem::exists(candidate)) {
return candidate.string();
}
}
if (const char* self_dir_env = std::getenv("DMF_STUDIO_BIN_DIR")) {
auto candidate = std::filesystem::path(self_dir_env) / binary_name;
if (std::filesystem::exists(candidate)) {
return candidate.string();
}
}
return binary_name;
}
bool ProcessManager::start_node(GraphNode& node, const std::string& mxl_domain, uint16_t base_port) {
if (node.state == NodeState::Running) {
spdlog::warn("ProcessManager: node '{}' is already running", node.id);
return true;
}
auto binary = find_node_binary(node.type);
node.control_port = base_port;
pid_t pid = fork();
if (pid < 0) {
spdlog::error("ProcessManager: fork() failed for node '{}': {}", node.id, strerror(errno));
return false;
}
if (pid == 0) {
std::string port_str = std::to_string(node.control_port);
std::string config_str = node.config.dump();
execlp(binary.c_str(), binary.c_str(),
"--node-id", node.id.c_str(),
"--control-port", port_str.c_str(),
"--mxl-domain", mxl_domain.c_str(),
"--config", config_str.c_str(),
nullptr);
spdlog::error("ProcessManager: execlp failed for '{}': {}", binary, strerror(errno));
_exit(1);
}
node.pid = pid;
node.state = NodeState::Running;
spdlog::info("ProcessManager: started node '{}' pid={} port={}", node.id, pid, node.control_port);
return true;
}
bool ProcessManager::stop_node(GraphNode& node) {
if (node.state != NodeState::Running || node.pid <= 0) {
spdlog::warn("ProcessManager: node '{}' is not running", node.id);
return true;
}
if (kill(node.pid, SIGTERM) != 0) {
spdlog::error("ProcessManager: failed to send SIGTERM to node '{}': {}", node.id, strerror(errno));
return false;
}
int status = 0;
waitpid(node.pid, &status, 0);
spdlog::info("ProcessManager: stopped node '{}' pid={}", node.id, node.pid);
node.pid = 0;
node.state = NodeState::Stopped;
return true;
}
void ProcessManager::stop_all() {
}
bool ProcessManager::is_running(const NodeId& node_id) const {
return false;
}
} // namespace dmf_engine
+17
View File
@@ -0,0 +1,17 @@
add_library(dmf-node STATIC
src/control_server.cpp
src/node_runner.cpp
)
target_include_directories(dmf-node PUBLIC
include
${CMAKE_SOURCE_DIR}/extern/mxl/lib/include
)
target_link_libraries(dmf-node PUBLIC
mxl
nlohmann_json::nlohmann_json
spdlog::spdlog
fmt::fmt
websockets
)
@@ -0,0 +1,35 @@
#pragma once
#include <dmf-node/node.hpp>
#include <functional>
#include <string>
#include <unordered_map>
#include <vector>
struct lws;
namespace dmf_node {
using CommandHandler = std::function<void(const nlohmann::json& payload)>;
using StatusCallback = std::function<void(const nlohmann::json& event)>;
class ControlServer {
public:
ControlServer(uint16_t port, StatusCallback on_event);
~ControlServer();
ControlServer(const ControlServer&) = delete;
ControlServer& operator=(const ControlServer&) = delete;
void register_command(const std::string& cmd, CommandHandler handler);
void send_event(const nlohmann::json& event);
void poll(int timeout_ms);
private:
struct Impl;
std::unique_ptr<Impl> impl_;
};
} // namespace dmf_node
+32
View File
@@ -0,0 +1,32 @@
#pragma once
#include <dmf-node/port.hpp>
#include <dmf-node/types.hpp>
#include <mxl/flow.h>
#include <mxl/mxl.h>
#include <nlohmann/json.hpp>
#include <memory>
#include <vector>
namespace dmf_node {
class Node {
public:
virtual ~Node() = default;
virtual std::string type() const = 0;
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_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;
};
} // namespace dmf_node
@@ -0,0 +1,39 @@
#pragma once
#include <dmf-node/node.hpp>
#include <dmf-node/control_server.hpp>
#include <mxl/mxl.h>
#include <atomic>
#include <memory>
#include <string>
namespace dmf_node {
class NodeRunner {
public:
template <typename NodeType>
static int run(int argc, char* argv[]) {
NodeRunner runner;
if (!runner.parse_args(argc, argv)) {
return 1;
}
auto node = std::make_unique<NodeType>();
return runner.exec(std::move(node));
}
private:
bool parse_args(int argc, char* argv[]);
int exec(std::unique_ptr<Node> node);
std::string node_id_;
uint16_t control_port_ = 0;
std::string mxl_domain_ = "/dev/shm/mxl";
std::string config_str_;
mxlInstance mxl_instance_ = nullptr;
std::atomic<bool> running_{false};
};
} // namespace dmf_node
+25
View File
@@ -0,0 +1,25 @@
#pragma once
#include <cstdint>
#include <string>
namespace dmf_node {
enum class MediaType : uint8_t {
VideoV210,
AudioFloat32,
AncData,
};
enum class PortDirection : uint8_t {
Input,
Output,
};
struct PortDef {
std::string id;
PortDirection direction;
MediaType media_type;
};
} // namespace dmf_node
+11
View File
@@ -0,0 +1,11 @@
#pragma once
#include <string>
namespace dmf_node {
using NodeId = std::string;
using PortId = std::string;
using FlowId = std::string;
} // namespace dmf_node
+169
View File
@@ -0,0 +1,169 @@
#include <dmf-node/control_server.hpp>
#include <dmf-node/node.hpp>
#include <libwebsockets.h>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
#include <cstring>
#include <mutex>
#include <string>
#include <unordered_map>
#include <vector>
namespace dmf_node {
struct ControlServerData {
StatusCallback on_event;
std::unordered_map<std::string, CommandHandler> commands;
std::mutex send_mutex;
std::vector<std::string> send_queue;
struct lws* client_wsi = nullptr;
};
struct ControlServer::Impl {
uint16_t port;
std::unique_ptr<ControlServerData> data;
struct lws_context* context = nullptr;
};
static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason,
void* user, void* in, size_t len);
static const struct lws_protocols protocols[] = {
{
"dmf-control",
callback_ws,
sizeof(ControlServerData*),
65536,
},
{nullptr, nullptr, 0, 0},
};
static int callback_ws(struct lws* wsi, enum lws_callback_reasons reason,
void* user, void* in, size_t len) {
auto** ppdata = static_cast<ControlServerData**>(user);
ControlServerData* data = ppdata ? *ppdata : nullptr;
switch (reason) {
case LWS_CALLBACK_ESTABLISHED: {
auto* vhost = lws_get_vhost(wsi);
data = static_cast<ControlServerData*>(lws_vhost_user(vhost));
if (ppdata) {
*ppdata = data;
}
if (data) {
data->client_wsi = wsi;
}
spdlog::info("Control WS: client connected");
break;
}
case LWS_CALLBACK_RECEIVE: {
if (!data) {
auto* vhost = lws_get_vhost(wsi);
data = static_cast<ControlServerData*>(lws_vhost_user(vhost));
}
if (!data) {
break;
}
try {
auto msg = nlohmann::json::parse(static_cast<char*>(in), static_cast<char*>(in) + len);
if (!msg.contains("cmd")) {
spdlog::warn("Control WS: message missing 'cmd' field");
break;
}
auto cmd = msg["cmd"].get<std::string>();
auto it = data->commands.find(cmd);
if (it != data->commands.end()) {
it->second(msg);
} else {
spdlog::warn("Control WS: unknown command '{}'", cmd);
}
} catch (const nlohmann::json::parse_error& e) {
spdlog::warn("Control WS: JSON parse error: {}", e.what());
}
break;
}
case LWS_CALLBACK_SERVER_WRITEABLE: {
if (!data) {
auto* vhost = lws_get_vhost(wsi);
data = static_cast<ControlServerData*>(lws_vhost_user(vhost));
}
if (!data) {
break;
}
std::lock_guard<std::mutex> lock(data->send_mutex);
while (!data->send_queue.empty()) {
auto& msg = data->send_queue.back();
std::vector<uint8_t> buf(LWS_PRE + msg.size());
std::memcpy(buf.data() + LWS_PRE, msg.data(), msg.size());
lws_write(wsi, buf.data() + LWS_PRE, msg.size(), LWS_WRITE_TEXT);
data->send_queue.pop_back();
}
break;
}
case LWS_CALLBACK_CLOSED: {
if (data) {
data->client_wsi = nullptr;
}
spdlog::info("Control WS: client disconnected");
break;
}
default:
break;
}
return 0;
}
ControlServer::ControlServer(uint16_t port, StatusCallback on_event)
: impl_(std::make_unique<Impl>()) {
impl_->port = port;
impl_->data = std::make_unique<ControlServerData>();
impl_->data->on_event = std::move(on_event);
struct lws_context_creation_info info;
std::memset(&info, 0, sizeof(info));
info.port = port;
info.protocols = protocols;
info.user = impl_->data.get();
info.gid = -1;
info.uid = -1;
impl_->context = lws_create_context(&info);
if (!impl_->context) {
spdlog::error("Failed to create WS context on port {}", port);
throw std::runtime_error("Failed to create WS context");
}
spdlog::info("Control WS: listening on port {}", port);
}
ControlServer::~ControlServer() {
if (impl_->context) {
lws_context_destroy(impl_->context);
}
}
void ControlServer::register_command(const std::string& cmd, CommandHandler handler) {
impl_->data->commands[cmd] = std::move(handler);
}
void ControlServer::send_event(const nlohmann::json& event) {
auto data = event.dump();
{
std::lock_guard<std::mutex> lock(impl_->data->send_mutex);
impl_->data->send_queue.push_back(data);
}
if (impl_->data->client_wsi) {
lws_callback_on_writable(impl_->data->client_wsi);
}
}
void ControlServer::poll(int timeout_ms) {
lws_service(impl_->context, timeout_ms);
}
} // namespace dmf_node
+184
View File
@@ -0,0 +1,184 @@
#include <dmf-node/node_runner.hpp>
#include <dmf-node/control_server.hpp>
#include <mxl/flow.h>
#include <mxl/mxl.h>
#include <nlohmann/json.hpp>
#include <spdlog/spdlog.h>
#include <cstring>
#include <csignal>
#include <filesystem>
#include <thread>
namespace dmf_node {
static std::atomic<bool> g_running{true};
static void signal_handler(int /*signum*/) {
g_running = false;
}
bool NodeRunner::parse_args(int argc, char* argv[]) {
for (int i = 1; i < argc; ++i) {
std::string arg = argv[i];
if ((arg == "--node-id" || arg == "-n") && i + 1 < argc) {
node_id_ = argv[++i];
} else if ((arg == "--control-port" || arg == "-p") && i + 1 < argc) {
control_port_ = static_cast<uint16_t>(std::stoi(argv[++i]));
} else if ((arg == "--mxl-domain" || arg == "-d") && i + 1 < argc) {
mxl_domain_ = argv[++i];
} else if ((arg == "--config" || arg == "-c") && i + 1 < argc) {
config_str_ = argv[++i];
} else if (arg == "--help" || arg == "-h") {
spdlog::info("Usage: {} [options]", argv[0]);
spdlog::info(" --node-id, -n Node instance ID");
spdlog::info(" --control-port, -p WebSocket control port");
spdlog::info(" --mxl-domain, -d MXL domain path (default: /dev/shm/mxl)");
spdlog::info(" --config, -c Node configuration JSON");
return false;
}
}
if (node_id_.empty()) {
spdlog::error("--node-id is required");
return false;
}
if (control_port_ == 0) {
spdlog::error("--control-port is required");
return false;
}
return true;
}
int NodeRunner::exec(std::unique_ptr<Node> node) {
std::signal(SIGINT, signal_handler);
std::signal(SIGTERM, signal_handler);
spdlog::info("Starting node '{}' type='{}'", node_id_, node->type());
if (!std::filesystem::exists(mxl_domain_)) {
spdlog::error("MXL domain path does not exist: {}", mxl_domain_);
return 1;
}
mxl_instance_ = mxlCreateInstance(mxl_domain_.c_str(), nullptr);
if (!mxl_instance_) {
spdlog::error("Failed to create MXL instance on domain: {}", mxl_domain_);
return 1;
}
spdlog::info("MXL instance created on domain: {}", mxl_domain_);
mxlGarbageCollectFlows(mxl_instance_);
if (!config_str_.empty()) {
try {
auto cfg = nlohmann::json::parse(config_str_);
node->configure(cfg);
} catch (const nlohmann::json::parse_error& e) {
spdlog::error("Failed to parse config JSON: {}", e.what());
mxlDestroyInstance(mxl_instance_);
return 1;
}
}
auto control_server = std::make_unique<ControlServer>(control_port_, [](const nlohmann::json& /*event*/) {});
auto mk_ports = [](const std::vector<PortDef>& ports) {
auto arr = nlohmann::json::array();
for (const auto& p : ports) {
arr.push_back({{"id", p.id}, {"direction", p.direction == PortDirection::Input ? "input" : "output"}, {"media_type", static_cast<int>(p.media_type)}});
}
return arr;
};
control_server->register_command("add_writer", [&, w = std::unordered_map<std::string, mxlFlowWriter>{}](const nlohmann::json& msg) mutable {
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();
mxlFlowWriter writer = nullptr;
bool created = false;
mxlFlowConfigInfo config_info{};
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;
}
spdlog::info("Created flow writer on port '{}' flow {} (created={})", port_id, flow_id, created);
node->on_add_writer(port_id, writer);
});
control_server->register_command("add_reader", [&](const nlohmann::json& msg) {
auto flow_id = msg["flow_id"].get<std::string>();
auto port_id = msg["port_id"].get<std::string>();
mxlFlowReader reader = nullptr;
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;
}
spdlog::info("Created flow reader on port '{}' flow {}", port_id, flow_id);
node->on_add_reader(port_id, reader);
});
control_server->register_command("remove_writer", [&](const nlohmann::json& msg) {
auto port_id = msg["port_id"].get<std::string>();
node->on_remove_writer(port_id);
spdlog::info("Removed writer on port '{}'", port_id);
});
control_server->register_command("remove_reader", [&](const nlohmann::json& msg) {
auto port_id = msg["port_id"].get<std::string>();
node->on_remove_reader(port_id);
spdlog::info("Removed reader on port '{}'", port_id);
});
control_server->register_command("configure", [&](const nlohmann::json& msg) {
if (msg.contains("params")) {
node->configure(msg["params"]);
spdlog::info("Reconfigured node '{}'", node_id_);
}
});
control_server->register_command("shutdown", [&](const nlohmann::json& /*msg*/) {
spdlog::info("Shutdown command received");
g_running = false;
});
nlohmann::json ready_event;
ready_event["event"] = "ready";
ready_event["type"] = node->type();
ready_event["node_id"] = node_id_;
ready_event["ports"] = mk_ports(node->input_ports());
for (const auto& p : node->output_ports()) {
ready_event["ports"].push_back({{"id", p.id}, {"direction", "output"}, {"media_type", static_cast<int>(p.media_type)}});
}
control_server->send_event(ready_event);
running_ = true;
spdlog::info("Node '{}' entering process loop", node_id_);
while (g_running && running_) {
control_server->poll(1);
node->process();
static int status_counter = 0;
if (++status_counter % 100 == 0) {
nlohmann::json status_event;
status_event["event"] = "status";
status_event["node_id"] = node_id_;
status_event["data"] = node->status();
control_server->send_event(status_event);
}
}
spdlog::info("Node '{}' shutting down", node_id_);
mxlDestroyInstance(mxl_instance_);
return 0;
}
} // namespace dmf_node
+6
View File
@@ -0,0 +1,6 @@
add_executable(dmf-node-passthrough
src/main.cpp
src/passthrough_node.cpp
)
target_link_libraries(dmf-node-passthrough PRIVATE dmf-node)
+6
View File
@@ -0,0 +1,6 @@
#include <dmf-node/node_runner.hpp>
#include "passthrough_node.hpp"
int main(int argc, char* argv[]) {
return dmf_node::NodeRunner::run<dmf_node::PassthroughNode>(argc, argv);
}
@@ -0,0 +1,94 @@
#include "passthrough_node.hpp"
#include <mxl/flow.h>
#include <mxl/mxl.h>
#include <spdlog/spdlog.h>
#include <cstring>
namespace dmf_node {
void PassthroughNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
if (port_id == "video_out") {
writer_ = writer;
spdlog::info("Passthrough: writer added on video_out");
}
}
void PassthroughNode::on_add_reader(const std::string& port_id, mxlFlowReader reader) {
if (port_id == "video_in") {
reader_ = reader;
spdlog::info("Passthrough: reader added on video_in");
}
}
void PassthroughNode::on_remove_writer(const std::string& port_id) {
if (port_id == "video_out") {
writer_.reset();
spdlog::info("Passthrough: writer removed from video_out");
}
}
void PassthroughNode::on_remove_reader(const std::string& port_id) {
if (port_id == "video_in") {
reader_.reset();
spdlog::info("Passthrough: reader removed from video_in");
}
}
void PassthroughNode::process() {
if (!reader_ || !writer_) {
return;
}
mxlGrainInfo grain_info{};
uint8_t* payload = nullptr;
auto status = mxlFlowReaderGetGrain(*reader_, read_index_, 100000000ULL, &grain_info, &payload);
if (status != MXL_STATUS_OK) {
if (status == MXL_ERR_TIMEOUT) {
return;
}
if (status == MXL_ERR_OUT_OF_RANGE_TOO_LATE) {
mxlFlowRuntimeInfo runtime{};
mxlFlowReaderGetRuntimeInfo(*reader_, &runtime);
read_index_ = runtime.headIndex;
return;
}
return;
}
mxlGrainInfo out_grain{};
uint8_t* out_payload = nullptr;
status = mxlFlowWriterOpenGrain(*writer_, write_index_, &out_grain, &out_payload);
if (status != MXL_STATUS_OK) {
spdlog::warn("Passthrough: failed to open output grain at index {}: {}", write_index_, static_cast<int>(status));
read_index_++;
return;
}
auto copy_size = std::min(grain_info.grainSize, out_grain.grainSize);
std::memcpy(out_payload, payload, copy_size);
out_grain.validSlices = grain_info.validSlices;
out_grain.flags = grain_info.flags;
mxlFlowWriterCommitGrain(*writer_, &out_grain);
read_index_++;
write_index_++;
grains_processed_++;
}
nlohmann::json PassthroughNode::status() const {
return {
{"type", "passthrough"},
{"grains_processed", grains_processed_},
{"read_index", read_index_},
{"write_index", write_index_},
{"has_reader", reader_.has_value()},
{"has_writer", writer_.has_value()},
};
}
} // namespace dmf_node
@@ -0,0 +1,45 @@
#pragma once
#include <dmf-node/node.hpp>
#include <mxl/flow.h>
#include <unordered_map>
#include <optional>
namespace dmf_node {
class PassthroughNode : public Node {
public:
PassthroughNode() = default;
std::string type() const override { return "passthrough"; }
std::vector<PortDef> input_ports() const override {
return {{"video_in", PortDirection::Input, MediaType::VideoV210}};
}
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:
std::optional<mxlFlowReader> reader_;
std::optional<mxlFlowWriter> writer_;
mxlInstance mxl_instance_ = nullptr;
uint64_t read_index_ = 0;
uint64_t write_index_ = 0;
uint64_t grains_processed_ = 0;
};
} // namespace dmf_node
+8
View File
@@ -0,0 +1,8 @@
add_executable(dmf-test-graph
test_graph.cpp
)
target_link_libraries(dmf-test-graph PRIVATE
dmf-engine
Catch2::Catch2WithMain
)
+88
View File
@@ -0,0 +1,88 @@
#include <dmf-engine/graph.hpp>
#include <dmf-engine/flow_manager.hpp>
#include <catch2/catch_test_macros.hpp>
TEST_CASE("Graph add and remove nodes", "[graph]") {
dmf_engine::Graph graph;
auto id1 = graph.add_node("passthrough");
auto id2 = graph.add_node("test-source");
REQUIRE(graph.get_node(id1) != nullptr);
REQUIRE(graph.get_node(id2) != nullptr);
REQUIRE(graph.get_node(id1)->type == "passthrough");
REQUIRE(graph.get_node(id2)->type == "test-source");
auto nodes = graph.get_nodes();
REQUIRE(nodes.size() == 2);
graph.remove_node(id1);
REQUIRE(graph.get_node(id1) == nullptr);
REQUIRE(graph.get_nodes().size() == 1);
}
TEST_CASE("Graph add and remove edges", "[graph]") {
dmf_engine::Graph graph;
auto n1 = graph.add_node("test-source");
auto n2 = graph.add_node("passthrough");
auto edge_id = graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {});
REQUIRE(graph.get_edges().size() == 1);
graph.remove_edge(edge_id);
REQUIRE(graph.get_edges().size() == 0);
}
TEST_CASE("Graph remove node also removes edges", "[graph]") {
dmf_engine::Graph graph;
auto n1 = graph.add_node("test-source");
auto n2 = graph.add_node("passthrough");
graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {});
graph.add_edge(n2, "video_out", n1, "video_in", "flow2", {});
REQUIRE(graph.get_edges().size() == 2);
graph.remove_node(n1);
REQUIRE(graph.get_edges().size() == 0);
}
TEST_CASE("FlowManager creates unique IDs", "[flow_manager]") {
dmf_engine::FlowManager fm1;
dmf_engine::FlowManager fm2;
auto id1 = fm1.create_flow_id();
auto id2 = fm1.create_flow_id();
auto id3 = fm2.create_flow_id();
REQUIRE(id1 != id2);
REQUIRE(id2 != id3);
}
TEST_CASE("FlowManager creates V210 flow definition", "[flow_manager]") {
dmf_engine::FlowManager fm;
auto flow_id = fm.create_flow_id();
auto def = fm.create_v210_flow_def(flow_id, 1920, 1080, 50, 1);
REQUIRE(def["id"] == flow_id);
REQUIRE(def["format"] == "video/v210");
REQUIRE(def["width"] == 1920);
REQUIRE(def["height"] == 1080);
}
TEST_CASE("Graph serialize", "[graph]") {
dmf_engine::Graph graph;
auto n1 = graph.add_node("test-source");
auto n2 = graph.add_node("passthrough");
graph.add_edge(n1, "video_out", n2, "video_in", "flow1", {});
auto j = graph.serialize();
REQUIRE(j.contains("nodes"));
REQUIRE(j.contains("edges"));
REQUIRE(j["nodes"].size() == 2);
REQUIRE(j["edges"].size() == 1);
}
+20
View File
@@ -0,0 +1,20 @@
{
"name": "dmf-studio",
"version": "0.1.0",
"dependencies": [
"libwebsockets",
"nlohmann-json",
"spdlog",
"fmt",
"catch2",
{
"name": "stduuid",
"version>=": "1.2.3",
"features": ["system-gen", "gsl-span"]
},
"picojson",
"cli11",
"ada-url"
],
"builtin-baseline": "432412eecac55981cf608cc12e5b4bd91768ec57"
}