From a74fcf4c9f1bf40dde48cb50fe5d9991a296ce25 Mon Sep 17 00:00:00 2001 From: Michael Orlov Date: Sat, 22 Jul 2023 12:30:00 -0700 Subject: [PATCH 1/2] Add BagSplitInfo service call on bag close (#1422) - Note: The `BagSplitInfo::opened_file` will have empty string to indicate that it was "bag close" and not bag split event. Signed-off-by: Michael Orlov (cherry picked from commit ba199d05954d6e51975c47acd3725cac6267778f) # Conflicts: # rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp --- .../include/rosbag2_cpp/bag_events.hpp | 8 +++ .../rosbag2_cpp/writers/sequential_writer.cpp | 12 +++- .../rosbag2_cpp/test_sequential_writer.cpp | 72 +++++++++++++++++++ 3 files changed, 91 insertions(+), 1 deletion(-) 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 a7bf406d7d..591661d7f5 100644 --- a/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp +++ b/rosbag2_cpp/src/rosbag2_cpp/writers/sequential_writer.cpp @@ -61,6 +61,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(); } @@ -161,7 +165,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 9a2636c24f..7c30879896 100644 --- a/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp +++ b/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp @@ -566,6 +566,7 @@ TEST_F(SequentialWriterTest, split_event_calls_callback) EXPECT_EQ(opened_file, fake_storage_uri_); } +<<<<<<< HEAD class ManualSplitWriter : public rosbag2_cpp::writers::SequentialWriter { @@ -587,6 +588,77 @@ void write_sample_split_bag( msg_content.c_str(), msg_length); std::string topic_name = "testtopic"; +======= +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}, + {300, 2}, + {200, 3}, + {500, 4}, + {400, 5}, + {600, 6} + }; +>>>>>>> ba199d0 (Add BagSplitInfo service call on bag close (#1422)) rosbag2_storage::StorageOptions storage_options; storage_options.uri = uri; storage_options.storage_id = "sqlite3"; From e00d091534c7151059112ac06e51186243526ab2 Mon Sep 17 00:00:00 2001 From: Michael Orlov Date: Mon, 6 May 2024 01:06:52 -0700 Subject: [PATCH 2/2] Fix merge conflicts - Ensure that writer_ is destructed before intercepted fake_metadata_ Signed-off-by: Michael Orlov --- .../rosbag2_cpp/test_sequential_writer.cpp | 59 ++++++++----------- 1 file changed, 24 insertions(+), 35 deletions(-) diff --git a/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp b/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp index 7c30879896..b1674c1b09 100644 --- a/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp +++ b/rosbag2_cpp/test/rosbag2_cpp/test_sequential_writer.cpp @@ -78,11 +78,13 @@ class SequentialWriterTest : public Test std::shared_ptr> storage_; std::shared_ptr> converter_factory_; std::unique_ptr metadata_io_; - std::unique_ptr writer_; + rosbag2_storage::StorageOptions storage_options_; std::atomic fake_storage_size_{0}; // Need to be atomic for cache update since it // uses in callback from cache_consumer thread rosbag2_storage::BagMetadata fake_metadata_; + // Ensure writer_ is destructed before intercepted fake_metadata_ + std::unique_ptr writer_; std::string fake_storage_uri_; }; @@ -566,29 +568,6 @@ TEST_F(SequentialWriterTest, split_event_calls_callback) EXPECT_EQ(opened_file, fake_storage_uri_); } -<<<<<<< HEAD - -class ManualSplitWriter : public rosbag2_cpp::writers::SequentialWriter -{ -public: - // makes the method public for manual splitting - void split() - { - split_bagfile(); - } -}; - -void write_sample_split_bag( - const std::string & uri, - const std::vector> & message_timestamps_by_file) -{ - std::string msg_content = "Hello"; - auto msg_length = msg_content.length(); - std::shared_ptr fake_data = rosbag2_storage::make_serialized_message( - msg_content.c_str(), msg_length); - std::string topic_name = "testtopic"; - -======= TEST_F(SequentialWriterTest, split_event_calls_on_writer_close) { const int message_count = 7; @@ -636,7 +615,7 @@ TEST_F(SequentialWriterTest, split_event_calls_on_writer_close) writer_->add_event_callbacks(callbacks); writer_->open(storage_options_, {"rmw_format", "rmw_format"}); - writer_->create_topic({"test_topic", "test_msgs/BasicTypes", "", "", ""}); + writer_->create_topic({"test_topic", "test_msgs/BasicTypes", "", ""}); for (auto i = 0; i < message_count; ++i) { writer_->write(message); @@ -649,16 +628,26 @@ TEST_F(SequentialWriterTest, split_event_calls_on_writer_close) EXPECT_TRUE(opened_file.empty()); } -TEST_P(ParametrizedTemporaryDirectoryFixture, split_bag_metadata_has_full_duration) { - const std::vector> fake_messages { - {100, 1}, - {300, 2}, - {200, 3}, - {500, 4}, - {400, 5}, - {600, 6} - }; ->>>>>>> ba199d0 (Add BagSplitInfo service call on bag close (#1422)) +class ManualSplitWriter : public rosbag2_cpp::writers::SequentialWriter +{ +public: + // makes the method public for manual splitting + void split() + { + split_bagfile(); + } +}; + +void write_sample_split_bag( + const std::string & uri, + const std::vector> & message_timestamps_by_file) +{ + std::string msg_content = "Hello"; + auto msg_length = msg_content.length(); + std::shared_ptr fake_data = rosbag2_storage::make_serialized_message( + msg_content.c_str(), msg_length); + std::string topic_name = "testtopic"; + rosbag2_storage::StorageOptions storage_options; storage_options.uri = uri; storage_options.storage_id = "sqlite3";