JohannesItten
6c6225d831
load_graph/stop_node/start_node each called notify() (push to all clients) and then sent a direct response — the requesting client got two identical messages, breaking test recv sequencing. Use thread_local tl_requester to skip the requesting connection in the push. notify() is called synchronously from the command, so the thread_local is visible from the status_cb. Other clients still receive the push; the requester gets only the direct response. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
114 lines
4.1 KiB
C++
114 lines
4.1 KiB
C++
#include <filesystem>
|
|
#include <mutex>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <unordered_set>
|
|
#include <crow.h>
|
|
#include "StudioManager.hpp"
|
|
|
|
namespace fs = std::filesystem;
|
|
|
|
int main(int argc, char* argv[]) {
|
|
dmf::install_signal_handlers();
|
|
|
|
const fs::path bin_dir = fs::path(argv[0]).parent_path();
|
|
|
|
#ifdef __APPLE__
|
|
const std::string domain = "/tmp/dmf-studio";
|
|
#else
|
|
const std::string domain = "/dev/shm/dmf-studio";
|
|
#endif
|
|
|
|
std::error_code ec;
|
|
fs::create_directories(domain, ec);
|
|
if (ec) {
|
|
fprintf(stderr, "[studio-manager] cannot create domain dir: %s\n", ec.message().c_str());
|
|
return 1;
|
|
}
|
|
|
|
dmf::StudioManager manager(bin_dir, domain);
|
|
|
|
if (argc > 1) {
|
|
auto r = manager.load_graph_file(argv[1]);
|
|
if (r.value("type", "") == "error") {
|
|
fprintf(stderr, "[studio-manager] %s\n",
|
|
r.value("message", "load failed").c_str());
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
// ── WebSocket server ──────────────────────────────────────────────────────
|
|
|
|
crow::SimpleApp app;
|
|
app.signal_clear(); // let our signal handler manage SIGTERM/SIGINT
|
|
|
|
std::mutex ws_mutex;
|
|
std::unordered_set<crow::websocket::connection*> clients;
|
|
|
|
// Push status to every connected client except the one whose command triggered
|
|
// the change — that client gets the direct response from the handler instead.
|
|
// tl_requester is thread-local: notify() is called synchronously from within
|
|
// the manager command, so it runs on the same thread as onmessage.
|
|
thread_local crow::websocket::connection* tl_requester = nullptr;
|
|
|
|
manager.on_status_change([&](nlohmann::json status) {
|
|
std::lock_guard lk(ws_mutex);
|
|
const std::string msg = status.dump();
|
|
for (auto* c : clients)
|
|
if (c != tl_requester) c->send_text(msg);
|
|
});
|
|
|
|
CROW_WEBSOCKET_ROUTE(app, "/ws")
|
|
.onopen([&](crow::websocket::connection& conn) {
|
|
{
|
|
std::lock_guard lk(ws_mutex);
|
|
clients.insert(&conn);
|
|
}
|
|
// get_status() acquires manager mutex — must NOT hold ws_mutex here
|
|
conn.send_text(manager.get_status().dump());
|
|
fprintf(stderr, "[studio-manager] client connected\n");
|
|
})
|
|
.onclose([&](crow::websocket::connection& conn, const std::string&) {
|
|
std::lock_guard lk(ws_mutex);
|
|
clients.erase(&conn);
|
|
fprintf(stderr, "[studio-manager] client disconnected\n");
|
|
})
|
|
.onmessage([&](crow::websocket::connection& conn, const std::string& data, bool) {
|
|
auto j = nlohmann::json::parse(data, nullptr, false);
|
|
if (j.is_discarded()) {
|
|
conn.send_text(R"({"type":"error","message":"invalid JSON"})");
|
|
return;
|
|
}
|
|
|
|
nlohmann::json response;
|
|
const std::string type = j.value("type", "");
|
|
tl_requester = &conn;
|
|
try {
|
|
if (type == "load_graph") response = manager.load_graph(j.at("graph"));
|
|
else if (type == "stop_node") response = manager.stop_node(j.at("id"));
|
|
else if (type == "start_node") response = manager.start_node(j.at("id"));
|
|
else if (type == "get_status") response = manager.get_status();
|
|
else response = {{"type","error"},{"message","unknown command: " + type}};
|
|
} catch (const std::exception& e) {
|
|
response = {{"type","error"},{"message", e.what()}};
|
|
}
|
|
tl_requester = nullptr;
|
|
conn.send_text(response.dump());
|
|
});
|
|
|
|
// ── Monitor thread — detects crashes, stops Crow on shutdown ─────────────
|
|
|
|
std::thread monitor([&] {
|
|
manager.run_monitor();
|
|
app.stop();
|
|
});
|
|
|
|
fprintf(stderr, "[studio-manager] WebSocket API at ws://0.0.0.0:7070/ws\n");
|
|
app.port(7070).multithreaded().run();
|
|
|
|
manager.shutdown();
|
|
monitor.join();
|
|
fprintf(stderr, "[studio-manager] done\n");
|
|
return 0;
|
|
}
|