Generalize reader and writer creation into common factory

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

Author: emersonknapp

rosbag2_py/src/rosbag2_py/_transport.cpp

@@ -20,16 +20,10 @@
 #include <utility>
 #include <vector>
 
-#include "rosbag2_compression/compression_options.hpp"
-#include "rosbag2_compression/sequential_compression_reader.hpp"
-#include "rosbag2_compression/sequential_compression_writer.hpp"
-#include "rosbag2_cpp/reader.hpp"
-#include "rosbag2_cpp/readers/sequential_reader.hpp"
-#include "rosbag2_cpp/writer.hpp"
-#include "rosbag2_cpp/writers/sequential_writer.hpp"
 #include "rosbag2_storage/storage_options.hpp"
 #include "rosbag2_transport/play_options.hpp"
 #include "rosbag2_transport/player.hpp"
+#include "rosbag2_transport/reader_writer_factory.hpp"
 #include "rosbag2_transport/record_options.hpp"
 #include "rosbag2_transport/recorder.hpp"
 
@@ -116,24 +110,7 @@ class Player
     const rosbag2_storage::StorageOptions & storage_options,
     PlayOptions & play_options)
   {
-    std::unique_ptr<rosbag2_cpp::Reader> reader = nullptr;
-    // Determine whether to build compression or regular reader
-    {
-      rosbag2_storage::MetadataIo metadata_io{};
-      rosbag2_storage::BagMetadata metadata{};
-      if (metadata_io.metadata_file_exists(storage_options.uri)) {
-        metadata = metadata_io.read_metadata(storage_options.uri);
-        if (!metadata.compression_format.empty()) {
-          reader = std::make_unique<rosbag2_cpp::Reader>(
-            std::make_unique<rosbag2_compression::SequentialCompressionReader>());
-        }
-      }
-      if (reader == nullptr) {
-        reader = std::make_unique<rosbag2_cpp::Reader>(
-          std::make_unique<rosbag2_cpp::readers::SequentialReader>());
-      }
-    }
-
+    auto reader = rosbag2_transport::ReaderWriterFactory::make_reader(storage_options);
     auto player = std::make_shared<rosbag2_transport::Player>(
       std::move(reader), storage_options, play_options);
 
@@ -175,31 +152,11 @@ class Recorder
     const rosbag2_storage::StorageOptions & storage_options,
     RecordOptions & record_options)
   {
-    rosbag2_compression::CompressionOptions compression_options {
-      record_options.compression_format,
-      rosbag2_compression::compression_mode_from_string(record_options.compression_mode),
-      record_options.compression_queue_size,
-      record_options.compression_threads
-    };
-    if (compression_options.compression_threads < 1) {
-      compression_options.compression_threads = std::thread::hardware_concurrency();
-    }
-
     if (record_options.rmw_serialization_format.empty()) {
       record_options.rmw_serialization_format = std::string(rmw_get_serialization_format());
     }
 
-
-    std::unique_ptr<rosbag2_cpp::Writer> writer = nullptr;
-    // Change writer based on recording options
-    if (!record_options.compression_format.empty()) {
-      writer = std::make_unique<rosbag2_cpp::Writer>(
-        std::make_unique<rosbag2_compression::SequentialCompressionWriter>(compression_options));
-    } else {
-      writer = std::make_unique<rosbag2_cpp::Writer>(
-        std::make_unique<rosbag2_cpp::writers::SequentialWriter>());
-    }
-
+    auto writer = rosbag2_transport::ReaderWriterFactory::make_writer(record_options);
     auto recorder = std::make_shared<rosbag2_transport::Recorder>(
       std::move(writer), storage_options, record_options);
     recorder->record();

rosbag2_transport/CMakeLists.txt

@@ -39,6 +39,7 @@ add_library(${PROJECT_NAME} SHARED
   src/rosbag2_transport/bag_rewrite.cpp
   src/rosbag2_transport/player.cpp
   src/rosbag2_transport/qos.cpp
+  src/rosbag2_transport/reader_writer_factory.cpp
   src/rosbag2_transport/recorder.cpp
   src/rosbag2_transport/record_options.cpp
   src/rosbag2_transport/topic_filter.cpp)

rosbag2_transport/include/rosbag2_transport/bag_rewrite.hpp

@@ -32,28 +32,16 @@ namespace rosbag2_transport
 /// - compress
 /// - serialization format conversion
 ///
-/// \param input_bags - a vector of Readers for bag inputs. Must already be "open"ed
-/// \param output_bags - full "recording" configuration of the bag(s) to output.
-///   Must already be "open"ed
-///   Each output bag will be passed every message from each input bag in timestamp order,
-///   and is responsible for being configured to flter/process as needed.
-void bag_rewrite(
-  const std::vector<std::unique_ptr<rosbag2_cpp::Reader>> & input_bags,
-  const std::vector<
-    std::pair<std::unique_ptr<rosbag2_cpp::Writer>, rosbag2_transport::RecordOptions>
-  > & output_bags
-);
-
-/// See above bag_rewrite documentation
-/// Convenience signature to create and open necessary Readers and Writers, given
-/// the relevant Options structs.
+/// \param input_options - a vector of Readers for bags to read messages from
+/// \param output_bags - full "recording" configuration of the bag(s) to write messages to
+///   Each output bag will be passed messages from every input bag, on topics that pass its
+///   filtering options
 void bag_rewrite(
   const std::vector<rosbag2_storage::StorageOptions> & input_options,
   const std::vector<
     std::pair<rosbag2_storage::StorageOptions, rosbag2_transport::RecordOptions>
   > & output_options
 );
-
 }  // namespace rosbag2_transport
 
 #endif  // ROSBAG2_TRANSPORT__BAG_REWRITE_HPP_

