diff --git a/.bdignore b/.bdignore new file mode 100644 index 0000000..5e06ba1 --- /dev/null +++ b/.bdignore @@ -0,0 +1,3 @@ +# BlackDuck scan exclusions +# Vendored third-party source — attributed in NOTICE and LICENSE files +src/vendor/ diff --git a/.github/skills/openspec-coverage/openspec_coverage.py b/.github/skills/openspec-coverage/openspec_coverage.py index 3d9cef9..704e6af 100644 --- a/.github/skills/openspec-coverage/openspec_coverage.py +++ b/.github/skills/openspec-coverage/openspec_coverage.py @@ -231,10 +231,15 @@ def extract_methods_from_code(file_path): method_pattern = re.compile( r"^\s*(?:[\w:<>,~]+\s+)+([\w:~]+)\s*\([^)]*\)\s*(const)?\s*[{;]" ) + conversion_operator_pattern = re.compile(r"\boperator\s+[^\s(]+\s*\(") methods = set() try: with open(file_path, encoding="utf-8", errors="ignore") as f: for line in f: + # Skip conversion operators (e.g., 'explicit operator bool() const'), + # which do not have a standalone method identifier to report. + if conversion_operator_pattern.search(line): + continue m = method_pattern.match(line) if m: methods.add(m.group(1)) diff --git a/.version b/.version new file mode 100644 index 0000000..18efdb9 --- /dev/null +++ b/.version @@ -0,0 +1 @@ +1.1.8 diff --git a/LICENSE b/LICENSE index ad69171..31a5bb5 100644 --- a/LICENSE +++ b/LICENSE @@ -202,25 +202,3 @@ limitations under the License. =========================================================================== - -MIT License - -Copyright (c) YEAR COPYRIGHT HOLDER - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. diff --git a/README.md b/README.md index b636394..7d1f536 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,7 @@ if (value) { - If the header is present, its value is returned. - If the header is not present, `std::nullopt` is returned. +- Header-name lookup is case-sensitive; use the same capitalization as received from the server. #### Thread Safety Header operations are thread-safe. Access to response headers is protected by a mutex internally. diff --git a/build.sh b/build.sh index fe04cd1..1ea0868 100755 --- a/build.sh +++ b/build.sh @@ -18,6 +18,39 @@ set -e +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +IMAGE="firebolt-cpp-transport-ci:local" + +# Pre-scan for --docker before arg parsing so forwarded args are preserved +use_docker=false +_forward_args=() +for _arg in "$@"; do + [[ "$_arg" == "--docker" ]] && { use_docker=true; continue; } + _forward_args+=("$_arg") +done + +if $use_docker; then + for _bdir in build build-dev; do + _cache="$SCRIPT_DIR/$_bdir/CMakeCache.txt" + if [[ -f "$_cache" ]]; then + _cached=$(grep '^CMAKE_HOME_DIRECTORY' "$_cache" 2>/dev/null | cut -d= -f2 || true) + # In Docker, the workspace is mounted at /workspace. Only wipe if the + # cache was configured for a different environment (e.g. a native host build). + if [[ -n "$_cached" && "$_cached" != "/workspace" ]]; then + echo "Wiping stale $_bdir (configured at $_cached, expected /workspace)..." + rm -rf "$SCRIPT_DIR/$_bdir" + fi + fi + done + if ! docker image inspect "$IMAGE" &>/dev/null; then + echo "Building CI Docker image (one-time)..." + docker build -t "$IMAGE" -f "$SCRIPT_DIR/.github/Dockerfile" "$SCRIPT_DIR" + fi + exec docker run --rm --user "$(id -u):$(id -g)" \ + -v "$SCRIPT_DIR:/workspace" -w /workspace \ + "$IMAGE" ./build.sh "${_forward_args[@]}" +fi + bdir="build" do_install=false params= @@ -50,12 +83,21 @@ while [[ ! -z $1 ]]; do esac; shift done -[[ ! -z $SYSROOT_PATH ]] || { echo "SYSROOT_PATH not set" >/dev/stderr; exit 1; } -[[ -e $SYSROOT_PATH ]] || { echo "SYSROOT_PATH not exist ($SYSROOT_PATH)" >/dev/stderr; exit 1; } +SYSROOT_PATH="${SYSROOT_PATH:-/}" +[[ -e $SYSROOT_PATH ]] || { echo "SYSROOT_PATH does not exist ($SYSROOT_PATH)" >/dev/stderr; exit 1; } $cleanFirst && rm -rf $bdir -if [[ ! -e "$bdir" || -n "$@" ]]; then +_cache="$SCRIPT_DIR/$bdir/CMakeCache.txt" +if [[ -f "$_cache" ]]; then + _cached=$(grep '^CMAKE_HOME_DIRECTORY' "$_cache" 2>/dev/null | cut -d= -f2 || true) + if [[ -n "$_cached" && "$_cached" != "$SCRIPT_DIR" ]]; then + echo "Wiping stale $bdir (configured at $_cached, now at $SCRIPT_DIR)..." + rm -rf "$SCRIPT_DIR/$bdir" + fi +fi + +if [[ ! -f "$bdir/CMakeCache.txt" || $# -gt 0 ]]; then params+=" -DCMAKE_EXPORT_COMPILE_COMMANDS=ON" command -v ccache >/dev/null 2>&1 && params+=" -DCMAKE_C_COMPILER_LAUNCHER=ccache -DCMAKE_CXX_COMPILER_LAUNCHER=ccache" cmake -B $bdir \ diff --git a/fmt.sh b/fmt.sh index cbd6973..0215d33 100755 --- a/fmt.sh +++ b/fmt.sh @@ -15,9 +15,9 @@ fi RUN="docker run --rm --user $(id -u):$(id -g) -v $SCRIPT_DIR:/workspace $IMAGE bash -c" if [[ "${1:-}" == "--fix" ]]; then - $RUN "find /workspace/src /workspace/include /workspace/test -type f \( -name '*.cpp' -o -name '*.h' \) | xargs clang-format -i" + $RUN "find /workspace/src /workspace/include /workspace/test -type f \( -name '*.cpp' -o -name '*.h' \) -print0 | xargs -0 clang-format -i" echo "Done. Files reformatted." else - $RUN "find /workspace/src /workspace/include /workspace/test -type f \( -name '*.cpp' -o -name '*.h' \) | xargs clang-format --dry-run --Werror" + $RUN "find /workspace/src /workspace/include /workspace/test -type f \( -name '*.cpp' -o -name '*.h' \) -print0 | xargs -0 clang-format --dry-run --Werror" echo "Formatting OK." fi diff --git a/include/firebolt/config.h.in b/include/firebolt/config.h.in index 8a374cc..665e33d 100644 --- a/include/firebolt/config.h.in +++ b/include/firebolt/config.h.in @@ -77,6 +77,45 @@ struct Config /** Watchdog polling cycle in milliseconds. Default: 500. */ unsigned watchdogCycle_ms = 500; + + /** + * @brief Retry policy for the initial gateway connection. + * + * On embedded devices the Firebolt client may start before the gateway + * daemon is ready (poor systemd ordering) or before the loopback interface + * has an IPv4 address assigned. Setting reconnect_max_attempts > 0 causes + * connect() to retry up to that many times, waiting reconnect_delay_ms + * between each attempt, before returning an error to the caller. + * + * Each attempt is bounded by connect_attempt_timeout_ms: if the underlying + * transport (DNS + TCP + WebSocket handshake) does not complete within that + * window, the attempt is aborted and counted as a failure. connect() + * therefore blocks for at most: + * (reconnect_max_attempts + 1) * connect_attempt_timeout_ms + * + reconnect_max_attempts * reconnect_delay_ms + * + * Capped internally at 100 retries regardless of this value. + * + * Default: 0 (single attempt, no retry; connect() returns the last + * transport error, e.g. Error::NotConnected, Error::Timedout, or + * Error::General, or Error::NotConnected on per-attempt timeout). + */ + unsigned reconnect_max_attempts = 0; + + /** Milliseconds to wait between reconnect attempts. Default: 1000. */ + unsigned reconnect_delay_ms = 1000; + + /** + * @brief Per-attempt timeout for the initial connection handshake (ms). + * + * If DNS resolution + TCP connect + WebSocket handshake do not complete + * within this window, the in-flight attempt is aborted and connect() + * either retries (if reconnect_max_attempts > 0) or returns + * Error::NotConnected to the caller. + * + * Default: 10000 (10 seconds). + */ + unsigned connect_attempt_timeout_ms = 10000; }; } // namespace Firebolt diff --git a/include/firebolt/gateway.h b/include/firebolt/gateway.h index b607745..0929351 100644 --- a/include/firebolt/gateway.h +++ b/include/firebolt/gateway.h @@ -24,6 +24,7 @@ #include #include #include +#include #include namespace Firebolt::Transport @@ -36,6 +37,9 @@ class IGateway public: virtual ~IGateway(); + // NOTE: onConnectionChange is invoked once with the final result on the + // connect() calling thread. All subsequent callbacks (disconnect, watchdog + // reconnect) fire on the websocketpp IO thread. Callers must be thread-safe. virtual Firebolt::Error connect(const Firebolt::Config& config, ConnectionChangeCallback onConnectionChange) = 0; virtual Firebolt::Error disconnect() = 0; diff --git a/presentations/openspec-with-copliot-experience/package.json b/presentations/openspec-with-copilot-experience/package.json similarity index 88% rename from presentations/openspec-with-copliot-experience/package.json rename to presentations/openspec-with-copilot-experience/package.json index b82945f..1defc59 100644 --- a/presentations/openspec-with-copliot-experience/package.json +++ b/presentations/openspec-with-copilot-experience/package.json @@ -1,5 +1,5 @@ { - "name": "openspec-with-copliot-experience", + "name": "openspec-with-copilot-experience", "private": true, "scripts": { "dev": "slidev --open", diff --git a/presentations/openspec-with-copliot-experience/slides.md b/presentations/openspec-with-copilot-experience/slides.md similarity index 97% rename from presentations/openspec-with-copliot-experience/slides.md rename to presentations/openspec-with-copilot-experience/slides.md index 4cf34a3..5b8ebff 100644 --- a/presentations/openspec-with-copliot-experience/slides.md +++ b/presentations/openspec-with-copilot-experience/slides.md @@ -278,10 +278,10 @@ https://github.com/rdkcentral/firebolt-cpp-transport/tree/feature/openspec - Current slides are Auto Generated from OpenSpec and Copilot chat conversations. - Multiple Reusable Skills generated for creating slides based on the effort ```sh - $ /init-slidev-presentation "openspec-with-copliot-experience" --content "slides.md" - $ /add-slide "openspec-with-copliot-experience" --content "slides.md" --slideTitle "Slide Title" - $ /add-diagram "openspec-with-copliot-experience" --content "slides.md" --diagramCode "mermaid code here" - $ /export-slidev-presentation "openspec-with-copliot-experience" --format pdf + $ /init-slidev-presentation "openspec-with-copilot-experience" --content "slides.md" + $ /add-slide "openspec-with-copilot-experience" --content "slides.md" --slideTitle "Slide Title" + $ /add-diagram "openspec-with-copilot-experience" --content "slides.md" --diagramCode "mermaid code here" + $ /export-slidev-presentation "openspec-with-copilot-experience" --format pdf ``` --- diff --git a/presentations/transport-slides/slides.md b/presentations/transport-slides/slides.md index 7324d1b..4b2cf45 100644 --- a/presentations/transport-slides/slides.md +++ b/presentations/transport-slides/slides.md @@ -132,76 +132,3 @@ layout: center # Thank You! [GitHub](https://github.com/rdkcentral/firebolt-cpp-transport) - ---- - -# Connection Protocol - -The Firebolt Transport Layer uses WebSocket as its underlying transport, with JSON-RPC layered on top for structured messaging and event notifications. - -**Message Flow:** - -``` -┌─────────────┐ WebSocket ┌─────────────┐ -│ Client │<------------------->│ Gateway │ -└─────────────┘ JSON-RPC └─────────────┘ -``` - -- Client connects to a configurable WebSocket URL (default: ws://127.0.0.1:9998) -- Connection state managed via callbacks (e.g., ConnectionChangeCallback) -- Supports reconnect, disconnect, and error handling -- All requests, responses, and events are JSON-RPC messages -- Legacy support via `legacyRPCv1` flag - ---- - -# Integration Points - -- Standalone library for JSON-RPC over WebSocket -- Agnostic to client usage and WebSocket implementation -- API schemas and business logic reside in separate repos (e.g., firebolt-cpp-client) -- IGateway and IHelper interfaces are designed for external use -- Initialization/configuration flows for Gateway are an open requirement - -**Conclusion:** -The transport library is decoupled from client and schema logic, providing a generic, extensible foundation for JSON-RPC over WebSocket. Integration patterns and initialization flows should be defined to support external consumers and SDKs. ---- - -# Firebolt C++ Transport - -A modern, efficient transport layer for Firebolt. - ---- - -# Project Overview - -- C++ implementation -- Modular design -- JSON-RPC support -- Logging and diagnostics - ---- - -# Architecture - -- Gateway -- Helpers -- Transport core -- Utilities - ---- - -# Key Features - -- High performance -- Extensible -- Easy integration -- Comprehensive unit tests - ---- -layout: center ---- - -# Thank You! - -[GitHub](https://github.com/rdkcentral/firebolt-cpp-transport) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index f0045ea..230e8cc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -82,6 +82,9 @@ target_link_libraries(${TARGET} target_include_directories(${TARGET} PRIVATE + # vendor/ must come before the system websocketpp headers so that + # our patched endpoint.hpp (numeric-IP resolver fix) takes priority. + $ $ $ PUBLIC diff --git a/src/gateway.cpp b/src/gateway.cpp index 21d2cf0..9a1e028 100644 --- a/src/gateway.cpp +++ b/src/gateway.cpp @@ -486,6 +486,7 @@ class GatewayImpl : public IGateway, private IClientTransport { private: ConnectionChangeCallback connectionChangeListener; + std::mutex connectionChangeListener_mtx; Transport transport; Client client; Server server; @@ -534,10 +535,16 @@ class GatewayImpl : public IGateway, private IClientTransport Firebolt::Logger::setFormat(cfg.log.format.ts, cfg.log.format.location, cfg.log.format.function, cfg.log.format.thread); - connectionChangeListener = onConnectionChange; + ConnectionChangeCallback previousConnectionChangeListener; + { + std::lock_guard lock(connectionChangeListener_mtx); + previousConnectionChangeListener = connectionChangeListener; + connectionChangeListener = std::move(onConnectionChange); + } runtime_waitTime_ms = cfg.waitTime_ms; legacyRPCv1 = cfg.legacyRPCv1; + watchdog_interval_ms = cfg.watchdogCycle_ms; FIREBOLT_LOG_NOTICE("Gateway", "[connect] config waitTime_ms=%u watchdog_interval_ms=%u headers=%zu", runtime_waitTime_ms, watchdog_interval_ms, cfg.headers.size()); @@ -579,12 +586,14 @@ class GatewayImpl : public IGateway, private IClientTransport } FIREBOLT_LOG_NOTICE("Gateway", "Connecting to url = %s", safeConnectUrl.c_str()); Firebolt::Error status = transport.connect( - url, [this](const nlohmann::json& message) { this->onMessage(message); }, + std::move(url), [this](const nlohmann::json& message) { this->onMessage(message); }, [this](const bool connected, Firebolt::Error error) { this->onConnectionChange(connected, error); }, transportLoggingInclude, transportLoggingExclude, cfg.headers); if (status != Firebolt::Error::None) { + std::lock_guard lock(connectionChangeListener_mtx); + connectionChangeListener = std::move(previousConnectionChangeListener); FIREBOLT_LOG_ERROR("Gateway", "[connect] transport connect failed status=%d", static_cast(status)); return status; } @@ -810,17 +819,41 @@ class GatewayImpl : public IGateway, private IClientTransport { if (message.contains("id") && (message.contains("result") || message.contains("error"))) { - FIREBOLT_LOG_DEBUG("Gateway", "[onMessage] classified as response id=%u", message["id"].get()); + std::string responseIdLog; + if (message["id"].is_number_unsigned()) + { + responseIdLog = std::to_string(message["id"].get()); + } + else if (message["id"].is_number_integer()) + { + responseIdLog = std::to_string(message["id"].get()); + } + else + { + responseIdLog = message["id"].dump(); + } + FIREBOLT_LOG_DEBUG("Gateway", "[onMessage] classified as response id=%s", responseIdLog.c_str()); + if (legacyRPCv1) { if (message.contains("result") && !message["result"].empty() && (!message["result"].is_object() || !message["result"].contains("listening"))) { - MessageID id = message["id"]; + std::optional id; + if (message["id"].is_number_unsigned()) + { + id = message["id"].get(); + } + else if (message["id"].is_number_integer() && (message["id"].get() >= 0)) + { + id = static_cast(message["id"].get()); + } + std::string eventName; + if (id.has_value()) { std::lock_guard lock(rpcv1_eventMap_mtx); - auto it = rpcv1_eventMap.find(id); + auto it = rpcv1_eventMap.find(id.value()); if (it != rpcv1_eventMap.end()) { eventName = it->second; @@ -896,7 +929,20 @@ class GatewayImpl : public IGateway, private IClientTransport FIREBOLT_LOG_NOTICE("Gateway", "[connection] state=%s error=%d suppressed_repeats=%zu", connected ? "connected" : "disconnected", static_cast(error), suppressedCount); } - connectionChangeListener(connected, error); + ConnectionChangeCallback listener; + { + std::lock_guard lock(connectionChangeListener_mtx); + listener = connectionChangeListener; + } + + if (listener) + { + listener(connected, error); + } + else + { + FIREBOLT_LOG_WARNING("Gateway", "[connection] no connectionChangeListener installed"); + } } MessageID getNextMessageID() override { return transport.getNextMessageID(); } diff --git a/src/logger.cpp b/src/logger.cpp index 27a15cb..c0a3690 100644 --- a/src/logger.cpp +++ b/src/logger.cpp @@ -45,7 +45,8 @@ namespace { std::string toLowerCopy(std::string value) { - std::transform(value.begin(), value.end(), value.begin(), [](unsigned char c) { return std::tolower(c); }); + std::transform(value.begin(), value.end(), value.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); return value; } @@ -293,7 +294,7 @@ bool tryWriteToConfiguredLogFile(const char* message) /* static */ bool Logger::formatter_addFunction = true; // clang-format off -std::map _logLevelNames = { +const std::map _logLevelNames = { {LogLevel::Error, "Error"}, {LogLevel::Warning, "Warning"}, {LogLevel::Notice, "Notice"}, @@ -304,7 +305,7 @@ std::map _logLevelNames = { #ifdef ENABLE_SYSLOG // clang-format off -std::map _logLevel2SysLog = { +const std::map _logLevel2SysLog = { {LogLevel::Error, LOG_ERR}, {LogLevel::Warning, LOG_WARNING}, {LogLevel::Notice, LOG_NOTICE}, @@ -392,7 +393,8 @@ void Logger::log(LogLevel logLevel, const std::string& module, const std::string time = timeBuf; } - const std::string levelName = _logLevelNames[logLevel]; + const auto levelNameIt = _logLevelNames.find(logLevel); + const std::string levelName = (levelNameIt != _logLevelNames.end()) ? levelNameIt->second : "Unknown"; std::string fileName; if (formatter_addLocation) @@ -449,7 +451,8 @@ void Logger::log(LogLevel logLevel, const std::string& module, const std::string if (!tryWriteToConfiguredLogFile(formattedMsg)) { #ifdef ENABLE_SYSLOG - syslog(_logLevel2SysLog[logLevel], "%s", formattedMsg); + const auto syslogLevel = _logLevel2SysLog.find(logLevel); + syslog(syslogLevel != _logLevel2SysLog.end() ? syslogLevel->second : LOG_ERR, "%s", formattedMsg); #else fprintf(stderr, "%s\n", formattedMsg); fflush(stderr); diff --git a/src/transport.cpp b/src/transport.cpp index e20b2af..34303ed 100644 --- a/src/transport.cpp +++ b/src/transport.cpp @@ -32,6 +32,7 @@ using message_ptr = websocketpp::config::asio_client::message_type::ptr; enum class Transport::TransportState { NotStarted, + Connecting, Connected, Disconnected, }; @@ -159,9 +160,10 @@ Firebolt::Error Transport::connect(std::string url, MessageCallback onMessage, C std::optional transportLoggingExclude, const std::map& headers) { - if (connectionStatus_ == TransportState::Connected) + const auto state = connectionStatus_.load(); + if (state == TransportState::Connected || state == TransportState::Connecting) { - FIREBOLT_LOG_WARNING("Transport", "Connect called when already connected. Ignoring"); + FIREBOLT_LOG_WARNING("Transport", "Connect called while connection is already active/in-progress. Ignoring"); return Firebolt::Error::AlreadyConnected; } @@ -250,10 +252,18 @@ Firebolt::Error Transport::connect(std::string url, MessageCallback onMessage, C } // Inject custom headers before connecting - for (const auto& header : headers) + try + { + for (const auto& header : headers) + { + con->replace_header(header.first, header.second); + FIREBOLT_LOG_DEBUG("Transport", "[connect] injected header '%s'", header.first.c_str()); + } + } + catch (const std::exception& ex) { - con->replace_header(header.first, header.second); - FIREBOLT_LOG_DEBUG("Transport", "[connect] injected header '%s'", header.first.c_str()); + FIREBOLT_LOG_ERROR("Transport", "Failed to inject custom headers: %s", ex.what()); + return Firebolt::Error::NotConnected; } connectionHandle_ = con->get_handle(); @@ -268,6 +278,7 @@ Firebolt::Error Transport::connect(std::string url, MessageCallback onMessage, C con->set_message_handler(websocketpp::lib::bind(&Transport::onMessage, this, websocketpp::lib::placeholders::_1, websocketpp::lib::placeholders::_2)); + connectionStatus_ = TransportState::Connecting; client_->connect(con); FIREBOLT_LOG_DEBUG("Transport", "[connect] connect() dispatched to websocket client"); @@ -283,7 +294,8 @@ Firebolt::Error Transport::disconnect() } client_->stop_perpetual(); - if (connectionStatus_ == TransportState::Connected) + const bool closeHandshakeInitiated = (connectionStatus_ == TransportState::Connected); + if (closeHandshakeInitiated) { // Shorten the close-handshake timeout so that join() below does not block // for the full websocketpp default (5 s) if the gateway is unresponsive. @@ -308,8 +320,14 @@ Firebolt::Error Transport::disconnect() FIREBOLT_LOG_ERROR("Transport", "Error closing connection: %s", ec.message().c_str()); } } + else + { + // Force-stop pending async ops (DNS/connect/handshake) so run() can exit promptly. + client_->stop(); + } - FIREBOLT_LOG_DEBUG("Transport", "[disconnect] waiting for connectionThread join (close handshake in progress)..."); + FIREBOLT_LOG_DEBUG("Transport", "[disconnect] waiting for connectionThread join (%s)...", + closeHandshakeInitiated ? "close handshake path" : "force-stop path"); auto t0_ct = std::chrono::steady_clock::now(); if (connectionThread_ && connectionThread_->joinable()) { @@ -329,6 +347,10 @@ Firebolt::Error Transport::disconnect() .count())); client_ = std::make_unique(); + { + std::lock_guard lock(responseHeadersMutex_); + responseHeaders_.clear(); + } connectionStatus_ = TransportState::NotStarted; FIREBOLT_LOG_DEBUG("Transport", "[disconnect] transport reset complete, state=%d", static_cast(connectionStatus_.load())); @@ -395,34 +417,46 @@ void Transport::onMessage(websocketpp::connection_hdl /* hdl */, FIREBOLT_LOG_WARNING("Transport", "Received a message while stopping the message worker. Ignoring"); return; } + const size_t payloadSize = msg->get_payload().size(); + size_t queueDepth = 0; { std::lock_guard lock(messageQueueMutex_); messageQueue_.push(msg->get_payload()); - FIREBOLT_LOG_DEBUG("Transport", "[onMessage] enqueued payload bytes=%zu, queue_depth=%zu", - msg->get_payload().size(), messageQueue_.size()); + queueDepth = messageQueue_.size(); } + FIREBOLT_LOG_DEBUG("Transport", "[onMessage] enqueued payload bytes=%zu, queue_depth=%zu", payloadSize, queueDepth); messageQueueCv_.notify_one(); } void Transport::onOpen(websocketpp::client* c, websocketpp::connection_hdl hdl) { - connectionStatus_ = TransportState::Connected; - - client::connection_ptr con = c->get_con_from_hdl(hdl); // Populate responseHeaders_ from the connection's response headers { std::lock_guard lock(responseHeadersMutex_); responseHeaders_.clear(); - if (con) + try { - const auto& headers = con->get_response().get_headers(); - for (const auto& header : headers) + client::connection_ptr con = c->get_con_from_hdl(hdl); + if (con) + { + const auto& headers = con->get_response().get_headers(); + for (const auto& header : headers) + { + responseHeaders_[header.first] = header.second; + } + FIREBOLT_LOG_DEBUG("Transport", "[onOpen] stored response headers=%zu", responseHeaders_.size()); + } + else { - responseHeaders_[header.first] = header.second; + FIREBOLT_LOG_WARNING("Transport", "[onOpen] connection handle resolved to null"); } - FIREBOLT_LOG_DEBUG("Transport", "[onOpen] stored response headers=%zu", responseHeaders_.size()); + } + catch (const std::exception& ex) + { + FIREBOLT_LOG_WARNING("Transport", "[onOpen] failed to resolve connection handle (%s)", ex.what()); } } + connectionStatus_ = TransportState::Connected; FIREBOLT_LOG_NOTICE("Transport", "Connection opened (state=%d)", static_cast(connectionStatus_.load())); connectionReceiver_(true, Firebolt::Error::None); } @@ -430,6 +464,10 @@ void Transport::onOpen(websocketpp::client* c, void Transport::onClose(websocketpp::client* c, websocketpp::connection_hdl hdl) { connectionStatus_ = TransportState::Disconnected; + { + std::lock_guard lock(responseHeadersMutex_); + responseHeaders_.clear(); + } Firebolt::Error mappedError = Firebolt::Error::General; try { @@ -455,6 +493,10 @@ void Transport::onClose(websocketpp::client* c void Transport::onFail(websocketpp::client* c, websocketpp::connection_hdl hdl) { connectionStatus_ = TransportState::Disconnected; + { + std::lock_guard lock(responseHeadersMutex_); + responseHeaders_.clear(); + } Firebolt::Error mappedError = Firebolt::Error::General; try { diff --git a/src/transport.h b/src/transport.h index 6f717d2..6b5d1ee 100644 --- a/src/transport.h +++ b/src/transport.h @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include diff --git a/src/vendor/websocketpp/transport/asio/endpoint.hpp b/src/vendor/websocketpp/transport/asio/endpoint.hpp new file mode 100644 index 0000000..76f6d5f --- /dev/null +++ b/src/vendor/websocketpp/transport/asio/endpoint.hpp @@ -0,0 +1,1197 @@ +/* + * Copyright (c) 2015, Peter Thorson. All rights reserved. + * + * SPDX-License-Identifier: BSD-3-Clause + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of the WebSocket++ Project nor the + * names of its contributors may be used to endorse or promote products + * derived from this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL PETER THORSON BE LIABLE FOR ANY + * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND + * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef WEBSOCKETPP_TRANSPORT_ASIO_HPP +#define WEBSOCKETPP_TRANSPORT_ASIO_HPP + +#include +#include +#include + +#include +#include + +#include +#include + +#include +#include + +namespace websocketpp { +namespace transport { +namespace asio { + +/// Asio based endpoint transport component +/** + * transport::asio::endpoint implements an endpoint transport component using + * Asio. + */ +template +class endpoint : public config::socket_type { +public: + /// Type of this endpoint transport component + typedef endpoint type; + + /// Type of the concurrency policy + typedef typename config::concurrency_type concurrency_type; + /// Type of the socket policy + typedef typename config::socket_type socket_type; + /// Type of the error logging policy + typedef typename config::elog_type elog_type; + /// Type of the access logging policy + typedef typename config::alog_type alog_type; + + /// Type of the socket connection component + typedef typename socket_type::socket_con_type socket_con_type; + /// Type of a shared pointer to the socket connection component + typedef typename socket_con_type::ptr socket_con_ptr; + + /// Type of the connection transport component associated with this + /// endpoint transport component + typedef asio::connection transport_con_type; + /// Type of a shared pointer to the connection transport component + /// associated with this endpoint transport component + typedef typename transport_con_type::ptr transport_con_ptr; + + /// Type of a pointer to the ASIO io_service being used + typedef lib::asio::io_service * io_service_ptr; + /// Type of a shared pointer to the acceptor being used + typedef lib::shared_ptr acceptor_ptr; + /// Type of a shared pointer to the resolver being used + typedef lib::shared_ptr resolver_ptr; + /// Type of timer handle + typedef lib::shared_ptr timer_ptr; + /// Type of a shared pointer to an io_service work object + typedef lib::shared_ptr work_ptr; + + /// Type of socket pre-bind handler + typedef lib::function tcp_pre_bind_handler; + + // generate and manage our own io_service + explicit endpoint() + : m_io_service(NULL) + , m_external_io_service(false) + , m_listen_backlog(lib::asio::socket_base::max_connections) + , m_reuse_addr(false) + , m_state(UNINITIALIZED) + { + //std::cout << "transport::asio::endpoint constructor" << std::endl; + } + + ~endpoint() { + // clean up our io_service if we were initialized with an internal one. + + // Explicitly destroy local objects + m_acceptor.reset(); + m_resolver.reset(); + m_work.reset(); + if (m_state != UNINITIALIZED && !m_external_io_service) { + delete m_io_service; + } + } + + /// transport::asio objects are moveable but not copyable or assignable. + /// The following code sets this situation up based on whether or not we + /// have C++11 support or not +#ifdef _WEBSOCKETPP_DEFAULT_DELETE_FUNCTIONS_ + endpoint(const endpoint & src) = delete; + endpoint& operator= (const endpoint & rhs) = delete; +#else +private: + endpoint(const endpoint & src); + endpoint & operator= (const endpoint & rhs); +public: +#endif // _WEBSOCKETPP_DEFAULT_DELETE_FUNCTIONS_ + +#ifdef _WEBSOCKETPP_MOVE_SEMANTICS_ + endpoint (endpoint && src) + : config::socket_type(std::move(src)) + , m_tcp_pre_init_handler(src.m_tcp_pre_init_handler) + , m_tcp_post_init_handler(src.m_tcp_post_init_handler) + , m_io_service(src.m_io_service) + , m_external_io_service(src.m_external_io_service) + , m_acceptor(src.m_acceptor) + , m_listen_backlog(lib::asio::socket_base::max_connections) + , m_reuse_addr(src.m_reuse_addr) + , m_elog(src.m_elog) + , m_alog(src.m_alog) + , m_state(src.m_state) + { + src.m_io_service = NULL; + src.m_external_io_service = false; + src.m_acceptor = NULL; + src.m_state = UNINITIALIZED; + } + + /*endpoint & operator= (const endpoint && rhs) { + if (this != &rhs) { + m_io_service = rhs.m_io_service; + m_external_io_service = rhs.m_external_io_service; + m_acceptor = rhs.m_acceptor; + m_listen_backlog = rhs.m_listen_backlog; + m_reuse_addr = rhs.m_reuse_addr; + m_state = rhs.m_state; + + rhs.m_io_service = NULL; + rhs.m_external_io_service = false; + rhs.m_acceptor = NULL; + rhs.m_listen_backlog = lib::asio::socket_base::max_connections; + rhs.m_state = UNINITIALIZED; + + // TODO: this needs to be updated + } + return *this; + }*/ +#endif // _WEBSOCKETPP_MOVE_SEMANTICS_ + + /// Return whether or not the endpoint produces secure connections. + bool is_secure() const { + return socket_type::is_secure(); + } + + /// initialize asio transport with external io_service (exception free) + /** + * Initialize the ASIO transport policy for this endpoint using the provided + * io_service object. asio_init must be called exactly once on any endpoint + * that uses transport::asio before it can be used. + * + * @param ptr A pointer to the io_service to use for asio events + * @param ec Set to indicate what error occurred, if any. + */ + void init_asio(io_service_ptr ptr, lib::error_code & ec) { + if (m_state != UNINITIALIZED) { + m_elog->write(log::elevel::library, + "asio::init_asio called from the wrong state"); + using websocketpp::error::make_error_code; + ec = make_error_code(websocketpp::error::invalid_state); + return; + } + + m_alog->write(log::alevel::devel,"asio::init_asio"); + + m_io_service = ptr; + m_external_io_service = true; + m_acceptor.reset(new lib::asio::ip::tcp::acceptor(*m_io_service)); + + m_state = READY; + ec = lib::error_code(); + } + + /// initialize asio transport with external io_service + /** + * Initialize the ASIO transport policy for this endpoint using the provided + * io_service object. asio_init must be called exactly once on any endpoint + * that uses transport::asio before it can be used. + * + * @param ptr A pointer to the io_service to use for asio events + */ + void init_asio(io_service_ptr ptr) { + lib::error_code ec; + init_asio(ptr,ec); + if (ec) { throw exception(ec); } + } + + /// Initialize asio transport with internal io_service (exception free) + /** + * This method of initialization will allocate and use an internally managed + * io_service. + * + * @see init_asio(io_service_ptr ptr) + * + * @param ec Set to indicate what error occurred, if any. + */ + void init_asio(lib::error_code & ec) { + // Use a smart pointer until the call is successful and ownership has + // successfully been taken. Use unique_ptr when available. + // TODO: remove the use of auto_ptr when C++98/03 support is no longer + // necessary. +#ifdef _WEBSOCKETPP_CPP11_MEMORY_ + lib::unique_ptr service(new lib::asio::io_service()); +#else + lib::auto_ptr service(new lib::asio::io_service()); +#endif + init_asio(service.get(), ec); + if( !ec ) service.release(); // Call was successful, transfer ownership + m_external_io_service = false; + } + + /// Initialize asio transport with internal io_service + /** + * This method of initialization will allocate and use an internally managed + * io_service. + * + * @see init_asio(io_service_ptr ptr) + */ + void init_asio() { + // Use a smart pointer until the call is successful and ownership has + // successfully been taken. Use unique_ptr when available. + // TODO: remove the use of auto_ptr when C++98/03 support is no longer + // necessary. +#ifdef _WEBSOCKETPP_CPP11_MEMORY_ + lib::unique_ptr service(new lib::asio::io_service()); +#else + lib::auto_ptr service(new lib::asio::io_service()); +#endif + init_asio( service.get() ); + // If control got this far without an exception, then ownership has successfully been taken + service.release(); + m_external_io_service = false; + } + + /// Sets the tcp pre bind handler + /** + * The tcp pre bind handler is called after the listen acceptor has + * been created but before the socket bind is performed. + * + * @since 0.8.0 + * + * @param h The handler to call on tcp pre bind init. + */ + void set_tcp_pre_bind_handler(tcp_pre_bind_handler h) { + m_tcp_pre_bind_handler = h; + } + + /// Sets the tcp pre init handler + /** + * The tcp pre init handler is called after the raw tcp connection has been + * established but before any additional wrappers (proxy connects, TLS + * handshakes, etc) have been performed. + * + * @since 0.3.0 + * + * @param h The handler to call on tcp pre init. + */ + void set_tcp_pre_init_handler(tcp_init_handler h) { + m_tcp_pre_init_handler = h; + } + + /// Sets the tcp pre init handler (deprecated) + /** + * The tcp pre init handler is called after the raw tcp connection has been + * established but before any additional wrappers (proxy connects, TLS + * handshakes, etc) have been performed. + * + * @deprecated Use set_tcp_pre_init_handler instead + * + * @param h The handler to call on tcp pre init. + */ + void set_tcp_init_handler(tcp_init_handler h) { + set_tcp_pre_init_handler(h); + } + + /// Sets the tcp post init handler + /** + * The tcp post init handler is called after the tcp connection has been + * established and all additional wrappers (proxy connects, TLS handshakes, + * etc have been performed. This is fired before any bytes are read or any + * WebSocket specific handshake logic has been performed. + * + * @since 0.3.0 + * + * @param h The handler to call on tcp post init. + */ + void set_tcp_post_init_handler(tcp_init_handler h) { + m_tcp_post_init_handler = h; + } + + /// Sets the maximum length of the queue of pending connections. + /** + * Sets the maximum length of the queue of pending connections. Increasing + * this will allow WebSocket++ to queue additional incoming connections. + * Setting it higher may prevent failed connections at high connection rates + * but may cause additional latency. + * + * For this value to take effect you may need to adjust operating system + * settings. + * + * New values affect future calls to listen only. + * + * The default value is specified as *::asio::socket_base::max_connections + * which uses the operating system defined maximum queue length. Your OS + * may restrict or silently lower this value. A value of zero may cause + * all connections to be rejected. + * + * @since 0.3.0 + * + * @param backlog The maximum length of the queue of pending connections + */ + void set_listen_backlog(int backlog) { + m_listen_backlog = backlog; + } + + /// Sets whether to use the SO_REUSEADDR flag when opening listening sockets + /** + * Specifies whether or not to use the SO_REUSEADDR TCP socket option. What + * this flag does depends on your operating system. + * + * Please consult operating system documentation for more details. There + * may be security consequences to enabling this option. + * + * New values affect future calls to listen only so set this value prior to + * calling listen. + * + * The default is false. + * + * @since 0.3.0 + * + * @param value Whether or not to use the SO_REUSEADDR option + */ + void set_reuse_addr(bool value) { + m_reuse_addr = value; + } + + /// Retrieve a reference to the endpoint's io_service + /** + * The io_service may be an internal or external one. This may be used to + * call methods of the io_service that are not explicitly wrapped by the + * endpoint. + * + * This method is only valid after the endpoint has been initialized with + * `init_asio`. No error will be returned if it isn't. + * + * @return A reference to the endpoint's io_service + */ + lib::asio::io_service & get_io_service() { + return *m_io_service; + } + + /// Get local TCP endpoint + /** + * Extracts the local endpoint from the acceptor. This represents the + * address that WebSocket++ is listening on. + * + * Sets a bad_descriptor error if the acceptor is not currently listening + * or otherwise unavailable. + * + * @since 0.7.0 + * + * @param ec Set to indicate what error occurred, if any. + * @return The local endpoint + */ + lib::asio::ip::tcp::endpoint get_local_endpoint(lib::asio::error_code & ec) { + if (m_acceptor) { + return m_acceptor->local_endpoint(ec); + } else { + ec = lib::asio::error::make_error_code(lib::asio::error::bad_descriptor); + return lib::asio::ip::tcp::endpoint(); + } + } + + /// Set up endpoint for listening manually (exception free) + /** + * Bind the internal acceptor using the specified settings. The endpoint + * must have been initialized by calling init_asio before listening. + * + * @param ep An endpoint to read settings from + * @param ec Set to indicate what error occurred, if any. + */ + void listen(lib::asio::ip::tcp::endpoint const & ep, lib::error_code & ec) + { + if (m_state != READY) { + m_elog->write(log::elevel::library, + "asio::listen called from the wrong state"); + using websocketpp::error::make_error_code; + ec = make_error_code(websocketpp::error::invalid_state); + return; + } + + m_alog->write(log::alevel::devel,"asio::listen"); + + lib::asio::error_code bec; + + m_acceptor->open(ep.protocol(),bec); + if (bec) {ec = clean_up_listen_after_error(bec);return;} + + m_acceptor->set_option(lib::asio::socket_base::reuse_address(m_reuse_addr),bec); + if (bec) {ec = clean_up_listen_after_error(bec);return;} + + // if a TCP pre-bind handler is present, run it + if (m_tcp_pre_bind_handler) { + ec = m_tcp_pre_bind_handler(m_acceptor); + if (ec) { + ec = clean_up_listen_after_error(ec); + return; + } + } + + m_acceptor->bind(ep,bec); + if (bec) {ec = clean_up_listen_after_error(bec);return;} + + m_acceptor->listen(m_listen_backlog,bec); + if (bec) {ec = clean_up_listen_after_error(bec);return;} + + // Success + m_state = LISTENING; + ec = lib::error_code(); + } + + + + /// Set up endpoint for listening manually + /** + * Bind the internal acceptor using the settings specified by the endpoint e + * + * @param ep An endpoint to read settings from + */ + void listen(lib::asio::ip::tcp::endpoint const & ep) { + lib::error_code ec; + listen(ep,ec); + if (ec) { throw exception(ec); } + } + + /// Set up endpoint for listening with protocol and port (exception free) + /** + * Bind the internal acceptor using the given internet protocol and port. + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * Common options include: + * - IPv6 with mapped IPv4 for dual stack hosts lib::asio::ip::tcp::v6() + * - IPv4 only: lib::asio::ip::tcp::v4() + * + * @param internet_protocol The internet protocol to use. + * @param port The port to listen on. + * @param ec Set to indicate what error occurred, if any. + */ + template + void listen(InternetProtocol const & internet_protocol, uint16_t port, + lib::error_code & ec) + { + lib::asio::ip::tcp::endpoint ep(internet_protocol, port); + listen(ep,ec); + } + + /// Set up endpoint for listening with protocol and port + /** + * Bind the internal acceptor using the given internet protocol and port. + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * Common options include: + * - IPv6 with mapped IPv4 for dual stack hosts lib::asio::ip::tcp::v6() + * - IPv4 only: lib::asio::ip::tcp::v4() + * + * @param internet_protocol The internet protocol to use. + * @param port The port to listen on. + */ + template + void listen(InternetProtocol const & internet_protocol, uint16_t port) + { + lib::asio::ip::tcp::endpoint ep(internet_protocol, port); + listen(ep); + } + + /// Set up endpoint for listening on a port (exception free) + /** + * Bind the internal acceptor using the given port. The IPv6 protocol with + * mapped IPv4 for dual stack hosts will be used. If you need IPv4 only use + * the overload that allows specifying the protocol explicitly. + * + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * @param port The port to listen on. + * @param ec Set to indicate what error occurred, if any. + */ + void listen(uint16_t port, lib::error_code & ec) { + listen(lib::asio::ip::tcp::v6(), port, ec); + } + + /// Set up endpoint for listening on a port + /** + * Bind the internal acceptor using the given port. The IPv6 protocol with + * mapped IPv4 for dual stack hosts will be used. If you need IPv4 only use + * the overload that allows specifying the protocol explicitly. + * + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * @param port The port to listen on. + * @param ec Set to indicate what error occurred, if any. + */ + void listen(uint16_t port) { + listen(lib::asio::ip::tcp::v6(), port); + } + + /// Set up endpoint for listening on a host and service (exception free) + /** + * Bind the internal acceptor using the given host and service. More details + * about what host and service can be are available in the Asio + * documentation for ip::basic_resolver_query::basic_resolver_query's + * constructors. + * + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * @param host A string identifying a location. May be a descriptive name or + * a numeric address string. + * @param service A string identifying the requested service. This may be a + * descriptive name or a numeric string corresponding to a port number. + * @param ec Set to indicate what error occurred, if any. + */ + void listen(std::string const & host, std::string const & service, + lib::error_code & ec) + { + using lib::asio::ip::tcp; + tcp::resolver r(*m_io_service); + tcp::resolver::query query(host, service); + tcp::resolver::iterator endpoint_iterator = r.resolve(query); + tcp::resolver::iterator end; + if (endpoint_iterator == end) { + m_elog->write(log::elevel::library, + "asio::listen could not resolve the supplied host or service"); + ec = make_error_code(error::invalid_host_service); + return; + } + listen(*endpoint_iterator,ec); + } + + /// Set up endpoint for listening on a host and service + /** + * Bind the internal acceptor using the given host and service. More details + * about what host and service can be are available in the Asio + * documentation for ip::basic_resolver_query::basic_resolver_query's + * constructors. + * + * The endpoint must have been initialized by calling init_asio before + * listening. + * + * @param host A string identifying a location. May be a descriptive name or + * a numeric address string. + * @param service A string identifying the requested service. This may be a + * descriptive name or a numeric string corresponding to a port number. + * @param ec Set to indicate what error occurred, if any. + */ + void listen(std::string const & host, std::string const & service) + { + lib::error_code ec; + listen(host,service,ec); + if (ec) { throw exception(ec); } + } + + /// Stop listening (exception free) + /** + * Stop listening and accepting new connections. This will not end any + * existing connections. + * + * @since 0.3.0-alpha4 + * @param ec A status code indicating an error, if any. + */ + void stop_listening(lib::error_code & ec) { + if (m_state != LISTENING) { + m_elog->write(log::elevel::library, + "asio::listen called from the wrong state"); + using websocketpp::error::make_error_code; + ec = make_error_code(websocketpp::error::invalid_state); + return; + } + + m_acceptor->close(); + m_state = READY; + ec = lib::error_code(); + } + + /// Stop listening + /** + * Stop listening and accepting new connections. This will not end any + * existing connections. + * + * @since 0.3.0-alpha4 + */ + void stop_listening() { + lib::error_code ec; + stop_listening(ec); + if (ec) { throw exception(ec); } + } + + /// Check if the endpoint is listening + /** + * @return Whether or not the endpoint is listening. + */ + bool is_listening() const { + return (m_state == LISTENING); + } + + /// wraps the run method of the internal io_service object + std::size_t run() { + return m_io_service->run(); + } + + /// wraps the run_one method of the internal io_service object + /** + * @since 0.3.0-alpha4 + */ + std::size_t run_one() { + return m_io_service->run_one(); + } + + /// wraps the stop method of the internal io_service object + void stop() { + m_io_service->stop(); + } + + /// wraps the poll method of the internal io_service object + std::size_t poll() { + return m_io_service->poll(); + } + + /// wraps the poll_one method of the internal io_service object + std::size_t poll_one() { + return m_io_service->poll_one(); + } + + /// wraps the reset method of the internal io_service object + void reset() { + m_io_service->reset(); + } + + /// wraps the stopped method of the internal io_service object + bool stopped() const { + return m_io_service->stopped(); + } + + /// Marks the endpoint as perpetual, stopping it from exiting when empty + /** + * Marks the endpoint as perpetual. Perpetual endpoints will not + * automatically exit when they run out of connections to process. To stop + * a perpetual endpoint call `end_perpetual`. + * + * An endpoint may be marked perpetual at any time by any thread. It must be + * called either before the endpoint has run out of work or before it was + * started + * + * @since 0.3.0 + */ + void start_perpetual() { + m_work.reset(new lib::asio::io_service::work(*m_io_service)); + } + + /// Clears the endpoint's perpetual flag, allowing it to exit when empty + /** + * Clears the endpoint's perpetual flag. This will cause the endpoint's run + * method to exit normally when it runs out of connections. If there are + * currently active connections it will not end until they are complete. + * + * @since 0.3.0 + */ + void stop_perpetual() { + m_work.reset(); + } + + /// Call back a function after a period of time. + /** + * Sets a timer that calls back a function after the specified period of + * milliseconds. Returns a handle that can be used to cancel the timer. + * A cancelled timer will return the error code error::operation_aborted + * A timer that expired will return no error. + * + * @param duration Length of time to wait in milliseconds + * @param callback The function to call back when the timer has expired + * @return A handle that can be used to cancel the timer if it is no longer + * needed. + */ + timer_ptr set_timer(long duration, timer_handler callback) { + timer_ptr new_timer = lib::make_shared( + *m_io_service, + lib::asio::milliseconds(duration) + ); + + new_timer->async_wait( + lib::bind( + &type::handle_timer, + this, + new_timer, + callback, + lib::placeholders::_1 + ) + ); + + return new_timer; + } + + /// Timer handler + /** + * The timer pointer is included to ensure the timer isn't destroyed until + * after it has expired. + * + * @param t Pointer to the timer in question + * @param callback The function to call back + * @param ec A status code indicating an error, if any. + */ + void handle_timer(timer_ptr, timer_handler callback, + lib::asio::error_code const & ec) + { + if (ec) { + if (ec == lib::asio::error::operation_aborted) { + callback(make_error_code(transport::error::operation_aborted)); + } else { + m_elog->write(log::elevel::info, + "asio handle_timer error: "+ec.message()); + log_err(log::elevel::info,"asio handle_timer",ec); + callback(socket_con_type::translate_ec(ec)); + } + } else { + callback(lib::error_code()); + } + } + + /// Accept the next connection attempt and assign it to con (exception free) + /** + * @param tcon The connection to accept into. + * @param callback The function to call when the operation is complete. + * @param ec A status code indicating an error, if any. + */ + void async_accept(transport_con_ptr tcon, accept_handler callback, + lib::error_code & ec) + { + if (m_state != LISTENING || !m_acceptor) { + using websocketpp::error::make_error_code; + ec = make_error_code(websocketpp::error::async_accept_not_listening); + return; + } + + m_alog->write(log::alevel::devel, "asio::async_accept"); + + if (config::enable_multithreading) { + m_acceptor->async_accept( + tcon->get_raw_socket(), + tcon->get_strand()->wrap(lib::bind( + &type::handle_accept, + this, + callback, + lib::placeholders::_1 + )) + ); + } else { + m_acceptor->async_accept( + tcon->get_raw_socket(), + lib::bind( + &type::handle_accept, + this, + callback, + lib::placeholders::_1 + ) + ); + } + } + + /// Accept the next connection attempt and assign it to con. + /** + * @param tcon The connection to accept into. + * @param callback The function to call when the operation is complete. + */ + void async_accept(transport_con_ptr tcon, accept_handler callback) { + lib::error_code ec; + async_accept(tcon,callback,ec); + if (ec) { throw exception(ec); } + } +protected: + /// Initialize logging + /** + * The loggers are located in the main endpoint class. As such, the + * transport doesn't have direct access to them. This method is called + * by the endpoint constructor to allow shared logging from the transport + * component. These are raw pointers to member variables of the endpoint. + * In particular, they cannot be used in the transport constructor as they + * haven't been constructed yet, and cannot be used in the transport + * destructor as they will have been destroyed by then. + */ + void init_logging(const lib::shared_ptr& a, const lib::shared_ptr& e) { + m_alog = a; + m_elog = e; + } + + void handle_accept(accept_handler callback, lib::asio::error_code const & + asio_ec) + { + lib::error_code ret_ec; + + m_alog->write(log::alevel::devel, "asio::handle_accept"); + + if (asio_ec) { + if (asio_ec == lib::asio::errc::operation_canceled) { + ret_ec = make_error_code(websocketpp::error::operation_canceled); + } else { + log_err(log::elevel::info,"asio handle_accept",asio_ec); + ret_ec = socket_con_type::translate_ec(asio_ec); + } + } + + callback(ret_ec); + } + + /// Initiate a new connection + // TODO: there have to be some more failure conditions here + void async_connect(transport_con_ptr tcon, uri_ptr u, connect_handler cb) { + using namespace lib::asio::ip; + + // Create a resolver + if (!m_resolver) { + m_resolver.reset(new lib::asio::ip::tcp::resolver(*m_io_service)); + } + + tcon->set_uri(u); + + std::string proxy = tcon->get_proxy(); + std::string host; + std::string port; + + if (proxy.empty()) { + host = u->get_host(); + port = u->get_port_str(); + } else { + lib::error_code ec; + + uri_ptr pu = lib::make_shared(proxy); + + if (!pu->get_valid()) { + cb(make_error_code(error::proxy_invalid)); + return; + } + + ec = tcon->proxy_init(u->get_authority()); + if (ec) { + cb(ec); + return; + } + + host = pu->get_host(); + port = pu->get_port_str(); + } + + // If the host is a numeric IP (e.g. 127.0.0.1) use the numeric_host + // resolver flag so that getaddrinfo() uses AI_NUMERICHOST and does NOT + // apply AI_ADDRCONFIG. AI_ADDRCONFIG causes resolution to fail on + // embedded devices where the loopback interface is not brought up when + // external network interfaces (eth0/wifi) are all down -- even though + // the gateway is reachable on 127.0.0.1 and the Firebolt gateway daemon + // is running locally. For genuine hostname strings the existing + // AI_ADDRCONFIG (address_configured) flag is preserved. + lib::asio::error_code m_numeric_ec; + lib::asio::ip::address::from_string(host, m_numeric_ec); + const tcp::resolver::query::flags resolve_flags = m_numeric_ec + ? tcp::resolver::query::address_configured + : tcp::resolver::query::numeric_host; + tcp::resolver::query query(host, port, resolve_flags); + + if (m_alog->static_test(log::alevel::devel)) { + m_alog->write(log::alevel::devel, + "starting async DNS resolve for "+host+":"+port); + } + + timer_ptr dns_timer; + + dns_timer = tcon->set_timer( + config::timeout_dns_resolve, + lib::bind( + &type::handle_resolve_timeout, + this, + dns_timer, + cb, + lib::placeholders::_1 + ) + ); + + if (config::enable_multithreading) { + m_resolver->async_resolve( + query, + tcon->get_strand()->wrap(lib::bind( + &type::handle_resolve, + this, + tcon, + dns_timer, + cb, + lib::placeholders::_1, + lib::placeholders::_2 + )) + ); + } else { + m_resolver->async_resolve( + query, + lib::bind( + &type::handle_resolve, + this, + tcon, + dns_timer, + cb, + lib::placeholders::_1, + lib::placeholders::_2 + ) + ); + } + } + + /// DNS resolution timeout handler + /** + * The timer pointer is included to ensure the timer isn't destroyed until + * after it has expired. + * + * @param dns_timer Pointer to the timer in question + * @param callback The function to call back + * @param ec A status code indicating an error, if any. + */ + void handle_resolve_timeout(timer_ptr, connect_handler callback, + lib::error_code const & ec) + { + lib::error_code ret_ec; + + if (ec) { + if (ec == transport::error::operation_aborted) { + m_alog->write(log::alevel::devel, + "asio handle_resolve_timeout timer cancelled"); + return; + } + + log_err(log::elevel::devel,"asio handle_resolve_timeout",ec); + ret_ec = ec; + } else { + ret_ec = make_error_code(transport::error::timeout); + } + + m_alog->write(log::alevel::devel,"DNS resolution timed out"); + m_resolver->cancel(); + callback(ret_ec); + } + + void handle_resolve(transport_con_ptr tcon, timer_ptr dns_timer, + connect_handler callback, lib::asio::error_code const & ec, + lib::asio::ip::tcp::resolver::iterator iterator) + { + if (ec == lib::asio::error::operation_aborted || + lib::asio::is_neg(dns_timer->expires_from_now())) + { + m_alog->write(log::alevel::devel,"async_resolve cancelled"); + return; + } + + dns_timer->cancel(); + + if (ec) { + log_err(log::elevel::info,"asio async_resolve",ec); + callback(socket_con_type::translate_ec(ec)); + return; + } + + if (m_alog->static_test(log::alevel::devel)) { + std::stringstream s; + s << "Async DNS resolve successful. Results: "; + + lib::asio::ip::tcp::resolver::iterator it, end; + for (it = iterator; it != end; ++it) { + s << (*it).endpoint() << " "; + } + + m_alog->write(log::alevel::devel,s.str()); + } + + m_alog->write(log::alevel::devel,"Starting async connect"); + + timer_ptr con_timer; + + con_timer = tcon->set_timer( + config::timeout_connect, + lib::bind( + &type::handle_connect_timeout, + this, + tcon, + con_timer, + callback, + lib::placeholders::_1 + ) + ); + + if (config::enable_multithreading) { + lib::asio::async_connect( + tcon->get_raw_socket(), + iterator, + tcon->get_strand()->wrap(lib::bind( + &type::handle_connect, + this, + tcon, + con_timer, + callback, + lib::placeholders::_1 + )) + ); + } else { + lib::asio::async_connect( + tcon->get_raw_socket(), + iterator, + lib::bind( + &type::handle_connect, + this, + tcon, + con_timer, + callback, + lib::placeholders::_1 + ) + ); + } + } + + /// Asio connect timeout handler + /** + * The timer pointer is included to ensure the timer isn't destroyed until + * after it has expired. + * + * @param tcon Pointer to the transport connection that is being connected + * @param con_timer Pointer to the timer in question + * @param callback The function to call back + * @param ec A status code indicating an error, if any. + */ + void handle_connect_timeout(transport_con_ptr tcon, timer_ptr, + connect_handler callback, lib::error_code const & ec) + { + lib::error_code ret_ec; + + if (ec) { + if (ec == transport::error::operation_aborted) { + m_alog->write(log::alevel::devel, + "asio handle_connect_timeout timer cancelled"); + return; + } + + log_err(log::elevel::devel,"asio handle_connect_timeout",ec); + ret_ec = ec; + } else { + ret_ec = make_error_code(transport::error::timeout); + } + + m_alog->write(log::alevel::devel,"TCP connect timed out"); + tcon->cancel_socket_checked(); + callback(ret_ec); + } + + void handle_connect(transport_con_ptr tcon, timer_ptr con_timer, + connect_handler callback, lib::asio::error_code const & ec) + { + if (ec == lib::asio::error::operation_aborted || + lib::asio::is_neg(con_timer->expires_from_now())) + { + m_alog->write(log::alevel::devel,"async_connect cancelled"); + return; + } + + con_timer->cancel(); + + if (ec) { + log_err(log::elevel::info,"asio async_connect",ec); + callback(socket_con_type::translate_ec(ec)); + return; + } + + if (m_alog->static_test(log::alevel::devel)) { + m_alog->write(log::alevel::devel, + "Async connect to "+tcon->get_remote_endpoint()+" successful."); + } + + callback(lib::error_code()); + } + + /// Initialize a connection + /** + * init is called by an endpoint once for each newly created connection. + * It's purpose is to give the transport policy the chance to perform any + * transport specific initialization that couldn't be done via the default + * constructor. + * + * @param tcon A pointer to the transport portion of the connection. + * + * @return A status code indicating the success or failure of the operation + */ + lib::error_code init(transport_con_ptr tcon) { + m_alog->write(log::alevel::devel, "transport::asio::init"); + + // Initialize the connection socket component + socket_type::init(lib::static_pointer_cast(tcon)); + + lib::error_code ec; + + ec = tcon->init_asio(m_io_service); + if (ec) {return ec;} + + tcon->set_tcp_pre_init_handler(m_tcp_pre_init_handler); + tcon->set_tcp_post_init_handler(m_tcp_post_init_handler); + + return lib::error_code(); + } +private: + /// Convenience method for logging the code and message for an error_code + template + void log_err(log::level l, char const * msg, error_type const & ec) { + std::stringstream s; + s << msg << " error: " << ec << " (" << ec.message() << ")"; + m_elog->write(l,s.str()); + } + + /// Helper for cleaning up in the listen method after an error + template + lib::error_code clean_up_listen_after_error(error_type const & ec) { + if (m_acceptor->is_open()) { + m_acceptor->close(); + } + log_err(log::elevel::info,"asio listen",ec); + return socket_con_type::translate_ec(ec); + } + + enum state { + UNINITIALIZED = 0, + READY = 1, + LISTENING = 2 + }; + + // Handlers + tcp_pre_bind_handler m_tcp_pre_bind_handler; + tcp_init_handler m_tcp_pre_init_handler; + tcp_init_handler m_tcp_post_init_handler; + + // Network Resources + io_service_ptr m_io_service; + bool m_external_io_service; + acceptor_ptr m_acceptor; + resolver_ptr m_resolver; + work_ptr m_work; + + // Network constants + int m_listen_backlog; + bool m_reuse_addr; + + lib::shared_ptr m_elog; + lib::shared_ptr m_alog; + + // Transport state + state m_state; +}; + +} // namespace asio +} // namespace transport +} // namespace websocketpp + +#endif // WEBSOCKETPP_TRANSPORT_ASIO_HPP diff --git a/test/integration/test-no-network.sh b/test/integration/test-no-network.sh new file mode 100755 index 0000000..72f2226 --- /dev/null +++ b/test/integration/test-no-network.sh @@ -0,0 +1,90 @@ +#!/usr/bin/env bash +# Copyright 2025 Comcast Cable Communications Management, LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# +# Integration test: ConnectViaNumericLoopbackIP with no external network. +# +# Reproduces the RDK STB failure (RDKEMW-16441): +# websocketpp used AI_ADDRCONFIG by default; on boxes with no active eth0/wifi, +# getaddrinfo("127.0.0.1") returned HOST_NOT_FOUND even though loopback is up. +# +# Isolation strategy (tried in order): +# 1. Docker --network none — loopback-only container; no special kernel privileges needed. +# Uses the same CI image as ./build.sh --docker. +# This is the preferred method and the pattern for future +# network-isolation tests in this repo. +# 2. unshare --net — kernel user namespace; requires +# sysctl kernel.unprivileged_userns_clone=1 (often restricted). +# +# Usage: +# ./test/integration/test-no-network.sh [path-to-utApp] +# +# The test binary defaults to build-dev/test/utApp (built with +tests). +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +BINARY="${1:-$REPO_ROOT/build-dev/test/utApp}" +IMAGE="firebolt-cpp-transport-ci:local" +FILTER="TransportNumericIPUTest.*:TransportIPv6UTest.ConnectViaIPv6LoopbackIP:TransportNumericIPResolverTest.ConnectFailureViaNumericIP" + +if [[ ! -x "$BINARY" ]]; then + echo "error: test binary not found: $BINARY" >&2 + echo " Build first: ./build.sh +tests (or ./build.sh --docker +tests)" >&2 + exit 1 +fi + +# Path of the binary relative to REPO_ROOT (used for Docker volume mount). +# If the binary is outside REPO_ROOT the Docker mount won't reach it — fall back +# to the default build-dev path which is always under REPO_ROOT. +if [[ "$BINARY" != "$REPO_ROOT"/* ]]; then + echo "warning: binary '$BINARY' is outside REPO_ROOT — using default build-dev/test/utApp for Docker mode" >&2 + BINARY="$REPO_ROOT/build-dev/test/utApp" +fi +REL_BIN="${BINARY#"$REPO_ROOT"/}" + +run_in_docker() { + if ! docker image inspect "$IMAGE" &>/dev/null; then + echo "Building CI Docker image (one-time)..." + docker build -t "$IMAGE" -f "$REPO_ROOT/.github/Dockerfile" "$REPO_ROOT" + fi + echo "==> Testing in Docker --network none: $BINARY" + docker run --rm --network none \ + --user "$(id -u):$(id -g)" \ + -v "$REPO_ROOT:/workspace" -w /workspace \ + "$IMAGE" \ + "./$REL_BIN" --gtest_filter="$FILTER" +} + +run_with_unshare() { + echo "==> Testing via unshare --net: $BINARY" + unshare --net bash -c " + set -euo pipefail + ip link set lo up + \"$BINARY\" --gtest_filter='$FILTER' + " +} + +if docker info &>/dev/null 2>&1; then + run_in_docker +elif unshare --net true 2>/dev/null; then + run_with_unshare +else + echo "error: neither Docker nor unshare --net is available." >&2 + echo " Option 1 (preferred): install/start Docker" >&2 + echo " Option 2: sudo sysctl -w kernel.unprivileged_userns_clone=1" >&2 + exit 1 +fi diff --git a/test/unit/gatewayTest.cpp b/test/unit/gatewayTest.cpp index 71e5bba..693117c 100644 --- a/test/unit/gatewayTest.cpp +++ b/test/unit/gatewayTest.cpp @@ -19,15 +19,92 @@ #include "firebolt/gateway.h" #include "firebolt/logger.h" #include "utils.h" +#include #include #include +#include #include +#include #include +#include #include #include using namespace Firebolt::Transport; +struct ReservedLoopbackPort +{ + int fd = -1; + uint16_t port = 49198; + + ReservedLoopbackPort() = default; + ReservedLoopbackPort(const ReservedLoopbackPort&) = delete; + ReservedLoopbackPort& operator=(const ReservedLoopbackPort&) = delete; + + ReservedLoopbackPort(ReservedLoopbackPort&& other) noexcept + : fd(other.fd), + port(other.port) + { + other.fd = -1; + } + + ReservedLoopbackPort& operator=(ReservedLoopbackPort&& other) noexcept + { + if (this != &other) + { + if (fd >= 0) + { + ::close(fd); + } + fd = other.fd; + port = other.port; + other.fd = -1; + } + return *this; + } + + ~ReservedLoopbackPort() + { + if (fd >= 0) + { + ::close(fd); + } + } +}; + +static ReservedLoopbackPort reserveLikelyUnusedLoopbackPort() +{ + ReservedLoopbackPort reserved; + reserved.fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (reserved.fd < 0) + { + return reserved; + } + + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + addr.sin_port = 0; + + if (::bind(reserved.fd, reinterpret_cast(&addr), sizeof(addr)) != 0) + { + ::close(reserved.fd); + reserved.fd = -1; + return reserved; + } + + socklen_t addrLen = sizeof(addr); + if (::getsockname(reserved.fd, reinterpret_cast(&addr), &addrLen) != 0) + { + ::close(reserved.fd); + reserved.fd = -1; + return reserved; + } + + reserved.port = ntohs(addr.sin_port); + return reserved; +} + TEST(GatewayUrlUtilsUTest, VerifyUrls) { bool legacyRPCv1 = false; @@ -924,32 +1001,34 @@ TEST_F(GatewayUTest, SendNotConnected) { // Don't start the server — connect will fail IGateway& gateway = GetGatewayInstance(); - // Don't connect, just try to send on a fresh instance (already disconnected from TearDown) - // The gateway instance is a singleton, so we need to rely on the state being disconnected - // after TearDown was called in a previous test - // Instead, let's connect to a non-existent server and verify send after connection failure - std::promise connPromise; - auto connFuture = connPromise.get_future(); - std::atomic promiseSet{false}; - Firebolt::Config cfg = getTestConfig(); - cfg.wsUrl = "ws://localhost:49199"; // No server here + ReservedLoopbackPort reservedPort = reserveLikelyUnusedLoopbackPort(); + ASSERT_GE(reservedPort.fd, 0) << "Failed to reserve loopback port for SendNotConnected"; + cfg.wsUrl = "ws://127.0.0.1:" + std::to_string(reservedPort.port); + std::promise connectFailure; + auto connectFailureFuture = connectFailure.get_future(); Firebolt::Error err = gateway.connect(cfg, - [&](bool connected, const Firebolt::Error&) + [&connectFailure](bool connected, const Firebolt::Error& cbErr) { - bool expected = false; - if (promiseSet.compare_exchange_strong(expected, true)) + if (!connected) { - connPromise.set_value(connected); + try + { + connectFailure.set_value(cbErr); + } + catch (const std::future_error&) + { + } } }); - ASSERT_EQ(err, Firebolt::Error::None); + ASSERT_TRUE(err == Firebolt::Error::General || err == Firebolt::Error::None); - // Wait for connection failure — assert so we don't proceed with dangling callback refs - auto status = connFuture.wait_for(std::chrono::milliseconds(2000)); - ASSERT_EQ(status, std::future_status::ready) << "Connection failure callback not received"; - EXPECT_FALSE(connFuture.get()); + if (err == Firebolt::Error::None) + { + ASSERT_EQ(connectFailureFuture.wait_for(std::chrono::seconds(2)), std::future_status::ready); + EXPECT_EQ(connectFailureFuture.get(), Firebolt::Error::General); + } // Send should fail with NotConnected err = gateway.send("test.method", {}); @@ -1206,25 +1285,35 @@ TEST_F(GatewayUTest, RequestFailsWhenSendErrors) // Don't start server — connect to a port nobody listens on // so transport is in Disconnected state after connection failure IGateway& gateway = GetGatewayInstance(); - std::promise connPromise; - auto connFuture = connPromise.get_future(); - std::atomic pset{false}; Firebolt::Config cfg = getTestConfig(); - cfg.wsUrl = "ws://localhost:49198"; + ReservedLoopbackPort reservedPort = reserveLikelyUnusedLoopbackPort(); + ASSERT_GE(reservedPort.fd, 0) << "Failed to reserve loopback port for RequestFailsWhenSendErrors"; + cfg.wsUrl = "ws://127.0.0.1:" + std::to_string(reservedPort.port); + std::promise connectFailure; + auto connectFailureFuture = connectFailure.get_future(); Firebolt::Error err = gateway.connect(cfg, - [&](bool connected, const Firebolt::Error&) + [&connectFailure](bool connected, const Firebolt::Error& cbErr) { - bool exp = false; - if (pset.compare_exchange_strong(exp, true)) - connPromise.set_value(connected); + if (!connected) + { + try + { + connectFailure.set_value(cbErr); + } + catch (const std::future_error&) + { + } + } }); - ASSERT_EQ(err, Firebolt::Error::None); + ASSERT_TRUE(err == Firebolt::Error::General || err == Firebolt::Error::None); - // Assert callback arrives — prevents proceeding with dangling refs - auto connStatus = connFuture.wait_for(std::chrono::milliseconds(2000)); - ASSERT_EQ(connStatus, std::future_status::ready) << "Connection failure callback not received"; + if (err == Firebolt::Error::None) + { + ASSERT_EQ(connectFailureFuture.wait_for(std::chrono::seconds(2)), std::future_status::ready); + EXPECT_EQ(connectFailureFuture.get(), Firebolt::Error::General); + } // Now request — transport.send will fail with NotConnected, // which triggers the error branch in Client::request (lines 136-141) diff --git a/test/unit/transportTest.cpp b/test/unit/transportTest.cpp index 6f9e19b..f48c242 100644 --- a/test/unit/transportTest.cpp +++ b/test/unit/transportTest.cpp @@ -18,12 +18,15 @@ #include "transport.h" #include "firebolt/logger.h" +#include #include +#include #include #include #include #include #include +#include #include #include #include @@ -130,7 +133,13 @@ TEST_F(TransportIntegrationUTest, ConnectAndDisconnect) { if (connected) { - connectionPromise.set_value(true); + try + { + connectionPromise.set_value(true); + } + catch (const std::future_error&) + { + } } }; @@ -159,7 +168,13 @@ TEST_F(TransportIntegrationUTest, SendAndReceiveMessage) { if (connected) { - connectionPromise.set_value(true); + try + { + connectionPromise.set_value(true); + } + catch (const std::future_error&) + { + } } }; @@ -269,6 +284,100 @@ TEST_F(TransportIntegrationUTest, ConnectWhenAlreadyConnected) TEST_F(TransportIntegrationUTest, HeaderInjectionAndResponseHeaderRetrieval) { + using local_server = websocketpp::server; + using local_connection_hdl = websocketpp::connection_hdl; + + local_server headerServer; + std::promise injectedHeaderSeenPromise; + auto injectedHeaderSeenFuture = injectedHeaderSeenPromise.get_future(); + + headerServer.init_asio(); + headerServer.set_reuse_addr(true); + + headerServer.set_validate_handler( + [&headerServer, &injectedHeaderSeenPromise](local_connection_hdl hdl) + { + try + { + auto con = headerServer.get_con_from_hdl(hdl); + if (con) + { + std::string headerValue; + bool found = false; + const auto& requestHeaders = con->get_request().get_headers(); + for (const auto& header : requestHeaders) + { + std::string keyLower = header.first; + std::transform(keyLower.begin(), keyLower.end(), keyLower.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + if (keyLower == "x-test-header") + { + headerValue = header.second; + found = true; + break; + } + } + injectedHeaderSeenPromise.set_value(found ? std::move(headerValue) : std::string()); + } + else + { + injectedHeaderSeenPromise.set_value(std::string()); + } + } + catch (const std::future_error&) + { + } + catch (...) + { + try + { + injectedHeaderSeenPromise.set_value(std::string()); + } + catch (const std::future_error&) + { + } + } + return true; + }); + + headerServer.set_message_handler( + [&headerServer](local_connection_hdl hdl, local_server::message_ptr msg) + { + try + { + headerServer.send(hdl, msg->get_payload(), msg->get_opcode()); + } + catch (...) + { + } + }); + + websocketpp::lib::error_code listenEc; + headerServer.listen(0, listenEc); + ASSERT_FALSE(listenEc) << "Failed to bind header test server to ephemeral port: " << listenEc.message(); + websocketpp::lib::asio::error_code endpointEc; + const uint16_t headerServerPort = headerServer.get_local_endpoint(endpointEc).port(); + ASSERT_FALSE(endpointEc) << "Failed to query header test server local endpoint: " << endpointEc.message(); + + headerServer.start_accept(); + std::thread headerServerThread([&headerServer]() { headerServer.run(); }); + + bool headerServerStopped = false; + auto stopHeaderServer = [&headerServer, &headerServerThread, &headerServerStopped]() + { + if (headerServerStopped) + { + return; + } + headerServer.stop_listening(); + headerServer.stop(); + if (headerServerThread.joinable()) + { + headerServerThread.join(); + } + headerServerStopped = true; + }; + Transport transport; std::promise connectionPromise; auto connectionFuture = connectionPromise.get_future(); @@ -277,35 +386,60 @@ TEST_F(TransportIntegrationUTest, HeaderInjectionAndResponseHeaderRetrieval) { if (connected) { - connectionPromise.set_value(true); + try + { + connectionPromise.set_value(true); + } + catch (const std::future_error&) + { + } } }; auto onMessage = [&](const nlohmann::json& /*msg*/) {}; - // Custom header to inject std::map customHeaders = {{"X-Test-Header", "HeaderValue"}}; + const std::string headerServerUri = "ws://localhost:" + std::to_string(headerServerPort); - Firebolt::Error err = - transport.connect(m_uri, onMessage, onConnectionChange, std::nullopt, std::nullopt, customHeaders); - ASSERT_EQ(err, Firebolt::Error::None); + Firebolt::Error err = transport.connect(std::move(headerServerUri), onMessage, onConnectionChange, std::nullopt, + std::nullopt, customHeaders); + EXPECT_EQ(err, Firebolt::Error::None); + if (err != Firebolt::Error::None) + { + stopHeaderServer(); + return; + } auto status = connectionFuture.wait_for(std::chrono::seconds(2)); - ASSERT_EQ(status, std::future_status::ready) << "Connection timed out"; + EXPECT_EQ(status, std::future_status::ready) << "Connection timed out"; + if (status != std::future_status::ready) + { + stopHeaderServer(); + return; + } EXPECT_TRUE(connectionFuture.get()); - // The server will echo back headers, but since we use websocketpp, only standard headers may be available. - // We check that getResponseHeader returns something for a standard header (e.g., Sec-WebSocket-Accept) - // and for our custom header (may be empty if server does not echo). + auto headerStatus = injectedHeaderSeenFuture.wait_for(std::chrono::seconds(2)); + EXPECT_EQ(headerStatus, std::future_status::ready) << "Server did not observe injected request header"; + if (headerStatus == std::future_status::ready) + { + EXPECT_EQ(injectedHeaderSeenFuture.get(), "HeaderValue"); + } + auto stdHeader = transport.getResponseHeader("Sec-WebSocket-Accept"); EXPECT_TRUE(stdHeader.has_value()); auto customHeader = transport.getResponseHeader("X-Test-Header"); - // Custom header may not be echoed by server, but should not crash - EXPECT_TRUE(customHeader == std::nullopt || customHeader.has_value()); + EXPECT_EQ(customHeader, std::nullopt); + + auto definitelyMissingHeader = transport.getResponseHeader("X-Definitely-Missing-Header"); + EXPECT_EQ(definitelyMissingHeader, std::nullopt); err = transport.disconnect(); EXPECT_EQ(err, Firebolt::Error::None); + EXPECT_EQ(transport.getResponseHeader("Sec-WebSocket-Accept"), std::nullopt); + + stopHeaderServer(); } class TransportCustomServerUTest : public ::testing::Test