server push

include/rpc/rpc_thread_manager.h

@@ -28,6 +28,8 @@ class RpcThreadManager {
     void queue_item(void* newFunc);
 
     torrent::Poll* poll();
+
+    void publish_ws_topic(std::string_view topic, std::string_view message);
 
 private:
 

include/websockets_thread.h

@@ -44,6 +44,8 @@ class lt_cacheline_aligned WebsocketsThread : public ProtocolThread {
 
   void queue_item(void*) override;
 
+  void publish_ws_topic(std::string_view topic, std::string_view message);
+
 private:
 
   std::unique_ptr<std::thread> m_websockets_thread = nullptr;

src/core/download_list.cc

@@ -30,15 +30,31 @@
 #include "core/download_list.h"
 #include "core/download_store.h"
 #include "ui/root.h"
+#include "nlohmann/json.hpp"
 
 #define DL_TRIGGER_EVENT(download, event_name)                                 \
   rpc::commands.call_catch(event_name,                                         \
                            rpc::make_target(download),                         \
                            torrent::Object(),                                  \
-                           "Event '" event_name "' failed: ");
+                           "Event '" event_name "' failed: ");                 \
+  publish_topic(download, event_name);
 
 namespace core {
 
+void publish_topic(Download* download, std::string_view topic) {
+  nlohmann::json message = {
+    {"id", nullptr},
+    {"jsonrpc", "2,0"},
+    {"result", {
+                  {"torrent_hash_value", torrent::utils::transform_hex_str(download->info()->hash().str())},
+                  {"event_name", topic},
+                  {"time_stamp", std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count()}
+                }
+    }
+  };
+  worker_thread->publish_ws_topic("event.*", message.dump());
+}
+
 #ifdef RT_USE_EXTRA_DEBUG
 inline void
 DownloadList::check_contains(Download* d) {

src/rpc/rpc_thread_manager.cc

@@ -46,4 +46,8 @@ void RpcThreadManager::queue_item(void* newFunc) {
 
 torrent::Poll* RpcThreadManager::poll() {
   return m_thread_worker->poll();
+}
+
+void RpcThreadManager::publish_ws_topic(std::string_view topic, std::string_view message) {
+  m_websockets_thread->publish_ws_topic(topic, message);
 }

src/websockets_thread.cc

@@ -64,10 +64,15 @@ WebsocketsThread::queue_item(void*) {
 void
 WebsocketsThread::start_thread() {
   auto create_ws_server_and_run = [&]() {
+    if (!listen_info) {
+      throw torrent::internal_error("Can't get listen info for websocket, please check network.websockets.open_local or network.websockets.open_port in rtorrent.rc !!!");
+    }
+
     m_websockets_app = new App();
 
     App::WebSocketBehavior<ConnectionData> behavior;
     behavior.open = [&](WebSocket<false, true, ConnectionData>* ws) {
+      ws->subscribe("event.*");
       all_connection.emplace_back(ws);
       ws->getUserData()->address = ws->getRemoteAddressAsText();
     };
@@ -108,4 +113,11 @@ WebsocketsThread::handle_request(const std::string_view& request) {
   rpc::rpc.dispatch(rpc::RpcManager::RPCType::JSON, request.data(), request.length(), [&](const char* response, uint32_t) {
     return m_websocket_connection->send(std::string_view(response), uWS::OpCode::TEXT);
   });
+}
+
+void WebsocketsThread::publish_ws_topic(std::string_view topic, std::string_view message) {
+  // `m_websockets_app` maybe uninitialized when some events happen so null check is required here
+  if (m_websockets_app) {
+    m_websockets_app->publish(topic, message, OpCode::TEXT);
+  }
 }