feat: Add subscriber plugins using pluginlib

CMakeLists.txt

@@ -46,6 +46,7 @@ add_library(${PROJECT_NAME} SHARED
   src/web_video_server.cpp
   src/multipart_stream.cpp
   src/streamer.cpp
+  src/subscriber.cpp
   src/utils.cpp
 )
 
@@ -61,14 +62,15 @@ target_link_libraries(${PROJECT_NAME}
     async_web_server_cpp::async_web_server_cpp
     pluginlib::pluginlib
     rclcpp::rclcpp
+    ${sensor_msgs_TARGETS}
     rmw::rmw
     Boost::boost
   PRIVATE
     rclcpp_components::component
 )
 
 add_library(${PROJECT_NAME}_streamers SHARED
-  src/streamers/image_transport_streamer.cpp
+  src/streamers/image_streamer.cpp
   src/streamers/libav_streamer.cpp
   src/streamers/h264_streamer.cpp
   src/streamers/jpeg_streamers.cpp
@@ -92,7 +94,6 @@ target_link_libraries(${PROJECT_NAME}_streamers
   ${PROJECT_NAME}
   async_web_server_cpp::async_web_server_cpp
   cv_bridge::cv_bridge
-  image_transport::image_transport
   pluginlib::pluginlib
   rclcpp::rclcpp
   ${sensor_msgs_TARGETS}
@@ -105,6 +106,25 @@ target_link_libraries(${PROJECT_NAME}_streamers
   ${swscale_LIBRARIES}
 )
 
+add_library(${PROJECT_NAME}_subscribers SHARED
+  src/subscribers/image_transport_subscriber.cpp
+)
+
+target_include_directories(${PROJECT_NAME}_subscribers
+  PUBLIC
+    "$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>"
+    "$<INSTALL_INTERFACE:include/${PROJECT_NAME}>"
+)
+
+target_link_libraries(${PROJECT_NAME}_subscribers
+  ${PROJECT_NAME}
+  async_web_server_cpp::async_web_server_cpp
+  image_transport::image_transport
+  pluginlib::pluginlib
+  rclcpp::rclcpp
+  ${sensor_msgs_TARGETS}
+)
+
 ## Declare a cpp executable
 add_executable(${PROJECT_NAME}_node
   src/web_video_server_node.cpp
@@ -116,7 +136,8 @@ target_link_libraries(${PROJECT_NAME}_node
 
 rclcpp_components_register_nodes(${PROJECT_NAME} "web_video_server::WebVideoServer")
 
-pluginlib_export_plugin_description_file(web_video_server plugins.xml)
+pluginlib_export_plugin_description_file(web_video_server streamer_plugins.xml)
+pluginlib_export_plugin_description_file(web_video_server subscriber_plugins.xml)
 
 #############
 ## Install ##
@@ -128,7 +149,7 @@ install(
 )
 
 install(
-  TARGETS ${PROJECT_NAME} ${PROJECT_NAME}_streamers
+  TARGETS ${PROJECT_NAME} ${PROJECT_NAME}_streamers ${PROJECT_NAME}_subscribers
   EXPORT export_${PROJECT_NAME}
   LIBRARY DESTINATION lib
   ARCHIVE DESTINATION lib

README.md

@@ -4,18 +4,22 @@ This node provides HTTP streaming of ROS topics in various formats, making it ea
 
 ## Features
 
+- Subscribe to ROS topics in multiple formats:
+  - image_transport
+
 - Stream ROS image topics over HTTP in multiple formats:
   - MJPEG (Motion JPEG)
   - VP8 (WebM)
   - VP9 (WebM)
   - H264 (MP4)
   - PNG streams
   - ROS compressed image streams
+
 - Query snapshots of image topics in multiple formats:
   - JPEG
   - PNG
   - ROS compressed image
-- Plugin-based architecture for easy addition of new streaming formats
+- Plugin-based architecture for easy addition of new subscribers or streamer formats
 - Adjustable quality, size, and other streaming parameters
 - Web interface to browse available image topics
 - Support for different QoS profiles in ROS 2
@@ -91,6 +95,7 @@ ros2 run web_video_server web_video_server
 | `server_threads` | int | 1 | 1+ | Number of server threads for handling HTTP requests |
 | `ros_threads` | int | 2 | 1+ | Number of threads for ROS message handling |
 | `verbose` | bool | false | true, false | Enable verbose logging |
+| `default_qos_profile` | string | "default" | "default", "system_default", "sensor_data", "services_default" | QoS profile for ROS 2 subscribers |
 | `default_stream_type` | string | "mjpeg" | "mjpeg", "vp8", "vp9", "h264", "png", "ros_compressed" | Default format for video streams |
 | `publish_rate` | double | -1.0 | -1.0 or positive value | Rate for republishing images (-1.0 means no republishing) |
 
@@ -181,6 +186,9 @@ http://localhost:8080/snapshot?topic=/camera/image_raw
 ## Creating custom streamer plugins
 See the [custom streamer plugin tutorial](doc/custom-streamer-plugin.md) for information on how to write your own streamer plugins.
 
+## Creating custom subscriber plugins
+See the [custom subscriber plugin tutorial](doc/custom-subscriber-plugin.md) for information on how to write your own subscriber plugins.
+
 ## About
 This project is released as part of the [Robot Web Tools](https://robotwebtools.github.io/) effort.
 

doc/custom-streamer-plugin.md

@@ -15,10 +15,7 @@ This tutorial will guide you through the steps to create a simple custom streame
 
 1. Add `TestStreamer` and `TestStreamerFactory` classes to `include/test_streamer_plugin/test_streamer_plugin.hpp` header file:
     ```cpp
-    #ifndef TEST_STREAMER_PLUGIN__TEST_STREAMER_PLUGIN_HPP_
-    #define TEST_STREAMER_PLUGIN__TEST_STREAMER_PLUGIN_HPP_
-
-    #include "test_streamer_plugin/visibility_control.h"
+    #pragma once
 
     #include "web_video_server/streamer.hpp"
 
@@ -52,7 +49,6 @@ This tutorial will guide you through the steps to create a simple custom streame
 
     }  // namespace test_streamer_plugin
 
-    #endif  // TEST_STREAMER_PLUGIN__TEST_STREAMER_PLUGIN_HPP_
     ```
 
 1. Implement the `TestStreamer` and `TestStreamerFactory` classes in `src/test_streamer_plugin.cpp`:

doc/custom-subscriber-plugin.md

@@ -0,0 +1,243 @@
+# How to write a custom subscriber plugin
+
+This tutorial will guide you through the steps to create a simple custom subscriber plugin for the `web_video_server` package in ROS 2. The example plugin will log messages when it is created, started, and when frames are restreamed.
+
+1. Create you local workspace if you don't have one:
+    ```bash
+    mkdir -p ~/ros_ws/src
+    cd ~/ros_ws/src
+    ```
+1. Create a new package for your custom subscriber plugin:
+    ```bash
+    ros2 pkg create --build-type ament_cmake test_subscriber_plugin --dependencies web_video_server pluginlib --library-name test_subscriber_plugin
+    cd test_subscriber_plugin
+    ```
+
+1. Add `TestSubscriber` and `TestSubscriberFactory` classes to `include/test_subscriber_plugin/test_subscriber_plugin.hpp` header file:
+    ```cpp
+    #pragma once
+
+    #include "web_video_server/subscriber.hpp"
+
+    #include <opencv2/opencv.hpp>
+    #ifdef CV_BRIDGE_USES_OLD_HEADERS
+    #include <cv_bridge/cv_bridge.h>
+    #else
+    #include <cv_bridge/cv_bridge.hpp>
+    #endif
+
+    // replace the following line with one appropriate for you data type
+    #include <std_msgs/msg/string.hpp>
+
+      namespace test_subscriber_plugin
+      {
+
+      class TestSubscriber : public web_video_server::SubscriberBase
+      {
+      public:
+        TestSubscriber(rclcpp::Node::SharedPtr node);
+
+        ~TestSubscriber();
+
+        void subscribe(const async_web_server_cpp::HttpRequest &request,
+                      const std::string& topic, 
+                      const web_video_server::ImageCallback& callback);
+
+      private:
+        // replace param in the following line with one appropriate for you data type  
+        void subscriber_callback(const std_msgs::msg::String::ConstSharedPtr &input_msg);
+
+        // replace param in the following line with one appropriate for you data type  
+        rclcpp::Subscription<std_msgs::msg::String>::SharedPtr sub_;
+        rclcpp::CallbackGroup::SharedPtr cbg_;   
+      };
+
+      class TestSubscriberFactory : public web_video_server::SubscriberFactoryInterface
+      {
+      public:
+        // replace the text string below with one appropriate for you data type
+        // it should agree with value returned by the rclcpp function 
+        // node.get_topic_names_and_types()
+        std::string get_type() override {return "std_msgs/msg/String";}
+
+        std::shared_ptr<web_video_server::SubscriberInterface> create_subscriber(
+            rclcpp::Node::SharedPtr node);
+
+        std::vector<std::string> get_available_topics(rclcpp::Node & node);
+      };
+
+      }  // namespace test_subscriber_plugin
+    ```
+
+1. Implement the `TestSubscriber` and `TestSubscriberFactory` classes in `src/test_subscriber_plugin.cpp`:
+    ```cpp
+    #include "test_subscriber_plugin/test_subscriber_plugin.hpp"
+
+    namespace test_subscriber_plugin
+    {
+
+    TestSubscriber::TestSubscriber(rclcpp::Node::SharedPtr node)
+    : web_video_server::SubscriberBase(node, "test_subscriber")
+    {
+      const std::scoped_lock lock(subscriber_mutex);
+
+      RCLCPP_INFO(logger_, "TestSubscriber created!");
+
+      // Declare any new parameters required for this subscriber
+      if (!node_->has_parameter("test_parameter")) node_->declare_parameter("test_parameter", "default");
+    }
+
+    TestSubscriber::~TestSubscriber()
+    {
+      const std::scoped_lock lock(subscriber_mutex);
+      inactive_ = true;
+
+      RCLCPP_INFO(logger_, "TestSubscriber destroyed!");
+    }
+
+    void TestSubscriber::subscribe(const async_web_server_cpp::HttpRequest &request,
+                                   const std::string& topic, 
+                                   const web_video_server::ImageCallback& callback)
+    {
+      const std::scoped_lock lock(subscriber_mutex);
+
+      callback_ = callback;
+
+      RCLCPP_INFO(logger_, "TestSubscriber started for topic: %s", topic.c_str());
+
+      // Load parameters used by this subscriber
+      std::string default_test_parameter = node_->get_parameter("test_parameter").as_string();  
+      std::string test_parameter = request.get_query_param_value_or_default("test_parameter", default_test_parameter);
+
+      std::string default_qos_profile = node_->get_parameter("default_qos_profile").as_string();    
+      auto qos_profile_name = request.get_query_param_value_or_default("qos_profile", default_qos_profile);
+
+      // Get QoS profile from query parameter
+      RCLCPP_INFO(
+        logger_, "Streaming topic %s with QoS profile %s", topic.c_str(),
+        qos_profile_name.c_str());
+      auto qos_profile = web_video_server::get_qos_profile_from_name(qos_profile_name);
+      if (!qos_profile) {
+        qos_profile = rmw_qos_profile_default;
+        RCLCPP_ERROR(
+          logger_,
+          "Invalid QoS profile %s specified. Using default profile.",
+          qos_profile_name.c_str());
+      }
+
+      const auto qos = rclcpp::QoS(
+        rclcpp::QoSInitialization(qos_profile.value().history, 1),
+        qos_profile.value());
+
+      cbg_ = node_->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive);    
+      rclcpp::SubscriptionOptions options;
+      options.callback_group = cbg_;  
+
+      // Create subscriber (update as appropriate for your subscriber)
+      sub_ = node_->create_subscription<std_msgs::msg::String>(
+        topic, qos, std::bind(&TestSubscriber::subscriber_callback, this, std::placeholders::_1), options
+      );
+    }
+
+    void TestSubscriber::subscriber_callback(const std_msgs::msg::String::ConstSharedPtr &input_msg)
+    {
+      const std::scoped_lock lock(subscriber_mutex);
+
+      if(inactive_) return;      
+
+      RCLCPP_INFO_STREAM(logger_,  "New TestSubscriber msg: " << input_msg->data);
+
+      // Convert input msg to image
+      cv::Mat image(500, 1000, CV_8UC3, cv::Scalar(0, 0, 0));
+      cv:putText(image, input_msg->data, cv::Point(30,250), cv::FONT_HERSHEY_SIMPLEX, 1.0, cv::Scalar(255, 0, 0), 2, cv::LINE_AA);
+
+      // Send to streamer using callback      
+      cv_bridge::CvImage bridge_image(std_msgs::msg::Header(), sensor_msgs::image_encodings::RGB8, image);
+      sensor_msgs::msg::Image output_msg;
+      bridge_image.toImageMsg(output_msg);
+      sensor_msgs::msg::Image::ConstSharedPtr output_ptr = std::make_shared<sensor_msgs::msg::Image>(output_msg);
+      try_forward_image(output_ptr);
+    }
+
+    std::shared_ptr<web_video_server::SubscriberInterface> TestSubscriberFactory::create_subscriber(
+      rclcpp::Node::SharedPtr node
+    ) {
+      return std::make_shared<TestSubscriber>(node);
+    }
+
+    std::vector<std::string> TestSubscriberFactory::get_available_topics(rclcpp::Node & node)
+    {
+      std::vector<std::string> result;
+      auto topic_names_and_types = node.get_topic_names_and_types();
+      for (const auto & topic_and_types : topic_names_and_types) {
+        for (const auto & type : topic_and_types.second) {
+          if (type == this->get_type()) {
+            result.push_back(topic_and_types.first);
+            break;
+          }
+        }
+      }
+      return result;
+    }
+
+    }  // namespace test_subscriber_plugin
+
+    #include "pluginlib/class_list_macros.hpp"
+
+    PLUGINLIB_EXPORT_CLASS(
+      test_subscriber_plugin::TestSubscriberFactory,
+      web_video_server::SubscriberFactoryInterface)
+    ```
+
+1. Add `plugins.xml` file with plugin description:
+    ```xml
+    <library path="test_subscriber_plugin">
+      <class name="test_subscriber_plugin/subscriber/test"
+        type="test_subscriber_plugin::TestSubscriberFactory"
+        base_class_type="web_video_server::SubscriberFactoryInterface">
+        <description>Test subscriber implementation</description>
+      </class>
+    </library>
+    ```
+
+1. Update `CMakeLists.txt` to export the plugin description file (Add this anywhere after `find_package` section):
+    ```cmake
+    pluginlib_export_plugin_description_file(web_video_server plugins.xml)
+    ```
+
+1. Build your package:
+    ```bash
+    cd ~/ros_ws
+    colcon build --packages-select test_subscriber_plugin
+    source install/setup.bash
+    ```
+
+1. Run the `web_video_server` node and test your custom subscriber plugin by accessing a topic of the appropriate format:
+    ```bash
+    ros2 topic pub /your_topic std_msgs/msg/String "data: test"
+    ros2 run web_video_server web_video_server --ros-args -p port:=8082 -p address:=localhost
+    ```
+    Then open your web browser and navigate to:
+    ```
+    http://localhost:8082/stream?topic=/your_topic
+    ```
+
+## Implementation hints
+- You can access query parameters from the HTTP request in your subscriber constructor using `request.get_query_param_value_or_default` method.
+- Use `logger_` member variable from the base `SubscriberBase` class for logging.
+- Link specific targets in `CMakeLists.txt`. For example, replace:
+    ```cmake
+    target_link_libraries(
+      test_subscriber_plugin PUBLIC
+      ${web_video_server_TARGETS}
+      ${pluginlib_TARGETS}
+    )
+    ```
+    with:
+    ```cmake
+    target_link_libraries(
+      test_subscriber_plugin
+      web_video_server::web_video_server
+      pluginlib::pluginlib
+    )
+    ```