Fixes an init race condition

rosbag2/include/rosbag2/writer.hpp

@@ -77,6 +77,16 @@ class ROSBAG2_PUBLIC Writer
    */
   virtual void create_topic(const TopicMetadata & topic_with_type);
 
+  /**
+   * Remove a new topic in the underlying storage.
+   * If creation of subscription fails remove the topic
+   * from the db (more of cleanup)
+   *
+   * \param topic_with_type name and type identifier of topic to be created
+   * \throws runtime_error if the Writer is not open.
+   */
+  virtual void remove_topic(const TopicMetadata & topic_with_type);
+
   /**
    * Write a message to a bagfile. The topic needs to have been created before writing is possible.
    *

rosbag2/src/rosbag2/writer.cpp

@@ -75,6 +75,15 @@ void Writer::create_topic(const TopicMetadata & topic_with_type)
   storage_->create_topic(topic_with_type);
 }
 
+void Writer::remove_topic(const TopicMetadata & topic_with_type)
+{
+  if (!storage_) {
+    throw std::runtime_error("Bag is not open. Call open() before removing.");
+  }
+
+  storage_->remove_topic(topic_with_type);
+}
+
 void Writer::write(std::shared_ptr<SerializedBagMessage> message)
 {
   if (!storage_) {

rosbag2/test/rosbag2/mock_storage.hpp

@@ -31,6 +31,7 @@ class MockStorage : public rosbag2_storage::storage_interfaces::ReadWriteInterfa
 public:
   MOCK_METHOD2(open, void(const std::string &, rosbag2_storage::storage_interfaces::IOFlag));
   MOCK_METHOD1(create_topic, void(const rosbag2_storage::TopicMetadata &));
+  MOCK_METHOD1(remove_topic, void(const rosbag2_storage::TopicMetadata &));
   MOCK_METHOD0(has_next, bool());
   MOCK_METHOD0(read_next, std::shared_ptr<rosbag2_storage::SerializedBagMessage>());
   MOCK_METHOD1(write, void(std::shared_ptr<const rosbag2_storage::SerializedBagMessage>));

rosbag2_storage/include/rosbag2_storage/storage_interfaces/base_write_interface.hpp

@@ -36,6 +36,8 @@ class ROSBAG2_STORAGE_PUBLIC BaseWriteInterface
   virtual void write(std::shared_ptr<const SerializedBagMessage> msg) = 0;
 
   virtual void create_topic(const TopicMetadata & topic) = 0;
+
+  virtual void remove_topic(const TopicMetadata & topic) = 0;
 };
 
 }  // namespace storage_interfaces

rosbag2_storage/test/rosbag2_storage/test_plugin.cpp

@@ -56,6 +56,11 @@ void TestPlugin::create_topic(const rosbag2_storage::TopicMetadata & topic)
   std::cout << "Created topic with name =" << topic.name << " and type =" << topic.type << ".\n";
 }
 
+void TestPlugin::remove_topic(const rosbag2_storage::TopicMetadata & topic)
+{
+  std::cout << "Removed topic with name =" << topic.name << " and type =" << topic.type << ".\n";
+}
+
 void TestPlugin::write(const std::shared_ptr<const rosbag2_storage::SerializedBagMessage> msg)
 {
   (void) msg;

rosbag2_storage/test/rosbag2_storage/test_plugin.hpp

@@ -33,6 +33,8 @@ class TestPlugin : public rosbag2_storage::storage_interfaces::ReadWriteInterfac
 
   void create_topic(const rosbag2_storage::TopicMetadata & topic) override;
 
+  void remove_topic(const rosbag2_storage::TopicMetadata & topic) override;
+
   bool has_next() override;
 
   std::shared_ptr<rosbag2_storage::SerializedBagMessage> read_next() override;

rosbag2_storage_default_plugins/include/rosbag2_storage_default_plugins/sqlite/sqlite_storage.hpp

@@ -50,6 +50,8 @@ class ROSBAG2_STORAGE_DEFAULT_PLUGINS_PUBLIC SqliteStorage
     rosbag2_storage::storage_interfaces::IOFlag io_flag =
     rosbag2_storage::storage_interfaces::IOFlag::READ_WRITE) override;
 
+  void remove_topic(const rosbag2_storage::TopicMetadata & topic) override;
+
   void create_topic(const rosbag2_storage::TopicMetadata & topic) override;
 
   void write(std::shared_ptr<const rosbag2_storage::SerializedBagMessage> message) override;

rosbag2_storage_default_plugins/src/rosbag2_storage_default_plugins/sqlite/sqlite_storage.cpp

@@ -161,6 +161,18 @@ void SqliteStorage::create_topic(const rosbag2_storage::TopicMetadata & topic)
   }
 }
 
+void SqliteStorage::remove_topic(const rosbag2_storage::TopicMetadata & topic)
+{
+  if (topics_.find(topic.name) != std::end(topics_)) {
+    auto delete_topic =
+      database_->prepare_statement(
+      "DELETE FROM topics where name = ? and type = ? and serialization_format = ?");
+    delete_topic->bind(topic.name, topic.type, topic.serialization_format);
+    delete_topic->execute_and_reset();
+    topics_.erase(topic.name);
+  }
+}
+
 void SqliteStorage::prepare_for_writing()
 {
   write_statement_ = database_->prepare_statement(

rosbag2_storage_default_plugins/test/rosbag2_storage_default_plugins/sqlite/test_sqlite_storage.cpp

@@ -180,3 +180,22 @@ TEST_F(StorageTestFixture, get_metadata_returns_correct_struct_if_no_messages) {
   ));
   EXPECT_THAT(metadata.duration, Eq(std::chrono::seconds(0)));
 }
+
+TEST_F(StorageTestFixture, remove_topics_and_types_returns_the_empty_vector) {
+  std::unique_ptr<rosbag2_storage::storage_interfaces::ReadWriteInterface> writable_storage =
+    std::make_unique<rosbag2_storage_plugins::SqliteStorage>();
+  writable_storage->open(temporary_dir_path_);
+  writable_storage->create_topic({"topic1", "type1", "rmw1"});
+  metadata_io_.write_metadata(temporary_dir_path_, writable_storage->get_metadata());
+  writable_storage->remove_topic({"topic1", "type1", "rmw1"});
+  writable_storage.reset();
+
+  // Remove topics
+
+  auto readable_storage = std::make_unique<rosbag2_storage_plugins::SqliteStorage>();
+  readable_storage->open(
+    temporary_dir_path_, rosbag2_storage::storage_interfaces::IOFlag::READ_ONLY);
+  auto topics_and_types = readable_storage->get_all_topics_and_types();
+
+  EXPECT_THAT(topics_and_types, IsEmpty());
+}

rosbag2_transport/src/rosbag2_transport/recorder.cpp

@@ -107,11 +107,15 @@ void Recorder::subscribe_topics(
 void Recorder::subscribe_topic(const rosbag2::TopicMetadata & topic)
 {
   auto subscription = create_subscription(topic.name, topic.type);
+
   if (subscription) {
+    writer_->create_topic(topic);
     subscribed_topics_.insert(topic.name);
     subscriptions_.push_back(subscription);
-    writer_->create_topic(topic);
     ROSBAG2_TRANSPORT_LOG_INFO_STREAM("Subscribed to topic '" << topic.name << "'");
+  } else {
+    writer_->remove_topic(topic);
+    subscribed_topics_.erase(topic.name);
   }
 }