Refactors the socket broadcast client

rmf_websocket/CMakeLists.txt

@@ -21,11 +21,15 @@ find_package(nlohmann_json REQUIRED)
 find_package(nlohmann_json_schema_validator_vendor REQUIRED)
 find_package(nlohmann_json_schema_validator REQUIRED)
 find_package(websocketpp REQUIRED)
-find_package(Boost COMPONENTS system filesystem REQUIRED)
+find_package(Boost COMPONENTS system REQUIRED)
 find_package(Threads)
 
-file(GLOB_RECURSE core_lib_srcs "src/rmf_websocket/*.cpp")
-add_library(rmf_websocket SHARED ${core_lib_srcs})
+
+add_library(rmf_websocket SHARED
+  src/rmf_websocket/client/ClientWebSocketEndpoint.cpp
+  src/rmf_websocket/BroadcastClient.cpp
+  src/rmf_websocket/BroadcastServer.cpp
+)
 
 target_link_libraries(rmf_websocket
   PUBLIC
@@ -35,8 +39,7 @@ target_link_libraries(rmf_websocket
     nlohmann_json::nlohmann_json
     nlohmann_json_schema_validator
   PRIVATE
-    ${Boost_FILESYSTEM_LIBRARY}
-    ${Boost_SYSTEM_LIBRARY}
+    Boost::system
     Threads::Threads
 )
 
@@ -48,6 +51,26 @@ target_include_directories(rmf_websocket
     ${WEBSOCKETPP_INCLUDE_DIR}
 )
 
+add_executable(example_client
+  examples/client.cpp)
+
+target_link_libraries(example_client
+  PUBLIC
+  rmf_websocket
+    ${websocketpp_LIBRARIES}
+  PRIVATE
+    Threads::Threads
+)
+
+target_include_directories(example_client
+  PUBLIC
+    $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+    $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
+    ${rclcpp_INCLUDE_DIRS}
+    ${WEBSOCKETPP_INCLUDE_DIR}
+)
+
+
 ament_export_targets(rmf_websocket HAS_LIBRARY_TARGET)
 ament_export_dependencies(rmf_traffic rclcpp nlohmann_json websocketpp)
 
@@ -65,4 +88,98 @@ install(
   ARCHIVE DESTINATION lib
 )
 
+if(BUILD_TESTING)
+  find_package(ament_cmake_uncrustify REQUIRED)
+  find_file(uncrustify_config_file
+    NAMES "rmf_code_style.cfg"
+    PATHS "${rmf_utils_DIR}/../../../share/rmf_utils/")
+
+  ament_uncrustify(
+    include src
+    CONFIG_FILE ${uncrustify_config_file}
+    LANGUAGE C++
+    MAX_LINE_LENGTH 80
+  )
+
+  # unit test
+  find_package(ament_cmake_catch2 REQUIRED)
+  ament_add_catch2(test_ring_buffer
+    src/rmf_websocket/utils/RingBuffer_TEST.cpp
+    TIMEOUT 300)
+  target_link_libraries(test_ring_buffer
+    PRIVATE
+      rmf_utils::rmf_utils
+      Boost::system
+    )
+
+#integration test
+  find_package(OpenSSL REQUIRED)
+  ament_add_catch2(test_client
+    test/test_client.cpp
+    TIMEOUT 300)
+  target_link_libraries(test_client
+    PUBLIC
+    ${websocketpp_LIBRARIES}
+      ${rclcpp_LIBRARIES}
+      rmf_utils::rmf_utils
+      Boost::system
+      rmf_websocket
+      PRIVATE
+      Threads::Threads
+      OpenSSL::Crypto
+    )
+  target_include_directories(test_client
+    PUBLIC
+      $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+      $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
+      ${rclcpp_INCLUDE_DIRS}
+      ${WEBSOCKETPP_INCLUDE_DIR}
+  )
+
+
+  ament_add_catch2(test_client_no_server
+    test/test_client_no_server.cpp
+    TIMEOUT 300)
+  target_link_libraries(test_client_no_server
+    PUBLIC
+      ${websocketpp_LIBRARIES}
+      ${rclcpp_LIBRARIES}
+      rmf_utils::rmf_utils
+      Boost::system
+      rmf_websocket
+      PRIVATE
+      Threads::Threads
+      OpenSSL::Crypto
+  )
+  target_include_directories(test_client_no_server
+    PUBLIC
+      $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+      $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
+      ${rclcpp_INCLUDE_DIRS}
+      ${WEBSOCKETPP_INCLUDE_DIR}
+  )
+
+  ament_add_catch2(test_client_with_update_cb
+    test/test_client_with_update_cb.cpp
+    TIMEOUT 300)
+  target_link_libraries(test_client_with_update_cb
+    PUBLIC
+      ${websocketpp_LIBRARIES}
+      ${rclcpp_LIBRARIES}
+      rmf_utils::rmf_utils
+      Boost::system
+      rmf_websocket
+      PRIVATE
+      Threads::Threads
+      OpenSSL::Crypto
+  )
+  target_include_directories(test_client_with_update_cb
+    PUBLIC
+      $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>
+      $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
+      ${rclcpp_INCLUDE_DIRS}
+      ${WEBSOCKETPP_INCLUDE_DIR}
+  )
+endif()
+
 ament_package()

rmf_websocket/examples/client.cpp

@@ -0,0 +1,53 @@
+#include <rmf_websocket/BroadcastClient.hpp>
+#include <rclcpp/rclcpp.hpp>
+
+using namespace rmf_websocket;
+using namespace std::chrono_literals;
+
+std::vector<nlohmann::json> new_connection_data()
+{
+  std::vector<nlohmann::json> msgs;
+  nlohmann::json json;
+  json["hi"] = "Hello";
+  msgs.push_back(json);
+  return msgs;
+}
+
+class MinimalPublisher : public rclcpp::Node
+{
+public:
+  MinimalPublisher()
+  : Node("web_socket_test_node"), count_(0)
+  {
+    timer_ = this->create_wall_timer(
+      10ms, std::bind(&MinimalPublisher::timer_callback, this));
+  }
+
+private:
+  void timer_callback()
+  {
+    if (!client_)
+    {
+      client_ = BroadcastClient::make(
+        "ws://127.0.0.1:8000/",
+        shared_from_this(),
+        &new_connection_data
+      );
+    }
+    obj["count"] = count_++;
+    client_->publish(obj);
+  }
+  rclcpp::TimerBase::SharedPtr timer_;
+  nlohmann::json obj {{"Otototo", true}};
+  std::size_t count_ = 0;
+  std::shared_ptr<BroadcastClient> client_ = nullptr;
+};
+
+
+int main(int argc, char* argv[])
+{
+  rclcpp::init(argc, argv);
+  rclcpp::spin(std::make_shared<MinimalPublisher>());
+  rclcpp::shutdown();
+  return 0;
+}

rmf_websocket/include/rmf_websocket/BroadcastClient.hpp

@@ -24,11 +24,6 @@
 #include <rmf_utils/impl_ptr.hpp>
 
 #include <memory>
-#include <set>
-#include <queue>
-#include <mutex>
-#include <thread>
-#include <atomic>
 
 namespace rmf_websocket {
 //==============================================================================