From cca0cd810a2ce362c2bf3d64c3a810a42a71f2a9 Mon Sep 17 00:00:00 2001 From: Johanness Date: Tue, 16 Jun 2026 11:08:06 +0300 Subject: [PATCH] feat: fabric-bridge auto-fetch target_info from remote target node Initiator can now auto-discover target_info by polling the target's HTTP control server. Just set target_host and target_port in config: --config '{"mode":"initiator",...,"target_host":"10.10.100.1","target_port":9100}' No more manual copy-paste of target_info JSON between hosts. --- .../fabric-bridge/src/fabric_bridge_node.cpp | 118 +++++++++++++++++- .../fabric-bridge/src/fabric_bridge_node.hpp | 5 + 2 files changed, 117 insertions(+), 6 deletions(-) diff --git a/nodes/fabric-bridge/src/fabric_bridge_node.cpp b/nodes/fabric-bridge/src/fabric_bridge_node.cpp index 127d446..530dc69 100644 --- a/nodes/fabric-bridge/src/fabric_bridge_node.cpp +++ b/nodes/fabric-bridge/src/fabric_bridge_node.cpp @@ -4,6 +4,14 @@ #include +#ifndef _WIN32 +#include +#include +#include +#include +#include +#endif + namespace dmf_node { FabricBridgeNode::~FabricBridgeNode() { @@ -49,6 +57,12 @@ void FabricBridgeNode::configure(const nlohmann::json& params) { spdlog::info("Fabric-bridge: target_info set, signaling startup"); } } + + if (params.contains("target_host")) { + target_host_ = params["target_host"].get(); + target_port_ = params.value("target_port", 9100); + spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_); + } } void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) { @@ -85,14 +99,106 @@ void FabricBridgeNode::on_remove_reader(const std::string& port_id) { } } +void FabricBridgeNode::fetch_target_info() { + if (target_host_.empty()) return; + + spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_); + + std::string request = "POST /cmd HTTP/1.1\r\n" + "Host: " + target_host_ + "\r\n" + "Content-Type: application/json\r\n" + "Connection: close\r\n" + "Content-Length: 20\r\n" + "\r\n" + "{\"cmd\":\"status\"}"; + + for (int attempt = 0; attempt < 30 && running_ && target_info_str_.empty(); ++attempt) { + int sock = socket(AF_INET, SOCK_STREAM, 0); + if (sock < 0) { + spdlog::warn("Fabric-bridge: socket() failed, retry {}...", attempt + 1); + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + continue; + } + + struct hostent* he = gethostbyname(target_host_.c_str()); + if (!he) { + spdlog::warn("Fabric-bridge: cannot resolve {}, retry {}...", target_host_, attempt + 1); + close(sock); + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + continue; + } + + struct sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons(target_port_); + std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length); + + if (connect(sock, reinterpret_cast(&addr), sizeof(addr)) < 0) { + spdlog::warn("Fabric-bridge: connect to {}:{} failed, retry {}...", target_host_, target_port_, attempt + 1); + close(sock); + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + continue; + } + + if (send(sock, request.c_str(), request.size(), 0) < 0) { + spdlog::warn("Fabric-bridge: send failed, retry {}...", attempt + 1); + close(sock); + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + continue; + } + + std::string response; + char buf[4096]; + ssize_t n; + while ((n = recv(sock, buf, sizeof(buf), 0)) > 0) { + response.append(buf, n); + } + close(sock); + + auto body_pos = response.find("\r\n\r\n"); + if (body_pos == std::string::npos) { + spdlog::warn("Fabric-bridge: no HTTP body in response, retry {}...", attempt + 1); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + continue; + } + + try { + auto json = nlohmann::json::parse(response.substr(body_pos + 4)); + if (json.contains("data") && json["data"].contains("target_info")) { + auto ti = json["data"]["target_info"].get(); + if (!ti.empty()) { + target_info_str_ = ti; + start_requested_.store(true); + spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size()); + return; + } + } + } catch (const nlohmann::json::exception& e) { + spdlog::warn("Fabric-bridge: JSON parse error: {}", e.what()); + } + + spdlog::info("Fabric-bridge: target_info not ready yet, retry {}...", attempt + 1); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + } + + if (target_info_str_.empty()) { + spdlog::error("Fabric-bridge: failed to fetch target_info after 30 attempts"); + } +} + void FabricBridgeNode::process() { if (!worker_thread_.has_value() && start_requested_.load()) { - running_ = true; - spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target"); - if (mode_ == FabricMode::Initiator) { - worker_thread_ = std::thread(&FabricBridgeNode::run_initiator, this); - } else { - worker_thread_ = std::thread(&FabricBridgeNode::run_target, this); + if (mode_ == FabricMode::Initiator && !target_host_.empty() && target_info_str_.empty()) { + fetch_target_info(); + } + if (start_requested_.load()) { + running_ = true; + spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target"); + if (mode_ == FabricMode::Initiator) { + worker_thread_ = std::thread(&FabricBridgeNode::run_initiator, this); + } else { + worker_thread_ = std::thread(&FabricBridgeNode::run_target, this); + } } } std::this_thread::sleep_for(std::chrono::milliseconds(100)); diff --git a/nodes/fabric-bridge/src/fabric_bridge_node.hpp b/nodes/fabric-bridge/src/fabric_bridge_node.hpp index ed52c58..1a5ef7f 100644 --- a/nodes/fabric-bridge/src/fabric_bridge_node.hpp +++ b/nodes/fabric-bridge/src/fabric_bridge_node.hpp @@ -51,6 +51,11 @@ private: std::string target_info_str_; + std::string target_host_; + uint16_t target_port_ = 9100; + + void fetch_target_info(); + mxlFabricsInstance fabrics_instance_ = nullptr; mxlFabricsInitiator initiator_ = nullptr; mxlFabricsTarget target_ = nullptr;