diff --git a/rclcpp/src/rclcpp/subscription_base.cpp b/rclcpp/src/rclcpp/subscription_base.cpp index 2ea4664f52..4a9ce761b2 100644 --- a/rclcpp/src/rclcpp/subscription_base.cpp +++ b/rclcpp/src/rclcpp/subscription_base.cpp @@ -99,6 +99,23 @@ SubscriptionBase::SubscriptionBase( SubscriptionBase::~SubscriptionBase() { + // The middleware may be invoking this callback from one of its own threads. + // Unregister it before rcl_subscription_fini tears down the RMW subscription. + try { + clear_on_new_message_callback(); + } catch (const std::exception & exception) { + RCLCPP_ERROR_STREAM( + node_logger_, + "rclcpp::SubscriptionBase@" << this << + " failed to clear the on new message callback during destruction: " << + exception.what()); + } catch (...) { + RCLCPP_ERROR_STREAM( + node_logger_, + "rclcpp::SubscriptionBase@" << this << + " failed to clear the on new message callback during destruction"); + } + if (!use_intra_process_) { return; } diff --git a/rclcpp/test/rclcpp/test_subscription.cpp b/rclcpp/test/rclcpp/test_subscription.cpp index fb7a57a1f7..360f726134 100644 --- a/rclcpp/test/rclcpp/test_subscription.cpp +++ b/rclcpp/test/rclcpp/test_subscription.cpp @@ -316,6 +316,39 @@ TEST_F(TestSubscription, rcl_subscription_fini_error) { node_->create_subscription("topic", 10, callback).reset()); } +TEST_F(TestSubscription, on_new_message_callback_cleared_before_fini) { + initialize(rclcpp::NodeOptions().use_intra_process_comms(false)); + using test_msgs::msg::Empty; + + bool callback_registered = false; + auto mock = mocking_utils::patch( + "lib:rclcpp", rcl_subscription_set_on_new_message_callback, + [&callback_registered]( + const rcl_subscription_t *, rcl_event_callback_t callback, const void * user_data) + { + (void)user_data; + callback_registered = callback != nullptr; + return RCL_RET_OK; + }); + + auto sub = node_->create_subscription( + "topic", 10, [](Empty::ConstSharedPtr) {}); + sub->set_on_new_message_callback([](size_t) {}); + ASSERT_TRUE(callback_registered); + + auto fini_mock = mocking_utils::patch( + "lib:rclcpp", rcl_subscription_fini, + [&callback_registered](rcl_subscription_t *, rcl_node_t *) + { + EXPECT_FALSE(callback_registered); + return RCL_RET_OK; + }); + + sub.reset(); + + EXPECT_FALSE(callback_registered); +} + TEST_F(TestSubscription, rcl_subscription_get_actual_qos_error) { initialize(); auto callback = [](std::shared_ptr) {};