Fix transient local IPC publish (#2708)

* Fix transient local publish when inter and intra process communications are both present.

* Apply the fix to TypeAdapted signature

* Add an executor to intra_process_inter_process_mix_transient_local test case to enable inter process publishing test

Signed-off-by: Jeffery Hsu <[email protected]>
Co-authored-by: Tomoya Fujita <[email protected]>

Author: jefferyyjhsu

rclcpp/include/rclcpp/publisher.hpp

@@ -235,8 +235,12 @@ class Publisher : public PublisherBase
     // interprocess publish, resulting in lower publish-to-subscribe latency.
     // It's not possible to do that with an unique_ptr,
     // as do_intra_process_publish takes the ownership of the message.
+
+    // When durability is set to TransientLocal (i.e. there is a buffer),
+    // inter process publish should always take place to ensure
+    // late joiners receive past data.
     bool inter_process_publish_needed =
-      get_subscription_count() > get_intra_process_subscription_count();
+      get_subscription_count() > get_intra_process_subscription_count() || buffer_;
 
     if (inter_process_publish_needed) {
       auto shared_msg =
@@ -313,8 +317,11 @@ class Publisher : public PublisherBase
       return;
     }
 
+    // When durability is set to TransientLocal (i.e. there is a buffer),
+    // inter process publish should always take place to ensure
+    // late joiners receive past data.
     bool inter_process_publish_needed =
-      get_subscription_count() > get_intra_process_subscription_count();
+      get_subscription_count() > get_intra_process_subscription_count() || buffer_;
 
     if (inter_process_publish_needed) {
       auto ros_msg_ptr = std::make_shared<ROSMessageType>();

rclcpp/test/rclcpp/test_publisher.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <future>
 #include <chrono>
 #include <memory>
 #include <string>
@@ -725,11 +726,7 @@ TEST_F(TestPublisher, intra_process_transient_local) {
   auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {};
   struct IntraProcessCallback
   {
-    void callback_fun(size_t s)
-    {
-      (void) s;
-      called = true;
-    }
+    void callback_fun(size_t) {called = true;}
     bool called = false;
   };
   rclcpp::SubscriptionOptions sub_options_ipm_disabled;
@@ -786,3 +783,44 @@ TEST_F(TestPublisher, intra_process_transient_local) {
   EXPECT_FALSE(callback3.called);
   EXPECT_FALSE(callback4.called);
 }
+
+TEST_F(TestPublisher, intra_process_inter_process_mix_transient_local) {
+  constexpr auto history_depth = 10u;
+  initialize(rclcpp::NodeOptions().use_intra_process_comms(true));
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> pub_options_ipm_enabled;
+  pub_options_ipm_enabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
+
+  auto pub_ipm_enabled_transient_local_enabled = node->create_publisher<test_msgs::msg::Empty>(
+    "topic1",
+    rclcpp::QoS(rclcpp::KeepLast(history_depth)).transient_local(), pub_options_ipm_enabled);
+
+  test_msgs::msg::Empty msg;
+  pub_ipm_enabled_transient_local_enabled->publish(msg);
+
+  auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {};
+  struct SubscriptionCallback
+  {
+    void callback_fun(size_t) {called.set_value();}
+    std::promise<void> called;
+  };
+  rclcpp::SubscriptionOptions sub_options_ipm_disabled;
+  sub_options_ipm_disabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Disable;
+  SubscriptionCallback intra_callback, inter_callback;
+  std::future<void> intra_callback_future = intra_callback.called.get_future(),
+    inter_callback_future = inter_callback.called.get_future();
+  auto sub_ipm_disabled_transient_local_enabled = node->create_subscription<test_msgs::msg::Empty>(
+    "topic1",
+    rclcpp::QoS(rclcpp::KeepLast(history_depth)).transient_local(),
+    do_nothing, sub_options_ipm_disabled);
+  sub_ipm_disabled_transient_local_enabled->set_on_new_intra_process_message_callback(
+    std::bind(&SubscriptionCallback::callback_fun, &intra_callback, std::placeholders::_1));
+  sub_ipm_disabled_transient_local_enabled->set_on_new_message_callback(
+    std::bind(&SubscriptionCallback::callback_fun, &inter_callback, std::placeholders::_1));
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+  EXPECT_EQ(executor.spin_until_future_complete(inter_callback_future,
+    std::chrono::milliseconds(100)), rclcpp::FutureReturnCode::SUCCESS);
+  EXPECT_EQ(executor.spin_until_future_complete(intra_callback_future,
+    std::chrono::milliseconds(100)), rclcpp::FutureReturnCode::TIMEOUT);
+}