Add async future client example

.clang-tidy

@@ -22,7 +22,6 @@ cppcoreguidelines-*,\
 -cppcoreguidelines-pro-type-reinterpret-cast,\
 hicpp-*,\
 misc-*,\
--misc-const-correctness,\
 -misc-include-cleaner,\
 -misc-no-recursion,\
 modernize-*,\

.pre-commit-config.yaml

@@ -15,15 +15,15 @@ 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]
         exclude: docs/TODO.json
 
   # CMake linting and formatting
   - repo: https://github.com/BlankSpruce/gersemi
-    rev: 0.22.3
+    rev: 0.23.2
     hooks:
       - id: gersemi
         name: CMake linting

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,40 +150,60 @@ 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 $<TARGET_FILE:async_tcp_echo_client> #
-            --server $<TARGET_FILE:async_tcp_echo_server>
-    )
-    add_test(
-        NAME async_tcp_echo_client-test-no_server
-        COMMAND
-            ${PYTHON_EXECUTABLE} #
-            ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py #
-            --client $<TARGET_FILE:async_tcp_echo_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 $<TARGET_FILE:async_future_client> #
+        --server $<TARGET_FILE:async_tcp_echo_server>
+)
+add_test(
+    NAME async_future_client-test-no_server
+    COMMAND
+        ${PYTHON_EXECUTABLE} #
+        ${CMAKE_CURRENT_SOURCE_DIR}/run_test.py #
+        --client $<TARGET_FILE:async_future_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 $<TARGET_FILE:async_tcp_echo_client> #
+        --server $<TARGET_FILE:async_tcp_echo_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 $<TARGET_FILE:async_tcp_echo_client> #
+        --timeout 9
+)
+
+if(BUILD_EXAMPLES AND NOT ENABLE_TEST_COVERAGE)
     add_executable(
         blocking_tcp_echo_client
         examples/blocking_tcp_echo_client.cpp
     )
     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(

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"
       }
     },
     {

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

async_rrcp_client.hpp

@@ -34,6 +34,8 @@
 #include <memory>
 #include <string>
 #include <thread>
+#include <unordered_map>
+#include <utility>
 
 #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};