web_server 初步改

web_server/src/Graph_WebSocket.cpp

@@ -3,6 +3,8 @@
 #include <algorithm>
 #include <chrono>
 #include <string>
+#include <unordered_map>
+#include <mutex>
 namespace aethera::web {
 namespace {
 std::string graph_id_from_path(std::string_view path) { const auto split = path.find_last_of('/'); return split == std::string_view::npos ? std::string{} : std::string(path.substr(split + 1)); }
@@ -12,12 +14,33 @@ struct Graph_WebSocket::Private {
     std::shared_ptr<Graph_Session> session;                   /* 连接期间保持图会话存活。 */
     const void* owner{};                                     /* 会话订阅表使用的稳定连接身份。 */
     bool attached{};                                         /* 是否已安装像素订阅。 */
+    std::mutex callback_mutex;
+    std::unordered_map<std::string, Callback> callbacks;
 };
-Graph_WebSocket::Graph_WebSocket(drogon::WebSocketConnectionPtr connection, std::shared_ptr<Graph_Session> session) : d(std::make_unique<Private>(Private{connection, std::move(session), connection.get(), false})) {}
+Graph_WebSocket::Graph_WebSocket(drogon::WebSocketConnectionPtr connection, std::shared_ptr<Graph_Session> session) : d(std::make_unique<Private>()) { d->connection = connection; d->session = std::move(session); d->owner = connection.get(); }
 Graph_WebSocket::~Graph_WebSocket() { close(); }
 void Graph_WebSocket::start() { if (d->attached) return; const auto weak = weak_from_this(); d->session->attach(d->owner, [weak](std::string pixels) { const auto socket = weak.lock(); if (!socket) return; const auto connection = socket->d->connection.lock(); if (connection && connection->connected()) connection->send(pixels.data(), pixels.size(), drogon::WebSocketMessageType::Binary); }); d->attached = true; }
 void Graph_WebSocket::receive(std::string_view message) { const auto json = nlohmann::json::parse(message, nullptr, false); if (json.is_discarded() || !json.is_object()) return; Graph_Event event; if (const auto value = json.find("time"); value != json.end() && value->is_number()) event.time_milliseconds = value->get<double>(); if (const auto value = json.find("width"); value != json.end() && value->is_number_unsigned()) event.width = std::clamp(value->get<std::uint32_t>(), 160U, 1920U); if (const auto value = json.find("height"); value != json.end() && value->is_number_unsigned()) event.height = std::clamp(value->get<std::uint32_t>(), 120U, 1080U); d->session->submit(event); }
 void Graph_WebSocket::close() { if (!d->attached) return; d->session->detach(d->owner); d->attached = false; }
+void Graph_WebSocket::register_callback(std::string key, Callback callback) {
+    std::lock_guard lock(d->callback_mutex);
+    d->callbacks.insert_or_assign(std::move(key), std::move(callback));
+}
+void Graph_WebSocket::unregister_callback(std::string_view key) {
+    std::lock_guard lock(d->callback_mutex);
+    d->callbacks.erase(std::string(key));
+}
+void Graph_WebSocket::send(std::string_view key, std::string_view frame) {
+    Callback callback;
+    {
+        std::lock_guard lock(d->callback_mutex);
+        const auto iterator = d->callbacks.find(std::string(key));
+        if (iterator == d->callbacks.end()) return;
+        callback = iterator->second;
+    }
+    callback(frame);
+}
+
 Graph_WebSocket_Controller::Graph_WebSocket_Controller(std::shared_ptr<Graph_Registry> value) : registry(std::move(value)) {}
 void Graph_WebSocket_Controller::handleNewConnection(const drogon::HttpRequestPtr& request, const drogon::WebSocketConnectionPtr& connection) { auto session = registry->acquire(graph_id_from_path(request->path())); if (!session) { connection->shutdown(drogon::CloseCode::kViolation, "Unknown Aethera graph"); return; } auto socket = std::make_shared<Graph_WebSocket>(connection, std::move(session)); connection->setContext(socket); connection->setPingMessage("aethera-gallery", std::chrono::seconds(20)); socket->start(); }
 void Graph_WebSocket_Controller::handleNewMessage(const drogon::WebSocketConnectionPtr& connection, std::string&& message, const drogon::WebSocketMessageType& type) { if (type != drogon::WebSocketMessageType::Text || message.size() > 64 * 1024) return; if (const auto socket = connection->getContext<Graph_WebSocket>()) socket->receive(message); }

web_server/src/Graph_WebSocket.hpp

@@ -1,7 +1,9 @@
 #pragma once
 #include "Graph_Session.hpp"
 #include <drogon/WebSocketController.h>
+#include <functional>
 #include <memory>
+#include <string>
 namespace aethera::web {
 class Graph_WebSocket final : public std::enable_shared_from_this<Graph_WebSocket> {
 public:
@@ -9,9 +11,13 @@ public:
     ~Graph_WebSocket();
     Graph_WebSocket(const Graph_WebSocket&) = delete;
     Graph_WebSocket& operator=(const Graph_WebSocket&) = delete;
+    using Callback = std::function<void(std::string_view)>;
     void start();
     void receive(std::string_view message);
     void close();
+    void register_callback(std::string key, Callback callback);
+    void unregister_callback(std::string_view key);
+    void send(std::string_view key, std::string_view frame);
 private:
     struct Private;
     std::unique_ptr<Private> d;                              /* Drogon 连接与图会话绑定。 */

web_server/src/Plot.cpp

@@ -0,0 +1,6 @@
+#include "Plot.hpp"
+namespace aethera::web {
+Plot::Plot(std::string key_value, std::shared_ptr<Graph_WebSocket> websocket_value) : websocket(std::move(websocket_value)), key(std::move(key_value)) { websocket->register_callback(key, [this](std::string_view event) { on_event(event); }); }
+Plot::~Plot() { websocket->unregister_callback(key); }
+void Plot::send_frame(std::string_view frame) { websocket->send(key, frame); }
+}

web_server/src/Plot.hpp

@@ -0,0 +1,18 @@
+#pragma once
+#include "Graph_WebSocket.hpp"
+#include <memory>
+#include <string>
+namespace aethera::web {
+class Plot {
+public:
+    Plot(std::string key, std::shared_ptr<Graph_WebSocket> websocket);
+    virtual ~Plot();
+    Plot(const Plot&) = delete;
+    Plot& operator=(const Plot&) = delete;
+protected:
+    virtual void on_event(std::string_view event) = 0;
+    void send_frame(std::string_view frame);
+    std::shared_ptr<Graph_WebSocket> websocket;
+    std::string key;
+};
+}