fix: Start streamers only in executor thread

Author: bjsowa

include/web_video_server/web_video_server.hpp

@@ -101,10 +101,12 @@ class WebVideoServer : public rclcpp::Node
 
 private:
   void restream_frames(std::chrono::duration<double> max_age);
+  void activate_pending_streamers();
   void cleanup_inactive_streams();
+  void resourse_management_timer_callback();
 
   rclcpp::TimerBase::SharedPtr restream_timer_;
-  rclcpp::TimerBase::SharedPtr cleanup_timer_;
+  rclcpp::TimerBase::SharedPtr resource_management_timer_;
 
   // Parameters
   double publish_rate_;
@@ -118,11 +120,13 @@ class WebVideoServer : public rclcpp::Node
   async_web_server_cpp::HttpRequestHandlerGroup handler_group_;
 
   std::vector<std::shared_ptr<StreamerInterface>> streamers_;
+  std::vector<std::shared_ptr<StreamerInterface>> pending_streamers_;
   pluginlib::ClassLoader<StreamerFactoryInterface> streamer_factory_loader_;
   std::map<std::string, std::shared_ptr<StreamerFactoryInterface>> streamer_factories_;
   pluginlib::ClassLoader<SnapshotStreamerFactoryInterface> snapshot_streamer_factory_loader_;
   std::map<std::string, std::shared_ptr<StreamerFactoryInterface>> snapshot_streamer_factories_;
   std::mutex streamers_mutex_;
+  std::mutex pending_mutex_;
 };
 
 }  // namespace web_video_server

src/web_video_server.cpp

@@ -34,6 +34,7 @@
 #include <chrono>
 #include <cstring>
 #include <exception>
+#include <iterator>
 #include <map>
 #include <memory>
 #include <mutex>
@@ -147,7 +148,9 @@ WebVideoServer::WebVideoServer(const rclcpp::NodeOptions & options)
       [this]() {restream_frames(1s / publish_rate_);});
   }
 
-  cleanup_timer_ = create_wall_timer(500ms, [this]() {cleanup_inactive_streams();});
+  resource_management_timer_ = create_wall_timer(100ms,
+      [this]() {resourse_management_timer_callback();}
+  );
 
   server_->run();
 }
@@ -166,6 +169,27 @@ void WebVideoServer::restream_frames(std::chrono::duration<double> max_age)
   }
 }
 
+void WebVideoServer::activate_pending_streamers()
+{
+  std::vector<std::shared_ptr<StreamerInterface>> to_activate;
+  {
+    const std::scoped_lock lock(pending_mutex_);
+    to_activate.swap(pending_streamers_);
+  }
+
+  for (auto & streamer : to_activate) {
+    streamer->start();
+  }
+
+  if (!to_activate.empty()) {
+    const std::scoped_lock lock(streamers_mutex_);
+    streamers_.insert(
+      streamers_.end(),
+      std::make_move_iterator(to_activate.begin()),
+      std::make_move_iterator(to_activate.end()));
+  }
+}
+
 void WebVideoServer::cleanup_inactive_streams()
 {
   const std::unique_lock lock(streamers_mutex_, std::try_to_lock);
@@ -182,6 +206,12 @@ void WebVideoServer::cleanup_inactive_streams()
   }
 }
 
+void WebVideoServer::resourse_management_timer_callback()
+{
+  activate_pending_streamers();
+  cleanup_inactive_streams();
+}
+
 bool WebVideoServer::handle_request(
   const async_web_server_cpp::HttpRequest & request,
   async_web_server_cpp::HttpConnectionPtr connection, const char * begin,
@@ -208,9 +238,8 @@ bool WebVideoServer::handle_stream(
   if (streamer_factories_.find(type) != streamer_factories_.end()) {
     const std::shared_ptr<StreamerInterface> streamer = streamer_factories_[type]->create_streamer(
       request, connection, weak_from_this());
-    streamer->start();
-    const std::scoped_lock lock(streamers_mutex_);
-    streamers_.push_back(streamer);
+    const std::scoped_lock lock(pending_mutex_);
+    pending_streamers_.push_back(streamer);
   } else {
     async_web_server_cpp::HttpReply::stock_reply(async_web_server_cpp::HttpReply::not_found)(
       request, connection, begin, end);
@@ -228,9 +257,8 @@ bool WebVideoServer::handle_snapshot(
     const std::shared_ptr<StreamerInterface> streamer =
       snapshot_streamer_factories_[type]->create_streamer(
       request, connection, weak_from_this());
-    streamer->start();
-    const std::scoped_lock lock(streamers_mutex_);
-    streamers_.push_back(streamer);
+    const std::scoped_lock lock(pending_mutex_);
+    pending_streamers_.push_back(streamer);
   } else {
     async_web_server_cpp::HttpReply::stock_reply(async_web_server_cpp::HttpReply::not_found)(
       request, connection, begin, end);