fix: engine REST API - handle GET/DELETE requests immediately, use user-provided node IDs
- Use lws_http_get_uri_and_method() for proper HTTP method detection (GET, POST, PUT, DELETE all supported) - Handle GET/DELETE in LWS_CALLBACK_HTTP without waiting for body - Support user-provided node IDs via config.id field - Fixes all REST API endpoints hanging on GET requests
This commit is contained in:
@@ -30,43 +30,40 @@ struct HttpRequest {
|
|||||||
std::string method;
|
std::string method;
|
||||||
std::string path;
|
std::string path;
|
||||||
std::string body;
|
std::string body;
|
||||||
bool body_done = false;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
static ApiServerImpl* g_impl = nullptr;
|
static ApiServerImpl* g_impl = nullptr;
|
||||||
|
|
||||||
static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
static int send_json_response(struct lws* wsi, const std::string& status_str,
|
||||||
void* user, void* in, size_t len) {
|
const std::string& json_body) {
|
||||||
auto* req = static_cast<HttpRequest*>(user);
|
auto headers = "HTTP/1.1 " + status_str + "\r\n"
|
||||||
|
"Content-Type: application/json\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(json_body.size()) + "\r\n"
|
||||||
|
"\r\n";
|
||||||
|
|
||||||
switch (reason) {
|
std::vector<uint8_t> buf(LWS_PRE + headers.size() + json_body.size());
|
||||||
case LWS_CALLBACK_HTTP: {
|
std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size());
|
||||||
new (req) HttpRequest();
|
std::memcpy(buf.data() + LWS_PRE + headers.size(), json_body.data(), json_body.size());
|
||||||
|
|
||||||
if (lws_hdr_total_length(wsi, WSI_TOKEN_POST_URI) > 0) {
|
lws_write(wsi, buf.data() + LWS_PRE, headers.size() + json_body.size(), LWS_WRITE_HTTP);
|
||||||
req->method = "POST";
|
|
||||||
char buf[256] = {};
|
if (lws_http_transaction_completed(wsi)) {
|
||||||
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;
|
return -1;
|
||||||
}
|
}
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
std::string status_str, content_type, response_body;
|
static void handle_request(const std::string& method, const std::string& path,
|
||||||
|
const std::string& body,
|
||||||
|
std::string& status_str, std::string& response_body) {
|
||||||
|
if (!g_impl || !g_impl->graph) {
|
||||||
|
status_str = "500 Internal Server Error";
|
||||||
|
response_body = nlohmann::json({{"error", "Server not initialized"}}).dump();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
auto ok = [&](const nlohmann::json& j) {
|
auto ok = [&](const nlohmann::json& j) {
|
||||||
status_str = "200 OK";
|
status_str = "200 OK";
|
||||||
@@ -81,18 +78,16 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
|||||||
response_body = nlohmann::json({{"error", msg}}).dump();
|
response_body = nlohmann::json({{"error", msg}}).dump();
|
||||||
};
|
};
|
||||||
|
|
||||||
content_type = "application/json";
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
nlohmann::json req_body = req->body.empty() ? nlohmann::json::object() : nlohmann::json::parse(req->body);
|
nlohmann::json req_body = body.empty() ? nlohmann::json::object() : nlohmann::json::parse(body);
|
||||||
|
|
||||||
auto& graph = *g_impl->graph;
|
auto& graph = *g_impl->graph;
|
||||||
auto& fm = *g_impl->flow_manager;
|
auto& fm = *g_impl->flow_manager;
|
||||||
auto& pm = *g_impl->process_manager;
|
auto& pm = *g_impl->process_manager;
|
||||||
|
|
||||||
if (req->path == "/api/graph" && req->method == "GET") {
|
if (path == "/api/graph" && method == "GET") {
|
||||||
ok(graph.serialize());
|
ok(graph.serialize());
|
||||||
} else if (req->path == "/api/graph/nodes" && req->method == "POST") {
|
} else if (path == "/api/graph/nodes" && method == "POST") {
|
||||||
if (!req_body.contains("type")) {
|
if (!req_body.contains("type")) {
|
||||||
error_resp(400, "Missing 'type' field");
|
error_resp(400, "Missing 'type' field");
|
||||||
} else {
|
} else {
|
||||||
@@ -101,14 +96,14 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
|||||||
auto id = graph.add_node(type, config);
|
auto id = graph.add_node(type, config);
|
||||||
created({{"id", id}});
|
created({{"id", id}});
|
||||||
}
|
}
|
||||||
} else if (req->path.find("/api/graph/nodes/") == 0 && req->method == "DELETE") {
|
} else if (path.find("/api/graph/nodes/") == 0 && method == "DELETE") {
|
||||||
auto node_id = req->path.substr(std::string("/api/graph/nodes/").length());
|
auto node_id = path.substr(std::string("/api/graph/nodes/").length());
|
||||||
if (graph.remove_node(node_id)) {
|
if (graph.remove_node(node_id)) {
|
||||||
ok({{"deleted", node_id}});
|
ok({{"deleted", node_id}});
|
||||||
} else {
|
} else {
|
||||||
error_resp(404, "Node not found: " + node_id);
|
error_resp(404, "Node not found: " + node_id);
|
||||||
}
|
}
|
||||||
} else if (req->path == "/api/graph/edges" && req->method == "POST") {
|
} else if (path == "/api/graph/edges" && method == "POST") {
|
||||||
if (!req_body.contains("from_node") || !req_body.contains("to_node")) {
|
if (!req_body.contains("from_node") || !req_body.contains("to_node")) {
|
||||||
error_resp(400, "Missing from_node/to_node");
|
error_resp(400, "Missing from_node/to_node");
|
||||||
} else {
|
} else {
|
||||||
@@ -123,53 +118,83 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
|||||||
auto edge_id = graph.add_edge(from_node, from_port, to_node, to_port, flow_id, flow_def);
|
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}});
|
created({{"id", edge_id}, {"flow_id", flow_id}});
|
||||||
}
|
}
|
||||||
} else if (req->path.find("/api/graph/edges/") == 0 && req->method == "DELETE") {
|
} else if (path.find("/api/graph/edges/") == 0 && method == "DELETE") {
|
||||||
auto edge_id = req->path.substr(std::string("/api/graph/edges/").length());
|
auto edge_id = path.substr(std::string("/api/graph/edges/").length());
|
||||||
if (graph.remove_edge(edge_id)) {
|
if (graph.remove_edge(edge_id)) {
|
||||||
ok({{"deleted", edge_id}});
|
ok({{"deleted", edge_id}});
|
||||||
} else {
|
} else {
|
||||||
error_resp(404, "Edge not found: " + edge_id);
|
error_resp(404, "Edge not found: " + edge_id);
|
||||||
}
|
}
|
||||||
} else if (req->path == "/api/graph/start" && req->method == "POST") {
|
} else if (path == "/api/graph/start" && method == "POST") {
|
||||||
auto nodes = graph.get_nodes();
|
auto nodes = graph.get_nodes();
|
||||||
uint16_t port = 9100;
|
uint16_t port = 9100;
|
||||||
for (auto& node : nodes) {
|
for (auto& node : nodes) {
|
||||||
pm.start_node(const_cast<GraphNode&>(node), "/dev/shm/mxl", port++);
|
pm.start_node(const_cast<GraphNode&>(node), "/dev/shm/mxl", port++);
|
||||||
}
|
}
|
||||||
ok({{"status", "started"}});
|
ok({{"status", "started"}});
|
||||||
} else if (req->path == "/api/graph/stop" && req->method == "POST") {
|
} else if (path == "/api/graph/stop" && method == "POST") {
|
||||||
auto nodes = graph.get_nodes();
|
auto nodes = graph.get_nodes();
|
||||||
for (auto& node : nodes) {
|
for (auto& node : nodes) {
|
||||||
pm.stop_node(const_cast<GraphNode&>(node));
|
pm.stop_node(const_cast<GraphNode&>(node));
|
||||||
}
|
}
|
||||||
ok({{"status", "stopped"}});
|
ok({{"status", "stopped"}});
|
||||||
} else {
|
} else {
|
||||||
error_resp(404, "Not found: " + req->method + " " + req->path);
|
error_resp(404, "Not found: " + method + " " + path);
|
||||||
}
|
}
|
||||||
} catch (const nlohmann::json::exception& e) {
|
} catch (const nlohmann::json::exception& e) {
|
||||||
error_resp(400, std::string("JSON error: ") + e.what());
|
error_resp(400, std::string("JSON error: ") + e.what());
|
||||||
} catch (const std::exception& e) {
|
} catch (const std::exception& e) {
|
||||||
error_resp(500, e.what());
|
error_resp(500, e.what());
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
auto headers = "HTTP/1.1 " + status_str + "\r\n"
|
static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
||||||
"Content-Type: " + content_type + "\r\n"
|
void* user, void* in, size_t len) {
|
||||||
"Access-Control-Allow-Origin: *\r\n"
|
auto* req = static_cast<HttpRequest*>(user);
|
||||||
"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());
|
switch (reason) {
|
||||||
std::memcpy(buf.data() + LWS_PRE, headers.data(), headers.size());
|
case LWS_CALLBACK_HTTP: {
|
||||||
std::memcpy(buf.data() + LWS_PRE + headers.size(), response_body.data(), response_body.size());
|
new (req) HttpRequest();
|
||||||
|
|
||||||
lws_write(wsi, buf.data() + LWS_PRE, headers.size() + response_body.size(), LWS_WRITE_HTTP);
|
char* uri_ptr = nullptr;
|
||||||
|
int uri_len = 0;
|
||||||
|
int method = lws_http_get_uri_and_method(wsi, &uri_ptr, &uri_len);
|
||||||
|
|
||||||
if (lws_http_transaction_completed(wsi)) {
|
switch (method) {
|
||||||
return -1;
|
case LWSHUMETH_GET: req->method = "GET"; break;
|
||||||
|
case LWSHUMETH_POST: req->method = "POST"; break;
|
||||||
|
case LWSHUMETH_PUT: req->method = "PUT"; break;
|
||||||
|
case LWSHUMETH_DELETE: req->method = "DELETE"; break;
|
||||||
|
default: req->method = "GET"; break;
|
||||||
}
|
}
|
||||||
return 0;
|
|
||||||
|
if (uri_ptr && uri_len > 0) {
|
||||||
|
req->path.assign(uri_ptr, uri_len);
|
||||||
|
}
|
||||||
|
|
||||||
|
if (req->method == "GET" || req->method == "DELETE") {
|
||||||
|
std::string status_str, response_body;
|
||||||
|
handle_request(req->method, req->path, "", status_str, response_body);
|
||||||
|
return send_json_response(wsi, status_str, response_body);
|
||||||
|
}
|
||||||
|
|
||||||
|
int body_len = lws_hdr_total_length(wsi, WSI_TOKEN_HTTP_CONTENT_LENGTH);
|
||||||
|
if (body_len == 0) {
|
||||||
|
std::string status_str, response_body;
|
||||||
|
handle_request(req->method, req->path, "", status_str, response_body);
|
||||||
|
return send_json_response(wsi, status_str, response_body);
|
||||||
|
}
|
||||||
|
req->body.reserve(body_len);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
case LWS_CALLBACK_HTTP_BODY: {
|
||||||
|
req->body.append(static_cast<char*>(in), len);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
case LWS_CALLBACK_HTTP_BODY_COMPLETION: {
|
||||||
|
std::string status_str, response_body;
|
||||||
|
handle_request(req->method, req->path, req->body, status_str, response_body);
|
||||||
|
return send_json_response(wsi, status_str, response_body);
|
||||||
}
|
}
|
||||||
default:
|
default:
|
||||||
break;
|
break;
|
||||||
@@ -178,7 +203,7 @@ static int callback_http(struct lws* wsi, enum lws_callback_reasons reason,
|
|||||||
}
|
}
|
||||||
|
|
||||||
static const struct lws_protocols protocols[] = {
|
static const struct lws_protocols protocols[] = {
|
||||||
{"http-api", callback_http, sizeof(HttpRequest), 0},
|
{"http-api", callback_http, sizeof(HttpRequest), 4096},
|
||||||
{nullptr, nullptr, 0, 0},
|
{nullptr, nullptr, 0, 0},
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -7,7 +7,12 @@
|
|||||||
namespace dmf_engine {
|
namespace dmf_engine {
|
||||||
|
|
||||||
NodeId Graph::add_node(const std::string& type, const nlohmann::json& config) {
|
NodeId Graph::add_node(const std::string& type, const nlohmann::json& config) {
|
||||||
auto id = type + "_" + std::to_string(next_node_num_++);
|
std::string id;
|
||||||
|
if (config.contains("id") && config["id"].is_string()) {
|
||||||
|
id = config["id"].get<std::string>();
|
||||||
|
} else {
|
||||||
|
id = type + "_" + std::to_string(next_node_num_++);
|
||||||
|
}
|
||||||
GraphNode node;
|
GraphNode node;
|
||||||
node.id = id;
|
node.id = id;
|
||||||
node.type = type;
|
node.type = type;
|
||||||
|
|||||||
Reference in New Issue
Block a user