rosbag2_transport/include/rosbag2_transport/reader_writer_factory.hpp

@@ -0,0 +1,41 @@
+// Copyright 2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef ROSBAG2_TRANSPORT__READER_WRITER_FACTORY_HPP_
+#define ROSBAG2_TRANSPORT__READER_WRITER_FACTORY_HPP_
+
+#include <memory>
+
+#include "rosbag2_cpp/reader.hpp"
+#include "rosbag2_cpp/writer.hpp"
+#include "rosbag2_storage/storage_options.hpp"
+#include "rosbag2_transport/record_options.hpp"
+#include "rosbag2_transport/visibility_control.hpp"
+
+namespace rosbag2_transport
+{
+class ROSBAG2_TRANSPORT_PUBLIC ReaderWriterFactory
+{
+public:
+  /// Create a Reader with the appropriate underlying implementation.
+  static std::unique_ptr<rosbag2_cpp::Reader> make_reader(
+    const rosbag2_storage::StorageOptions & storage_options);
+
+  /// Create a Writer with the appropriate underlying implementation.
+  static std::unique_ptr<rosbag2_cpp::Writer> make_writer(
+    const rosbag2_transport::RecordOptions & record_options);
+};
+}  // namespace rosbag2_transport
+
+#endif  // ROSBAG2_TRANSPORT__READER_WRITER_FACTORY_HPP_

rosbag2_transport/src/rosbag2_transport/bag_rewrite.cpp

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "rosbag2_transport/bag_rewrite.hpp"
+
 #include <memory>
 #include <string>
 #include <unordered_map>
@@ -26,59 +28,14 @@
 #include "rosbag2_cpp/readers/sequential_reader.hpp"
 #include "rosbag2_cpp/writer.hpp"
 #include "rosbag2_cpp/writers/sequential_writer.hpp"
