Refactors the socket broadcast client

rmf_websocket/CMakeLists.txt

@@ -24,8 +24,12 @@ find_package(websocketpp REQUIRED)
 find_package(Boost COMPONENTS system filesystem 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
@@ -48,6 +52,28 @@ 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
+    ${Boost_FILESYSTEM_LIBRARY}
+    ${Boost_SYSTEM_LIBRARY}
+    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 +91,27 @@ 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
+  )
+
+  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
+    )
+endif()
+
 ament_package()

rmf_websocket/examples/client.cpp

@@ -0,0 +1,54 @@
+#include <rmf_websocket/BroadcastClient.hpp>
+#include <thread>
+#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/examples/test_harness.py

@@ -0,0 +1,12 @@
+from fastapi import FastAPI, WebSocket
+from fastapi.responses import HTMLResponse
+
+app = FastAPI()
+
+@app.websocket("/")
+async def websocket_endpoint(websocket: WebSocket):
+    await websocket.accept()
+    while True:
+        data = await websocket.receive_text()
+        print(f"Message text was: {data}")
+        #await websocket.send_text(f"Message text was: {data}")

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 {
 //==============================================================================