From 5398b4d29d95119a214e9c417110f4ff5358f21f Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 02:38:53 +0300 Subject: [PATCH 1/5] feat(bridges): add protocol v1 named pipe bridge --- .github/workflows/ubuntu-smoke.yml | 6 + .github/workflows/windows-smoke.yml | 4 + CMakeLists.txt | 54 + examples/README.md | 3 + .../protocol_v1_named_pipe_bridge_smoke.cpp | 382 ++++ external/SimpleNamedPipe | 2 +- .../envelope-and-transport.md | 3 + .../envelope-and-transport.ru.md | 3 + guides/bridge-taxonomy.md | 4 +- include/optionx_cpp/bridges/protocol_v1.hpp | 5 + .../BridgeProtocolNamedPipeBridge.hpp | 1584 +++++++++++++++++ .../BridgeProtocolNamedPipeConfig.hpp | 153 ++ .../detail/BridgeProtocolServerUtils.hpp | 1 + include/optionx_cpp/data/trading/enums.hpp | 9 +- tests/bridge_umbrella_include_test.cpp | 9 + tests/protocol_v1_bridge_test.cpp | 192 ++ 16 files changed, 2408 insertions(+), 6 deletions(-) create mode 100644 examples/protocol_v1_named_pipe_bridge_smoke.cpp create mode 100644 include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp create mode 100644 include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeConfig.hpp diff --git a/.github/workflows/ubuntu-smoke.yml b/.github/workflows/ubuntu-smoke.yml index 6adcf20a..1c016ae1 100644 --- a/.github/workflows/ubuntu-smoke.yml +++ b/.github/workflows/ubuntu-smoke.yml @@ -108,6 +108,12 @@ jobs: - name: Run protocol_v1_bridge_smoke run: ./build-linux/protocol_v1_bridge_smoke --self-test + - name: Build protocol_v1_named_pipe_bridge_smoke + run: cmake --build build-linux --target protocol_v1_named_pipe_bridge_smoke -j + + - name: Run protocol_v1_named_pipe_bridge_smoke + run: ./build-linux/protocol_v1_named_pipe_bridge_smoke --self-test + - name: Build market_data_subscription_contract_test run: cmake --build build-linux --target market_data_subscription_contract_test -j diff --git a/.github/workflows/windows-smoke.yml b/.github/workflows/windows-smoke.yml index 3aa3f4f3..9fa675c5 100644 --- a/.github/workflows/windows-smoke.yml +++ b/.github/workflows/windows-smoke.yml @@ -55,6 +55,9 @@ jobs: - name: Build Bridge Protocol v1 smoke example run: cmake --build build-windows --config Debug --target protocol_v1_bridge_smoke + - name: Build Bridge Protocol v1 named-pipe smoke example + run: cmake --build build-windows --config Debug --target protocol_v1_named_pipe_bridge_smoke + - name: Run MetaTrader file tests shell: pwsh run: | @@ -72,6 +75,7 @@ jobs: .\build-windows\Debug\bot_binary_command_builder_smoke.exe --self-test .\build-windows\Debug\named_pipe_bridge_smoke.exe --self-test .\build-windows\Debug\protocol_v1_bridge_smoke.exe --self-test + .\build-windows\Debug\protocol_v1_named_pipe_bridge_smoke.exe --self-test - name: Test MetaEditor compile smoke script shell: pwsh diff --git a/CMakeLists.txt b/CMakeLists.txt index a96caa77..e5a66859 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -736,6 +736,60 @@ if(OPTIONX_BUILD_EXAMPLES) -P "${CMAKE_CURRENT_SOURCE_DIR}/cmake/copy_runtime_dlls.cmake" ) endif() + + add_executable(protocol_v1_named_pipe_bridge_smoke examples/protocol_v1_named_pipe_bridge_smoke.cpp) + target_compile_features(protocol_v1_named_pipe_bridge_smoke PRIVATE cxx_std_17) + + target_include_directories(protocol_v1_named_pipe_bridge_smoke PRIVATE + ${EXAMPLE_INCLUDE_DIRS} + ${EXAMPLE_DEPS_INCLUDE_DIRS} + ${CMAKE_CURRENT_SOURCE_DIR}/external/Simple-Web-Server + ${CMAKE_CURRENT_SOURCE_DIR}/external/SimpleNamedPipe/include + ) + + target_link_directories(protocol_v1_named_pipe_bridge_smoke PRIVATE ${EXAMPLE_LIBRARY_DIRS}) + target_compile_definitions( + protocol_v1_named_pipe_bridge_smoke PRIVATE + ${EXAMPLE_DEFINES} + LOGIT_BASE_PATH="${LOGIT_BASE_PATH_FWD}" + ) + if(MINGW) + target_compile_options(protocol_v1_named_pipe_bridge_smoke PRIVATE -Wa,-mbig-obj) + elseif(MSVC) + target_compile_options(protocol_v1_named_pipe_bridge_smoke PRIVATE /bigobj) + endif() + if(OPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS) + target_link_libraries( + protocol_v1_named_pipe_bridge_smoke PRIVATE + ${EXAMPLE_LIBS} + ${OPTIONX_NETWORK_LIBS} + ) + if(WIN32) + target_link_libraries(protocol_v1_named_pipe_bridge_smoke PRIVATE ${OPTIONX_WINDOWS_SYSTEM_LIBS}) + endif() + else() + target_link_libraries(protocol_v1_named_pipe_bridge_smoke PRIVATE ${EXAMPLE_LIBS} optionx_cpp) + endif() + + if(OPTIONX_BUILD_DEPS AND NOT OPTIONX_LIGHTWEIGHT_BRIDGE_SMOKE_TESTS) + add_dependencies(protocol_v1_named_pipe_bridge_smoke mdbx-static AES) + endif() + + foreach(dll ${EXAMPLE_DLL_FILES}) + add_custom_command(TARGET protocol_v1_named_pipe_bridge_smoke POST_BUILD + COMMAND ${CMAKE_COMMAND} -E copy_if_different + "${dll}" "$" + ) + endforeach() + + if(WIN32) + add_custom_command(TARGET protocol_v1_named_pipe_bridge_smoke POST_BUILD + COMMAND ${CMAKE_COMMAND} + -DOPTIONX_RUNTIME_DLL_DIR="${EXAMPLE_BUILD_LIBS_DIR}/bin" + -DOPTIONX_RUNTIME_TARGET_DIR="$" + -P "${CMAKE_CURRENT_SOURCE_DIR}/cmake/copy_runtime_dlls.cmake" + ) + endif() endif() if(OPTIONX_BUILD_TESTS) diff --git a/examples/README.md b/examples/README.md index fdee2a50..18e8e292 100644 --- a/examples/README.md +++ b/examples/README.md @@ -35,6 +35,9 @@ Currently maintained examples: - `protocol_v1_bridge_smoke.cpp` starts the Bridge Protocol v1 HTTP/WebSocket server bridge and can run `--self-test` to POST a `trade.open` JSON-RPC command to itself. +- `protocol_v1_named_pipe_bridge_smoke.cpp` starts Bridge Protocol v1 over a + local named pipe and, on Windows, can run `--self-test` with a local pipe + client. - `metatrader_file_bridge_smoke.cpp` runs the C++ side of the MetaTrader Common\Files bridge against a temporary command/event layout. - `metatrader_file_command_writer_smoke.cpp` demonstrates the C++ command-writer diff --git a/examples/protocol_v1_named_pipe_bridge_smoke.cpp b/examples/protocol_v1_named_pipe_bridge_smoke.cpp new file mode 100644 index 00000000..ddf62c06 --- /dev/null +++ b/examples/protocol_v1_named_pipe_bridge_smoke.cpp @@ -0,0 +1,382 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(_WIN32) +#include +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif +#include +#endif + +namespace { + +using optionx::bridges::protocol_v1::BridgeProtocolNamedPipeBridge; +using optionx::bridges::protocol_v1::BridgeProtocolNamedPipeConfig; + +std::atomic_bool g_stop_requested{false}; +std::atomic_int g_interrupt_count{0}; + +class DemoAccountInfo final : public optionx::BaseAccountInfoData { +public: + std::int64_t user_id = 7; + double balance = 1000.0; + optionx::CurrencyType currency = optionx::CurrencyType::USD; + optionx::AccountType account_type = optionx::AccountType::DEMO; + + std::unique_ptr clone_unique() const override { + return std::make_unique(*this); + } + + std::shared_ptr clone_shared() const override { + return std::make_shared(*this); + } + +private: + bool get_info_bool(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::CONNECTION_STATUS; + } + + std::int64_t get_info_int64(const optionx::AccountInfoRequest& request) const override { + switch (request.type) { + case optionx::AccountInfoType::USER_ID: + return user_id; + case optionx::AccountInfoType::CONNECTION_STATUS: + return 1; + case optionx::AccountInfoType::ACCOUNT_TYPE: + return static_cast(account_type); + case optionx::AccountInfoType::CURRENCY: + return static_cast(currency); + default: + return 0; + } + } + + double get_info_f64(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::BALANCE ? balance : 0.0; + } + + std::string get_info_str(const optionx::AccountInfoRequest& request) const override { + return request.type == optionx::AccountInfoType::USER_ID + ? std::to_string(user_id) + : std::string(); + } + + optionx::AccountType get_info_account_type( + const optionx::AccountInfoRequest&) const override { + return account_type; + } + + optionx::CurrencyType get_info_currency( + const optionx::AccountInfoRequest&) const override { + return currency; + } +}; + +std::string unique_pipe_name() { + const auto stamp = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + return "OptionXProtocolV1PipeSmoke_" + std::to_string(stamp); +} + +BridgeProtocolNamedPipeConfig default_config() { + BridgeProtocolNamedPipeConfig config; + config.named_pipe = unique_pipe_name(); + config.bridge_id = 3; + config.request_body_limit = 8192; + return config; +} + +void request_stop_from_interrupt() { + const auto count = g_interrupt_count.fetch_add(1) + 1; + if (count == 1) { + g_stop_requested.store(true); + return; + } + std::_Exit(130); +} + +#ifdef _WIN32 +BOOL WINAPI console_ctrl_handler(DWORD event_type) { + switch (event_type) { + case CTRL_C_EVENT: + case CTRL_BREAK_EVENT: + case CTRL_CLOSE_EVENT: + case CTRL_LOGOFF_EVENT: + case CTRL_SHUTDOWN_EVENT: + request_stop_from_interrupt(); + return TRUE; + default: + return FALSE; + } +} +#else +void signal_handler(int) { + request_stop_from_interrupt(); +} +#endif + +void install_stop_handlers() { + g_stop_requested.store(false); + g_interrupt_count.store(0); +#ifdef _WIN32 + SetConsoleCtrlHandler(console_ctrl_handler, TRUE); +#else + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); +#endif +} + +void print_usage() { + std::cout + << "Usage: protocol_v1_named_pipe_bridge_smoke [--self-test] [--pipe name]\n" + << "Starts Bridge Protocol v1 over a local named pipe.\n"; +} + +nlohmann::json make_self_test_command() { + return nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "protocol-v1-pipe-smoke-trade"}, + {"method", "trade.open"}, + {"params", { + {"context", { + {"idempotency_key", "protocol-v1-pipe-smoke-trade"}, + {"valid_until_ms", + optionx::bridges::metatrader_file::detail::unix_time_ms() + 60000} + }}, + {"routing", { + {"selector", { + {"kind", "account"}, + {"account_id", "7"} + }} + }}, + {"identity", { + {"unique_hash", "protocol-v1-pipe-smoke"}, + {"signal_name", "protocol_v1_pipe_smoke"} + }}, + {"trade", { + {"symbol", "EURUSD"}, + {"order_type", "BUY"}, + {"option_type", "SPRINT"}, + {"amount", { + {"value", "1.00"}, + {"currency", "USD"} + }}, + {"expiry", { + {"kind", "duration"}, + {"duration_ms", 60000} + }} + }} + }} + }; +} + +#if defined(_WIN32) +bool wait_for_flag( + std::mutex& mutex, + std::condition_variable& cv, + bool& value, + const std::chrono::seconds timeout) { + std::unique_lock lock(mutex); + return cv.wait_for(lock, timeout, [&value]() { + return value; + }); +} + +nlohmann::json pipe_request( + SimpleNamedPipe::NamedPipeClient& client, + const nlohmann::json& request) { + std::error_code ec; + if (!client.write(request.dump(-1), &ec)) { + throw std::runtime_error("write failed: " + ec.message()); + } + + const auto expected_id = request.at("id"); + std::string buffered; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + while (std::chrono::steady_clock::now() < deadline) { + std::string chunk; + if (!client.read(chunk, 250, &ec)) { + continue; + } + buffered += chunk; + for (;;) { + const auto newline = buffered.find('\n'); + if (newline == std::string::npos) { + break; + } + auto parsed = nlohmann::json::parse(buffered.substr(0, newline)); + buffered.erase(0, newline + 1); + if (parsed.contains("id") && parsed.at("id") == expected_id) { + return parsed; + } + } + } + throw std::runtime_error("read timed out"); +} +#endif + +} // namespace + +int main(int argc, char** argv) { + if (argc > 1 && std::string(argv[1]) == "--help") { + print_usage(); + return 0; + } + + const bool self_test = + argc > 1 && std::string(argv[1]) == "--self-test"; + +#if !defined(_WIN32) + if (self_test) { + std::cout << "Bridge Protocol v1 named-pipe transport is Windows-only; self-test skipped.\n"; + return 0; + } +#endif + + install_stop_handlers(); + + auto config = default_config(); + for (int i = 1; i + 1 < argc; ++i) { + if (std::string(argv[i]) == "--pipe") { + config.named_pipe = argv[i + 1]; + } + } + + std::atomic next_signal_id{1}; + std::mutex mutex; + std::condition_variable cv; + bool server_started = false; + bool signal_received = false; + + BridgeProtocolNamedPipeBridge bridge; + if (!bridge.configure(std::make_unique(config))) { + std::cerr << "Bridge configuration failed\n"; + return 2; + } + + struct BridgeCleanup { + BridgeProtocolNamedPipeBridge& bridge; + + ~BridgeCleanup() noexcept { + try { + bridge.shutdown(); + bridge.on_status_update() = {}; + bridge.on_trade_signal() = {}; + bridge.on_signal_id() = {}; + } catch (...) { + } + } + } cleanup{bridge}; + + bridge.on_signal_id() = [&next_signal_id]() { + return next_signal_id.fetch_add(1); + }; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + std::cout << "status=" << optionx::to_str(update.status); + if (!update.connection_id.empty()) { + std::cout << " connection=" << update.connection_id; + } + if (!update.message.empty()) { + std::cout << " message=" << update.message; + } + std::cout << '\n'; + + { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + server_started = true; + } + } + cv.notify_all(); + }; + + bridge.on_trade_signal() = [&](std::unique_ptr signal) { + if (!signal) { + return; + } + nlohmann::json json = *signal; + std::cout << "signal:\n" << json.dump(2) << '\n'; + + optionx::TradeResult result; + result.trade_id = signal->signal_id; + result.option_id = 1000 + signal->signal_id; + result.amount = signal->amount; + result.profit = signal->amount * 0.82; + result.payout = 0.82; + result.trade_state = optionx::TradeState::WIN; + bridge.update_trade_result(signal->to_trade_request(), result); + + { + std::lock_guard lock(mutex); + signal_received = true; + } + cv.notify_all(); + }; + + bridge.update_account_info(optionx::AccountInfoUpdate( + std::make_shared(), + optionx::AccountUpdateStatus::BALANCE_UPDATED)); + bridge.run(); + std::cout << "Bridge Protocol v1 named pipe: " << config.named_pipe << '\n'; + +#if defined(_WIN32) + if (self_test) { + if (!wait_for_flag(mutex, cv, server_started, std::chrono::seconds(3))) { + std::cerr << "Named-pipe protocol server did not start\n"; + return 3; + } + + SimpleNamedPipe::ClientConfig client_config( + config.named_pipe, + config.buffer_size, + 3000); + SimpleNamedPipe::NamedPipeClient client(client_config); + std::error_code ec; + if (!client.connect(&ec)) { + std::cerr << "Client connect failed: " << ec.message() << '\n'; + return 4; + } + + try { + const auto response = pipe_request(client, make_self_test_command()); + std::cout << "self-test response: " << response.dump(2) << '\n'; + if (response.at("result").at("status").get() != "accepted") { + std::cerr << "Unexpected response status\n"; + client.close(); + return 5; + } + } catch (const std::exception& ex) { + std::cerr << "Self-test request failed: " << ex.what() << '\n'; + client.close(); + return 6; + } + + if (!wait_for_flag(mutex, cv, signal_received, std::chrono::seconds(1))) { + std::cerr << "Bridge did not publish a signal\n"; + client.close(); + return 7; + } + + client.close(); + return 0; + } +#endif + + std::cout << "Press Ctrl+C to stop...\n"; + while (!g_stop_requested.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + return 0; +} diff --git a/external/SimpleNamedPipe b/external/SimpleNamedPipe index 179e5cc4..422612a2 160000 --- a/external/SimpleNamedPipe +++ b/external/SimpleNamedPipe @@ -1 +1 @@ -Subproject commit 179e5cc48ad830236e93520f5cf0f1f0538813db +Subproject commit 422612a29e8514c3bb9fbf09970db1407f49ec4a diff --git a/guides/bridge-protocol-v1/envelope-and-transport.md b/guides/bridge-protocol-v1/envelope-and-transport.md index 6af30c1b..65e21405 100644 --- a/guides/bridge-protocol-v1/envelope-and-transport.md +++ b/guides/bridge-protocol-v1/envelope-and-transport.md @@ -114,6 +114,9 @@ Rules: - Named-pipe and other session transports may use `protocol.hello` as the initial protocol-version handshake, then keep the selected version for the session. +- The C++ named-pipe transport frames JSON-RPC responses and notifications as + UTF-8 newline-delimited JSON. Clients must read until `\n` instead of + assuming one pipe read equals one JSON message. - If a session transport already selected v1 but `protocol.hello` requests only unsupported versions, the bridge must return `unsupported_protocol_version`. - Business command `params` must not contain a separate `protocol_version`, diff --git a/guides/bridge-protocol-v1/envelope-and-transport.ru.md b/guides/bridge-protocol-v1/envelope-and-transport.ru.md index 3e4d7d4f..a1b939de 100644 --- a/guides/bridge-protocol-v1/envelope-and-transport.ru.md +++ b/guides/bridge-protocol-v1/envelope-and-transport.ru.md @@ -115,6 +115,9 @@ server error. - Named-pipe и другие session transports могут использовать `protocol.hello` как initial protocol-version handshake и затем сохранять выбранную версию на всю session. +- C++ named-pipe transport frames JSON-RPC responses and notifications as UTF-8 + newline-delimited JSON. Clients must read until `\n` instead of assuming one + pipe read equals one JSON message. - Если session transport уже выбрал v1, но `protocol.hello` запрашивает только неподдерживаемые версии, bridge должен вернуть `unsupported_protocol_version`. - Business command `params` не должны содержать отдельный `protocol_version`, diff --git a/guides/bridge-taxonomy.md b/guides/bridge-taxonomy.md index e9707d83..95d3e4a9 100644 --- a/guides/bridge-taxonomy.md +++ b/guides/bridge-taxonomy.md @@ -23,7 +23,7 @@ OptionX protocol together. | Family | Public include | Role | Transports | |---|---|---|---| -| Native OptionX API | `optionx_cpp/bridges/protocol_v1.hpp` | JSON-RPC Bridge Protocol v1 for clients that can speak OptionX directly. | HTTP, WebSocket; named pipe planned. | +| Native OptionX API | `optionx_cpp/bridges/protocol_v1.hpp` | JSON-RPC Bridge Protocol v1 for clients that can speak OptionX directly. | HTTP, WebSocket, named pipe. | | MetaTrader Common Files | `optionx_cpp/bridges/metatrader_file.hpp` | MT4/MT5 file-command bridge and command writer for `Common\Files`. | Files. | | TradingView extension | `optionx_cpp/bridges/trading_view.hpp` | Adapter for payloads emitted by `browser_extensions/tradingview-alert-extension`. | HTTP. | | BinaryBot/BotBinary | `optionx_cpp/bridges/bot_binary.hpp` | Formatter/parser helpers for observed BinaryBot-compatible command strings. | HTTP query value, file-signal name. | @@ -36,7 +36,7 @@ does not need to be the same for every family. ## Layout Rules - Add new native OptionX transports under `bridges/protocol_v1/`. - For example, a named-pipe transport for Bridge Protocol v1 should be + For example, the named-pipe transport for Bridge Protocol v1 is `bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp`. - Keep legacy or platform-specific adapters under their adapter family even when their transport overlaps a native transport. For example, the legacy diff --git a/include/optionx_cpp/bridges/protocol_v1.hpp b/include/optionx_cpp/bridges/protocol_v1.hpp index efa4e245..9d77e4bf 100644 --- a/include/optionx_cpp/bridges/protocol_v1.hpp +++ b/include/optionx_cpp/bridges/protocol_v1.hpp @@ -31,6 +31,9 @@ #include #include #include +#if defined(_WIN32) +#include +#endif #include "data/bridge.hpp" #include "data/trading.hpp" @@ -38,8 +41,10 @@ #include "BaseBridge.hpp" #include "metatrader_file.hpp" +#include "protocol_v1/BridgeProtocolNamedPipeConfig.hpp" #include "protocol_v1/BridgeProtocolServerConfig.hpp" #include "protocol_v1/detail/BridgeProtocolServerUtils.hpp" +#include "protocol_v1/BridgeProtocolNamedPipeBridge.hpp" #include "protocol_v1/BridgeProtocolServerBridge.hpp" #endif // OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp new file mode 100644 index 00000000..97f57a4a --- /dev/null +++ b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp @@ -0,0 +1,1584 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_BRIDGE_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_BRIDGE_HPP_INCLUDED + +/// \file BridgeProtocolNamedPipeBridge.hpp +/// \brief Defines the Bridge Protocol v1 named-pipe bridge. + +namespace optionx::bridges::protocol_v1 { + + /// \class BridgeProtocolNamedPipeBridge + /// \brief Serves JSON-RPC Bridge Protocol v1 commands over a local named pipe. + class BridgeProtocolNamedPipeBridge final : public BaseBridge { + private: + struct StoredOperation { + std::string fingerprint; + nlohmann::json result; + std::int64_t completed_at_ms = 0; + std::size_t byte_size = 0; + bool dispatching = false; + }; + + enum class RuntimePhase { + Stopped, + Running, + Stopping + }; + + struct RuntimeState { + std::mutex mutex; + std::condition_variable lifecycle_cv; + std::shared_ptr config; + bridge_status_callback_t status_callback; + BaseBridge::trade_signal_callback_t trade_signal_callback; + BaseBridge::signal_report_callback_t signal_report_callback; + BaseBridge::signal_id_allocator_t signal_id_allocator; + std::shared_ptr account_info; + std::unordered_map operations; + std::unordered_map event_revisions; + std::deque operation_order; + std::string stream_id; + std::size_t operation_cache_bytes = 0; + std::size_t active_transport_callbacks = 0; + std::uint64_t event_seq = 0; + RuntimePhase phase = RuntimePhase::Stopped; + bool stop_notified = false; + bool pending_callback_shutdown = false; + bool transport_callback_admission_closed = false; +# if defined(_WIN32) + std::shared_ptr server; + std::set client_ids; +# endif + }; + + inline static thread_local std::vector s_callback_stack; + + class CallbackScope final { + public: + explicit CallbackScope(std::shared_ptr state) + : m_state(std::move(state)), + m_identity(m_state.get()) { + if (m_identity) { + s_callback_stack.push_back(m_identity); + } + } + + ~CallbackScope() { + if (!m_identity) { + return; + } + if (!s_callback_stack.empty() && + s_callback_stack.back() == m_identity) { + s_callback_stack.pop_back(); + return; + } + const auto it = std::find( + s_callback_stack.rbegin(), + s_callback_stack.rend(), + m_identity); + if (it != s_callback_stack.rend()) { + s_callback_stack.erase(std::next(it).base()); + } + } + + private: + std::shared_ptr m_state; + const RuntimeState* m_identity = nullptr; + }; + + class TransportCallbackScope final { + public: + explicit TransportCallbackScope(std::shared_ptr state) + : m_state(std::move(state)), + m_identity(m_state.get()) { + if (!m_identity) { + return; + } + { + std::lock_guard lock(m_state->mutex); + if (m_state->transport_callback_admission_closed || + m_state->phase != RuntimePhase::Running) { + return; + } + ++m_state->active_transport_callbacks; + m_admitted = true; + } + s_callback_stack.push_back(m_identity); + } + + ~TransportCallbackScope() { + if (!m_admitted) { + return; + } + if (!s_callback_stack.empty() && + s_callback_stack.back() == m_identity) { + s_callback_stack.pop_back(); + } else { + const auto it = std::find( + s_callback_stack.rbegin(), + s_callback_stack.rend(), + m_identity); + if (it != s_callback_stack.rend()) { + s_callback_stack.erase(std::next(it).base()); + } + } + + bool should_drain = false; + { + std::lock_guard lock(m_state->mutex); + if (m_state->active_transport_callbacks > 0) { + --m_state->active_transport_callbacks; + } + should_drain = + m_state->active_transport_callbacks == 0 && + m_state->pending_callback_shutdown; + } + if (should_drain) { + BridgeProtocolNamedPipeBridge::drain_pending_callback_shutdown(m_state); + } + } + + bool admitted() const noexcept { + return m_admitted; + } + + private: + std::shared_ptr m_state; + const RuntimeState* m_identity = nullptr; + bool m_admitted = false; + }; + + public: + /// \brief Constructs an unconfigured named-pipe protocol bridge. + BridgeProtocolNamedPipeBridge() + : m_state(std::make_shared()) {} + + /// \brief Stops the named-pipe transport before destruction. + ~BridgeProtocolNamedPipeBridge() override { + shutdown(); + } + + /// \brief Configures the bridge with Bridge Protocol v1 named-pipe settings. + bool configure(std::unique_ptr config) override { + if (!config) return false; + + const auto* typed = + dynamic_cast(config.get()); + if (!typed) { + config->dispatch_callbacks( + false, + "Invalid Bridge Protocol v1 named-pipe config type."); + return false; + } + + auto next_config = std::make_shared(*typed); + const auto validation = next_config->validate(); + config->dispatch_callbacks(validation.first, validation.second); + if (!validation.first) { + return false; + } + + std::unique_lock lock(m_state->mutex); + while (m_state->phase == RuntimePhase::Stopping) { + m_state->lifecycle_cv.wait(lock); + } + if (m_state->phase != RuntimePhase::Stopped) { + config->dispatch_callbacks( + false, + "Bridge Protocol v1 named-pipe bridge cannot be reconfigured while running."); + return false; + } + m_state->config = std::move(next_config); + m_state->stream_id = make_stream_id(); + return true; + } + + /// \brief Returns the status callback slot. + bridge_status_callback_t& on_status_update() override { + return m_state->status_callback; + } + + /// \brief Returns the trade signal callback slot. + trade_signal_callback_t& on_trade_signal() override { + return m_state->trade_signal_callback; + } + + /// \brief Returns the signal diagnostic callback slot. + signal_report_callback_t& on_signal_report() override { + return m_state->signal_report_callback; + } + + /// \brief Returns the signal ID allocator slot. + signal_id_allocator_t& on_signal_id() override { + return m_state->signal_id_allocator; + } + + /// \brief Updates the account snapshot used by `account.balance.get`. + void update_account_info(const AccountInfoUpdate& info) override { + if (!info.account_info) return; + + std::shared_ptr config; + { + std::lock_guard lock(m_state->mutex); + m_state->account_info = info.account_info; + config = m_state->config; + } + if (!config) { + return; + } + + broadcast_notification(make_balance_updated_notification(*config, *info.account_info)); + } + + /// \brief Broadcasts `trade.updated` notifications to connected named-pipe clients. + void update_trade_result( + const TradeRequest& request, + const TradeResult& result) override { + auto config = get_config(); + if (!config) { + return; + } + const auto now = metatrader_file::detail::unix_time_ms(); + const auto coord = next_event_coordinate(trade_revision_key(request, result)); + auto notification = metatrader_file::detail::make_trade_updated_notification( + make_event_id("trade", coord.stream_id, coord.seq), + source_uri(*config), + coord.stream_id, + coord.seq, + result.close_date > 0 ? result.close_date : now, + now, + request, + result, + coord.revision); + broadcast_notification(std::move(notification)); + } + + /// \brief Starts the named-pipe transport. + void run() override { + auto config = get_config_or_throw(); +# if defined(_WIN32) + { + std::unique_lock lock(m_state->mutex); + while (m_state->phase == RuntimePhase::Stopping) { + m_state->lifecycle_cv.wait(lock); + } + if (m_state->phase == RuntimePhase::Running) { + return; + } + } + + SimpleNamedPipe::ServerConfig server_config( + config->named_pipe, + config->buffer_size, + config->pipe_timeout_ms); + auto server = std::make_shared(server_config); + configure_server_callbacks(server); + + { + std::lock_guard lock(m_state->mutex); + if (m_state->phase != RuntimePhase::Stopped) { + return; + } + m_state->phase = RuntimePhase::Running; + m_state->server = server; + m_state->client_ids.clear(); + m_state->operation_order.clear(); + m_state->operations.clear(); + m_state->operation_cache_bytes = 0; + m_state->event_revisions.clear(); + m_state->event_seq = 0; + m_state->stream_id = make_stream_id(); + m_state->stop_notified = false; + m_state->pending_callback_shutdown = false; + m_state->transport_callback_admission_closed = false; + m_state->lifecycle_cv.notify_all(); + } + + try { + server->start(true); + } catch (const std::exception& ex) { + clear_runtime_after_start_failure(); + notify_status(BridgeStatus::SERVER_START_FAILED, {}, ex.what()); + } +# else + (void)config; + notify_status( + BridgeStatus::SERVER_START_FAILED, + {}, + "Bridge Protocol v1 named-pipe transport is available only on Windows."); +# endif + } + + /// \brief Stops the named-pipe transport and drains active callbacks. + void shutdown() override { +# if defined(_WIN32) + bool should_drain = false; + std::shared_ptr server; + { + std::unique_lock lock(m_state->mutex); + if (m_state->phase == RuntimePhase::Stopping) { + if (is_inside_callback()) { + return; + } + while (m_state->phase == RuntimePhase::Stopping) { + m_state->lifecycle_cv.wait(lock); + } + return; + } + if (m_state->phase == RuntimePhase::Stopped) { + return; + } + + m_state->phase = RuntimePhase::Stopping; + m_state->transport_callback_admission_closed = true; + if (m_state->config) { + fail_dispatching_operations_locked( + m_state, + *m_state->config, + server_stopped_result()); + } + if (is_inside_callback()) { + m_state->pending_callback_shutdown = true; + should_drain = m_state->active_transport_callbacks == 0; + } else { + server = collect_server_locked(m_state); + } + } + + if (should_drain) { + drain_pending_callback_shutdown(m_state); + return; + } + if (server) { + finalize_shutdown(m_state, std::move(server)); + } +# else + notify_status(BridgeStatus::SERVER_STOPPED); +# endif + } + + private: + std::shared_ptr m_state; + + std::shared_ptr get_config() const { + std::lock_guard lock(m_state->mutex); + return m_state->config; + } + + std::shared_ptr get_config_or_throw() const { + auto config = get_config(); + if (!config) { + throw std::invalid_argument("Bridge Protocol v1 named-pipe bridge is not configured."); + } + return config; + } + + bool is_inside_callback() const noexcept { + return std::find( + s_callback_stack.begin(), + s_callback_stack.end(), + m_state.get()) != s_callback_stack.end(); + } + + static std::string source_uri(const BridgeProtocolNamedPipeConfig& config) { + return "optionx://bridge/protocol_v1/" + + config.installation_id + + "/" + + config.server_instance_id + + "/" + + std::to_string(config.bridge_id); + } + + static std::string make_stream_id() { + std::ostringstream out; + out << "bridge-protocol-v1-pipe-" + << std::chrono::steady_clock::now().time_since_epoch().count() + << '-' + << std::this_thread::get_id(); + return out.str(); + } + + static std::string make_event_id( + const std::string& prefix, + const std::string& stream_id, + const std::uint64_t seq) { + return "evt-" + stream_id + "-" + prefix + "-" + std::to_string(seq); + } + + struct EventCoordinate { + std::string stream_id; + std::uint64_t seq = 0; + std::uint64_t revision = 1; + }; + + EventCoordinate next_event_coordinate(const std::string& subject_key) { + std::lock_guard lock(m_state->mutex); + EventCoordinate coordinate; + coordinate.stream_id = m_state->stream_id; + coordinate.seq = ++m_state->event_seq; + if (!subject_key.empty()) { + coordinate.revision = ++m_state->event_revisions[subject_key]; + } + return coordinate; + } + + static std::string trade_revision_key( + const TradeRequest& request, + const TradeResult& result) { + const auto trade_id = result.trade_id != 0 ? result.trade_id : request.trade_id; + if (trade_id != 0) { + return "trade:" + std::to_string(trade_id); + } + if (request.signal_id != 0) { + return "signal:" + std::to_string(request.signal_id); + } + if (!request.unique_hash.empty()) { + return "unique_hash:" + request.unique_hash; + } + return "trade:unknown"; + } + + nlohmann::json make_balance_updated_notification( + const BridgeProtocolNamedPipeConfig& config, + const BaseAccountInfoData& account) { + const auto now = metatrader_file::detail::unix_time_ms(); + const auto account_id = metatrader_file::detail::account_id_string(account); + const auto coord = next_event_coordinate("account:" + account_id); + return metatrader_file::detail::make_balance_updated_notification( + make_event_id("balance", coord.stream_id, coord.seq), + source_uri(config), + coord.stream_id, + coord.seq, + now, + now, + account_id, + metatrader_file::detail::safe_account_balance(account), + metatrader_file::detail::safe_account_currency(account), + coord.revision); + } + + void notify_status( + BridgeStatus status, + std::string connection_id = {}, + std::string message = {}) const { + notify_status_from_state( + m_state, + status, + std::move(connection_id), + std::move(message)); + } + + static void notify_status_from_state( + const std::shared_ptr& state, + BridgeStatus status, + std::string connection_id, + std::string message = {}) { + bridge_status_callback_t callback; + { + std::lock_guard lock(state->mutex); + callback = state->status_callback; + } + if (callback) { + try { + CallbackScope scope(state); + callback(BridgeStatusUpdate{ + status, + std::move(connection_id), + std::move(message) + }); + } catch (...) { + } + } + } + + std::string current_stream_id() const { + std::lock_guard lock(m_state->mutex); + return m_state->stream_id; + } + + static std::string frame_message(nlohmann::json message) { + return message.dump(-1) + "\n"; + } + + static bool is_valid_jsonrpc_id(const nlohmann::json& id) { + return id.is_string() || + id.is_null() || + id.is_number_integer() || + id.is_number_unsigned(); + } + + static nlohmann::json request_id_or_null(const nlohmann::json& request) { + if (!request.is_object() || !request.contains("id")) { + return nullptr; + } + const auto& id = request.at("id"); + return is_valid_jsonrpc_id(id) ? id : nlohmann::json(nullptr); + } + + static bool hello_accepts_v1(const nlohmann::json& params) { + if (!params.is_object() || !params.contains("requested_protocol_versions")) { + return true; + } + const auto& versions = params.at("requested_protocol_versions"); + if (!versions.is_array()) { + return false; + } + for (const auto& version : versions) { + if (version.is_string() && version.get() == "1") { + return true; + } + if (version.is_number_integer() && version.get() == 1) { + return true; + } + if (version.is_number_unsigned() && version.get() == 1) { + return true; + } + } + return false; + } + + nlohmann::json handle_message_body( + const BridgeProtocolNamedPipeConfig& config, + const std::string& body) { + if (body.size() > config.request_body_limit) { + return detail::jsonrpc_error( + nullptr, + detail::jsonrpc_invalid_request, + "Message is too large.", + nlohmann::json{{"code", "request_body_too_large"}}); + } + + try { + return handle_jsonrpc_command(config, nlohmann::json::parse(body)); + } catch (const nlohmann::json::parse_error& ex) { + return detail::jsonrpc_error( + nullptr, + detail::jsonrpc_parse_error, + ex.what()); + } catch (const std::exception& ex) { + return detail::jsonrpc_error( + nullptr, + detail::jsonrpc_internal_error, + ex.what()); + } + } + + nlohmann::json handle_jsonrpc_command( + const BridgeProtocolNamedPipeConfig& config, + const nlohmann::json& request) { + const auto id = request_id_or_null(request); + if (!request.is_object() || + !request.contains("jsonrpc") || + !request.at("jsonrpc").is_string() || + request.at("jsonrpc").get() != "2.0" || + !request.contains("id") || + !is_valid_jsonrpc_id(request.at("id")) || + !request.contains("method") || + !request.at("method").is_string()) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_request, + "Invalid JSON-RPC request."); + } + if (!id.is_null() && id.dump(-1).size() > config.max_jsonrpc_id_bytes) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_request, + "JSON-RPC id exceeds configured size limit.", + nlohmann::json{{"code", "jsonrpc_id_too_large"}}); + } + + const auto method = request.at("method").get(); + const auto params = + request.contains("params") ? request.at("params") : nlohmann::json::object(); + if (!params.is_object()) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + "JSON-RPC params must be an object."); + } + + if (method == "protocol.hello") { + if (!hello_accepts_v1(params)) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_unsupported_protocol_version, + "Unsupported protocol version.", + nlohmann::json{{"code", "unsupported_protocol_version"}}); + } + return detail::jsonrpc_result( + id, + nlohmann::json{ + {"selected_protocol_version", "1"}, + {"installation_id", config.installation_id}, + {"server_instance_id", config.server_instance_id}, + {"session_id", current_stream_id()} + }); + } + if (method == "protocol.capabilities.get") { + auto result = detail::capabilities_snapshot(); + result["server_instance_id"] = config.server_instance_id; + result["features"]["http"] = false; + result["features"]["websocket"] = false; + result["features"]["named_pipe"] = true; + result["limits"]["max_message_bytes"] = config.request_body_limit; + result["limits"]["max_jsonrpc_id_bytes"] = config.max_jsonrpc_id_bytes; + result["limits"]["max_idempotency_key_bytes"] = + config.max_idempotency_key_bytes; + result["limits"]["max_operation_fingerprint_bytes"] = + config.max_operation_fingerprint_bytes; + result["limits"]["max_operation_cache_bytes"] = + config.max_operation_cache_bytes; + result["limits"]["operation_cache_retention_ms"] = + config.operation_cache_retention_ms; + return detail::jsonrpc_result(id, std::move(result)); + } + if (method == "account.balance.get") { + return handle_account_balance_get(id); + } + if (method == "signal.submit" || method == "trade.open") { + return handle_trade_affecting_command( + config, + id, + method, + params, + method == "trade.open"); + } + + return detail::jsonrpc_error( + id, + detail::jsonrpc_method_not_found, + "Method not found.", + nlohmann::json{{"method", method}}); + } + + nlohmann::json handle_account_balance_get(const nlohmann::json& id) { + std::shared_ptr account; + { + std::lock_guard lock(m_state->mutex); + account = m_state->account_info; + } + if (!account) { + return detail::jsonrpc_result( + id, + nlohmann::json{ + {"status", "unavailable"}, + {"final", true}, + {"reason", { + {"code", "account_snapshot_unavailable"}, + {"message", "No account snapshot is available."} + }} + }); + } + return detail::jsonrpc_result( + id, + nlohmann::json{ + {"status", "completed"}, + {"final", true}, + {"account", metatrader_file::detail::account_snapshot_json(*account)} + }); + } + + static bool is_known_key( + const std::unordered_set& keys, + const std::string& key) { + return keys.find(key) != keys.end(); + } + + static std::pair reject_unknown_keys( + const nlohmann::json& object, + const std::unordered_set& keys, + const std::string& path) { + if (!object.is_object()) { + return {false, path + " must be an object."}; + } + for (const auto& item : object.items()) { + if (!is_known_key(keys, item.key())) { + return { + false, + path + " contains unsupported field: " + item.key() + "." + }; + } + } + return {true, {}}; + } + + static std::pair validate_known_trade_schema( + const nlohmann::json& params, + const bool direct_trade_open) { + static const std::unordered_set top_level_keys{ + "context", + "routing", + "identity", + "sizing", + "signal", + "trade", + "metadata", + "extensions" + }; + static const std::unordered_set context_keys{ + "idempotency_key", + "valid_until_ms", + "client_created_at_ms", + "file_seq", + "transport" + }; + static const std::unordered_set identity_keys{ + "signal_name", "unique_hash", "unique_id", "user_data" + }; + static const std::unordered_set trade_keys{ + "symbol", + "order_type", + "direction", + "action", + "option_type", + "amount", + "currency", + "expiry", + "duration_ms", + "duration", + "duration_sec", + "expires_at_ms", + "expiry_time", + "refund", + "min_payout", + "comment", + "account_id", + "account_type", + "signal_name", + "unique_hash", + "unique_id", + "user_data", + "metadata", + "extensions" + }; + static const std::unordered_set amount_keys{"value", "currency"}; + static const std::unordered_set expiry_keys{ + "kind", "duration_ms", "expires_at_ms" + }; + static const std::unordered_set sizing_keys{ + "mode", + "type", + "amount", + "currency", + "step", + "group_id", + "group_hash", + "group_name", + "params", + "metadata", + "extensions" + }; + + if (!params.is_object()) { + return {false, "Command params must be an object."}; + } + const auto top = reject_unknown_keys(params, top_level_keys, "Command params"); + if (!top.first) { + return top; + } + if (params.contains("context")) { + const auto context = + reject_unknown_keys(params.at("context"), context_keys, "Command context"); + if (!context.first) { + return context; + } + } + if (params.contains("identity")) { + const auto identity = + reject_unknown_keys(params.at("identity"), identity_keys, "Command identity"); + if (!identity.first) { + return identity; + } + } + if (params.contains("sizing")) { + const auto sizing = + reject_unknown_keys(params.at("sizing"), sizing_keys, "Command sizing"); + if (!sizing.first) { + return sizing; + } + } + + const auto trade_key = direct_trade_open ? "trade" : "signal"; + if (!params.contains(trade_key) || !params.at(trade_key).is_object()) { + return { + false, + std::string("Command ") + trade_key + " must be an object." + }; + } + const auto& trade = params.at(trade_key); + const auto trade_schema = + reject_unknown_keys(trade, trade_keys, "Command " + std::string(trade_key)); + if (!trade_schema.first) { + return trade_schema; + } + if (trade.contains("amount") && trade.at("amount").is_object()) { + const auto amount = + reject_unknown_keys(trade.at("amount"), amount_keys, "Command amount"); + if (!amount.first) { + return amount; + } + } + if (trade.contains("expiry")) { + const auto expiry = + reject_unknown_keys(trade.at("expiry"), expiry_keys, "Command expiry"); + if (!expiry.first) { + return expiry; + } + } + return {true, {}}; + } + + nlohmann::json handle_trade_affecting_command( + const BridgeProtocolNamedPipeConfig& config, + const nlohmann::json& id, + const std::string& method, + const nlohmann::json& params, + const bool direct_trade_open) { + const auto schema_validation = + validate_known_trade_schema(params, direct_trade_open); + if (!schema_validation.first) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + schema_validation.second); + } + + const auto routing_validation = + validate_supported_routing(params, direct_trade_open); + if (!routing_validation.first) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + routing_validation.second); + } + + const auto idempotency_key = + metatrader_file::detail::context_idempotency_key(params); + if (idempotency_key.empty()) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + "Trade-affecting commands require context.idempotency_key."); + } + if (idempotency_key.size() > config.max_idempotency_key_bytes) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + "context.idempotency_key exceeds configured size limit.", + nlohmann::json{{"code", "idempotency_key_too_large"}}); + } + + const auto operation_key = method + "\n" + idempotency_key; + const auto fingerprint = + metatrader_file::detail::canonical_trade_command_payload( + params, + direct_trade_open).dump(-1); + if (fingerprint.size() > config.max_operation_fingerprint_bytes) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + "Canonical trade command fingerprint exceeds configured size limit.", + nlohmann::json{{"code", "operation_fingerprint_too_large"}}); + } + + { + std::unique_lock lock(m_state->mutex); + nlohmann::json cached_response; + if (try_cached_operation_response_locked( + config, + id, + operation_key, + fingerprint, + idempotency_key, + cached_response)) { + return cached_response; + } + } + + std::unique_ptr signal; + try { + signal = metatrader_file::detail::parse_signal_params(params, direct_trade_open); + } catch (const std::exception& ex) { + return detail::jsonrpc_error(id, detail::jsonrpc_invalid_params, ex.what()); + } + + const auto& context = metatrader_file::detail::object_member_or_empty(params, "context"); + if (!context.contains("valid_until_ms")) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_invalid_params, + "Trade-affecting commands require context.valid_until_ms."); + } + std::int64_t valid_until_ms = 0; + try { + valid_until_ms = metatrader_file::detail::context_valid_until_ms(params); + } catch (const std::exception& ex) { + return detail::jsonrpc_error(id, detail::jsonrpc_invalid_params, ex.what()); + } + if (valid_until_ms <= 0 || + metatrader_file::detail::unix_time_ms() > valid_until_ms) { + const auto result = nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "stale_request"}, + {"message", "Command valid_until_ms is in the past."} + }} + }; + if (!remember_operation(config, operation_key, fingerprint, result)) { + return detail::jsonrpc_result(id, idempotency_cache_full_result()); + } + return detail::jsonrpc_result(id, result); + } + + signal_id_allocator_t allocator; + trade_signal_callback_t callback; + { + std::lock_guard lock(m_state->mutex); + allocator = m_state->signal_id_allocator; + callback = m_state->trade_signal_callback; + } + if (!allocator || !callback) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_internal_error, + "Trade signal callback and signal ID allocator must be configured."); + } + + const auto signal_id = allocator(); + if (signal_id == 0) { + return detail::jsonrpc_error( + id, + detail::jsonrpc_internal_error, + "Signal ID allocator returned zero."); + } + + signal->bridge_id = config.bridge_id; + signal->signal_id = signal_id; + auto result = nlohmann::json{ + {"status", "accepted"}, + {"final", false}, + {"operation_id", "mem:" + std::to_string(config.bridge_id) + ":" + idempotency_key} + }; + if (direct_trade_open) { + result["trade_refs"] = nlohmann::json::array({ + nlohmann::json{ + {"status", "pending"}, + {"signal_id", std::to_string(signal_id)} + } + }); + } else { + result["signal_ref"] = nlohmann::json{ + {"signal_id", std::to_string(signal_id)}, + {"unique_hash", signal->unique_hash}, + {"signal_name", signal->signal_name} + }; + result["trade_refs"] = nlohmann::json::array(); + } + + { + std::unique_lock lock(m_state->mutex); + nlohmann::json cached_response; + if (try_cached_operation_response_locked( + config, + id, + operation_key, + fingerprint, + idempotency_key, + cached_response)) { + return cached_response; + } + + const auto byte_size = operation_byte_size(operation_key, fingerprint, result); + prune_completed_operations_locked(m_state, config, byte_size, 1); + if (m_state->operations.size() >= config.dedupe_cache_size) { + return detail::jsonrpc_result(id, idempotency_cache_full_result()); + } + if (byte_size > config.max_operation_cache_bytes || + m_state->operation_cache_bytes > + config.max_operation_cache_bytes - byte_size) { + return detail::jsonrpc_result(id, idempotency_cache_full_result()); + } + + m_state->operation_order.push_back(operation_key); + StoredOperation operation; + operation.fingerprint = fingerprint; + operation.result = result; + operation.byte_size = byte_size; + operation.completed_at_ms = 0; + operation.dispatching = true; + m_state->operations.emplace(operation_key, std::move(operation)); + m_state->operation_cache_bytes += byte_size; + } + + if (metatrader_file::detail::unix_time_ms() > valid_until_ms) { + const auto stale = nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "stale_request"}, + {"message", "Command valid_until_ms expired before dispatch."} + }} + }; + complete_reserved_operation(config, operation_key, stale); + return detail::jsonrpc_result(id, stale); + } + + try { + callback(std::move(signal)); + } catch (const std::exception& ex) { + const auto failed = dispatch_failed_result(ex.what()); + complete_reserved_operation(config, operation_key, failed); + return detail::jsonrpc_result(id, failed); + } catch (...) { + const auto failed = dispatch_failed_result("Trade signal callback failed."); + complete_reserved_operation(config, operation_key, failed); + return detail::jsonrpc_result(id, failed); + } + + complete_reserved_operation(config, operation_key, result); + return detail::jsonrpc_result(id, result); + } + + static std::pair validate_supported_routing( + const nlohmann::json& params, + const bool direct_trade_open) { + if (!params.is_object() || !params.contains("routing")) { + return {true, {}}; + } + const auto& routing = params.at("routing"); + if (!routing.is_object()) { + return {false, "Command routing must be an object."}; + } + if (routing.empty()) { + return {true, {}}; + } + if (routing.contains("policy")) { + return { + false, + direct_trade_open + ? "trade.open routing.policy is not supported." + : "signal.submit routing.policy is not supported by this bridge yet." + }; + } + + for (const auto& item : routing.items()) { + if (item.key() != "selector" && item.key() != "platform_type") { + return { + false, + "Command routing contains unsupported field: " + item.key() + "." + }; + } + } + + if (!routing.contains("selector")) { + return {true, {}}; + } + const auto& selector = routing.at("selector"); + if (!selector.is_object()) { + return {false, "Command routing.selector must be an object."}; + } + if (selector.empty()) { + return {true, {}}; + } + + for (const auto& item : selector.items()) { + if (item.key() != "kind" && item.key() != "account_id") { + return { + false, + "Command routing.selector contains unsupported field: " + item.key() + "." + }; + } + } + + auto kind = metatrader_file::detail::string_value(selector, "kind", "default"); + kind = metatrader_file::detail::lower_ascii_copy( + metatrader_file::detail::trim_ascii_copy(kind)); + if (kind.empty()) { + kind = "default"; + } + if (kind == "accounts" || kind == "all") { + return { + false, + direct_trade_open + ? "trade.open supports only default or account routing." + : "signal.submit accounts/all routing is not supported by this bridge yet." + }; + } + if (kind != "default" && kind != "account") { + return {false, "Command routing.selector.kind is unsupported."}; + } + if (kind == "default" && selector.contains("account_id")) { + return {false, "Command default routing must not include account_id."}; + } + if (kind == "account") { + if (!selector.contains("account_id")) { + return {false, "Command account routing requires account_id."}; + } + try { + static_cast( + metatrader_file::detail::int64_value(selector, "account_id", 0)); + } catch (...) { + return { + false, + "Command account routing requires a numeric account_id in this bridge." + }; + } + } + return {true, {}}; + } + + static nlohmann::json idempotency_conflict_result(std::string message) { + return nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "idempotency_conflict"}, + {"message", std::move(message)} + }} + }; + } + + static nlohmann::json idempotency_cache_full_result() { + return nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "idempotency_cache_full"}, + {"message", "Bridge Protocol v1 idempotency cache is full."} + }} + }; + } + + static nlohmann::json operation_in_progress_result( + const BridgeProtocolNamedPipeConfig& config, + const std::string& idempotency_key) { + return nlohmann::json{ + {"status", "processing"}, + {"final", false}, + {"operation_id", "mem:" + std::to_string(config.bridge_id) + ":" + idempotency_key}, + {"reason", { + {"code", "operation_in_progress"}, + {"message", "The idempotent operation is still being dispatched."} + }} + }; + } + + static nlohmann::json dispatch_failed_result(std::string message) { + if (message.size() > 1024) { + message.resize(1024); + } + return nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "dispatch_failed"}, + {"message", std::move(message)} + }} + }; + } + + static nlohmann::json server_stopped_result() { + return nlohmann::json{ + {"status", "rejected"}, + {"final", true}, + {"reason", { + {"code", "server_stopped"}, + {"message", "Bridge Protocol v1 named-pipe bridge stopped before dispatch completed."} + }} + }; + } + + static nlohmann::json server_stopping_response() { + return detail::jsonrpc_error( + nullptr, + detail::jsonrpc_internal_error, + "Bridge Protocol v1 named-pipe bridge is stopping.", + nlohmann::json{{"code", "server_stopping"}}); + } + + static std::size_t operation_byte_size( + const std::string& operation_key, + const std::string& fingerprint, + const nlohmann::json& result) { + return operation_key.size() + fingerprint.size() + result.dump(-1).size(); + } + + static void touch_operation_locked( + const std::shared_ptr& state, + const std::string& operation_key) { + const auto it = std::find( + state->operation_order.begin(), + state->operation_order.end(), + operation_key); + if (it == state->operation_order.end()) { + state->operation_order.push_back(operation_key); + return; + } + auto key = std::move(*it); + state->operation_order.erase(it); + state->operation_order.push_back(std::move(key)); + } + + bool try_cached_operation_response_locked( + const BridgeProtocolNamedPipeConfig& config, + const nlohmann::json& id, + const std::string& operation_key, + const std::string& fingerprint, + const std::string& idempotency_key, + nlohmann::json& response) { + prune_completed_operations_locked(m_state, config); + const auto existing = m_state->operations.find(operation_key); + if (existing == m_state->operations.end()) { + return false; + } + if (existing->second.fingerprint != fingerprint) { + response = detail::jsonrpc_result(id, idempotency_conflict_result( + "The same idempotency_key was used with a different payload.")); + return true; + } + if (existing->second.dispatching) { + response = detail::jsonrpc_result( + id, + operation_in_progress_result(config, idempotency_key)); + return true; + } + touch_operation_locked(m_state, operation_key); + response = detail::jsonrpc_result(id, existing->second.result); + return true; + } + + bool remember_operation( + const BridgeProtocolNamedPipeConfig& config, + const std::string& operation_key, + const std::string& fingerprint, + const nlohmann::json& result) { + std::lock_guard lock(m_state->mutex); + const auto byte_size = operation_byte_size(operation_key, fingerprint, result); + auto existing = m_state->operations.find(operation_key); + prune_completed_operations_locked( + m_state, + config, + byte_size, + existing == m_state->operations.end() ? 1 : 0); + existing = m_state->operations.find(operation_key); + const auto old_size = existing == m_state->operations.end() + ? std::size_t{0} + : existing->second.byte_size; + if (byte_size > config.max_operation_cache_bytes || + m_state->operation_cache_bytes - old_size > + config.max_operation_cache_bytes - byte_size) { + return false; + } + if (existing == m_state->operations.end()) { + if (m_state->operations.size() >= config.dedupe_cache_size) { + return false; + } + m_state->operation_order.push_back(operation_key); + } + auto& operation = m_state->operations[operation_key]; + m_state->operation_cache_bytes = + m_state->operation_cache_bytes - operation.byte_size + byte_size; + operation.fingerprint = fingerprint; + operation.result = result; + operation.byte_size = byte_size; + operation.completed_at_ms = metatrader_file::detail::unix_time_ms(); + operation.dispatching = false; + return true; + } + + void complete_reserved_operation( + const BridgeProtocolNamedPipeConfig& config, + const std::string& operation_key, + const nlohmann::json& result) { + std::lock_guard lock(m_state->mutex); + const auto existing = m_state->operations.find(operation_key); + if (existing == m_state->operations.end()) { + return; + } + const auto byte_size = + operation_byte_size(operation_key, existing->second.fingerprint, result); + m_state->operation_cache_bytes = + m_state->operation_cache_bytes - existing->second.byte_size + byte_size; + existing->second.result = result; + existing->second.byte_size = byte_size; + existing->second.completed_at_ms = metatrader_file::detail::unix_time_ms(); + existing->second.dispatching = false; + touch_operation_locked(m_state, operation_key); + if (byte_size > config.max_operation_cache_bytes) { + m_state->operation_cache_bytes -= existing->second.byte_size; + m_state->operations.erase(existing); + const auto order_it = std::find( + m_state->operation_order.begin(), + m_state->operation_order.end(), + operation_key); + if (order_it != m_state->operation_order.end()) { + m_state->operation_order.erase(order_it); + } + return; + } + prune_completed_operations_locked(m_state, config); + } + + static void fail_dispatching_operations_locked( + const std::shared_ptr& state, + const BridgeProtocolNamedPipeConfig& config, + const nlohmann::json& result) { + for (auto& item : state->operations) { + auto& operation = item.second; + if (!operation.dispatching) { + continue; + } + const auto byte_size = + operation_byte_size(item.first, operation.fingerprint, result); + state->operation_cache_bytes = + state->operation_cache_bytes - operation.byte_size + byte_size; + operation.result = result; + operation.byte_size = byte_size; + operation.completed_at_ms = metatrader_file::detail::unix_time_ms(); + operation.dispatching = false; + } + prune_completed_operations_locked(state, config); + } + + static bool evict_oldest_completed_operation_locked( + const std::shared_ptr& state) { + for (auto it = state->operation_order.begin(); + it != state->operation_order.end();) { + const auto& key = *it; + auto existing = state->operations.find(key); + if (existing == state->operations.end()) { + it = state->operation_order.erase(it); + continue; + } + if (existing->second.dispatching) { + ++it; + continue; + } + state->operation_cache_bytes -= existing->second.byte_size; + state->operations.erase(existing); + state->operation_order.erase(it); + return true; + } + return false; + } + + static void prune_completed_operations_locked( + const std::shared_ptr& state, + const BridgeProtocolNamedPipeConfig& config, + const std::size_t required_bytes = 0, + const std::size_t required_records = 0) { + const auto now = metatrader_file::detail::unix_time_ms(); + for (auto it = state->operation_order.begin(); it != state->operation_order.end();) { + auto existing = state->operations.find(*it); + if (existing == state->operations.end()) { + it = state->operation_order.erase(it); + continue; + } + const auto& operation = existing->second; + if (!operation.dispatching && + operation.completed_at_ms > 0 && + now - operation.completed_at_ms >= config.operation_cache_retention_ms) { + state->operation_cache_bytes -= operation.byte_size; + state->operations.erase(existing); + it = state->operation_order.erase(it); + continue; + } + ++it; + } + + while (state->operations.size() + required_records > + config.dedupe_cache_size) { + if (!evict_oldest_completed_operation_locked(state)) { + break; + } + } + while (required_bytes <= config.max_operation_cache_bytes && + state->operation_cache_bytes > + config.max_operation_cache_bytes - required_bytes) { + if (!evict_oldest_completed_operation_locked(state)) { + break; + } + } + } + +# if defined(_WIN32) + void configure_server_callbacks( + const std::shared_ptr& server) { + server->on_connected = [state = m_state](int client_id) { + TransportCallbackScope scope(state); + if (!scope.admitted()) { + std::shared_ptr current_server; + { + std::lock_guard lock(state->mutex); + current_server = state->server; + } + if (current_server) { + current_server->close(client_id); + } + return; + } + + { + std::lock_guard lock(state->mutex); + state->client_ids.insert(client_id); + } + + notify_status_from_state( + state, + BridgeStatus::CLIENT_CONNECTED, + connection_id(client_id)); + }; + + server->on_start = [state = m_state](const SimpleNamedPipe::ServerConfig&) { + CallbackScope scope(state); + notify_status_from_state(state, BridgeStatus::SERVER_STARTED, "pipe"); + }; + + server->on_disconnected = + [state = m_state](int client_id, const std::error_code& ec) { + CallbackScope scope(state); + { + std::lock_guard lock(state->mutex); + state->client_ids.erase(client_id); + } + notify_status_from_state( + state, + BridgeStatus::CLIENT_DISCONNECTED, + connection_id(client_id), + ec.message()); + }; + + server->on_message = + [this](int client_id, const std::string& message) { + handle_message(client_id, message); + }; + + server->on_error = [state = m_state](const std::error_code& ec) { + CallbackScope scope(state); + notify_status_from_state( + state, + BridgeStatus::CONNECTION_ERROR, + {}, + ec.message()); + }; + } + + static std::string connection_id(int client_id) { + return std::to_string(client_id); + } + + void handle_message(int client_id, const std::string& message) { + TransportCallbackScope scope(m_state); + std::shared_ptr server; + { + std::lock_guard lock(m_state->mutex); + server = m_state->server; + } + if (!server) { + return; + } + if (!scope.admitted()) { + send_pipe_text(server, client_id, frame_message(server_stopping_response())); + return; + } + + auto config = get_config_or_throw(); + const auto response = frame_message(handle_message_body(*config, message)); + send_pipe_text(server, client_id, response); + } + + void broadcast_notification(nlohmann::json notification) { + std::shared_ptr server; + std::vector clients; + { + std::lock_guard lock(m_state->mutex); + server = m_state->server; + clients.assign(m_state->client_ids.begin(), m_state->client_ids.end()); + } + if (!server) { + return; + } + const auto text = frame_message(std::move(notification)); + for (const int client_id : clients) { + send_pipe_text(server, client_id, text); + } + } + + static void send_pipe_text( + const std::shared_ptr& server, + const int client_id, + const std::string& text) { + server->send_to(client_id, text); + } + + void clear_runtime_after_start_failure() { + std::lock_guard lock(m_state->mutex); + m_state->server.reset(); + m_state->client_ids.clear(); + m_state->phase = RuntimePhase::Stopped; + m_state->transport_callback_admission_closed = false; + m_state->pending_callback_shutdown = false; + m_state->lifecycle_cv.notify_all(); + } + + static std::shared_ptr collect_server_locked( + const std::shared_ptr& state) { + auto server = state->server; + state->server.reset(); + state->client_ids.clear(); + if (!state->stop_notified) { + state->stop_notified = true; + } + return server; + } + + static void drain_pending_callback_shutdown( + const std::shared_ptr& state) { + std::shared_ptr server; + { + std::lock_guard lock(state->mutex); + if (!state->pending_callback_shutdown || + state->active_transport_callbacks != 0 || + state->phase != RuntimePhase::Stopping) { + return; + } + state->pending_callback_shutdown = false; + server = collect_server_locked(state); + } + std::thread([state, server = std::move(server)]() mutable { + finalize_shutdown(state, std::move(server)); + }).detach(); + } + + static void finalize_shutdown( + const std::shared_ptr& state, + std::shared_ptr server) { + if (server) { + server->stop(); + } + notify_status_from_state(state, BridgeStatus::SERVER_STOPPED, {}); + { + std::lock_guard lock(state->mutex); + state->phase = RuntimePhase::Stopped; + state->transport_callback_admission_closed = false; + state->pending_callback_shutdown = false; + state->lifecycle_cv.notify_all(); + } + } +# else + void broadcast_notification(nlohmann::json notification) { + (void)notification; + } + + static void drain_pending_callback_shutdown( + const std::shared_ptr& state) { + (void)state; + } +# endif + }; + +} // namespace optionx::bridges::protocol_v1 + +#endif // OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_BRIDGE_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeConfig.hpp b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeConfig.hpp new file mode 100644 index 00000000..09000c13 --- /dev/null +++ b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeConfig.hpp @@ -0,0 +1,153 @@ +#pragma once +#ifndef OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_CONFIG_HPP_INCLUDED +#define OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_CONFIG_HPP_INCLUDED + +/// \file BridgeProtocolNamedPipeConfig.hpp +/// \brief Defines configuration for Bridge Protocol v1 over a named pipe. + +namespace optionx::bridges::protocol_v1 { + + /// \class BridgeProtocolNamedPipeConfig + /// \brief Configuration for serving Bridge Protocol v1 over local named pipes. + class BridgeProtocolNamedPipeConfig final : public IBridgeConfig { + public: + std::string named_pipe = "optionx_bridge_protocol_v1"; ///< Named pipe endpoint. + BridgeId bridge_id = 0; ///< Source bridge ID assigned to emitted signals. + std::string installation_id = "optionx-local"; ///< Stable installation identifier. + std::string server_instance_id = "optionx-bridge-named-pipe"; ///< Runtime server identifier. + std::size_t buffer_size = 64 * 1024; ///< Named-pipe read/write buffer size. + std::size_t pipe_timeout_ms = 50; ///< Named-pipe internal wait timeout. + std::size_t request_body_limit = 1024 * 1024; ///< Maximum JSON-RPC message size. + std::size_t dedupe_cache_size = 4096; ///< In-memory idempotency/result cache size. + std::size_t max_jsonrpc_id_bytes = 256; ///< Maximum serialized JSON-RPC `id` size. + std::size_t max_idempotency_key_bytes = 512; ///< Maximum logical idempotency key size. + std::size_t max_operation_fingerprint_bytes = 64 * 1024; ///< Maximum canonical payload bytes. + std::size_t max_operation_cache_bytes = 4 * 1024 * 1024; ///< Maximum result cache bytes. + std::int64_t operation_cache_retention_ms = 15 * 60 * 1000; ///< Completed result TTL. + + /// \brief Serializes configuration to JSON. + void to_json(nlohmann::json& j) const override { + j = nlohmann::json{ + {"named_pipe", named_pipe}, + {"bridge_id", bridge_id}, + {"installation_id", installation_id}, + {"server_instance_id", server_instance_id}, + {"buffer_size", buffer_size}, + {"pipe_timeout_ms", pipe_timeout_ms}, + {"request_body_limit", request_body_limit}, + {"dedupe_cache_size", dedupe_cache_size}, + {"max_jsonrpc_id_bytes", max_jsonrpc_id_bytes}, + {"max_idempotency_key_bytes", max_idempotency_key_bytes}, + {"max_operation_fingerprint_bytes", max_operation_fingerprint_bytes}, + {"max_operation_cache_bytes", max_operation_cache_bytes}, + {"operation_cache_retention_ms", operation_cache_retention_ms} + }; + } + + /// \brief Loads configuration from JSON. + void from_json(const nlohmann::json& j) override { + if (j.contains("named_pipe")) named_pipe = j.at("named_pipe").get(); + if (j.contains("bridge_id")) bridge_id = j.at("bridge_id").get(); + if (j.contains("installation_id")) { + installation_id = j.at("installation_id").get(); + } + if (j.contains("server_instance_id")) { + server_instance_id = j.at("server_instance_id").get(); + } + if (j.contains("buffer_size")) buffer_size = j.at("buffer_size").get(); + if (j.contains("pipe_timeout_ms")) { + pipe_timeout_ms = j.at("pipe_timeout_ms").get(); + } + if (j.contains("request_body_limit")) { + request_body_limit = j.at("request_body_limit").get(); + } + if (j.contains("dedupe_cache_size")) { + dedupe_cache_size = j.at("dedupe_cache_size").get(); + } + if (j.contains("max_jsonrpc_id_bytes")) { + max_jsonrpc_id_bytes = j.at("max_jsonrpc_id_bytes").get(); + } + if (j.contains("max_idempotency_key_bytes")) { + max_idempotency_key_bytes = + j.at("max_idempotency_key_bytes").get(); + } + if (j.contains("max_operation_fingerprint_bytes")) { + max_operation_fingerprint_bytes = + j.at("max_operation_fingerprint_bytes").get(); + } + if (j.contains("max_operation_cache_bytes")) { + max_operation_cache_bytes = + j.at("max_operation_cache_bytes").get(); + } + if (j.contains("operation_cache_retention_ms")) { + operation_cache_retention_ms = + j.at("operation_cache_retention_ms").get(); + } + } + + /// \brief Creates a unique copy. + std::unique_ptr clone_unique() const override { + return std::make_unique(*this); + } + + /// \brief Creates a shared copy. + std::shared_ptr clone_shared() const override { + return std::make_shared(*this); + } + + /// \return `BridgeType::BRIDGE_PROTOCOL_V1_NAMED_PIPE`. + BridgeType bridge_type() const override { + return BridgeType::BRIDGE_PROTOCOL_V1_NAMED_PIPE; + } + + /// \brief Validates configuration. + std::pair validate() const override { + if (named_pipe.empty()) { + return {false, "Bridge Protocol v1 named_pipe must not be empty."}; + } + if (bridge_id == 0) { + return {false, "Bridge Protocol v1 bridge_id is required."}; + } + if (installation_id.empty()) { + return {false, "Bridge Protocol v1 installation_id must not be empty."}; + } + if (server_instance_id.empty()) { + return {false, "Bridge Protocol v1 server_instance_id must not be empty."}; + } + if (buffer_size == 0) { + return {false, "Bridge Protocol v1 buffer_size must be positive."}; + } + if (pipe_timeout_ms == 0) { + return {false, "Bridge Protocol v1 pipe_timeout_ms must be positive."}; + } + if (request_body_limit == 0) { + return {false, "Bridge Protocol v1 request_body_limit must be positive."}; + } + if (dedupe_cache_size == 0) { + return {false, "Bridge Protocol v1 dedupe_cache_size must be positive."}; + } + if (max_jsonrpc_id_bytes == 0) { + return {false, "Bridge Protocol v1 max_jsonrpc_id_bytes must be positive."}; + } + if (max_idempotency_key_bytes == 0) { + return {false, "Bridge Protocol v1 max_idempotency_key_bytes must be positive."}; + } + if (max_operation_fingerprint_bytes == 0) { + return { + false, + "Bridge Protocol v1 max_operation_fingerprint_bytes must be positive." + }; + } + if (max_operation_cache_bytes == 0) { + return {false, "Bridge Protocol v1 max_operation_cache_bytes must be positive."}; + } + if (operation_cache_retention_ms <= 0) { + return {false, "Bridge Protocol v1 operation_cache_retention_ms must be positive."}; + } + return {true, {}}; + } + }; + +} // namespace optionx::bridges::protocol_v1 + +#endif // OPTIONX_HEADER_BRIDGES_PROTOCOL_V1_BRIDGE_PROTOCOL_NAMED_PIPE_CONFIG_HPP_INCLUDED diff --git a/include/optionx_cpp/bridges/protocol_v1/detail/BridgeProtocolServerUtils.hpp b/include/optionx_cpp/bridges/protocol_v1/detail/BridgeProtocolServerUtils.hpp index 43af5b02..c17dc503 100644 --- a/include/optionx_cpp/bridges/protocol_v1/detail/BridgeProtocolServerUtils.hpp +++ b/include/optionx_cpp/bridges/protocol_v1/detail/BridgeProtocolServerUtils.hpp @@ -98,6 +98,7 @@ namespace optionx::bridges::protocol_v1::detail { {"features", { {"http", true}, {"websocket", true}, + {"named_pipe", false}, {"subscriptions", false}, {"event_replay", false}, {"trade_open_batch", false} diff --git a/include/optionx_cpp/data/trading/enums.hpp b/include/optionx_cpp/data/trading/enums.hpp index a2eb22c4..2d16d353 100644 --- a/include/optionx_cpp/data/trading/enums.hpp +++ b/include/optionx_cpp/data/trading/enums.hpp @@ -98,7 +98,8 @@ namespace optionx { LEGACY_TRADING_NAMED_PIPE, ///< Legacy named-pipe trading bridge. TRADING_VIEW_EXTENSION_HTTP, ///< TradingView browser extension HTTP bridge. METATRADER_FILE_TRANSPORT, ///< MetaTrader common-files JSON-RPC bridge transport. - BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET ///< Bridge Protocol v1 HTTP/WebSocket server. + BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET, ///< Bridge Protocol v1 HTTP/WebSocket server. + BRIDGE_PROTOCOL_V1_NAMED_PIPE ///< Bridge Protocol v1 named-pipe server. }; /// \brief Converts BridgeType to its string representation. @@ -111,7 +112,8 @@ namespace optionx { "LEGACY_TRADING_NAMED_PIPE", "TRADING_VIEW_EXTENSION_HTTP", "METATRADER_FILE_TRANSPORT", - "BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET" + "BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET", + "BRIDGE_PROTOCOL_V1_NAMED_PIPE" }; return utils::enum_string_or_unknown(str_data, static_cast(value)); } @@ -126,7 +128,8 @@ namespace optionx { {"LEGACY_TRADING_NAMED_PIPE", BridgeType::LEGACY_TRADING_NAMED_PIPE}, {"TRADING_VIEW_EXTENSION_HTTP", BridgeType::TRADING_VIEW_EXTENSION_HTTP}, {"METATRADER_FILE_TRANSPORT", BridgeType::METATRADER_FILE_TRANSPORT}, - {"BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET", BridgeType::BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET} + {"BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET", BridgeType::BRIDGE_PROTOCOL_V1_HTTP_WEBSOCKET}, + {"BRIDGE_PROTOCOL_V1_NAMED_PIPE", BridgeType::BRIDGE_PROTOCOL_V1_NAMED_PIPE} }; auto it = str_data.find(utils::to_upper_case(str)); if (it != str_data.end()) { diff --git a/tests/bridge_umbrella_include_test.cpp b/tests/bridge_umbrella_include_test.cpp index ddfbcbc2..bb166e34 100644 --- a/tests/bridge_umbrella_include_test.cpp +++ b/tests/bridge_umbrella_include_test.cpp @@ -76,6 +76,15 @@ TEST(BridgeUmbrellaIncludeTest, ExposesBridgeProtocolV1ServerFamily) { EXPECT_EQ(bridge.bound_websocket_port(), 0); } +TEST(BridgeUmbrellaIncludeTest, ExposesBridgeProtocolV1NamedPipeFamily) { + optionx::bridges::protocol_v1::BridgeProtocolNamedPipeConfig config; + optionx::bridges::protocol_v1::BridgeProtocolNamedPipeBridge bridge; + + EXPECT_EQ( + config.bridge_type(), + optionx::BridgeType::BRIDGE_PROTOCOL_V1_NAMED_PIPE); +} + TEST(BridgeUmbrellaIncludeTest, ExposesSignalReportApi) { optionx::BridgeSignalReport report; report.status = optionx::BridgeSignalReportStatus::REJECTED; diff --git a/tests/protocol_v1_bridge_test.cpp b/tests/protocol_v1_bridge_test.cpp index adc298f5..f3828fa3 100644 --- a/tests/protocol_v1_bridge_test.cpp +++ b/tests/protocol_v1_bridge_test.cpp @@ -5,6 +5,9 @@ #include #include +#if defined(_WIN32) +#include +#endif #include #include @@ -90,6 +93,17 @@ optionx::bridges::protocol_v1::BridgeProtocolServerConfig test_config() { return config; } +optionx::bridges::protocol_v1::BridgeProtocolNamedPipeConfig test_pipe_config() { + optionx::bridges::protocol_v1::BridgeProtocolNamedPipeConfig config; + const auto stamp = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); + config.named_pipe = "OptionxBridgeProtocolV1PipeTest_" + std::to_string(stamp); + config.bridge_id = 4; + config.request_body_limit = 8192; + config.dedupe_cache_size = 32; + return config; +} + bool wait_for_http_port( const optionx::bridges::protocol_v1::BridgeProtocolServerBridge& bridge) { for (int i = 0; i < 500; ++i) { @@ -193,6 +207,46 @@ nlohmann::json post_json_without_auth( return nlohmann::json::parse(response->content.string()); } +#if defined(_WIN32) +nlohmann::json pipe_json( + SimpleNamedPipe::NamedPipeClient& client, + const nlohmann::json& request) { + std::error_code ec; + EXPECT_TRUE(client.write(request.dump(-1), &ec)) << ec.message(); + const auto expected_id = request.at("id"); + + std::string buffered; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + while (std::chrono::steady_clock::now() < deadline) { + std::string chunk; + if (!client.read(chunk, 250, &ec)) { + continue; + } + buffered += chunk; + for (;;) { + const auto newline = buffered.find('\n'); + if (newline == std::string::npos) { + break; + } + const auto response = buffered.substr(0, newline); + buffered.erase(0, newline + 1); + nlohmann::json parsed; + try { + parsed = nlohmann::json::parse(response); + } catch (...) { + ADD_FAILURE() << "Invalid named-pipe response: " << response; + throw; + } + if (parsed.contains("id") && parsed.at("id") == expected_id) { + return parsed; + } + } + } + ADD_FAILURE() << "Buffered named-pipe response without matching JSON-RPC id: " << buffered; + throw std::runtime_error("Named-pipe JSON-RPC response was not received."); +} +#endif + } // namespace TEST(BridgeProtocolServerConfig, RoundTripsAndValidates) { @@ -248,6 +302,32 @@ TEST(BridgeProtocolServerConfig, RoundTripsAndValidates) { EXPECT_FALSE(unsafe.validate().first); } +TEST(BridgeProtocolNamedPipeConfig, RoundTripsAndValidates) { + namespace proto = optionx::bridges::protocol_v1; + + proto::BridgeProtocolNamedPipeConfig config = test_pipe_config(); + nlohmann::json json; + config.to_json(json); + + proto::BridgeProtocolNamedPipeConfig restored; + restored.from_json(json); + + EXPECT_TRUE(restored.validate().first); + EXPECT_EQ(restored.bridge_type(), optionx::BridgeType::BRIDGE_PROTOCOL_V1_NAMED_PIPE); + EXPECT_EQ(restored.named_pipe, config.named_pipe); + EXPECT_EQ(restored.bridge_id, 4u); + EXPECT_GT(restored.buffer_size, 0u); + EXPECT_GT(restored.pipe_timeout_ms, 0u); + EXPECT_GT(restored.max_jsonrpc_id_bytes, 0u); + EXPECT_GT(restored.max_idempotency_key_bytes, 0u); + EXPECT_GT(restored.max_operation_fingerprint_bytes, 0u); + EXPECT_GT(restored.max_operation_cache_bytes, 0u); + EXPECT_GT(restored.operation_cache_retention_ms, 0); + + restored.named_pipe.clear(); + EXPECT_FALSE(restored.validate().first); +} + TEST(BridgeProtocolServerBridge, AcceptsHttpJsonRpcCommands) { namespace proto = optionx::bridges::protocol_v1; @@ -326,6 +406,118 @@ TEST(BridgeProtocolServerBridge, AcceptsHttpJsonRpcCommands) { bridge.shutdown(); } +TEST(BridgeProtocolNamedPipeBridge, AcceptsJsonRpcCommands) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto config = test_pipe_config(); + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + std::mutex mutex; + std::condition_variable cv; + bool server_started = false; + std::string status_error; + std::atomic next_signal_id{50}; + std::atomic signal_count{0}; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + server_started = true; + } + if (update.status == optionx::BridgeStatus::SERVER_START_FAILED || + update.status == optionx::BridgeStatus::CONNECTION_ERROR) { + status_error = update.message; + } + cv.notify_all(); + }; + bridge.on_signal_id() = [&next_signal_id]() { + return next_signal_id.fetch_add(1); + }; + bridge.on_trade_signal() = [&signal_count](std::unique_ptr signal) { + ASSERT_EQ(signal->symbol, "EURUSD"); + ++signal_count; + }; + bridge.update_account_info(optionx::AccountInfoUpdate( + std::make_shared(), + optionx::AccountUpdateStatus::BALANCE_UPDATED)); + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return server_started || !status_error.empty(); + })) << "Named-pipe protocol bridge did not start."; + ASSERT_TRUE(status_error.empty()) << status_error; + } + + SimpleNamedPipe::ClientConfig client_config( + config.named_pipe, + config.buffer_size, + 3000); + SimpleNamedPipe::NamedPipeClient client(client_config); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + + const auto hello = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-hello"}, + {"method", "protocol.hello"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + + const auto capabilities = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-capabilities"}, + {"method", "protocol.capabilities.get"}, + {"params", nlohmann::json::object()} + }); + EXPECT_TRUE(capabilities.at("result").at("features").at("named_pipe").get()); + EXPECT_FALSE(capabilities.at("result").at("features").at("http").get()); + EXPECT_FALSE(capabilities.at("result").at("features").at("websocket").get()); + + const auto balance = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-balance"}, + {"method", "account.balance.get"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(balance.at("result").at("status").get(), "completed"); + EXPECT_EQ(balance.at("result").at("account").at("account_id").get(), "99"); + + const auto accepted = pipe_json(client, trade_command("pipe-trade", "pipe-idem")); + EXPECT_EQ(accepted.at("result").at("status").get(), "accepted"); + EXPECT_EQ(signal_count.load(), 1); + + const auto retry = pipe_json(client, trade_command("pipe-trade-retry", "pipe-idem")); + EXPECT_EQ( + retry.at("result").at("operation_id").get(), + accepted.at("result").at("operation_id").get()); + EXPECT_EQ(signal_count.load(), 1); + + const auto conflict = pipe_json( + client, + trade_command("pipe-trade-conflict", "pipe-idem", "GBPUSD")); + EXPECT_EQ(conflict.at("result").at("status").get(), "rejected"); + EXPECT_EQ( + conflict.at("result").at("reason").at("code").get(), + "idempotency_conflict"); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + TEST(BridgeProtocolServerBridge, ReturnsMethodSpecificAcceptedShapes) { namespace proto = optionx::bridges::protocol_v1; From fdf17514eb5a4d4f756e9be335538a3c031ce849 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 04:38:33 +0300 Subject: [PATCH 2/5] fix(bridges): harden protocol named pipe transport --- external/SimpleNamedPipe | 2 +- .../BridgeProtocolNamedPipeBridge.hpp | 133 ++++++++-- tests/protocol_v1_bridge_test.cpp | 234 ++++++++++++++++++ 3 files changed, 354 insertions(+), 15 deletions(-) diff --git a/external/SimpleNamedPipe b/external/SimpleNamedPipe index 422612a2..a19b4d85 160000 --- a/external/SimpleNamedPipe +++ b/external/SimpleNamedPipe @@ -1 +1 @@ -Subproject commit 422612a29e8514c3bb9fbf09970db1407f49ec4a +Subproject commit a19b4d850d52d51d5896ca9dfc5a837c112de328 diff --git a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp index 97f57a4a..3de78a88 100644 --- a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp +++ b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp @@ -21,6 +21,7 @@ namespace optionx::bridges::protocol_v1 { enum class RuntimePhase { Stopped, + Starting, Running, Stopping }; @@ -259,7 +260,8 @@ namespace optionx::bridges::protocol_v1 { # if defined(_WIN32) { std::unique_lock lock(m_state->mutex); - while (m_state->phase == RuntimePhase::Stopping) { + while (m_state->phase == RuntimePhase::Starting || + m_state->phase == RuntimePhase::Stopping) { m_state->lifecycle_cv.wait(lock); } if (m_state->phase == RuntimePhase::Running) { @@ -271,6 +273,7 @@ namespace optionx::bridges::protocol_v1 { config->named_pipe, config->buffer_size, config->pipe_timeout_ms); + server_config.write_limits.max_message_size = max_pipe_frame_bytes(*config); auto server = std::make_shared(server_config); configure_server_callbacks(server); @@ -279,7 +282,7 @@ namespace optionx::bridges::protocol_v1 { if (m_state->phase != RuntimePhase::Stopped) { return; } - m_state->phase = RuntimePhase::Running; + m_state->phase = RuntimePhase::Starting; m_state->server = server; m_state->client_ids.clear(); m_state->operation_order.clear(); @@ -297,7 +300,7 @@ namespace optionx::bridges::protocol_v1 { try { server->start(true); } catch (const std::exception& ex) { - clear_runtime_after_start_failure(); + clear_runtime_after_start_failure(server.get()); notify_status(BridgeStatus::SERVER_START_FAILED, {}, ex.what()); } # else @@ -500,6 +503,14 @@ namespace optionx::bridges::protocol_v1 { return message.dump(-1) + "\n"; } + static std::size_t max_pipe_frame_bytes( + const BridgeProtocolNamedPipeConfig& config) { + if (config.request_body_limit == (std::numeric_limits::max)()) { + return config.request_body_limit; + } + return config.request_body_limit + 1; + } + static bool is_valid_jsonrpc_id(const nlohmann::json& id) { return id.is_string() || id.is_null() || @@ -1406,6 +1417,7 @@ namespace optionx::bridges::protocol_v1 { # if defined(_WIN32) void configure_server_callbacks( const std::shared_ptr& server) { + auto* const server_identity = server.get(); server->on_connected = [state = m_state](int client_id) { TransportCallbackScope scope(state); if (!scope.admitted()) { @@ -1431,8 +1443,17 @@ namespace optionx::bridges::protocol_v1 { connection_id(client_id)); }; - server->on_start = [state = m_state](const SimpleNamedPipe::ServerConfig&) { - CallbackScope scope(state); + server->on_start = [state = m_state, server_identity]( + const SimpleNamedPipe::ServerConfig&) { + { + std::lock_guard lock(state->mutex); + if (state->server.get() != server_identity || + state->phase != RuntimePhase::Starting) { + return; + } + state->phase = RuntimePhase::Running; + state->lifecycle_cv.notify_all(); + } notify_status_from_state(state, BridgeStatus::SERVER_STARTED, "pipe"); }; @@ -1455,13 +1476,59 @@ namespace optionx::bridges::protocol_v1 { handle_message(client_id, message); }; - server->on_error = [state = m_state](const std::error_code& ec) { - CallbackScope scope(state); + server->on_error = [state = m_state, server_identity](const std::error_code& ec) { + bool start_failed = false; + std::shared_ptr server_to_finalize; + { + std::lock_guard lock(state->mutex); + if (state->server.get() != server_identity) { + return; + } + if (state->phase == RuntimePhase::Starting) { + start_failed = true; + state->phase = RuntimePhase::Stopping; + state->transport_callback_admission_closed = true; + server_to_finalize = state->server; + state->server.reset(); + state->client_ids.clear(); + state->pending_callback_shutdown = false; + } + } notify_status_from_state( state, - BridgeStatus::CONNECTION_ERROR, + start_failed ? BridgeStatus::SERVER_START_FAILED + : BridgeStatus::CONNECTION_ERROR, {}, ec.message()); + if (server_to_finalize) { + std::thread([state, server = std::move(server_to_finalize)]() mutable { + finalize_shutdown(state, std::move(server)); + }).detach(); + } + }; + + server->on_stop = [state = m_state, server_identity]( + const SimpleNamedPipe::ServerConfig&) { + bool should_notify = false; + { + std::lock_guard lock(state->mutex); + if (state->server.get() != server_identity) { + return; + } + state->server.reset(); + state->client_ids.clear(); + state->transport_callback_admission_closed = false; + state->pending_callback_shutdown = false; + if (state->phase != RuntimePhase::Stopped) { + state->phase = RuntimePhase::Stopped; + should_notify = !state->stop_notified; + state->stop_notified = true; + } + state->lifecycle_cv.notify_all(); + } + if (should_notify) { + notify_status_from_state(state, BridgeStatus::SERVER_STOPPED, {}); + } }; } @@ -1480,41 +1547,79 @@ namespace optionx::bridges::protocol_v1 { return; } if (!scope.admitted()) { - send_pipe_text(server, client_id, frame_message(server_stopping_response())); + auto config = get_config(); + if (config) { + send_pipe_text( + m_state, + *config, + server, + client_id, + frame_message(server_stopping_response())); + } return; } auto config = get_config_or_throw(); const auto response = frame_message(handle_message_body(*config, message)); - send_pipe_text(server, client_id, response); + send_pipe_text(m_state, *config, server, client_id, response); } void broadcast_notification(nlohmann::json notification) { std::shared_ptr server; std::vector clients; + std::shared_ptr config; { std::lock_guard lock(m_state->mutex); server = m_state->server; + config = m_state->config; clients.assign(m_state->client_ids.begin(), m_state->client_ids.end()); } - if (!server) { + if (!server || !config) { return; } const auto text = frame_message(std::move(notification)); for (const int client_id : clients) { - send_pipe_text(server, client_id, text); + send_pipe_text(m_state, *config, server, client_id, text); } } static void send_pipe_text( + const std::shared_ptr& state, + const BridgeProtocolNamedPipeConfig& config, const std::shared_ptr& server, const int client_id, const std::string& text) { - server->send_to(client_id, text); + if (text.size() > max_pipe_frame_bytes(config)) { + notify_status_from_state( + state, + BridgeStatus::CONNECTION_ERROR, + connection_id(client_id), + "Named-pipe message exceeds configured protocol frame limit."); + return; + } + + server->send_to( + client_id, + text, + [state, client_id](const std::error_code& ec) { + if (!ec) { + return; + } + notify_status_from_state( + state, + BridgeStatus::CONNECTION_ERROR, + connection_id(client_id), + ec.message()); + }); } - void clear_runtime_after_start_failure() { + void clear_runtime_after_start_failure( + const SimpleNamedPipe::NamedPipeServer* server_identity) { std::lock_guard lock(m_state->mutex); + if (m_state->server.get() != server_identity || + m_state->phase != RuntimePhase::Starting) { + return; + } m_state->server.reset(); m_state->client_ids.clear(); m_state->phase = RuntimePhase::Stopped; diff --git a/tests/protocol_v1_bridge_test.cpp b/tests/protocol_v1_bridge_test.cpp index f3828fa3..4a234897 100644 --- a/tests/protocol_v1_bridge_test.cpp +++ b/tests/protocol_v1_bridge_test.cpp @@ -208,6 +208,28 @@ nlohmann::json post_json_without_auth( } #if defined(_WIN32) +nlohmann::json read_pipe_json( + SimpleNamedPipe::NamedPipeClient& client, + const std::chrono::seconds timeout = std::chrono::seconds(3)) { + std::error_code ec; + std::string buffered; + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) { + std::string chunk; + if (!client.read(chunk, 250, &ec)) { + continue; + } + buffered += chunk; + const auto newline = buffered.find('\n'); + if (newline == std::string::npos) { + continue; + } + return nlohmann::json::parse(buffered.substr(0, newline)); + } + ADD_FAILURE() << "Named-pipe JSON frame was not received. buffered=" << buffered; + throw std::runtime_error("Named-pipe JSON frame was not received."); +} + nlohmann::json pipe_json( SimpleNamedPipe::NamedPipeClient& client, const nlohmann::json& request) { @@ -518,6 +540,218 @@ TEST(BridgeProtocolNamedPipeBridge, AcceptsJsonRpcCommands) { #endif } +TEST(BridgeProtocolNamedPipeBridge, HandlesFramesLargerThanTransportBuffer) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto config = test_pipe_config(); + config.buffer_size = 64; + config.request_body_limit = 64 * 1024; + + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + std::mutex mutex; + std::condition_variable cv; + bool server_started = false; + std::string status_error; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + server_started = true; + } + if (update.status == optionx::BridgeStatus::SERVER_START_FAILED || + update.status == optionx::BridgeStatus::CONNECTION_ERROR) { + status_error = update.message; + } + cv.notify_all(); + }; + bridge.on_signal_id() = []() { return optionx::SignalId{75}; }; + bridge.on_trade_signal() = [](std::unique_ptr) {}; + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return server_started || !status_error.empty(); + })); + ASSERT_TRUE(status_error.empty()) << status_error; + } + + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(config.named_pipe, 37, 3000)); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + + auto large_hello = nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-large-hello"}, + {"method", "protocol.hello"}, + {"params", { + {"requested_protocol_versions", nlohmann::json::array({"1"})}, + {"padding", std::string(2048, 'p')} + }} + }; + ASSERT_GT(large_hello.dump(-1).size(), config.buffer_size); + const auto hello = pipe_json(client, large_hello); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + EXPECT_GT(hello.dump(-1).size(), config.buffer_size); + + bridge.update_account_info(optionx::AccountInfoUpdate( + std::make_shared(), + optionx::AccountUpdateStatus::BALANCE_UPDATED)); + const auto notification = read_pipe_json(client); + ASSERT_TRUE(notification.contains("method")) << notification.dump(-1); + EXPECT_EQ(notification.at("method").get(), "balance.updated"); + EXPECT_GT(notification.dump(-1).size(), config.buffer_size); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + +TEST(BridgeProtocolNamedPipeBridge, AsyncStartupFailureReturnsToStoppedAndAllowsRestart) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto bad_config = test_pipe_config(); + bad_config.named_pipe = std::string(300, 'x'); + + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(bad_config))); + + std::mutex mutex; + std::condition_variable cv; + bool start_failed = false; + bool server_started = false; + std::string status_error; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_START_FAILED) { + start_failed = true; + status_error = update.message; + } + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + server_started = true; + } + cv.notify_all(); + }; + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { + return start_failed; + })) << "Expected async named-pipe startup failure."; + } + EXPECT_FALSE(status_error.empty()); + + auto good_config = test_pipe_config(); + ASSERT_TRUE(bridge.configure(std::make_unique(good_config))); + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return server_started; + })) << "Bridge did not restart after async startup failure."; + } + + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(good_config.named_pipe, good_config.buffer_size, 3000)); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + const auto hello = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-restart-hello"}, + {"method", "protocol.hello"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + +TEST(BridgeProtocolNamedPipeBridge, ShutdownFromServerStartedAllowsRestart) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto config = test_pipe_config(); + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + std::mutex mutex; + std::condition_variable cv; + int started_count = 0; + bool stopped = false; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + bool request_shutdown = false; + { + std::lock_guard lock(mutex); + ++started_count; + request_shutdown = started_count == 1; + } + cv.notify_all(); + if (request_shutdown) { + bridge.shutdown(); + } + return; + } + if (update.status == optionx::BridgeStatus::SERVER_STOPPED) { + std::lock_guard lock(mutex); + stopped = true; + cv.notify_all(); + } + }; + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { + return stopped; + })); + } + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return started_count >= 2; + })); + } + + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(config.named_pipe, config.buffer_size, 3000)); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + const auto hello = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-restarted-after-status-shutdown"}, + {"method", "protocol.hello"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + TEST(BridgeProtocolServerBridge, ReturnsMethodSpecificAcceptedShapes) { namespace proto = optionx::bridges::protocol_v1; From 08387c249b5cd852f37c23796f42c67504ffde11 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 06:00:31 +0300 Subject: [PATCH 3/5] fix(bridges): finalize named pipe stop safely --- external/SimpleNamedPipe | 2 +- .../BridgeProtocolNamedPipeBridge.hpp | 46 ++++-- tests/protocol_v1_bridge_test.cpp | 139 ++++++++++++++++++ 3 files changed, 172 insertions(+), 15 deletions(-) diff --git a/external/SimpleNamedPipe b/external/SimpleNamedPipe index a19b4d85..59515c49 160000 --- a/external/SimpleNamedPipe +++ b/external/SimpleNamedPipe @@ -1 +1 @@ -Subproject commit a19b4d850d52d51d5896ca9dfc5a837c112de328 +Subproject commit 59515c49af9fbf168f45882345332cfa5c07e5dc diff --git a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp index 3de78a88..75ae8921 100644 --- a/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp +++ b/include/optionx_cpp/bridges/protocol_v1/BridgeProtocolNamedPipeBridge.hpp @@ -159,6 +159,24 @@ namespace optionx::bridges::protocol_v1 { shutdown(); } +#ifdef OPTIONX_ENABLE_BRIDGE_PROTOCOL_TEST_HOOKS +# if defined(_WIN32) + /// \brief Stops the underlying server without going through bridge shutdown. + void simulate_unexpected_server_stop_for_test() { + std::shared_ptr server; + { + std::lock_guard lock(m_state->mutex); + server = m_state->server; + } + if (server) { + std::thread([server = std::move(server)]() mutable { + server->stop(); + }).detach(); + } + } +# endif +#endif + /// \brief Configures the bridge with Bridge Protocol v1 named-pipe settings. bool configure(std::unique_ptr config) override { if (!config) return false; @@ -1509,25 +1527,23 @@ namespace optionx::bridges::protocol_v1 { server->on_stop = [state = m_state, server_identity]( const SimpleNamedPipe::ServerConfig&) { - bool should_notify = false; + std::shared_ptr server_to_finalize; { std::lock_guard lock(state->mutex); if (state->server.get() != server_identity) { return; } + state->phase = RuntimePhase::Stopping; + state->transport_callback_admission_closed = true; + server_to_finalize = state->server; state->server.reset(); state->client_ids.clear(); - state->transport_callback_admission_closed = false; state->pending_callback_shutdown = false; - if (state->phase != RuntimePhase::Stopped) { - state->phase = RuntimePhase::Stopped; - should_notify = !state->stop_notified; - state->stop_notified = true; - } - state->lifecycle_cv.notify_all(); } - if (should_notify) { - notify_status_from_state(state, BridgeStatus::SERVER_STOPPED, {}); + if (server_to_finalize) { + std::thread([state, server = std::move(server_to_finalize)]() mutable { + finalize_shutdown(state, std::move(server)); + }).detach(); } }; } @@ -1633,9 +1649,6 @@ namespace optionx::bridges::protocol_v1 { auto server = state->server; state->server.reset(); state->client_ids.clear(); - if (!state->stop_notified) { - state->stop_notified = true; - } return server; } @@ -1663,14 +1676,19 @@ namespace optionx::bridges::protocol_v1 { if (server) { server->stop(); } - notify_status_from_state(state, BridgeStatus::SERVER_STOPPED, {}); + bool should_notify = false; { std::lock_guard lock(state->mutex); + should_notify = !state->stop_notified; + state->stop_notified = true; state->phase = RuntimePhase::Stopped; state->transport_callback_admission_closed = false; state->pending_callback_shutdown = false; state->lifecycle_cv.notify_all(); } + if (should_notify) { + notify_status_from_state(state, BridgeStatus::SERVER_STOPPED, {}); + } } # else void broadcast_notification(nlohmann::json notification) { diff --git a/tests/protocol_v1_bridge_test.cpp b/tests/protocol_v1_bridge_test.cpp index 4a234897..4aa4345c 100644 --- a/tests/protocol_v1_bridge_test.cpp +++ b/tests/protocol_v1_bridge_test.cpp @@ -752,6 +752,145 @@ TEST(BridgeProtocolNamedPipeBridge, ShutdownFromServerStartedAllowsRestart) { #endif } +TEST(BridgeProtocolNamedPipeBridge, UnexpectedServerStopAllowsRestart) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto config = test_pipe_config(); + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + std::mutex mutex; + std::condition_variable cv; + int started_count = 0; + int stopped_count = 0; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + ++started_count; + } + if (update.status == optionx::BridgeStatus::SERVER_STOPPED) { + ++stopped_count; + } + cv.notify_all(); + }; + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return started_count == 1; + })); + } + + bridge.simulate_unexpected_server_stop_for_test(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { + return stopped_count == 1; + })); + } + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return started_count == 2; + })); + } + + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(config.named_pipe, config.buffer_size, 3000)); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + const auto hello = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-restarted-after-unexpected-stop"}, + {"method", "protocol.hello"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + +TEST(BridgeProtocolNamedPipeBridge, ServerStoppedCallbackCanRestart) { + namespace proto = optionx::bridges::protocol_v1; + +#if defined(_WIN32) + auto config = test_pipe_config(); + proto::BridgeProtocolNamedPipeBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + + std::mutex mutex; + std::condition_variable cv; + int started_count = 0; + bool restart_requested = false; + + bridge.on_status_update() = [&](const optionx::BridgeStatusUpdate& update) { + bool should_restart = false; + { + std::lock_guard lock(mutex); + if (update.status == optionx::BridgeStatus::SERVER_STARTED) { + ++started_count; + } + if (update.status == optionx::BridgeStatus::SERVER_STOPPED && + !restart_requested) { + restart_requested = true; + should_restart = true; + } + } + cv.notify_all(); + if (should_restart) { + bridge.run(); + cv.notify_all(); + } + }; + + bridge.run(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(3), [&] { + return started_count == 1; + })); + } + + bridge.shutdown(); + { + std::unique_lock lock(mutex); + ASSERT_TRUE(cv.wait_for(lock, std::chrono::seconds(5), [&] { + return started_count == 2; + })); + } + + SimpleNamedPipe::NamedPipeClient client( + SimpleNamedPipe::ClientConfig(config.named_pipe, config.buffer_size, 3000)); + std::error_code ec; + ASSERT_TRUE(client.connect(&ec)) << ec.message(); + const auto hello = pipe_json( + client, + nlohmann::json{ + {"jsonrpc", "2.0"}, + {"id", "pipe-restarted-from-stopped-callback"}, + {"method", "protocol.hello"}, + {"params", nlohmann::json::object()} + }); + EXPECT_EQ(hello.at("result").at("selected_protocol_version").get(), "1"); + + client.close(); + bridge.shutdown(); +#else + GTEST_SKIP() << "Bridge Protocol v1 named-pipe transport is Windows-only."; +#endif +} + TEST(BridgeProtocolServerBridge, ReturnsMethodSpecificAcceptedShapes) { namespace proto = optionx::bridges::protocol_v1; From 3019f74a9bace753b98522054d9f19b7dbe3a81e Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 13:01:06 +0300 Subject: [PATCH 4/5] chore(deps): update SimpleNamedPipe test stub fix --- external/SimpleNamedPipe | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/external/SimpleNamedPipe b/external/SimpleNamedPipe index 59515c49..e3810acc 160000 --- a/external/SimpleNamedPipe +++ b/external/SimpleNamedPipe @@ -1 +1 @@ -Subproject commit 59515c49af9fbf168f45882345332cfa5c07e5dc +Subproject commit e3810accb9bf5910ec881383876ab836d47c3829 From 7e07cf625ed2b8c45f0f1650786effb30cefd231 Mon Sep 17 00:00:00 2001 From: Aster Seker Date: Wed, 22 Jul 2026 02:51:09 +0300 Subject: [PATCH 5/5] test(bridges): cover canonical idempotency aliases --- tests/protocol_v1_bridge_test.cpp | 92 +++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) diff --git a/tests/protocol_v1_bridge_test.cpp b/tests/protocol_v1_bridge_test.cpp index 4aa4345c..caa24d6d 100644 --- a/tests/protocol_v1_bridge_test.cpp +++ b/tests/protocol_v1_bridge_test.cpp @@ -1656,6 +1656,98 @@ TEST(BridgeProtocolServerBridge, CanonicalizesEquivalentDecimalRepresentations) EXPECT_EQ(signal_count.load(), 2); } +TEST(BridgeProtocolServerBridge, CanonicalizesBusinessAliasesAndIdentifiersInFingerprint) { + namespace proto = optionx::bridges::protocol_v1; + + auto config = test_config(); + config.enable_websocket = false; + + proto::BridgeProtocolServerBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { return optionx::SignalId{59}; }; + std::atomic signal_count{0}; + bridge.on_trade_signal() = [&signal_count](std::unique_ptr) { + ++signal_count; + }; + + bridge.run(); + ASSERT_TRUE(wait_for_http_port(bridge)); + + auto first_command = trade_command("alias-a", "idem-alias"); + first_command["params"]["identity"]["unique_id"] = 123; + first_command["params"]["trade"].erase("order_type"); + first_command["params"]["trade"]["direction"] = "buy"; + first_command["params"]["trade"]["option_type"] = "sprint"; + first_command["params"]["trade"]["amount"] = "1.00"; + first_command["params"]["trade"]["currency"] = "usd"; + first_command["params"]["trade"].erase("expiry"); + first_command["params"]["trade"]["duration_sec"] = 60; + + auto second_command = trade_command("alias-b", "idem-alias"); + second_command["params"]["identity"]["unique_id"] = "123"; + second_command["params"]["trade"]["order_type"] = "BUY"; + second_command["params"]["trade"]["option_type"] = "SPRINT"; + second_command["params"]["trade"]["amount"] = { + {"value", 1}, + {"currency", "USD"} + }; + second_command["params"]["trade"]["expiry"] = { + {"kind", "duration"}, + {"duration_ms", "60000"} + }; + + const auto first = post_json(config, bridge.bound_http_port(), first_command); + const auto second = post_json(config, bridge.bound_http_port(), second_command); + bridge.shutdown(); + + EXPECT_EQ(first.at("result").at("status").get(), "accepted"); + EXPECT_EQ(second.at("result").at("status").get(), "accepted"); + EXPECT_EQ( + second.at("result").at("operation_id").get(), + first.at("result").at("operation_id").get()); + EXPECT_EQ(signal_count.load(), 1); +} + +TEST(BridgeProtocolServerBridge, KeepsDurationAndAbsoluteExpiryDistinctInFingerprint) { + namespace proto = optionx::bridges::protocol_v1; + + auto config = test_config(); + config.enable_websocket = false; + + proto::BridgeProtocolServerBridge bridge; + ASSERT_TRUE(bridge.configure(std::make_unique(config))); + bridge.on_signal_id() = []() { return optionx::SignalId{60}; }; + std::atomic signal_count{0}; + bridge.on_trade_signal() = [&signal_count](std::unique_ptr) { + ++signal_count; + }; + + bridge.run(); + ASSERT_TRUE(wait_for_http_port(bridge)); + + auto duration_command = trade_command("expiry-kind-a", "idem-expiry-kind"); + duration_command["params"]["trade"]["expiry"] = { + {"kind", "duration"}, + {"duration_ms", 60000} + }; + auto absolute_command = trade_command("expiry-kind-b", "idem-expiry-kind"); + absolute_command["params"]["trade"]["expiry"] = { + {"kind", "absolute"}, + {"expires_at_ms", optionx::bridges::metatrader_file::detail::unix_time_ms() + 60000} + }; + + const auto first = post_json(config, bridge.bound_http_port(), duration_command); + const auto second = post_json(config, bridge.bound_http_port(), absolute_command); + bridge.shutdown(); + + EXPECT_EQ(first.at("result").at("status").get(), "accepted"); + EXPECT_EQ(second.at("result").at("status").get(), "rejected"); + EXPECT_EQ( + second.at("result").at("reason").at("code").get(), + "idempotency_conflict"); + EXPECT_EQ(signal_count.load(), 1); +} + TEST(BridgeProtocolServerBridge, CanonicalizesRoutingPlatformTypeInFingerprint) { namespace proto = optionx::bridges::protocol_v1;