-#include "rosbag2_transport/bag_rewrite.hpp"
+#include "rosbag2_transport/reader_writer_factory.hpp"
 
 #include "logging.hpp"
 #include "topic_filter.hpp"
 
 namespace
 {
 
-/// Create a Reader with the appropriate underlying implementation.
-std::unique_ptr<rosbag2_cpp::Reader> make_reader(
-  const rosbag2_storage::StorageOptions & storage_options)
-{
-  rosbag2_storage::MetadataIo metadata_io;
-  std::unique_ptr<rosbag2_cpp::reader_interfaces::BaseReaderInterface> reader_impl;
-
-  if (metadata_io.metadata_file_exists(storage_options.uri)) {
-    auto metadata = metadata_io.read_metadata(storage_options.uri);
-    if (!metadata.compression_format.empty()) {
-      reader_impl = std::make_unique<rosbag2_compression::SequentialCompressionReader>();
-    }
-  }
-  if (!reader_impl) {
-    reader_impl = std::make_unique<rosbag2_cpp::readers::SequentialReader>();
-  }
-
-  return std::make_unique<rosbag2_cpp::Reader>(std::move(reader_impl));
-}
-
-
-/// Create a Writer with the appropriate underlying implementation.
-std::unique_ptr<rosbag2_cpp::Writer> make_writer(
-  const rosbag2_transport::RecordOptions & record_options)
-{
-  std::unique_ptr<rosbag2_cpp::writer_interfaces::BaseWriterInterface> writer_impl;
-  if (!record_options.compression_format.empty()) {
-    rosbag2_compression::CompressionOptions compression_options {
-      record_options.compression_format,
-      rosbag2_compression::compression_mode_from_string(record_options.compression_mode),
-      record_options.compression_queue_size,
-      record_options.compression_threads
-    };
-    if (compression_options.compression_threads < 1) {
-      compression_options.compression_threads = std::thread::hardware_concurrency();
-    }
-    writer_impl = std::make_unique<rosbag2_compression::SequentialCompressionWriter>(
-      compression_options);
-  } else {
-    writer_impl = std::make_unique<rosbag2_cpp::writers::SequentialWriter>();
-  }
-
-  return std::make_unique<rosbag2_cpp::Writer>(std::move(writer_impl));
-}
-
 /// Find the next chronological message from all opened input bags.
 /// Updates the next_messages queue as necessary.
 /// next_messages is needed because Reader has no "peek" interface, we cannot put a message back.
@@ -207,12 +164,7 @@ setup_topic_filtering(
   return filtered_outputs;
 }
 
-}  // namespace
-
-namespace rosbag2_transport
-{
-
-void bag_rewrite(
+void perform_rewrite(
   const std::vector<std::unique_ptr<rosbag2_cpp::Reader>> & input_bags,
   const std::vector<
     std::pair<std::unique_ptr<rosbag2_cpp::Writer>, rosbag2_transport::RecordOptions>
@@ -239,6 +191,10 @@ void bag_rewrite(
   }
 }
 
+}  // namespace
+
+namespace rosbag2_transport
+{
 void bag_rewrite(
   const std::vector<rosbag2_storage::StorageOptions> & input_options,
   const std::vector<
@@ -252,7 +208,7 @@ void bag_rewrite(
   > output_bags;
 
   for (const auto & storage_options : input_options) {
-    auto reader = make_reader(storage_options);
+    auto reader = ReaderWriterFactory::make_reader(storage_options);
     reader->open(storage_options);
     input_bags.push_back(std::move(reader));
   }
@@ -265,12 +221,11 @@ void bag_rewrite(
     // room in the cache, which may require some new APIs
     auto modified_storage_options = storage_options;
     modified_storage_options.max_cache_size = 0u;
-    auto writer = make_writer(record_options);
+    auto writer = ReaderWriterFactory::make_writer(record_options);
     writer->open(modified_storage_options);
     output_bags.push_back(std::make_pair(std::move(writer), record_options));
   }
 
-  bag_rewrite(input_bags, output_bags);
+  perform_rewrite(input_bags, output_bags);
 }
-
 }  // namespace rosbag2_transport

rosbag2_transport/src/rosbag2_transport/reader_writer_factory.cpp

@@ -0,0 +1,70 @@
+// Copyright 2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "rosbag2_transport/reader_writer_factory.hpp"
+
+#include <memory>
+#include <utility>
+
+#include "rosbag2_compression/compression_options.hpp"
+#include "rosbag2_compression/sequential_compression_reader.hpp"
+#include "rosbag2_compression/sequential_compression_writer.hpp"
+#include "rosbag2_storage/metadata_io.hpp"
+
+namespace rosbag2_transport
+{
+
+std::unique_ptr<rosbag2_cpp::Reader> ReaderWriterFactory::make_reader(
+  const rosbag2_storage::StorageOptions & storage_options)
+{
+  rosbag2_storage::MetadataIo metadata_io;
+  std::unique_ptr<rosbag2_cpp::reader_interfaces::BaseReaderInterface> reader_impl;
+
+  if (metadata_io.metadata_file_exists(storage_options.uri)) {
+    auto metadata = metadata_io.read_metadata(storage_options.uri);
+    if (!metadata.compression_format.empty()) {
+      reader_impl = std::make_unique<rosbag2_compression::SequentialCompressionReader>();
+    }
+  }
+  if (!reader_impl) {
+    reader_impl = std::make_unique<rosbag2_cpp::readers::SequentialReader>();
+  }
+
+  return std::make_unique<rosbag2_cpp::Reader>(std::move(reader_impl));
+}
+
+std::unique_ptr<rosbag2_cpp::Writer> ReaderWriterFactory::make_writer(
+  const rosbag2_transport::RecordOptions & record_options)
+{
+  std::unique_ptr<rosbag2_cpp::writer_interfaces::BaseWriterInterface> writer_impl;
+  if (!record_options.compression_format.empty()) {
+    rosbag2_compression::CompressionOptions compression_options {
+      record_options.compression_format,
+      rosbag2_compression::compression_mode_from_string(record_options.compression_mode),
+      record_options.compression_queue_size,
+      record_options.compression_threads
+    };
+    if (compression_options.compression_threads < 1) {
+      compression_options.compression_threads = std::thread::hardware_concurrency();
+    }
+    writer_impl = std::make_unique<rosbag2_compression::SequentialCompressionWriter>(
+      compression_options);
+  } else {
+    writer_impl = std::make_unique<rosbag2_cpp::writers::SequentialWriter>();
+  }
+
+  return std::make_unique<rosbag2_cpp::Writer>(std::move(writer_impl));
+}
+
+}  // namespace rosbag2_transport