diff --git a/.clang-tidy b/.clang-tidy index 0392d1d..d4fc0b6 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -22,7 +22,6 @@ cppcoreguidelines-*,\ -cppcoreguidelines-pro-type-reinterpret-cast,\ hicpp-*,\ misc-*,\ --misc-const-correctness,\ -misc-include-cleaner,\ -misc-no-recursion,\ modernize-*,\ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 21cf883..29b4644 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -15,7 +15,7 @@ repos: # This brings in a portable version of clang-format. # See also: https://github.com/ssciwr/clang-format-wheel - repo: https://github.com/pre-commit/mirrors-clang-format - rev: v21.1.2 + rev: v21.1.6 hooks: - id: clang-format types_or: [c++, c, json] @@ -23,7 +23,7 @@ repos: # CMake linting and formatting - repo: https://github.com/BlankSpruce/gersemi - rev: 0.22.3 + rev: 0.23.2 hooks: - id: gersemi name: CMake linting diff --git a/CMakeLists.txt b/CMakeLists.txt index db374d1..ee77bac 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -36,6 +36,15 @@ endif() include(FetchContent) +FetchContent_Declare( + cxxopts + GIT_REPOSITORY https://github.com/jarro2783/cxxopts + GIT_TAG v3.3.1 + EXCLUDE_FROM_ALL + SYSTEM + FIND_PACKAGE_ARGS 3.3.1 HINTS $ENV{HOME}/.local +) + FetchContent_Declare( fmt GIT_TAG 12.0.0 @@ -45,7 +54,7 @@ FetchContent_Declare( FIND_PACKAGE_ARGS 12.0.0 NAMES fmt HINTS $ENV{HOME}/.local ) -FetchContent_MakeAvailable(fmt) +FetchContent_MakeAvailable(fmt cxxopts) # ---- default settings ---- @@ -141,27 +150,48 @@ target_link_libraries( # ---- simple rrcp client class usage examples ---- -if(BUILD_EXAMPLES AND NOT ENABLE_TEST_COVERAGE) - add_executable(async_tcp_echo_client examples/async_tcp_echo_client.cpp) - target_link_libraries(async_tcp_echo_client PRIVATE rrcp_helper) - do_test(async_tcp_echo_client --help Usage) - add_test( - NAME async_tcp_echo_client-test - COMMAND - ${PYTHON_EXECUTABLE} # - ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # - --client $ # - --server $ - ) - add_test( - NAME async_tcp_echo_client-test-no_server - COMMAND - ${PYTHON_EXECUTABLE} # - ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # - --client $ # - --timeout 9 - ) +add_executable(async_future_client examples/async_future_client.cpp) +target_link_libraries(async_future_client PRIVATE rrcp_helper) +do_test(async_future_client --help Usage) +add_test( + NAME async_future_client-test + COMMAND + ${PYTHON_EXECUTABLE} # + ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # + --client $ # + --server $ +) +add_test( + NAME async_future_client-test-no_server + COMMAND + ${PYTHON_EXECUTABLE} # + ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # + --client $ # + --timeout 9 +) +add_executable(async_tcp_echo_client examples/async_tcp_echo_client.cpp) +target_link_libraries(async_tcp_echo_client PRIVATE rrcp_helper) +do_test(async_tcp_echo_client --help Usage) +add_test( + NAME async_tcp_echo_client-test + COMMAND + ${PYTHON_EXECUTABLE} # + ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # + --client $ # + --server $ # + --input ${CMAKE_CURRENT_SOURCE_DIR}/rrcp.txt # +) +add_test( + NAME async_tcp_echo_client-test-no_server + COMMAND + ${PYTHON_EXECUTABLE} # + ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py # + --client $ # + --timeout 9 +) + +if(BUILD_EXAMPLES AND NOT ENABLE_TEST_COVERAGE) add_executable( blocking_tcp_echo_client examples/blocking_tcp_echo_client.cpp @@ -169,12 +199,11 @@ if(BUILD_EXAMPLES AND NOT ENABLE_TEST_COVERAGE) target_link_libraries(blocking_tcp_echo_client PRIVATE rrcp_helper) do_test(blocking_tcp_echo_client --help Usage) - add_executable(rrcp_client rrcp_client.cpp rrcp_message.hpp) + add_executable(rrcp_client examples/rrcp_client.cpp rrcp_message.hpp) target_link_libraries(rrcp_client PRIVATE rrcp_helper) do_test(rrcp_client --help Usage) - # TODO(CK): mv to examples too! - add_executable(timer timer.cpp) + add_executable(timer examples/timer.cpp) target_link_libraries( timer PUBLIC Threads::Threads ${BOOST_LIBRARIES} fmt::fmt-header-only @@ -185,7 +214,10 @@ endif() # ---- theadsafe rrcp client class usage examples main ---- add_executable(rrcp_async_tcp_client_threadsafe rrcp_async_tcp_client.cpp) -target_link_libraries(rrcp_async_tcp_client_threadsafe PRIVATE rrcp_helper) +target_link_libraries( + rrcp_async_tcp_client_threadsafe + PRIVATE rrcp_helper cxxopts::cxxopts +) do_test(rrcp_async_tcp_client_threadsafe --help Usage) add_test( NAME rrcp_async_tcp_client_threadsafe-test-no_server @@ -204,7 +236,10 @@ add_test( ) add_executable(rrcp_async_tcp_client rrcp_async_tcp_client.cpp) -target_link_libraries(rrcp_async_tcp_client PRIVATE rrcp_helper) +target_link_libraries( + rrcp_async_tcp_client + PRIVATE rrcp_helper cxxopts::cxxopts +) target_compile_definitions(rrcp_async_tcp_client PRIVATE USE_SIMPLE_RRCP_CLIENT) do_test(rrcp_async_tcp_client --help Usage) add_test( diff --git a/CMakePresets.json b/CMakePresets.json index a662c06..ff0c4b4 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -80,6 +80,11 @@ ], "cacheVariables": { "CMAKE_TOOLCHAIN_FILE": "infra/cmake/appleclang-toolchain.cmake" + }, + "condition": { + "type": "equals", + "lhs": "${hostSystemName}", + "rhs": "Darwin" } }, { @@ -91,6 +96,11 @@ ], "cacheVariables": { "CMAKE_TOOLCHAIN_FILE": "infra/cmake/appleclang-toolchain.cmake" + }, + "condition": { + "type": "equals", + "lhs": "${hostSystemName}", + "rhs": "Darwin" } }, { diff --git a/GNUmakefile b/GNUmakefile index 4739d12..c978143 100644 --- a/GNUmakefile +++ b/GNUmakefile @@ -46,14 +46,15 @@ $(BUILD_DIR): CMakeLists.txt cmake --preset $(PRESET_NAME) --log-level=VERBOSE # --fresh # -test -d build/Debug && ln -f -s $(CURDIR)/build/Debug $(CURDIR)/$(BUILD_DIR) -check: all +check: $(BUILD_DIR) # XXX all run-clang-tidy -p $(BUILD_DIR) $(CPPFILES) -fix: all +fix: $(BUILD_DIR) # XXX all run-clang-tidy -p $(BUILD_DIR) -fix -checks='-*,\ hicpp-explicit-conversions,\ hicpp-member-init,\ hicpp-named-parameter,\ +misc-const-correctness,\ modernize-deprecated-headers,\ modernize-loop-convert,\ modernize-return-braced-init-list,\ @@ -86,7 +87,10 @@ test: all # -$(BUILD_DIR)/ async_tcp_echo_client localhost # -echo | $(BUILD_DIR)/async_tcp_echo_client localhost 8001 # -killall async_tcp_echo_server - ctest --test-dir $(BUILD_DIR) --rerun-failed --output-on-failure + ctest --test-dir $(BUILD_DIR) --rerun-failed --output-on-failure --repeat-until-fail 1 + ctest --test-dir $(BUILD_DIR) --output-on-failure -R 'async_future_client-test$$' --repeat-until-fail 9 + ctest --test-dir $(BUILD_DIR) --output-on-failure -R 'rrcp_async_tcp_client-test$$' --repeat-until-fail 9 + ctest --test-dir $(BUILD_DIR) --output-on-failure -R 'rrcp_async_tcp_client_threadsafe-test$$' --repeat-until-fail 9 gcovr format: .clang-format diff --git a/async_rrcp_client.hpp b/async_rrcp_client.hpp index 447e639..bedb0a4 100644 --- a/async_rrcp_client.hpp +++ b/async_rrcp_client.hpp @@ -34,6 +34,8 @@ #include #include #include +#include +#include #include "rrcp_helper.hpp" @@ -85,6 +87,70 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client [[nodiscard]] auto connected() const -> bool { return connected_; } +#ifndef USE_OLD_WRITE + // Asynchronously sends a message and calls handler when the response arrives. + void async_write_message( + const std::string& message, std::function< void(const boost::system::error_code&, const std::string&) > handler) + { + if (stopped_) + { + boost::asio::post(io_context_, [handler]() -> void { handler(boost::asio::error::operation_aborted, {}); }); + return; + } + + // Generate a unique message ID + std::string msg_id_str; + msg_id_ = (msg_id_ + 1) % INVALID_ID; + auto command = rrcp::create_command_msg(message, msg_id_str, msg_id_); + + auto self = shared_from_this(); + + // Post the write to the io_context + boost::asio::post(io_context_, + [self, command, msg_id_str, handler = std::move(handler)]() mutable -> void + { + bool const write_in_progress = !self->write_msgs_.empty(); + self->write_msgs_.push_back(command); + + // Start the write loop if not already in progress + if (!write_in_progress) + { + self->deadline_.expires_after(TIMEOUT_DURATION); + self->do_write(); + } + + // Set up a response handler for this specific message ID + self->pending_responses_[msg_id_str] = handler; + }); + } + + void stop() + { + if (stopped_) + { + return; + } + + stopped_ = true; + + boost::asio::post(io_context_, + [self = shared_from_this()]() -> void + { + boost::system::error_code ec; + self->socket_.close(ec); + self->heartbeat_timer_.cancel(); + self->deadline_.cancel(); + + // Notify all pending response handlers about the error + for (auto& [id, handler] : self->pending_responses_) + { + handler(boost::asio::error::operation_aborted, {}); + } + self->pending_responses_.clear(); + }); + } + +#else // This function write the message into the send msg queue and starts the write actor. // It wait for the response message and return this. // @@ -150,8 +216,8 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client } } std::this_thread::sleep_for(125ms); - } while (!stopped_ && --count); - if (!count) + } while (!stopped_ && (--count != 0)); + if (count == 0) { fmt::print(stderr, "Error: Timeout read!\n"); } @@ -179,6 +245,7 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client deadline_.cancel(); }); } +#endif private: void do_read() @@ -188,10 +255,26 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client { if (!ec) { + std::string line; //========================== RRCP ============================ - std::string line = esc2char(self->input_buffer_.substr(1, length - 1)); // w/o START, STOP + if (self->input_buffer_[0] != START) + { + fmt::print(stderr, "Warning: Missing START delimiter (found 0x{:02X})\n", + static_cast< unsigned char >(self->input_buffer_[0])); + } + else + { + line = esc2char(self->input_buffer_.substr(1, length - 1)); + } self->input_buffer_.erase(0, length); //========================== END ============================ + if (line.empty()) + { + fmt::print(stderr, "Warning: Parsed empty message content - skipping\n"); + self->deadline_.expires_after(HEARTBEAT_INTERVAL + TIMEOUT_DURATION); + self->do_read(); + return; + } // TODO(CK): maby refactored to helper class? //========================== RRCP ============================ @@ -206,8 +289,9 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client { // Other responses than Trap and Ping messages fmt::print(stderr, "{}\n", line); // TRACE - self->read_msgs_.push_back(line); + self->handle_response(line); } + // Note: gPing messages are silently ignored (heartbeat responses) //========================== END ============================ self->deadline_.expires_after(HEARTBEAT_INTERVAL + TIMEOUT_DURATION); @@ -221,6 +305,30 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client }); } + void handle_response(std::string& response) + { +#ifndef USE_OLD_WRITE + // Check if this message matches a pending response + // XXX auto msg_id = rrcp::extract_msg_id(line); // implement helper to extract ID + // XXX auto it = pending_responses_.find(msg_id); + // XXX if (it != pending_responses_.end()) + + for (auto it = pending_responses_.begin(); it != pending_responses_.end(); ++it) + { + if (rrcp::find_response_msg(response, it->first)) + { + auto handler = std::move(it->second); + pending_responses_.erase(it); + // Call handler asynchronously + boost::asio::post(io_context_, [handler = std::move(handler), response]() -> void { handler({}, response); }); + break; + } + } +#else + read_msgs_.push_back(response); +#endif + } + void do_write() { boost::asio::async_write(socket_, boost::asio::buffer(write_msgs_.front()), @@ -297,9 +405,15 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client // I/O buffers (protected by io_context) std::string input_buffer_; - message_queue read_msgs_; message_queue write_msgs_; +#ifndef USE_OLD_WRITE + std::unordered_map< std::string, std::function< void(const boost::system::error_code& ec, std::string) > > + pending_responses_; +#else + message_queue read_msgs_; +#endif + // Thread-safe state std::atomic< bool > connected_{false}; std::atomic< bool > stopped_{false}; diff --git a/async_rrcp_client_threadsafe.hpp b/async_rrcp_client_threadsafe.hpp index 3336bd5..3667a3a 100644 --- a/async_rrcp_client_threadsafe.hpp +++ b/async_rrcp_client_threadsafe.hpp @@ -31,10 +31,9 @@ #include #include #include -#include #include #include -#include +#include // NOTE: for std::promise too #include #include #include @@ -170,55 +169,10 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client } private: -#define USE_PEDANTIC_CHECKES -#ifdef USE_PEDANTIC_CHECKES - // Helper method to safely parse RRCP message with bounds checking - static auto parse_rrcp_message(const std::string& buffer, std::size_t length, std::string& parsed_line) -> bool - { - // Minimum RRCP message: START + at least 1 char + STOP = 3 bytes - if (length < 3) - { - fmt::print(stderr, "Warning: Message too short (length={}) - expected minimum 3 bytes\n", length); - return false; - } - - // Validate buffer size - if (buffer.size() < length) - { - fmt::print(stderr, "Error: Buffer size ({}) smaller than expected length ({})\n", buffer.size(), length); - return false; - } - - // Check for START delimiter at beginning - if (buffer[0] != START) - { - fmt::print(stderr, "Warning: Missing START delimiter (found 0x{:02X})\n", static_cast< unsigned char >(buffer[0])); - return false; - } - - // Check for STOP delimiter at expected position - if (buffer[length - 1] != STOP) - { - fmt::print(stderr, "Warning: Missing STOP delimiter at position {} (found 0x{:02X})\n", length - 1, - static_cast< unsigned char >(buffer[length - 1])); - return false; - } - - // Extract message content (without START and STOP) - if (length >= 3) - { - parsed_line = esc2char(buffer.substr(1, length - 2)); - return true; - } - - return false; - } -#endif - void execute_write_request(const std::string& message, std::shared_ptr< response_promise_type > promise) { std::string msg_id_str; - int current_id = next_message_id_.fetch_add(1); + int const current_id = next_message_id_.fetch_add(1); auto command = rrcp::create_command_msg(message, msg_id_str, current_id); // Store promise for response correlation @@ -304,50 +258,46 @@ class async_rrcp_client : public std::enable_shared_from_this< async_rrcp_client { if (!ec) { - //========================== RRCP ============================ -#ifdef USE_PEDANTIC_CHECKES - std::string parsed_line; - // Use safe parsing helper with comprehensive bounds checking - if (!rrcp::async_rrcp_client::parse_rrcp_message(self->input_buffer_, length, parsed_line)) + std::string line; + //========================== RRCP ============================ + // Check for START delimiter at beginning + if (self->input_buffer_[0] != START) { - // Parsing failed - message was malformed, skip it - self->input_buffer_.erase(0, length); - self->deadline_.expires_after(HEARTBEAT_INTERVAL + TIMEOUT_DURATION); - self->do_read(); - return; + fmt::print(stderr, "Warning: Missing START delimiter (found 0x{:02X})\n", + static_cast< unsigned char >(self->input_buffer_[0])); + } + else + { + line = esc2char(self->input_buffer_.substr(1, length - 1)); } -#else - std::string parsed_line = esc2char(self->input_buffer_.substr(1, length - 1)); // TODO(CK): check START, STOP? -#endif + // Successfully parsed, remove processed data from buffer self->input_buffer_.erase(0, length); - //========================== END ============================ + //========================== END ============================ -#ifdef USE_PEDANTIC_CHECKES // Validate parsed content is not empty - if (parsed_line.empty()) + if (line.empty()) { fmt::print(stderr, "Warning: Parsed empty message content - skipping\n"); self->deadline_.expires_after(HEARTBEAT_INTERVAL + TIMEOUT_DURATION); self->do_read(); return; } -#endif // TODO(CK): maby refactored to helper class? //========================== RRCP ============================ // Process different message types - if (boost::algorithm::starts_with(parsed_line, "d")) // Trap data message + if (boost::algorithm::starts_with(line, "d")) // Trap data message { // Handle trap data messages - fmt::print(stderr, "trap data: {}\n", parsed_line); // TRACE - self->trap_handler_(parsed_line); + fmt::print(stderr, "trap data: {}\n", line); // TRACE + self->trap_handler_(line); } - else if (!boost::algorithm::starts_with(parsed_line, "gPing")) + else if (!boost::algorithm::starts_with(line, "gPing")) { // Handle response messages (but ignore heartbeat responses) - fmt::print(stderr, "{}\n", parsed_line); // TRACE - self->handle_response(parsed_line); + fmt::print(stderr, "{}\n", line); // TRACE + self->handle_response(line); } // Note: gPing messages are silently ignored (heartbeat responses) //========================== END ============================ diff --git a/examples/async_future_client.cpp b/examples/async_future_client.cpp new file mode 100644 index 0000000..8c4e2e1 --- /dev/null +++ b/examples/async_future_client.cpp @@ -0,0 +1,359 @@ +/*** + * async_future_client.cpp + * ~~~~~~~~~~~~~~~~~~~~~~~~~ + * + * Copyright (c) 2003-2024 Christopher M. Kohlhoff (chris at kohlhoff dot com) + * + * Distributed under the Boost Software License, Version 1.0. (See accompanying + * file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) + * + * Modernized from Claus Klein and ChatGPT + ***/ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/** + * @brief Protocol Requirements and Design Overview + * + * 1. Each request message starts with ASCII message number (zero-padded to 5 digits). + * 2. Messages prepend a 4-byte ASCII length header. + * 3. Responses follow the same format and must match the request msg_num. + * 4. The client stores pending requests in a map: std::map> pending_; + * 5. The receiver reads the length header, reads the exact message body, extracts msg_num, + * and fulfills the corresponding promise in the map. + */ +using namespace std::chrono_literals; +using boost::asio::ip::tcp; + +/** + * @brief Structure representing a response from the server + */ +struct response +{ + std::string msg_num_; /**< ASCII message number from server */ + std::vector< uint8_t > data_; /**< Payload bytes */ +}; + +/** + * @brief Asynchronous TCP client supporting future-based requests + * + * This client: + * - Sends requests asynchronously. + * - Reads responses and fulfills promises associated with the message number. + * - Supports multiple concurrent outstanding requests. + */ +class async_future_client : public std::enable_shared_from_this< async_future_client > +{ + public: + /** + * @brief Construct a new async_future_client + * @param ctx Reference to an existing Boost.Asio io_context + * @param host Hostname or IP address of the server + * @param port TCP port of the server + */ + async_future_client(boost::asio::io_context& ctx, const std::string& host, const std::string& port) + : io_context_(ctx), resolver_(ctx), socket_(ctx) + { + resolver_.async_resolve(host, port, + [this](boost::system::error_code ec, const tcp::resolver::results_type& results) -> void + { + if (!ec) + { + boost::asio::async_connect(socket_, results, + [this](boost::system::error_code ec, const tcp::endpoint&) -> void + { + if (!ec) + { + fmt::print(stderr, "Connected to server.\n"); + connected_ = true; + do_read_length(); + } + else + { + // No more endpoints to try; shut down client + stop(); + } + }); + } + }); + } + + /** + * @brief Send a request payload asynchronously + * + * The function returns a std::future which will be set once + * the corresponding response arrives from the server. + * + * @param payload Vector of bytes to send + * @return std::future Future to get the response + */ + auto send_request(const std::vector< uint8_t >& payload) -> std::future< response > + { +#ifdef HAS_ATOMIC_THREAD_FENCE + uint16_t num = msg_counter_.fetch_add(1, std::memory_order_relaxed) % std::numeric_limits< uint16_t >::max(); +#else + uint16_t num{}; + { + std::scoped_lock const lock(map_mutex_); + num = msg_counter_ + 1; + num = num % std::numeric_limits< uint16_t >::max(); + msg_counter_ = num; + } +#endif + + std::string msg_num = fmt::format("{:05}", num); // 5-digit ASCII message number + + // Build message: [msg_num + payload] + std::vector< uint8_t > body; + body.insert(body.end(), msg_num.begin(), msg_num.end()); + body.insert(body.end(), payload.begin(), payload.end()); + + // Prepend 4-byte ASCII length header + std::string len_str = fmt::format("{:04}", body.size()); + std::vector< uint8_t > full_msg; + full_msg.insert(full_msg.end(), len_str.begin(), len_str.end()); + full_msg.insert(full_msg.end(), body.begin(), body.end()); + + // Prepare promise and future + std::promise< response > prom; + auto fut = prom.get_future(); + { + std::scoped_lock const lock(map_mutex_); + pending_.emplace(msg_num, std::move(prom)); + } + + size_t counter{4}; + while (!connected_) + { + if (stopped_) + { + return fut; + } + fmt::print(stderr, "Client is not connected yet.\n"); + if (--counter == 0U) + { + return fut; + } + std::this_thread::sleep_for(125ms * counter); + } + + fmt::print("Send msg ({})\n", std::string(full_msg.begin(), full_msg.end())); + + boost::asio::async_write(socket_, boost::asio::buffer(full_msg), + [this, msg_num](std::error_code ec, std::size_t /*n*/) -> void + { + if (ec) + { + fmt::print(stderr, "Write Error: send_request({}) {}\n", msg_num, ec.message()); + + std::scoped_lock const lock(map_mutex_); + auto it = pending_.find(msg_num); + if (it != pending_.end()) + { + it->second.set_exception(std::make_exception_ptr(std::runtime_error(ec.message()))); + pending_.erase(it); + } + else + { + fmt::print(stderr, "Error: num not found at send_request({}) !\n", msg_num); + } + } + }); + + return fut; + } + + /** + * @brief Stop the client and close the socket + * + * This will cancel all ongoing operations and prevent new requests. + */ + void stop() + { + if (stopped_) + { + return; + } + + boost::asio::post(io_context_, + [this]() -> void + { + stopped_ = true; + connected_ = false; + boost::system::error_code ignored_error; + socket_.close(ignored_error); + }); + } + + /** + * @brief Example test function that repeatedly sends a payload + * + * Demonstrates how to use send_request() and wait for the future. + */ + void test() + { + std::vector< uint8_t > const payload{'H', 'e', 'l', 'l', 'o', ' ', 'T', 'e', 's', 't'}; + do + { + auto fut = send_request(payload); + if (fut.wait_for(1s) == std::future_status::ready) + { + auto resp = fut.get(); + fmt::print("Received response ({}): {}\n", resp.msg_num_, std::string(resp.data_.begin(), resp.data_.end())); + } + else + { + fmt::print(stderr, "Timeout waiting for response\n"); + break; + } + } while (true); + } + + private: + /** + * @brief Start reading the 4-byte length header + */ + void do_read_length() + { + boost::asio::async_read(socket_, boost::asio::buffer(len_buf_), + [this](std::error_code ec, std::size_t /*n*/) -> void + { + if (!ec) + { + std::string const len_str(len_buf_.begin(), len_buf_.end()); + std::size_t const msg_len = std::stoul(len_str, nullptr, 10); + do_read_body(msg_len); + } + else + { + fmt::print(stderr, "Read length Error: do_read_length() {}\n", ec.message()); + // XXX std::runtime_error(ec.message()); + // There are no more endpoints to try. Silently shut down the client. + stop(); + } + }); + } + + /** + * @brief Read the message body after reading the length header + * @param msg_len Length of the message body + */ + void do_read_body(std::size_t msg_len) + { + fmt::print(stderr, "do_read_body({})\n", msg_len); + + body_buf_.resize(msg_len); + boost::asio::async_read(socket_, boost::asio::buffer(body_buf_), + [this, msg_len](std::error_code ec, std::size_t /*n*/) -> void + { + if (!ec && body_buf_.size() >= 5) + { + std::string const msg_num(body_buf_.begin(), body_buf_.begin() + 5); + std::vector< uint8_t > payload(body_buf_.begin() + 5, body_buf_.end()); + handle_response(msg_num, std::move(payload)); + do_read_length(); + } + else if (ec) + { + fmt::print(stderr, "Read body Error: do_read_body({}) {}\n", msg_len, ec.message()); + std::runtime_error(ec.message()); + } + }); + } + + /** + * @brief Fulfill the promise associated with the received message number + * @param msg_num Message number as ASCII string + * @param data Payload data received + */ + void handle_response(const std::string& msg_num, std::vector< uint8_t > data) + { + fmt::print(stderr, "do_read_body({}) {}\n", msg_num, + std::string_view(reinterpret_cast< const char* >(data.data()), data.size())); + + std::scoped_lock const lock(map_mutex_); + auto it = pending_.find(msg_num); + if (it != pending_.end()) + { + it->second.set_value(response{msg_num, std::move(data)}); + pending_.erase(it); + } + else + { + fmt::print(stderr, "Error: num not found at do_read_body({}) !\n", msg_num); + } + } + + boost::asio::io_context& io_context_; /**< Reference to Boost.Asio io_context */ + tcp::resolver resolver_; /**< Resolver for hostname lookup */ + tcp::socket socket_; /**< TCP socket */ + std::array< char, 4 > len_buf_{}; /**< Buffer for 4-byte length header */ + std::vector< uint8_t > body_buf_; /**< Buffer for message body */ + +#ifdef HAS_ATOMIC_THREAD_FENCE + std::atomic< uint16_t > msg_counter_{0}; /**< Message counter (atomic) */ +#else + uint16_t msg_counter_{0}; /**< Message counter (protected by mutex) */ +#endif + + std::atomic< bool > connected_{false}; /**< Flag: connection established */ + std::atomic< bool > stopped_{false}; /**< Flag: client stopped */ + + std::mutex map_mutex_; /**< Protects access to pending_ map */ + std::map< std::string, std::promise< response > > pending_; /**< Map of pending requests */ +}; + +/** + * @brief Example + */ +auto main(int argc, char* argv[]) -> int +{ + if (argc != 3) + { + fmt::print("Usage: {} \n", argv[0]); + return EXIT_FAILURE; + } + + std::string const host = argv[1]; + std::string const port = argv[2]; + + try + { + boost::asio::io_context ctx; + auto client = std::make_shared< async_future_client >(ctx, host, port); + + std::thread io_thread([&ctx] -> void { ctx.run(); }); + + std::this_thread::sleep_for(1s); + + std::thread test1(&async_future_client::test, client); + // TODO(CK) std::thread test2(&async_future_client::test, client); + + test1.join(); + // TODO(CK) test2.join(); + + ctx.stop(); + io_thread.join(); + } + catch (const std::exception& ex) + { + fmt::print(stderr, "Exception: {}\n", ex.what()); + return EXIT_FAILURE; + } + + return EXIT_SUCCESS; +} diff --git a/examples/async_tcp_echo_client.cpp b/examples/async_tcp_echo_client.cpp index f2c60c0..fd14df6 100644 --- a/examples/async_tcp_echo_client.cpp +++ b/examples/async_tcp_echo_client.cpp @@ -12,6 +12,7 @@ #include +#include #include #include #include @@ -21,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -97,11 +99,19 @@ class asynchronous_tcp_client : public std::enable_shared_from_this< asynchronou // response to graceful termination or an unrecoverable error. void stop() { - boost::system::error_code ignored_error; - socket_.close(ignored_error); - timer_.cancel(); - connected_ = true; - stopped_ = true; + if (stopped_) + { + return; + } + boost::asio::post(io_context_, + [this, self = shared_from_this()]() -> void + { + stopped_ = true; + connected_ = false; + boost::system::error_code ignored_error; + socket_.close(ignored_error); + timer_.cancel(); + }); } private: @@ -188,19 +198,33 @@ class asynchronous_tcp_client : public std::enable_shared_from_this< asynchronou boost::asio::steady_timer timer_; std::string data_; message_queue write_msgs_; - bool connected_{false}; - bool stopped_{false}; + std::atomic< bool > connected_{false}; + std::atomic< bool > stopped_{false}; }; // NOLINTNEXTLINE(bugprone-exception-escape) auto main(int argc, char* argv[]) -> int { - if (argc != 3) + if (argc < 3) { - fmt::print(stderr, "Usage: {} \n", argv[0]); + fmt::print(stderr, "Usage: {} [input_file]\n", argv[0]); // NOLINT return EXIT_FAILURE; } + std::ifstream file; // persistent file object (if used) + std::istream* input_str = &std::cin; // pointer to chosen input stream + + if (argc == 4) + { + file.open(argv[3]); // NOLINT + if (!file) + { + fmt::print(stderr, "cannot open input file: {}\n", argv[3]); // NOLINT + return 2; + } + input_str = &file; + } + try { boost::asio::io_context io_context; @@ -209,7 +233,7 @@ auto main(int argc, char* argv[]) -> int std::thread io_thread([&io_context]() -> void { io_context.run(); }); - for (std::string line; std::getline(std::cin, line); fmt::print(stderr, "Enter command: ")) + for (std::string line; std::getline(*input_str, line); fmt::print(stderr, "Enter command: ")) { const std::string::size_type sz = line.find("//"); if ((sz != std::string::npos)) diff --git a/examples/async_tcp_echo_server.cpp b/examples/async_tcp_echo_server.cpp index 934daf9..2a90e74 100644 --- a/examples/async_tcp_echo_server.cpp +++ b/examples/async_tcp_echo_server.cpp @@ -49,15 +49,15 @@ class session : public std::enable_shared_from_this< session > { if (!ec) { - if ((std::string_view(data_.data(), length).contains("M:Utility")) || - (std::string_view(data_.data(), length).contains("M:A")) || - (std::string_view(data_.data(), length).contains("M:C"))) + std::string_view const msg{data_.data(), length}; + if ((msg.contains("M:Utility")) || (msg.contains("000")) || (msg.contains("M:A")) || (msg.contains("M:C"))) { + std::cerr << "do_read(len=" << length << "): " << msg << "\n"; do_write(length); } else { - std::size_t new_len = gen_random(length); + std::size_t const new_len = gen_random(length); #define CHANGE_ECHO_MSG #ifndef CHANGE_ECHO_MSG @@ -73,6 +73,8 @@ class session : public std::enable_shared_from_this< session > } #endif + std::string_view const msg{data_.data(), new_len}; + std::cerr << "do_write(len=" << length << "): " << msg << "\n"; do_write(new_len); } } @@ -100,7 +102,7 @@ class session : public std::enable_shared_from_this< session > }); } - static std::size_t gen_random(std::size_t input) + static auto gen_random(std::size_t input) -> std::size_t { static std::random_device rd; // a seed source for the random number engine static std::mt19937 gen(rd()); // mersenne_twister_engine seeded with rd() diff --git a/rrcp_client.cpp b/examples/rrcp_client.cpp similarity index 100% rename from rrcp_client.cpp rename to examples/rrcp_client.cpp diff --git a/timer.cpp b/examples/timer.cpp similarity index 100% rename from timer.cpp rename to examples/timer.cpp diff --git a/gcovr.cfg b/gcovr.cfg index c00744f..ad20de6 100644 --- a/gcovr.cfg +++ b/gcovr.cfg @@ -6,8 +6,9 @@ filter = RRCP* filter = async_* filter = examples/* filter = rrcp_* -# filter = tests/* +# NO! filter = tests/* exclude = tests +exclude = rrcp_message.hpp # exclude-directories = build/_deps exclude-directories = coverage diff --git a/rrcp_async_tcp_client.cpp b/rrcp_async_tcp_client.cpp index b3dd392..18268e6 100644 --- a/rrcp_async_tcp_client.cpp +++ b/rrcp_async_tcp_client.cpp @@ -15,6 +15,7 @@ #include #include #include +#include #include #include #include @@ -34,26 +35,84 @@ namespace void print(std::string msg) { fmt::print("{}\n", msg); } +// a little helper to return all args +struct options +{ + uint16_t port{8000}; + uint32_t timeout{3}; + bool verbose{true}; + std::string server{"localhost"}; + std::string filename{"-"}; +}; + +options getcxxopts(int argc, char** argv) +{ + options opt; + try + { + cxxopts::Options options(*argv, "rrcp"); + + // clang-format off + options.add_options() + ("h,help", "this help") + ("t,timeout", "Message timeout", cxxopts::value()->default_value("3")) + ("v,verbose", "trace notification handling", cxxopts::value()->default_value("true")) + ("p,port", "rrcpPort, default 8000", cxxopts::value()->default_value(std::to_string(opt.port))) + ("s,server", "Server name to connect", cxxopts::value()->default_value(opt.server)) + ("f,file", "File name to read from", cxxopts::value()->default_value(opt.filename)); + // clang-format on + + auto result = options.parse(argc, argv); + if (result.contains("help")) + { + std::cout << options.help() << '\n'; + exit(EXIT_SUCCESS); // NOLINT(concurrency-mt-unsafe) + } + if (result.contains("timeout")) + { + opt.timeout = result["timeout"].as< uint32_t >(); + } + if (result.contains("verbose")) + { + opt.verbose = result["verbose"].as< bool >(); + } + if (result.contains("port")) + { + opt.port = result["port"].as< uint16_t >(); + } + if (result.contains("server")) + { + opt.server = result["server"].as< std::string >(); + } + if (result.contains("file")) + { + opt.filename = result["file"].as< std::string >(); + } + } + catch (const cxxopts::exceptions::exception& e) + { + fmt::print(stderr, "cxxopt parse exception: {}", e.what()); + exit(EXIT_FAILURE); // NOLINT(concurrency-mt-unsafe) + } + return opt; +} + } // namespace // NOLINTNEXTLINE(bugprone-exception-escape) auto main(int argc, char* argv[]) -> int { - if (argc < 3) - { - fmt::print(stderr, "Usage: {} [input_file]\n", argv[0]); // NOLINT - return EXIT_FAILURE; - } + auto opts = getcxxopts(argc, argv); std::ifstream file; // persistent file object (if used) std::istream* input_str = &std::cin; // pointer to chosen input stream - if (argc == 4) + if (opts.filename != "-") { - file.open(argv[3]); // NOLINT + file.open(opts.filename); if (!file) { - fmt::print(stderr, "cannot open input file: {}\n", argv[3]); // NOLINT + fmt::print(stderr, "cannot open input file: {}\n", opts.filename); // NOLINT return 2; } input_str = &file; @@ -68,11 +127,11 @@ auto main(int argc, char* argv[]) -> int auto client = std::make_shared< async_rrcp_client >(io_context); client->register_trap_handler(&print); - client->start(resolver.resolve(argv[1], argv[2])); + client->start(resolver.resolve(opts.server, std::to_string(opts.port))); std::thread io_thread([&io_context]() -> void { io_context.run(); }); - std::this_thread::sleep_for(TIMEOUT_DURATION); // NOTE: only for gcov results! CK + std::this_thread::sleep_for(opts.timeout * 500ms); // NOTE: only for gcov results! CK for (std::string line; client->connected() && std::getline(*input_str, line); fmt::print(stderr, "Enter command: ")) { const std::string::size_type sz = line.find("//"); @@ -92,10 +151,25 @@ auto main(int argc, char* argv[]) -> int continue; } +#if defined(USE_SIMPLE_RRCP_CLIENT) && !defined(USE_OLD_WRITE) + client->async_write_message(line, + [](const boost::system::error_code& ec, const std::string& response) -> void + { + if (!ec) + { + fmt::print("Received response: {}\n", response); + } + else + { + fmt::print("Error: {}\n", ec.message()); + } + }); +#else const auto response = client->write(line); fmt::print("{}\n", response); +#endif } - std::this_thread::sleep_for(HEARTBEAT_INTERVAL); // NOTE: only for gcov results! CK + std::this_thread::sleep_for(opts.timeout * 500ms); // NOTE: only for gcov results! CK client->stop(); io_thread.join(); diff --git a/rrcp_helper.cpp b/rrcp_helper.cpp index 1248eb5..d52eae4 100644 --- a/rrcp_helper.cpp +++ b/rrcp_helper.cpp @@ -90,14 +90,14 @@ auto rrcp::insertAfterFirstWord(const std::string& input, const std::string& toI return input; // Nothing to do } - size_t firstSpace = input.find_first_of(" \t"); // Find first whitespace + size_t const firstSpace = input.find_first_of(" \t"); // Find first whitespace if (firstSpace == std::string::npos) { return input; // No spaces found, return original string } // NOTE: Only if Set/Get command request, NOT for Trap commands! - size_t nextNonSpace = input.find_first_of("SG", firstSpace); + size_t const nextNonSpace = input.find_first_of("SG", firstSpace); if (nextNonSpace == std::string::npos) { // If there's no second valid command, just return the input! @@ -158,7 +158,7 @@ auto rrcp::create_command_msg(const std::string& message, std::string& msg_id_st { msg_id_str.clear(); } - std::string msg = insertAfterFirstWord(message, msg_id_str); + std::string const msg = insertAfterFirstWord(message, msg_id_str); // DEBUG: fmt::print("rrcp MU to send({})\n", msg); diff --git a/tests/RRCP-test.cpp b/tests/RRCP-test.cpp index 4adc431..ef39110 100644 --- a/tests/RRCP-test.cpp +++ b/tests/RRCP-test.cpp @@ -11,6 +11,38 @@ namespace ut = boost::ut; +namespace +{ +// Helper to dump a string as hex +std::string to_hex(const std::string& s) +{ + std::ostringstream oss; + oss << std::hex << std::setfill('0'); + for (unsigned char c : s) + { + oss << std::setw(2) << static_cast< int >(c) << ' '; + } + return oss.str(); +} +} // namespace + +#if 0 +int main() { + using namespace ut; + + "hex dump example"_test = [] { + std::string data = "Hello\n"; + auto hex = to_hex(data); + + // You can print or assert + std::cout << "Hex dump: " << hex << std::endl; + + // Example assertion: expected specific hex string + expect(hex == "48 65 6c 6c 6f 0a "); + }; +} +#endif + ut::suite errors = [] -> void { using namespace ut; @@ -167,14 +199,14 @@ ut::suite errors = [] -> void auto quoted = rrcp::char2esc(BINARY); // NOTE: std::quoted works only with std::stringstream -#if defined(BOOST_UT_HAS_FORMAT) && defined(FIXME) // FIXME! +#if defined(BOOST_UT_HAS_FORMAT) std::ostringstream binary_bin; - binary_bin << std::quoted(binary); - ut::log("{} {}\n", binary.length(), binary_bin.str()); + binary_bin << std::quoted(BINARY); + ut::log("{}: {}\n", BINARY.length(), binary_bin.str()); std::ostringstream quoted_bin; quoted_bin << std::quoted(quoted); - ut::log("{} {}\n", quoted.length(), quoted_bin.str()); + ut::log("{}: {}\n", quoted.length(), quoted_bin.str()); #endif expect(BINARY == rrcp::esc2char(quoted)); @@ -183,6 +215,19 @@ ut::suite errors = [] -> void expect(quoted.length() == 33); }; + "special_quoteing"_test = [] -> void + { + constexpr std::string_view BINARY{"(\n\r\x1b)"sv}; + auto quoted = rrcp::char2esc(BINARY); + expect(BINARY.length() == 5); + expect(quoted.length() == 8); + expect(BINARY == rrcp::esc2char(quoted)); +#if defined(BOOST_UT_HAS_FORMAT) + auto hex = to_hex(quoted); + ut::log("{}: {}\n", quoted.length(), hex); +#endif + }; + // ============================================================ "rrcp_message"_test = [] -> void