Fix playback of compressed bagfiles

Signed-off-by: Emerson Knapp <[email protected]>

rosbag2_compression/src/rosbag2_compression/sequential_compression_reader.cpp

@@ -27,6 +27,21 @@
 
 #include "logging.hpp"
 
+namespace
+{
+void fill_topics_and_types(
+  const rosbag2_storage::BagMetadata & metadata,
+  std::vector<rosbag2_storage::TopicMetadata> & topics_and_types)
+{
+  topics_and_types.clear();
+  topics_and_types.reserve(metadata.topics_with_message_count.size());
+  for (const auto & topic_information : metadata.topics_with_message_count) {
+    topics_and_types.push_back(topic_information.topic_metadata);
+  }
+}
+}  // unnamed namespace
+
+
 namespace rosbag2_compression
 {
 
@@ -90,6 +105,7 @@ void SequentialCompressionReader::open(
     ROSBAG2_COMPRESSION_LOG_WARN("No topics were listed in metadata.");
     return;
   }
+  fill_topics_and_types(metadata_, topics_metadata_);
 
   // Currently a bag file can only be played if all topics have the same serialization format.
   check_topics_serialization_formats(topics);

rosbag2_compression/test/rosbag2_compression/test_sequential_compression_reader.cpp

@@ -54,6 +54,17 @@ class SequentialCompressionReaderTest : public Test
     ON_CALL(*storage_factory_, open_read_only(_, _)).WillByDefault(Return(storage_));
   }
 
+  rosbag2_storage::BagMetadata construct_default_bag_metadata() const
+  {
+    rosbag2_storage::BagMetadata metadata;
+    metadata.relative_file_paths = {"/path/to/storage"};
+    metadata.topics_with_message_count.push_back({{topic_with_type_}, 1});
+    metadata.compression_format = "zstd";
+    metadata.compression_mode =
+      rosbag2_compression::compression_mode_to_string(rosbag2_compression::CompressionMode::FILE);
+    return metadata;
+  }
+
   std::unique_ptr<StrictMock<MockStorageFactory>> storage_factory_;
   std::shared_ptr<NiceMock<MockStorage>> storage_;
   std::shared_ptr<StrictMock<MockConverterFactory>> converter_factory_;
@@ -65,12 +76,8 @@ class SequentialCompressionReaderTest : public Test
 
 TEST_F(SequentialCompressionReaderTest, open_throws_if_unsupported_compressor)
 {
-  rosbag2_storage::BagMetadata metadata;
-  metadata.relative_file_paths = {"/path/to/storage"};
-  metadata.topics_with_message_count.push_back({{topic_with_type_}, 1});
+  rosbag2_storage::BagMetadata metadata = construct_default_bag_metadata();
   metadata.compression_format = "bad_format";
-  metadata.compression_mode =
-    rosbag2_compression::compression_mode_to_string(rosbag2_compression::CompressionMode::FILE);
   EXPECT_CALL(*metadata_io_, read_metadata(_)).WillRepeatedly(Return(metadata));
   EXPECT_CALL(*metadata_io_, metadata_file_exists(_)).WillRepeatedly(Return(true));
   auto compression_factory = std::make_unique<rosbag2_compression::CompressionFactory>();
@@ -87,14 +94,36 @@ TEST_F(SequentialCompressionReaderTest, open_throws_if_unsupported_compressor)
     std::invalid_argument);
 }
 
+TEST_F(SequentialCompressionReaderTest, returns_all_topics_and_types)
+{
+  rosbag2_storage::BagMetadata metadata = construct_default_bag_metadata();
+  ON_CALL(*metadata_io_, read_metadata(_)).WillByDefault(Return(metadata));
+  ON_CALL(*metadata_io_, metadata_file_exists(_)).WillByDefault(Return(true));
+
+  auto decompressor = std::make_unique<NiceMock<MockDecompressor>>();
+  auto compression_factory = std::make_unique<StrictMock<MockCompressionFactory>>();
+
+  ON_CALL(*compression_factory, create_decompressor(_))
+  .WillByDefault(Return(ByMove(std::move(decompressor))));
+  EXPECT_CALL(*compression_factory, create_decompressor(_)).Times(1);
+  EXPECT_CALL(*storage_factory_, open_read_only(_, _)).Times(1);
+
+  auto compression_reader = std::make_unique<rosbag2_compression::SequentialCompressionReader>(
+    std::move(compression_factory),
+    std::move(storage_factory_),
+    converter_factory_,
+    std::move(metadata_io_));
+
+  compression_reader->open(
+    rosbag2_cpp::StorageOptions(), {"", storage_serialization_format_});
+
+  auto topics_and_types = compression_reader->get_all_topics_and_types();
+  EXPECT_FALSE(topics_and_types.empty());
+}
+
 TEST_F(SequentialCompressionReaderTest, open_supports_zstd_compressor)
 {
-  rosbag2_storage::BagMetadata metadata;
-  metadata.relative_file_paths = {"/path/to/storage"};
-  metadata.topics_with_message_count.push_back({{topic_with_type_}, 1});
-  metadata.compression_format = "zstd";
-  metadata.compression_mode =
-    rosbag2_compression::compression_mode_to_string(rosbag2_compression::CompressionMode::FILE);
+  rosbag2_storage::BagMetadata metadata = construct_default_bag_metadata();
   ON_CALL(*metadata_io_, read_metadata(_)).WillByDefault(Return(metadata));
   ON_CALL(*metadata_io_, metadata_file_exists(_)).WillByDefault(Return(true));
   auto compression_factory = std::make_unique<rosbag2_compression::CompressionFactory>();
@@ -114,12 +143,7 @@ TEST_F(SequentialCompressionReaderTest, open_supports_zstd_compressor)
 
 TEST_F(SequentialCompressionReaderTest, reader_calls_create_decompressor)
 {
-  rosbag2_storage::BagMetadata metadata;
-  metadata.relative_file_paths = {"/path/to/storage"};
-  metadata.topics_with_message_count.push_back({{topic_with_type_}, 1});
-  metadata.compression_format = "zstd";
-  metadata.compression_mode =
-    rosbag2_compression::compression_mode_to_string(rosbag2_compression::CompressionMode::FILE);
+  rosbag2_storage::BagMetadata metadata = construct_default_bag_metadata();
   ON_CALL(*metadata_io_, read_metadata(_)).WillByDefault(Return(metadata));
   ON_CALL(*metadata_io_, metadata_file_exists(_)).WillByDefault(Return(true));