Add pause and resume service calls for rosbag2 recorder (ros2#1131)

* feat(recorder): Add pause and resume service calls.

Add service calls to pause recording and resume recording.

Fixes ros2#1130

Signed-off-by: Rick Shanor <rickshanor@gmail.com>

* feat(recorder): pause, resume, and is_paused PR comments.

Address PR comments. Add is_paused service. Update tests accordingly.

Signed-off-by: Rick Shanor <rickshanor@gmail.com>

Signed-off-by: Rick Shanor <rickshanor@gmail.com>

rosbag2_transport/include/rosbag2_transport/recorder.hpp

@@ -30,6 +30,9 @@
 
 #include "rosbag2_cpp/writer.hpp"
 
+#include "rosbag2_interfaces/srv/is_paused.hpp"
+#include "rosbag2_interfaces/srv/pause.hpp"
+#include "rosbag2_interfaces/srv/resume.hpp"
 #include "rosbag2_interfaces/srv/snapshot.hpp"
 
 #include "rosbag2_interfaces/msg/write_split_event.hpp"
@@ -162,6 +165,9 @@ class Recorder : public rclcpp::Node
  std::string serialization_format_;
  std::unordered_map<std::string, rclcpp::QoS> topic_qos_profile_overrides_;
  std::unordered_set<std::string> topic_unknown_types_;
+ rclcpp::Service<rosbag2_interfaces::srv::IsPaused>::SharedPtr srv_is_paused_;
+ rclcpp::Service<rosbag2_interfaces::srv::Pause>::SharedPtr srv_pause_;
+ rclcpp::Service<rosbag2_interfaces::srv::Resume>::SharedPtr srv_resume_;
  rclcpp::Service<rosbag2_interfaces::srv::Snapshot>::SharedPtr srv_snapshot_;
  std::atomic<bool> paused_ = false;
 

rosbag2_transport/src/rosbag2_transport/recorder.cpp

@@ -173,6 +173,36 @@ void Recorder::record()
  split_event_pub_ =
  create_publisher<rosbag2_interfaces::msg::WriteSplitEvent>("events/write_split", 1);
 
+ srv_pause_ = create_service<rosbag2_interfaces::srv::Pause>(
+ "~/pause",
+ [this](
+ const std::shared_ptr<rmw_request_id_t>/* request_header */,
+ const std::shared_ptr<rosbag2_interfaces::srv::Pause::Request>/* request */,
+ const std::shared_ptr<rosbag2_interfaces::srv::Pause::Response>/* response */)
+ {
+ pause();
+ });
+
+ srv_resume_ = create_service<rosbag2_interfaces::srv::Resume>(
+ "~/resume",
+ [this](
+ const std::shared_ptr<rmw_request_id_t>/* request_header */,
+ const std::shared_ptr<rosbag2_interfaces::srv::Resume::Request>/* request */,
+ const std::shared_ptr<rosbag2_interfaces::srv::Resume::Response>/* response */)
+ {
+ resume();
+ });
+
+ srv_is_paused_ = create_service<rosbag2_interfaces::srv::IsPaused>(
+ "~/is_paused",
+ [this](
+ const std::shared_ptr<rmw_request_id_t>/* request_header */,
+ const std::shared_ptr<rosbag2_interfaces::srv::IsPaused::Request>/* request */,
+ const std::shared_ptr<rosbag2_interfaces::srv::IsPaused::Response> response)
+ {
+ response->paused = is_paused();
+ });
+
  // Start the thread that will publish events
  event_publisher_thread_ = std::thread(&Recorder::event_publisher_thread_main, this);
 

rosbag2_transport/test/rosbag2_transport/test_record_services.cpp

@@ -21,6 +21,9 @@
 
 #include "rclcpp/rclcpp.hpp"
 
+#include "rosbag2_interfaces/srv/is_paused.hpp"
+#include "rosbag2_interfaces/srv/pause.hpp"
+#include "rosbag2_interfaces/srv/resume.hpp"
 #include "rosbag2_interfaces/srv/snapshot.hpp"
 #include "rosbag2_transport/recorder.hpp"
 
@@ -35,6 +38,9 @@ using namespace ::testing; // NOLINT
 class RecordSrvsTest : public RecordIntegrationTestFixture
 {
 public:
+ using IsPaused = rosbag2_interfaces::srv::IsPaused;
+ using Pause = rosbag2_interfaces::srv::Pause;
+ using Resume = rosbag2_interfaces::srv::Resume;
  using Snapshot = rosbag2_interfaces::srv::Snapshot;
 
  RecordSrvsTest()
@@ -71,6 +77,9 @@ class RecordSrvsTest : public RecordIntegrationTestFixture
  pub_manager.setup_publisher(test_topic_, string_message, 10);
 
  const std::string ns = "/" + recorder_name_;
+ cli_is_paused_ = client_node_->create_client<IsPaused>(ns + "/is_paused");
+ cli_pause_ = client_node_->create_client<Pause>(ns + "/pause");
+ cli_resume_ = client_node_->create_client<Resume>(ns + "/resume");
  cli_snapshot_ = client_node_->create_client<Snapshot>(ns + "/snapshot");
 
  exec_ = std::make_shared<rclcpp::executors::SingleThreadedExecutor>();
@@ -126,6 +135,9 @@ class RecordSrvsTest : public RecordIntegrationTestFixture
 
  // Service clients
  rclcpp::Node::SharedPtr client_node_;
+ rclcpp::Client<IsPaused>::SharedPtr cli_is_paused_;
+ rclcpp::Client<Pause>::SharedPtr cli_pause_;
+ rclcpp::Client<Resume>::SharedPtr cli_resume_;
  rclcpp::Client<Snapshot>::SharedPtr cli_snapshot_;
 };
 
@@ -148,3 +160,20 @@ TEST_F(RecordSrvsTest, trigger_snapshot)
  successful_service_request<Snapshot>(cli_snapshot_);
  EXPECT_THAT(mock_writer.get_messages().size(), Ne(0u));
 }
+
+TEST_F(RecordSrvsTest, pause_resume)
+{
+ EXPECT_FALSE(recorder_->is_paused());
+ auto is_paused_response = successful_service_request<IsPaused>(cli_is_paused_);
+ EXPECT_FALSE(is_paused_response->paused);
+
+ successful_service_request<Pause>(cli_pause_);
+ EXPECT_TRUE(recorder_->is_paused());
+ is_paused_response = successful_service_request<IsPaused>(cli_is_paused_);
+ EXPECT_TRUE(is_paused_response->paused);
+
+ successful_service_request<Resume>(cli_resume_);
+ EXPECT_FALSE(recorder_->is_paused());
+ is_paused_response = successful_service_request<IsPaused>(cli_is_paused_);
+ EXPECT_FALSE(is_paused_response->paused);
+}