feat: Add shutdown handler for client to call and close alive sockets

README.md

@@ -178,6 +178,35 @@ http://localhost:8080/snapshot?topic=/camera/image_raw
 | `default_transport` | string | "raw" | "raw", "compressed", "theora" | Image transport to use |
 | `qos_profile` | string | "default" | "default", "system_default", "sensor_data", "services_default" | QoS profile for ROS 2 subscribers |
 
+### Stop an Active Stream
+
+To stop one or more active streams from the server side (e.g. when a UI component unmounts), use the `/shutdown` endpoint:
+
+```
+http://localhost:8080/shutdown?topic=/camera/image_raw
+```
+
+This closes all active streams for the given topic. An optional `client_id` parameter scopes the shutdown to a single named connection:
+
+```
+http://localhost:8080/shutdown?topic=/camera/image_raw&client_id=my-ui
+```
+
+To associate a stream with a `client_id`, pass it when opening the stream:
+
+```
+http://localhost:8080/stream?topic=/camera/image_raw&client_id=my-ui
+```
+
+The response is plain text in the form `stopped=<count>`, where `<count>` is the number of streams that were stopped. Returns `400 Bad Request` if `topic` is omitted.
+
+#### URL Parameters for Shutdown
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `topic` | string | (required) | The ROS topic whose streams should be stopped |
+| `client_id` | string | (none) | If provided, only the stream with this client_id is stopped |
+
 ## 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.
 

include/web_video_server/streamer.hpp

@@ -56,6 +56,11 @@ class StreamerInterface
    */
   virtual void start() = 0;
 
+  /**
+   * @brief Stops the streaming process and marks the streamer as inactive.
+   */
+  virtual void stop() = 0;
+
   /**
    * @brief Returns true if the streamer is inactive and should be deleted.
    *
@@ -73,6 +78,11 @@ class StreamerInterface
    * @brief Returns the topic being streamed.
    */
   virtual std::string get_topic() = 0;
+
+  /**
+   * @brief Returns the client_id associated with this stream, or an empty string if none.
+   */
+  virtual std::string get_client_id() = 0;
 };
 
 /**
@@ -87,6 +97,12 @@ class StreamerBase : public StreamerInterface
     rclcpp::Node::WeakPtr node,
     std::string logger_name = "streamer");
 
+  void stop() override
+  {
+    inactive_ = true;
+    connection_.reset();
+  }
+
   bool is_inactive() override
   {
     return inactive_;
@@ -97,6 +113,11 @@ class StreamerBase : public StreamerInterface
     return topic_;
   }
 
+  std::string get_client_id() override
+  {
+    return client_id_;
+  }
+
 protected:
   rclcpp::Node::SharedPtr lock_node() const;
 
@@ -106,6 +127,7 @@ class StreamerBase : public StreamerInterface
   rclcpp::Logger logger_;
   bool inactive_;
   std::string topic_;
+  std::string client_id_;
 };
 
 /**

include/web_video_server/web_video_server.hpp

@@ -89,6 +89,11 @@ class WebVideoServer : public rclcpp::Node
     async_web_server_cpp::HttpConnectionPtr connection,
     const char * begin, const char * end);
 
+  bool handle_shutdown(
+    const async_web_server_cpp::HttpRequest & request,
+    async_web_server_cpp::HttpConnectionPtr connection,
+    const char * begin, const char * end);
+
   bool handle_list_streams(
     const async_web_server_cpp::HttpRequest & request,
     async_web_server_cpp::HttpConnectionPtr connection,

src/streamer.cpp

@@ -51,7 +51,8 @@ StreamerBase::StreamerBase(
   std::string logger_name)
 : connection_(connection), request_(request), node_(std::move(node)),
   logger_(node_.lock()->get_logger().get_child(logger_name)), inactive_(false),
-  topic_(request.get_query_param_value_or_default("topic", ""))
+  topic_(request.get_query_param_value_or_default("topic", "")),
+  client_id_(request.get_query_param_value_or_default("client_id", ""))
 {
 }
 

src/web_video_server.cpp

@@ -120,6 +120,9 @@ WebVideoServer::WebVideoServer(const rclcpp::NodeOptions & options)
   handler_group_.addHandlerForPath(
     "/snapshot",
     boost::bind(&WebVideoServer::handle_snapshot, this, _1, _2, _3, _4));
+  handler_group_.addHandlerForPath(
+    "/shutdown",
+    boost::bind(&WebVideoServer::handle_shutdown, this, _1, _2, _3, _4));
 
   try {
     server_.reset(
@@ -263,6 +266,50 @@ bool WebVideoServer::handle_stream_viewer(
   return true;
 }
 
+bool WebVideoServer::handle_shutdown(
+  const async_web_server_cpp::HttpRequest & request,
+  async_web_server_cpp::HttpConnectionPtr connection, const char * begin,
+  const char * end)
+{
+  const std::string topic = request.get_query_param_value_or_default("topic", "");
+  if (topic.empty()) {
+    async_web_server_cpp::HttpReply::stock_reply(async_web_server_cpp::HttpReply::bad_request)(
+      request, connection, begin, end);
+    return true;
+  }
+
+  const std::string client_id = request.get_query_param_value_or_default("client_id", "");
+
+  int stopped = 0;
+  {
+    const std::scoped_lock lock(streamers_mutex_);
+    for (auto & streamer : streamers_) {
+      if (streamer->get_topic() == topic) {
+        if (client_id.empty() || streamer->get_client_id() == client_id) {
+          streamer->stop();
+          ++stopped;
+        }
+      }
+    }
+  }
+
+  if (verbose_) {
+    const std::string client_id_info = client_id.empty() ? "" : " (client_id='" + client_id + "')";
+    RCLCPP_INFO(
+      get_logger(), "Shutdown request for topic '%s'%s: stopped %d stream(s)",
+      topic.c_str(), client_id_info.c_str(), stopped);
+  }
+
+  async_web_server_cpp::HttpReply::builder(async_web_server_cpp::HttpReply::ok)
+  .header("Connection", "close")
+  .header("Server", "web_video_server")
+  .header("Content-type", "text/plain;")
+  .write(connection);
+
+  connection->write("stopped=" + std::to_string(stopped));
+  return true;
+}
+
 bool WebVideoServer::handle_list_streams(
   const async_web_server_cpp::HttpRequest & /* request */,
   async_web_server_cpp::HttpConnectionPtr connection, const char * /* begin */,