diff --git a/rosbag2_cpp/include/rosbag2_cpp/bag_events.hpp b/rosbag2_cpp/include/rosbag2_cpp/bag_events.hpp index 5650d8c22f..8e0ac48861 100644 --- a/rosbag2_cpp/include/rosbag2_cpp/bag_events.hpp +++ b/rosbag2_cpp/include/rosbag2_cpp/bag_events.hpp @@ -165,6 +165,14 @@ class EventCallbackManager return false; } + /** + * \brief Delete all callbacks + */ + void delete_all_callbacks() + { + callbacks_.clear(); + } + /** * \brief Execute all callbacks registered for the given event. * diff --git a/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp b/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp index 49aae2a460..d32951e6f7 100644 --- a/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp +++ b/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp @@ -62,6 +62,10 @@ SequentialWriter::SequentialWriter( SequentialWriter::~SequentialWriter() { + // Deleting all callbacks before calling close(). Calling callbacks from destructor is not safe. + // Callbacks likely was created after SequentialWriter object and may point to the already + // destructed objects. + callback_manager_.delete_all_callbacks(); close(); } @@ -172,7 +176,13 @@ void SequentialWriter::close() metadata_io_->write_metadata(base_folder_, metadata_); } - storage_.reset(); // Necessary to ensure that the storage is destroyed before the factory + if (storage_) { + auto info = std::make_shared(); + info->closed_file = storage_->get_relative_file_path(); + storage_.reset(); // Destroy storage before calling WRITE_SPLIT callback to make sure that + // bag file was closed before callback call. + callback_manager_.execute_callbacks(bag_events::BagEvent::WRITE_SPLIT, info); + } storage_factory_.reset(); } diff --git a/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp b/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp index 01650cd6de..1724cf2497 100644 --- a/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp +++ b/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp @@ -642,6 +642,66 @@ TEST_F(SequentialWriterTest, split_event_calls_callback) EXPECT_EQ(opened_file, fake_storage_uri_); } +TEST_F(SequentialWriterTest, split_event_calls_on_writer_close) +{ + const int message_count = 7; + + ON_CALL( + *storage_, + write(An>())).WillByDefault( + [this](std::shared_ptr) { + fake_storage_size_ += 1; + }); + + ON_CALL(*storage_, get_bagfile_size).WillByDefault( + [this]() { + return fake_storage_size_.load(); + }); + + ON_CALL(*metadata_io_, write_metadata).WillByDefault( + [this](const std::string &, const rosbag2_storage::BagMetadata & metadata) { + fake_metadata_ = metadata; + }); + + ON_CALL(*storage_, get_relative_file_path).WillByDefault( + [this]() { + return fake_storage_uri_; + }); + + auto sequential_writer = std::make_unique( + std::move(storage_factory_), converter_factory_, std::move(metadata_io_)); + writer_ = std::make_unique(std::move(sequential_writer)); + + auto message = std::make_shared(); + message->topic_name = "test_topic"; + + storage_options_.max_bagfile_size = 0; + + bool callback_called = false; + std::string closed_file, opened_file; + rosbag2_cpp::bag_events::WriterEventCallbacks callbacks; + callbacks.write_split_callback = + [&callback_called, &closed_file, &opened_file](rosbag2_cpp::bag_events::BagSplitInfo & info) { + closed_file = info.closed_file; + opened_file = info.opened_file; + callback_called = true; + }; + writer_->add_event_callbacks(callbacks); + + writer_->open(storage_options_, {"rmw_format", "rmw_format"}); + writer_->create_topic({"test_topic", "test_msgs/BasicTypes", "", "", ""}); + + for (auto i = 0; i < message_count; ++i) { + writer_->write(message); + } + writer_->close(); + + ASSERT_TRUE(callback_called); + auto expected_closed = rcpputils::fs::path(storage_options_.uri) / (storage_options_.uri + "_0"); + EXPECT_EQ(closed_file, expected_closed.string()); + EXPECT_TRUE(opened_file.empty()); +} + TEST_P(ParametrizedTemporaryDirectoryFixture, split_bag_metadata_has_full_duration) { const std::vector> fake_messages { {100, 1},