Refactor to use generic pub/sub from rclcpp

CMakeLists.txt

@@ -25,8 +25,6 @@ find_package(yaml_cpp_vendor REQUIRED)
 add_library(${PROJECT_NAME} SHARED
   src/${PROJECT_NAME}/domain_bridge.cpp
   src/${PROJECT_NAME}/domain_bridge_options.cpp
-  src/${PROJECT_NAME}/generic_publisher.cpp
-  src/${PROJECT_NAME}/generic_subscription.cpp
   src/${PROJECT_NAME}/parse_domain_bridge_yaml_config.cpp
   src/${PROJECT_NAME}/qos_options.cpp
   src/${PROJECT_NAME}/topic_bridge_options.cpp

doc/design.md

@@ -108,8 +108,7 @@ For convenience, a standalone binary executable is also provided for easy integr
 When bridging data, serialized data will be read from one domain and forwarded to another domain.
 As long as the bridge does not need access to the typed data, there is no need for deserialization and, therefore, C++ type support.
 In this way, we can handle arbitrary data coming from the middleware (as long as we load type support for the middleware).
-This is exactly how [rosbag2](https://github.com/ros2/rosbag2/tree/e4ce24cdfa7e24c6d2c025ecc38ab1157a0eecc8/rosbag2_transport) works; defining generic publishers and subscriptions.
-In fact, the generic publisher and subscription implementation is being moved to a common location that the domain bridge can leverage once available (see https://github.com/ros2/rclcpp/pull/1452).
+The domain bridge currently leverages the generic publisher and subscription implementations available in rclcpp.
 
 ### QoS mapping
 

src/domain_bridge/domain_bridge.cpp

@@ -32,12 +32,12 @@
 
 #include "rclcpp/executor.hpp"
 #include "rclcpp/expand_topic_or_service_name.hpp"
+#include "rclcpp/generic_publisher.hpp"
+#include "rclcpp/generic_subscription.hpp"
 #include "rclcpp/rclcpp.hpp"
 #include "rosbag2_cpp/typesupport_helpers.hpp"
 #include "rmw/types.h"
 
-#include "generic_publisher.hpp"
-#include "generic_subscription.hpp"
 #include "wait_for_qos_handler.hpp"
 
 namespace domain_bridge
@@ -50,7 +50,9 @@ class DomainBridgeImpl
   using NodeMap = std::unordered_map<std::size_t, std::shared_ptr<rclcpp::Node>>;
   using TopicBridgeMap = std::map<
     TopicBridge,
-    std::pair<std::shared_ptr<GenericPublisher>, std::shared_ptr<GenericSubscription>>>;
+    std::pair<
+      std::shared_ptr<rclcpp::GenericPublisher>,
+      std::shared_ptr<rclcpp::GenericSubscription>>>;
   using TypesupportMap = std::unordered_map<
     std::string, std::shared_ptr<rcpputils::SharedLibrary>>;
 
@@ -107,43 +109,34 @@ class DomainBridgeImpl
       type, "rosidl_typesupport_cpp");
   }
 
-  std::shared_ptr<GenericPublisher> create_publisher(
+  std::shared_ptr<rclcpp::GenericPublisher> create_publisher(
     rclcpp::Node::SharedPtr node,
     const std::string & topic_name,
+    const std::string & type,
     const rclcpp::QoS & qos,
-    const rosidl_message_type_support_t & typesupport_handle,
-    rclcpp::CallbackGroup::SharedPtr group)
+    rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> & options)
   {
-    auto publisher = std::make_shared<GenericPublisher>(
-      node->get_node_base_interface().get(),
-      typesupport_handle,
-      topic_name,
-      qos);
-    node->get_node_topics_interface()->add_publisher(publisher, std::move(group));
+    auto publisher = node->create_generic_publisher(topic_name, type, qos, options);
     return publisher;
   }
 
-  std::shared_ptr<GenericSubscription> create_subscription(
+  std::shared_ptr<rclcpp::GenericSubscription> create_subscription(
     rclcpp::Node::SharedPtr node,
-    std::shared_ptr<GenericPublisher> publisher,
+    std::shared_ptr<rclcpp::GenericPublisher> publisher,
     const std::string & topic_name,
+    const std::string & type,
     const rclcpp::QoS & qos,
-    const rosidl_message_type_support_t & typesupport_handle,
-    rclcpp::CallbackGroup::SharedPtr group)
+    rclcpp::SubscriptionOptionsWithAllocator<std::allocator<void>> & options)
   {
-    // Create subscription
-    auto subscription = std::make_shared<GenericSubscription>(
-      node->get_node_base_interface().get(),
-      typesupport_handle,
+    auto subscription = node->create_generic_subscription(
       topic_name,
+      type,
       qos,
       [publisher](std::shared_ptr<rclcpp::SerializedMessage> msg) {
         // Publish message into the other domain
-        auto serialized_data_ptr = std::make_shared<rcl_serialized_message_t>(
-          msg->get_rcl_serialized_message());
-        publisher->publish(serialized_data_ptr);
-      });
-    node->get_node_topics_interface()->add_subscription(subscription, std::move(group));
+        publisher->publish(*msg);
+      },
+      options);
     return subscription;
   }
 
@@ -239,27 +232,29 @@ class DomainBridgeImpl
           std::cerr << warning << std::endl;
         }
 
-        // Get typesupport handle
-        auto typesupport_handle = rosbag2_cpp::get_typesupport_handle(
-          type, "rosidl_typesupport_cpp", loaded_typesupports_.at(type));
+        rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> publisher_options;
+        rclcpp::SubscriptionOptionsWithAllocator<std::allocator<void>> subscription_options;
+
+        publisher_options.callback_group = topic_options.callback_group();
+        subscription_options.callback_group = topic_options.callback_group();
 
         // Create publisher for the 'to_domain'
         // The publisher should be created first so it is available to the subscription callback
         auto publisher = this->create_publisher(
           to_domain_node,
           topic_remapped,
+          type,
           qos,
-          *typesupport_handle,
-          topic_options.callback_group());
+          publisher_options);
 
         // Create subscription for the 'from_domain'
         auto subscription = this->create_subscription(
           from_domain_node,
           publisher,
           topic,
+          type,
           qos,
-          *typesupport_handle,
-          topic_options.callback_group());
+          subscription_options);
 
         this->bridged_topics_[topic_bridge] = {publisher, subscription};
       };