use public recorder api in tests (ros2#741)

* use public recorder api in tests

Signed-off-by: Karsten Knese <Karsten1987@users.noreply.github.com>

* fix segfault

Signed-off-by: Karsten Knese <Karsten1987@users.noreply.github.com>

* match rmw_connext*

Signed-off-by: Karsten Knese <Karsten1987@users.noreply.github.com>

rosbag2_test_common/include/rosbag2_test_common/wait_for.hpp

@@ -36,6 +36,21 @@ bool spin_and_wait_for(Timeout timeout, const Node & node, Condition condition)
   }
   return true;
 }
+
+template<typename Timeout, typename Condition>
+bool wait_until_shutdown(Timeout timeout, Condition condition)
+{
+  using clock = std::chrono::system_clock;
+  auto start = clock::now();
+  while (!condition()) {
+    if ((clock::now() - start) > timeout) {
+      return false;
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  }
+  rclcpp::shutdown();
+  return true;
+}
 }  // namespace rosbag2_test_common
 
 #endif  // ROSBAG2_TEST_COMMON__WAIT_FOR_HPP_

rosbag2_transport/CMakeLists.txt

@@ -79,7 +79,7 @@ function(create_tests_for_rmw_implementation)
   # disable the following tests for connext
   # due to slower discovery of nodes
   set(SKIP_TEST "")
-  if(${rmw_implementation} MATCHES "rmw_connext(.*)_cpp")
+  if(${rmw_implementation} MATCHES "rmw_connext(.*)")
     set(SKIP_TEST "SKIP_TEST")
   endif()
 

rosbag2_transport/include/rosbag2_transport/recorder.hpp

@@ -77,6 +77,9 @@ class Recorder : public rclcpp::Node
     return subscriptions_;
   }
 
+  ROSBAG2_TRANSPORT_PUBLIC
+  const rosbag2_cpp::Writer & get_writer_handle();
+
 private:
   void topics_discovery();
 

rosbag2_transport/src/rosbag2_transport/recorder.cpp

@@ -104,6 +104,11 @@ void Recorder::record()
   }
 }
 
+const rosbag2_cpp::Writer & Recorder::get_writer_handle()
+{
+  return *writer_;
+}
+
 void Recorder::topics_discovery()
 {
   while (rclcpp::ok() && stop_discovery_ == false) {

rosbag2_transport/test/rosbag2_transport/record_integration_fixture.hpp

@@ -42,24 +42,32 @@ class RecordIntegrationTestFixture : public Rosbag2TransportTestFixture
 public:
   RecordIntegrationTestFixture()
   : Rosbag2TransportTestFixture()
+  {
+  }
+
+  void SetUp() override
   {
     rclcpp::init(0, nullptr);
   }
 
-  void start_recording(const RecordOptions & options)
+  void TearDown() override
+  {
+    rclcpp::shutdown();
+  }
+
+  template<class T>
+  void start_async_spin(T node)
   {
-    // the future object returned from std::async needs to be stored not to block the execution
     future_ = std::async(
-      std::launch::async, [this, options]() {
-        rosbag2_transport::Recorder recorder(writer_);
-        recorder.record(storage_options_, options);
-      });
+      std::launch::async, [node]() -> void {rclcpp::spin(node);});
   }
 
-  void stop_recording()
+  void stop_spinning()
   {
     rclcpp::shutdown();
-    future_.get();
+    if (future_.valid()) {
+      future_.wait();
+    }
   }
 
   void run_publishers()