26.0.0

Changelog.
Signed-off-by: Chris Lalancette <clalancette@gmail.com>
2024-01-24 14:08:12 +00:00 · 2024-01-24 14:08:05 +00:00 · 2024-01-18 16:40:13 -05:00 · 2024-01-18 08:39:59 -08:00 · 2024-01-08 08:23:33 -05:00 · 2024-01-03 08:17:47 -05:00
165 changed files with 5125 additions and 2144 deletions
--- a/rclcpp/CHANGELOG.rst
+++ b/rclcpp/CHANGELOG.rst
@@ -2,6 +2,134 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^

+26.0.0 (2024-01-24)
+-------------------
+* Increase the cppcheck timeout to 600 seconds. (`#2409 <https://github.com/ros2/rclcpp/issues/2409>`_)
+* Add transient local durability support to publisher and subscriptions when using intra-process communication (`#2303 <https://github.com/ros2/rclcpp/issues/2303>`_)
+* Stop storing the context in the guard condition. (`#2400 <https://github.com/ros2/rclcpp/issues/2400>`_)
+* Contributors: Chris Lalancette, Jeffery Hsu
+
+25.0.0 (2023-12-26)
+-------------------
+* Updated GenericSubscription to AnySubscriptionCallback (`#1928 <https://github.com/ros2/rclcpp/issues/1928>`_)
+* make type support helper supported for service (`#2209 <https://github.com/ros2/rclcpp/issues/2209>`_)
+* Adding QoS to subscription options (`#2323 <https://github.com/ros2/rclcpp/issues/2323>`_)
+* Switch to target_link_libraries. (`#2374 <https://github.com/ros2/rclcpp/issues/2374>`_)
+* aligh with rcl that a rosout publisher of a node might not exist (`#2357 <https://github.com/ros2/rclcpp/issues/2357>`_)
+* Fix data race in EventHandlerBase (`#2349 <https://github.com/ros2/rclcpp/issues/2349>`_)
+* Support users holding onto shared pointers in the message memory pool (`#2336 <https://github.com/ros2/rclcpp/issues/2336>`_)
+* Contributors: Chen Lihui, Chris Lalancette, DensoADAS, Lucas Wendland, mauropasse
+
+24.0.0 (2023-11-06)
+-------------------
+* fix (signal_handler.hpp): spelling (`#2356 <https://github.com/ros2/rclcpp/issues/2356>`_)
+* Updates to not use std::move in some places. (`#2353 <https://github.com/ros2/rclcpp/issues/2353>`_)
+* rclcpp::Time::max() clock type support. (`#2352 <https://github.com/ros2/rclcpp/issues/2352>`_)
+* Serialized Messages with Topic Statistics (`#2274 <https://github.com/ros2/rclcpp/issues/2274>`_)
+* Add a custom deleter when constructing rcl_service_t (`#2351 <https://github.com/ros2/rclcpp/issues/2351>`_)
+* Disable the loaned messages inside the executor. (`#2335 <https://github.com/ros2/rclcpp/issues/2335>`_)
+* Use message_info in SubscriptionTopicStatistics instead of typed message (`#2337 <https://github.com/ros2/rclcpp/issues/2337>`_)
+* Add missing 'enable_rosout' comments (`#2345 <https://github.com/ros2/rclcpp/issues/2345>`_)
+* Adjust rclcpp usage of type description service (`#2344 <https://github.com/ros2/rclcpp/issues/2344>`_)
+* address rate related flaky tests. (`#2329 <https://github.com/ros2/rclcpp/issues/2329>`_)
+* Fixes pointed out by the clang analyzer. (`#2339 <https://github.com/ros2/rclcpp/issues/2339>`_)
+* Remove useless ROSRate class (`#2326 <https://github.com/ros2/rclcpp/issues/2326>`_)
+* Contributors: Alexey Merzlyakov, Chris Lalancette, Jiaqi Li, Lucas Wendland, Michael Carroll, Michael Orlov, Tomoya Fujita, Zard-C
+
+23.2.0 (2023-10-09)
+-------------------
+* add clients & services count (`#2072 <https://github.com/ros2/rclcpp/issues/2072>`_)
+* remove invalid sized allocation test for SerializedMessage. (`#2330 <https://github.com/ros2/rclcpp/issues/2330>`_)
+* Adding API to copy all parameters from one node to another (`#2304 <https://github.com/ros2/rclcpp/issues/2304>`_)
+* Contributors: Minju, Lee, Steve Macenski, Tomoya Fujita
+
+23.1.0 (2023-10-04)
+-------------------
+* Add locking to protect the TimeSource::NodeState::node_base\_ (`#2320 <https://github.com/ros2/rclcpp/issues/2320>`_)
+* Update SignalHandler get_global_signal_handler to avoid complex types in static memory (`#2316 <https://github.com/ros2/rclcpp/issues/2316>`_)
+* Removing Old Connext Tests (`#2313 <https://github.com/ros2/rclcpp/issues/2313>`_)
+* Documentation for list_parameters  (`#2315 <https://github.com/ros2/rclcpp/issues/2315>`_)
+* Decouple rosout publisher init from node init. (`#2174 <https://github.com/ros2/rclcpp/issues/2174>`_)
+* fix the depth to relative in list_parameters (`#2300 <https://github.com/ros2/rclcpp/issues/2300>`_)
+* Contributors: Chris Lalancette, Lucas Wendland, Minju, Lee, Tomoya Fujita, Tully Foote
+
+23.0.0 (2023-09-08)
+-------------------
+* Fix the return type of Rate::period. (`#2301 <https://github.com/ros2/rclcpp/issues/2301>`_)
+* Update API docs links in package READMEs (`#2302 <https://github.com/ros2/rclcpp/issues/2302>`_)
+* Cleanup flaky timers_manager tests. (`#2299 <https://github.com/ros2/rclcpp/issues/2299>`_)
+* Contributors: Chris Lalancette, Christophe Bedard
+
+22.2.0 (2023-09-07)
+-------------------
+* Topic correct typeadapter deduction (`#2294 <https://github.com/ros2/rclcpp/issues/2294>`_)
+* Fix C++20 allocator construct deprecation (`#2292 <https://github.com/ros2/rclcpp/issues/2292>`_)
+* Make Rate to select the clock to work with (`#2123 <https://github.com/ros2/rclcpp/issues/2123>`_)
+* Correct the position of a comment. (`#2290 <https://github.com/ros2/rclcpp/issues/2290>`_)
+* Remove unnecessary lambda captures in the tests. (`#2289 <https://github.com/ros2/rclcpp/issues/2289>`_)
+* Add rcl_logging_interface as an explicit dependency. (`#2284 <https://github.com/ros2/rclcpp/issues/2284>`_)
+* Revamp list_parameters to be more efficient and easier to read. (`#2282 <https://github.com/ros2/rclcpp/issues/2282>`_)
+* Contributors: AiVerisimilitude, Alexey Merzlyakov, Chen Lihui, Chris Lalancette, Jiaqi Li
+
+22.1.0 (2023-08-21)
+-------------------
+* Do not crash Executor when send_response fails due to client failure. (`#2276 <https://github.com/ros2/rclcpp/issues/2276>`_)
+* Adding Custom Unknown Type Error (`#2272 <https://github.com/ros2/rclcpp/issues/2272>`_)
+* Add a pimpl inside rclcpp::Node for future distro backports (`#2228 <https://github.com/ros2/rclcpp/issues/2228>`_)
+* Remove an unused variable from the events executor tests. (`#2270 <https://github.com/ros2/rclcpp/issues/2270>`_)
+* Add spin_all shortcut (`#2246 <https://github.com/ros2/rclcpp/issues/2246>`_)
+* Adding Missing Group Exceptions (`#2256 <https://github.com/ros2/rclcpp/issues/2256>`_)
+* Change associated clocks storage to unordered_set (`#2257 <https://github.com/ros2/rclcpp/issues/2257>`_)
+* associated clocks should be protected by mutex. (`#2255 <https://github.com/ros2/rclcpp/issues/2255>`_)
+* Instrument loaned message publication code path (`#2240 <https://github.com/ros2/rclcpp/issues/2240>`_)
+* Contributors: Chris Lalancette, Christophe Bedard, Emerson Knapp, Luca Della Vedova, Lucas Wendland, Tomoya Fujita, Tony Najjar
+
+22.0.0 (2023-07-11)
+-------------------
+* Implement get_node_type_descriptions_interface for lifecyclenode and add smoke test for it (`#2237 <https://github.com/ros2/rclcpp/issues/2237>`_)
+* Add new node interface TypeDescriptionsInterface to provide GetTypeDescription service (`#2224 <https://github.com/ros2/rclcpp/issues/2224>`_)
+* Move always_false_v to detail namespace (`#2232 <https://github.com/ros2/rclcpp/issues/2232>`_)
+* Revamp the test_subscription.cpp tests. (`#2227 <https://github.com/ros2/rclcpp/issues/2227>`_)
+* warning: comparison of integer expressions of different signedness (`#2219 <https://github.com/ros2/rclcpp/issues/2219>`_)
+* Modifies timers API to select autostart state (`#2005 <https://github.com/ros2/rclcpp/issues/2005>`_)
+* Enable callback group tests for connextdds (`#2182 <https://github.com/ros2/rclcpp/issues/2182>`_)
+* Contributors: Chris Lalancette, Christopher Wecht, Eloy Briceno, Emerson Knapp, Nathan Wiebe Neufeldt, Tomoya Fujita
+
+21.3.0 (2023-06-12)
+-------------------
+* Fix up misspellings of "receive". (`#2208 <https://github.com/ros2/rclcpp/issues/2208>`_)
+* Remove flaky stressAddRemoveNode test (`#2206 <https://github.com/ros2/rclcpp/issues/2206>`_)
+* Use TRACETOOLS\_ prefix for tracepoint-related macros (`#2162 <https://github.com/ros2/rclcpp/issues/2162>`_)
+* Contributors: Chris Lalancette, Christophe Bedard, Michael Carroll
+
+21.2.0 (2023-06-07)
+-------------------
+* remove nolint since ament_cpplint updated for the c++17 header (`#2198 <https://github.com/ros2/rclcpp/issues/2198>`_)
+* Feature/available capacity of ipm (`#2173 <https://github.com/ros2/rclcpp/issues/2173>`_)
+* add mutex to protect events_executor current entity collection (`#2187 <https://github.com/ros2/rclcpp/issues/2187>`_)
+* Declare rclcpp callbacks before the rcl entities (`#2024 <https://github.com/ros2/rclcpp/issues/2024>`_)
+* Contributors: Alberto Soragna, Chen Lihui, DensoADAS, mauropasse
+
+21.1.1 (2023-05-11)
+-------------------
+* Fix race condition in events-executor (`#2177 <https://github.com/ros2/rclcpp/issues/2177>`_)
+* Add missing stdexcept include (`#2186 <https://github.com/ros2/rclcpp/issues/2186>`_)
+* Fix a format-security warning when building with clang (`#2171 <https://github.com/ros2/rclcpp/issues/2171>`_)
+* Fix delivered message kind (`#2175 <https://github.com/ros2/rclcpp/issues/2175>`_)
+* Contributors: Alberto Soragna, Chris Lalancette, methylDragon, Øystein Sture
+
+21.1.0 (2023-04-27)
+-------------------
+
+21.0.0 (2023-04-18)
+-------------------
+* Add support for logging service. (`#2122 <https://github.com/ros2/rclcpp/issues/2122>`_)
+* Picking ABI-incompatible executor changes (`#2170 <https://github.com/ros2/rclcpp/issues/2170>`_)
+* add events-executor and timers-manager in rclcpp (`#2155 <https://github.com/ros2/rclcpp/issues/2155>`_)
+* Create common structures for executors to use (`#2143 <https://github.com/ros2/rclcpp/issues/2143>`_)
+* Implement deliver message kind (`#2168 <https://github.com/ros2/rclcpp/issues/2168>`_)
+* Contributors: Alberto Soragna, Lei Liu, Michael Carroll, methylDragon
+
 20.0.0 (2023-04-13)
 -------------------
 * applied tracepoints for ring_buffer (`#2091 <https://github.com/ros2/rclcpp/issues/2091>`_)
--- a/rclcpp/CMakeLists.txt
+++ b/rclcpp/CMakeLists.txt
@@ -10,11 +10,14 @@ find_package(builtin_interfaces REQUIRED)
 find_package(libstatistics_collector REQUIRED)
 find_package(rcl REQUIRED)
 find_package(rcl_interfaces REQUIRED)
+find_package(rcl_logging_interface REQUIRED)
 find_package(rcl_yaml_param_parser REQUIRED)
 find_package(rcpputils REQUIRED)
 find_package(rcutils REQUIRED)
 find_package(rmw REQUIRED)
 find_package(rosgraph_msgs REQUIRED)
+find_package(rosidl_dynamic_typesupport REQUIRED)
+find_package(rosidl_runtime_c REQUIRED)
 find_package(rosidl_runtime_cpp REQUIRED)
 find_package(rosidl_typesupport_c REQUIRED)
 find_package(rosidl_typesupport_cpp REQUIRED)
@@ -43,6 +46,7 @@ set(${PROJECT_NAME}_SRCS
  src/rclcpp/context.cpp
  src/rclcpp/contexts/default_context.cpp
  src/rclcpp/detail/add_guard_condition_to_rcl_wait_set.cpp
+  src/rclcpp/detail/resolve_intra_process_buffer_type.cpp
  src/rclcpp/detail/resolve_parameter_overrides.cpp
  src/rclcpp/detail/rmw_implementation_specific_payload.cpp
  src/rclcpp/detail/rmw_implementation_specific_publisher_payload.cpp
@@ -92,6 +96,7 @@ set(${PROJECT_NAME}_SRCS
  src/rclcpp/node_interfaces/node_time_source.cpp
  src/rclcpp/node_interfaces/node_timers.cpp
  src/rclcpp/node_interfaces/node_topics.cpp
+  src/rclcpp/node_interfaces/node_type_descriptions.cpp
  src/rclcpp/node_interfaces/node_waitables.cpp
  src/rclcpp/node_options.cpp
  src/rclcpp/parameter.cpp
@@ -105,6 +110,7 @@ set(${PROJECT_NAME}_SRCS
  src/rclcpp/qos.cpp
  src/rclcpp/event_handler.cpp
  src/rclcpp/qos_overriding_options.cpp
+  src/rclcpp/rate.cpp
  src/rclcpp/serialization.cpp
  src/rclcpp/serialized_message.cpp
  src/rclcpp/service.cpp
@@ -199,22 +205,28 @@ target_include_directories(${PROJECT_NAME} PUBLIC
  "$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/include>"
  "$<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}/include>"
  "$<INSTALL_INTERFACE:include/${PROJECT_NAME}>")
-target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})
-# specific order: dependents before dependencies
-ament_target_dependencies(${PROJECT_NAME}
-  "ament_index_cpp"
-  "libstatistics_collector"
-  "rcl"
-  "rcl_interfaces"
-  "rcl_yaml_param_parser"
-  "rcpputils"
-  "rcutils"
-  "builtin_interfaces"
-  "rosgraph_msgs"
-  "rosidl_typesupport_cpp"
-  "rosidl_runtime_cpp"
-  "statistics_msgs"
-  "tracetools"
+target_link_libraries(${PROJECT_NAME} PUBLIC
+  ${builtin_interfaces_TARGETS}
+  libstatistics_collector::libstatistics_collector
+  rcl::rcl
+  ${rcl_interfaces_TARGETS}
+  rcl_yaml_param_parser::rcl_yaml_param_parser
+  rcpputils::rcpputils
+  rcutils::rcutils
+  rmw::rmw
+  ${rosgraph_msgs_TARGETS}
+  rosidl_dynamic_typesupport::rosidl_dynamic_typesupport
+  rosidl_runtime_c::rosidl_runtime_c
+  rosidl_runtime_cpp::rosidl_runtime_cpp
+  rosidl_typesupport_cpp::rosidl_typesupport_cpp
+  ${statistics_msgs_TARGETS}
+  tracetools::tracetools
+  ${CMAKE_THREAD_LIBS_INIT}
+)
+
+target_link_libraries(${PROJECT_NAME} PRIVATE
+  ament_index_cpp::ament_index_cpp
+  rcl_logging_interface::rcl_logging_interface
 )

 # Causes the visibility macros to use dllexport rather than dllimport,
@@ -236,20 +248,23 @@ ament_export_libraries(${PROJECT_NAME})
 # Export modern CMake targets
 ament_export_targets(${PROJECT_NAME})

-# specific order: dependents before dependencies
-ament_export_dependencies(ament_index_cpp)
-ament_export_dependencies(libstatistics_collector)
-ament_export_dependencies(rcl)
-ament_export_dependencies(rcpputils)
-ament_export_dependencies(rcutils)
-ament_export_dependencies(builtin_interfaces)
-ament_export_dependencies(rosgraph_msgs)
-ament_export_dependencies(rosidl_typesupport_cpp)
-ament_export_dependencies(rosidl_typesupport_c)
-ament_export_dependencies(rosidl_runtime_cpp)
-ament_export_dependencies(rcl_yaml_param_parser)
-ament_export_dependencies(statistics_msgs)
-ament_export_dependencies(tracetools)
+ament_export_dependencies(
+  builtin_interfaces
+  libstatistics_collector
+  rcl
+  rcl_interfaces
+  rcl_yaml_param_parser
+  rcpputils
+  rcutils
+  rmw
+  rosgraph_msgs
+  rosidl_dynamic_typesupport
+  rosidl_runtime_c
+  rosidl_runtime_cpp
+  rosidl_typesupport_cpp
+  statistics_msgs
+  tracetools
+)

 if(BUILD_TESTING)
  find_package(ament_lint_auto REQUIRED)
@@ -267,7 +282,7 @@ install(

 if(TEST cppcheck)
  # must set the property after ament_package()
-  set_tests_properties(cppcheck PROPERTIES TIMEOUT 500)
+  set_tests_properties(cppcheck PROPERTIES TIMEOUT 600)
 endif()

 if(TEST cpplint)
--- a/rclcpp/README.md
+++ b/rclcpp/README.md
@@ -2,7 +2,7 @@

 The ROS client library in C++.

-Visit the [rclcpp API documentation](http://docs.ros2.org/latest/api/rclcpp/) for a complete list of its main components and features.
+The link to the latest rclcpp API documentation, which includes a complete list of its main components and features, can be found on the rclcpp package info page, at the [ROS Index](https://index.ros.org/p/rclcpp/).

 ## Quality Declaration

--- a/rclcpp/include/rclcpp/any_service_callback.hpp
+++ b/rclcpp/include/rclcpp/any_service_callback.hpp
@@ -156,7 +156,7 @@ public:
    const std::shared_ptr<rmw_request_id_t> & request_header,
    std::shared_ptr<typename ServiceT::Request> request)
  {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
    if (std::holds_alternative<std::monostate>(callback_)) {
      // TODO(ivanpauno): Remove the set method, and force the users of this class
      // to pass a callback at construnciton.
@@ -182,7 +182,7 @@ public:
      const auto & cb = std::get<SharedPtrWithRequestHeaderCallback>(callback_);
      cb(request_header, std::move(request), response);
    }
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
    return response;
  }

@@ -191,9 +191,9 @@ public:
 #ifndef TRACETOOLS_DISABLED
    std::visit(
      [this](auto && arg) {
-        if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+        if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
          char * symbol = tracetools::get_symbol(arg);
-          DO_TRACEPOINT(
+          TRACETOOLS_DO_TRACEPOINT(
            rclcpp_callback_register,
            static_cast<const void *>(this),
            symbol);
--- a/rclcpp/include/rclcpp/any_subscription_callback.hpp
+++ b/rclcpp/include/rclcpp/any_subscription_callback.hpp
@@ -20,7 +20,7 @@
 #include <stdexcept>
 #include <type_traits>
 #include <utility>
-#include <variant>  // NOLINT[build/include_order]
+#include <variant>

 #include "rosidl_runtime_cpp/traits.hpp"
 #include "tracetools/tracetools.h"
@@ -30,19 +30,20 @@
 #include "rclcpp/detail/subscription_callback_type_helper.hpp"
 #include "rclcpp/function_traits.hpp"
 #include "rclcpp/message_info.hpp"
+#include "rclcpp/serialization.hpp"
 #include "rclcpp/serialized_message.hpp"
 #include "rclcpp/type_adapter.hpp"


-template<class>
-inline constexpr bool always_false_v = false;
-
 namespace rclcpp
 {

 namespace detail
 {

+template<class>
+inline constexpr bool always_false_v = false;
+
 template<typename MessageT, typename AllocatorT>
 struct MessageDeleterHelper
 {
@@ -158,13 +159,14 @@ struct AnySubscriptionCallbackPossibleTypes
 template<
  typename MessageT,
  typename AllocatorT,
-  bool is_adapted_type = rclcpp::TypeAdapter<MessageT>::is_specialized::value
+  bool is_adapted_type = rclcpp::TypeAdapter<MessageT>::is_specialized::value,
+  bool is_serialized_type = serialization_traits::is_serialized_message_class<MessageT>::value
 >
 struct AnySubscriptionCallbackHelper;

 /// Specialization for when MessageT is not a TypeAdapter.
 template<typename MessageT, typename AllocatorT>
-struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false, false>
 {
  using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;

@@ -194,7 +196,7 @@ struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false>

 /// Specialization for when MessageT is a TypeAdapter.
 template<typename MessageT, typename AllocatorT>
-struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true, false>
 {
  using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;

@@ -232,6 +234,26 @@ struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, true>
  >;
 };

+/// Specialization for when MessageT is a SerializedMessage to exclude duplicated declarations.
+template<typename MessageT, typename AllocatorT>
+struct AnySubscriptionCallbackHelper<MessageT, AllocatorT, false, true>
+{
+  using CallbackTypes = AnySubscriptionCallbackPossibleTypes<MessageT, AllocatorT>;
+
+  using variant_type = std::variant<
+    typename CallbackTypes::ConstRefSerializedMessageCallback,
+    typename CallbackTypes::ConstRefSerializedMessageWithInfoCallback,
+    typename CallbackTypes::UniquePtrSerializedMessageCallback,
+    typename CallbackTypes::UniquePtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::SharedConstPtrSerializedMessageCallback,
+    typename CallbackTypes::SharedConstPtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::ConstRefSharedConstPtrSerializedMessageCallback,
+    typename CallbackTypes::ConstRefSharedConstPtrSerializedMessageWithInfoCallback,
+    typename CallbackTypes::SharedPtrSerializedMessageCallback,
+    typename CallbackTypes::SharedPtrSerializedMessageWithInfoCallback
+  >;
+};
+
 }  // namespace detail

 template<
@@ -487,12 +509,14 @@ public:
  }

  // Dispatch when input is a ros message and the output could be anything.
-  void
+  template<typename TMsg = ROSMessageType>
+  typename std::enable_if<!serialization_traits::is_serialized_message_class<TMsg>::value,
+    void>::type
  dispatch(
    std::shared_ptr<ROSMessageType> message,
    const rclcpp::MessageInfo & message_info)
  {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
    // Check if the variant is "unset", throw if it is.
    if (callback_variant_.index() == 0) {
      if (std::get<0>(callback_variant_) == nullptr) {
@@ -580,19 +604,19 @@ public:
        }
        // condition to catch unhandled callback types
        else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
        }
      }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
  }

  // Dispatch when input is a serialized message and the output could be anything.
  void
  dispatch(
-    std::shared_ptr<rclcpp::SerializedMessage> serialized_message,
+    std::shared_ptr<const rclcpp::SerializedMessage> serialized_message,
    const rclcpp::MessageInfo & message_info)
  {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), false);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), false);
    // Check if the variant is "unset", throw if it is.
    if (callback_variant_.index() == 0) {
      if (std::get<0>(callback_variant_) == nullptr) {
@@ -660,10 +684,10 @@ public:
        }
        // condition to catch unhandled callback types
        else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
        }
      }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
  }

  void
@@ -671,7 +695,7 @@ public:
    std::shared_ptr<const SubscribedType> message,
    const rclcpp::MessageInfo & message_info)
  {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), true);
    // Check if the variant is "unset", throw if it is.
    if (callback_variant_.index() == 0) {
      if (std::get<0>(callback_variant_) == nullptr) {
@@ -790,10 +814,10 @@ public:
        }
        // condition to catch unhandled callback types
        else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
        }
      }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
  }

  void
@@ -801,7 +825,7 @@ public:
    std::unique_ptr<SubscribedType, SubscribedTypeDeleter> message,
    const rclcpp::MessageInfo & message_info)
  {
-    TRACEPOINT(callback_start, static_cast<const void *>(this), true);
+    TRACETOOLS_TRACEPOINT(callback_start, static_cast<const void *>(this), true);
    // Check if the variant is "unset", throw if it is.
    if (callback_variant_.index() == 0) {
      if (std::get<0>(callback_variant_) == nullptr) {
@@ -924,10 +948,10 @@ public:
        }
        // condition to catch unhandled callback types
        else {  // NOLINT[readability/braces]
-          static_assert(always_false_v<T>, "unhandled callback type");
+          static_assert(detail::always_false_v<T>, "unhandled callback type");
        }
      }, callback_variant_);
-    TRACEPOINT(callback_end, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(callback_end, static_cast<const void *>(this));
  }

  constexpr
@@ -965,9 +989,9 @@ public:
 #ifndef TRACETOOLS_DISABLED
    std::visit(
      [this](auto && callback) {
-        if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+        if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
          char * symbol = tracetools::get_symbol(callback);
-          DO_TRACEPOINT(
+          TRACETOOLS_DO_TRACEPOINT(
            rclcpp_callback_register,
            static_cast<const void *>(this),
            symbol);
--- a/rclcpp/include/rclcpp/callback_group.hpp
+++ b/rclcpp/include/rclcpp/callback_group.hpp
@@ -129,8 +129,7 @@ public:
   * added to the executor in either case.
   *
   * \param[in] group_type The type of the callback group.
-   * \param[in] get_node_context Lambda to retrieve the node context when
-   *   checking that the creating node is valid and using the guard condition.
+   * \param[in] context A weak pointer to the context associated with this callback group.
   * \param[in] automatically_add_to_executor_with_node A boolean that
   *   determines whether a callback group is automatically added to an executor
   *   with the node with which it is associated.
@@ -138,7 +137,7 @@ public:
  RCLCPP_PUBLIC
  explicit CallbackGroup(
    CallbackGroupType group_type,
-    std::function<rclcpp::Context::SharedPtr(void)> get_node_context,
+    rclcpp::Context::WeakPtr context,
    bool automatically_add_to_executor_with_node = true);

  /// Default destructor.
@@ -180,6 +179,13 @@ public:
    return _find_ptrs_if_impl<rclcpp::Waitable, Function>(func, waitable_ptrs_);
  }

+  /// Get the total number of entities in this callback group.
+  /**
+   * \return the number of entities in the callback group.
+   */
+  RCLCPP_PUBLIC
+  size_t size() const;
+
  RCLCPP_PUBLIC
  std::atomic_bool &
  can_be_taken_from();
@@ -221,16 +227,6 @@ public:
  bool
  automatically_add_to_executor_with_node() const;

-  /// Retrieve the guard condition used to signal changes to this callback group.
-  /**
-   * \param[in] context_ptr context to use when creating the guard condition
-   * \return guard condition if it is valid, otherwise nullptr.
-   */
-  [[deprecated("Use get_notify_guard_condition() without arguments")]]
-  RCLCPP_PUBLIC
-  rclcpp::GuardCondition::SharedPtr
-  get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr);
-
  /// Retrieve the guard condition used to signal changes to this callback group.
  /**
   * \return guard condition if it is valid, otherwise nullptr.
@@ -290,7 +286,7 @@ protected:
  std::shared_ptr<rclcpp::GuardCondition> notify_guard_condition_ = nullptr;
  std::recursive_mutex notify_guard_condition_mutex_;

-  std::function<rclcpp::Context::SharedPtr(void)> get_context_;
+  rclcpp::Context::WeakPtr context_;

 private:
  template<typename TypeT, typename Function>
--- a/rclcpp/include/rclcpp/client.hpp
+++ b/rclcpp/include/rclcpp/client.hpp
@@ -20,13 +20,13 @@
 #include <future>
 #include <memory>
 #include <mutex>
-#include <optional>  // NOLINT, cpplint doesn't think this is a cpp std header
+#include <optional>
 #include <sstream>
 #include <string>
 #include <tuple>
 #include <unordered_map>
 #include <utility>
-#include <variant>  // NOLINT
+#include <variant>
 #include <vector>

 #include "rcl/client.h"
@@ -363,12 +363,16 @@ protected:
  std::shared_ptr<rclcpp::Context> context_;
  rclcpp::Logger node_logger_;

+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_response_callback_ before
+  // client_handle_, so on destruction the client is
+  // destroyed first. Otherwise, the rmw client callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_response_callback_{nullptr};
+  // Declare client_handle_ after callback
  std::shared_ptr<rcl_client_t> client_handle_;

  std::atomic<bool> in_use_by_wait_set_{false};
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_response_callback_{nullptr};
 };

 template<typename ServiceT>
--- a/rclcpp/include/rclcpp/context.hpp
+++ b/rclcpp/include/rclcpp/context.hpp
@@ -26,6 +26,7 @@
 #include <unordered_set>
 #include <utility>
 #include <vector>
+#include <stdexcept>

 #include "rcl/context.h"
 #include "rcl/guard_condition.h"
--- a/rclcpp/include/rclcpp/copy_all_parameter_values.hpp
+++ b/rclcpp/include/rclcpp/copy_all_parameter_values.hpp
@@ -0,0 +1,82 @@
+// Copyright 2023 Open Navigation LLC
+//
+// 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 RCLCPP__COPY_ALL_PARAMETER_VALUES_HPP_
+#define RCLCPP__COPY_ALL_PARAMETER_VALUES_HPP_
+
+#include <string>
+#include <vector>
+
+#include "rcl_interfaces/srv/list_parameters.hpp"
+#include "rcl_interfaces/msg/parameter_descriptor.hpp"
+#include "rcl_interfaces/msg/set_parameters_result.hpp"
+
+#include "rclcpp/parameter.hpp"
+#include "rclcpp/logger.hpp"
+#include "rclcpp/logging.hpp"
+
+namespace rclcpp
+{
+
+/**
+ * Copy all parameters from one source node to another destination node.
+ * May throw exceptions if parameters from source are uninitialized or undeclared.
+ * \param source Node to copy parameters from
+ * \param destination Node to copy parameters to
+ * \param override_existing_params Default false. Whether to override existing destination params
+ * if both the source and destination contain the same parameter.
+ */
+template<typename NodeT1, typename NodeT2>
+void
+copy_all_parameter_values(
+  const NodeT1 & source, const NodeT2 & destination, const bool override_existing_params = false)
+{
+  using Parameters = std::vector<rclcpp::Parameter>;
+  using Descriptions = std::vector<rcl_interfaces::msg::ParameterDescriptor>;
+  auto source_params = source->get_node_parameters_interface();
+  auto dest_params = destination->get_node_parameters_interface();
+  rclcpp::Logger logger = destination->get_node_logging_interface()->get_logger();
+
+  std::vector<std::string> param_names = source_params->list_parameters({}, 0).names;
+  Parameters params = source_params->get_parameters(param_names);
+  Descriptions descriptions = source_params->describe_parameters(param_names);
+
+  for (unsigned int idx = 0; idx != params.size(); idx++) {
+    if (!dest_params->has_parameter(params[idx].get_name())) {
+      dest_params->declare_parameter(
+        params[idx].get_name(), params[idx].get_parameter_value(), descriptions[idx]);
+    } else if (override_existing_params) {
+      try {
+        rcl_interfaces::msg::SetParametersResult result =
+          dest_params->set_parameters_atomically({params[idx]});
+        if (!result.successful) {
+          // Parameter update rejected or read-only
+          RCLCPP_WARN(
+            logger,
+            "Unable to set parameter (%s): %s!",
+            params[idx].get_name().c_str(), result.reason.c_str());
+        }
+      } catch (const rclcpp::exceptions::InvalidParameterTypeException & e) {
+        RCLCPP_WARN(
+          logger,
+          "Unable to set parameter (%s): incompatable parameter type (%s)!",
+          params[idx].get_name().c_str(), e.what());
+      }
+    }
+  }
+}
+
+}  // namespace rclcpp
+
+#endif  // RCLCPP__COPY_ALL_PARAMETER_VALUES_HPP_
--- a/rclcpp/include/rclcpp/create_generic_subscription.hpp
+++ b/rclcpp/include/rclcpp/create_generic_subscription.hpp
@@ -45,13 +45,15 @@ namespace rclcpp
 * Not all publisher options are currently respected, the only relevant options for this
 * publisher are `event_callbacks`, `use_default_callbacks`, and `%callback_group`.
 */
-template<typename AllocatorT = std::allocator<void>>
+template<
+  typename CallbackT,
+  typename AllocatorT = std::allocator<void>>
 std::shared_ptr<GenericSubscription> create_generic_subscription(
  rclcpp::node_interfaces::NodeTopicsInterface::SharedPtr topics_interface,
  const std::string & topic_name,
  const std::string & topic_type,
  const rclcpp::QoS & qos,
-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+  CallbackT && callback,
  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options = (
    rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>()
  )
@@ -60,13 +62,20 @@ std::shared_ptr<GenericSubscription> create_generic_subscription(
  auto ts_lib = rclcpp::get_typesupport_library(
    topic_type, "rosidl_typesupport_cpp");

+  auto allocator = options.get_allocator();
+
+  using rclcpp::AnySubscriptionCallback;
+  AnySubscriptionCallback<rclcpp::SerializedMessage, AllocatorT>
+  any_subscription_callback(*allocator);
+  any_subscription_callback.set(std::forward<CallbackT>(callback));
+
  auto subscription = std::make_shared<GenericSubscription>(
    topics_interface->get_node_base_interface(),
    std::move(ts_lib),
    topic_name,
    topic_type,
    qos,
-    callback,
+    any_subscription_callback,
    options);

  topics_interface->add_subscription(subscription, options.callback_group);
--- a/rclcpp/include/rclcpp/create_subscription.hpp
+++ b/rclcpp/include/rclcpp/create_subscription.hpp
@@ -50,8 +50,8 @@ template<
  typename SubscriptionT,
  typename MessageMemoryStrategyT,
  typename NodeParametersT,
-  typename NodeTopicsT,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType>
+  typename NodeTopicsT
+>
 typename std::shared_ptr<SubscriptionT>
 create_subscription(
  NodeParametersT & node_parameters,
@@ -70,7 +70,7 @@ create_subscription(
  using rclcpp::node_interfaces::get_node_topics_interface;
  auto node_topics_interface = get_node_topics_interface(node_topics);

-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
  subscription_topic_stats = nullptr;

  if (rclcpp::detail::resolve_enable_topic_statistics(
@@ -80,8 +80,7 @@ create_subscription(
    if (options.topic_stats_options.publish_period <= std::chrono::milliseconds(0)) {
      throw std::invalid_argument(
              "topic_stats_options.publish_period must be greater than 0, specified value of " +
-              std::to_string(options.topic_stats_options.publish_period.count()) +
-              " ms");
+              std::to_string(options.topic_stats_options.publish_period.count()) + " ms");
    }

    std::shared_ptr<Publisher<statistics_msgs::msg::MetricsMessage>>
@@ -89,14 +88,14 @@ create_subscription(
      node_parameters,
      node_topics_interface,
      options.topic_stats_options.publish_topic,
-      qos);
+      options.topic_stats_options.qos);

-    subscription_topic_stats = std::make_shared<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
-      >(node_topics_interface->get_node_base_interface()->get_name(), publisher);
+    subscription_topic_stats =
+      std::make_shared<rclcpp::topic_statistics::SubscriptionTopicStatistics>(
+      node_topics_interface->get_node_base_interface()->get_name(), publisher);

    std::weak_ptr<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
+      rclcpp::topic_statistics::SubscriptionTopicStatistics
    > weak_subscription_topic_stats(subscription_topic_stats);
    auto sub_call_back = [weak_subscription_topic_stats]() {
        auto subscription_topic_stats = weak_subscription_topic_stats.lock();
--- a/rclcpp/include/rclcpp/create_timer.hpp
+++ b/rclcpp/include/rclcpp/create_timer.hpp
@@ -90,7 +90,8 @@ create_timer(
  rclcpp::Clock::SharedPtr clock,
  rclcpp::Duration period,
  CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::CallbackGroup::SharedPtr group = nullptr,
+  bool autostart = true)
 {
  return create_timer(
    clock,
@@ -98,7 +99,8 @@ create_timer(
    std::forward<CallbackT>(callback),
    group,
    node_base.get(),
-    node_timers.get());
+    node_timers.get(),
+    autostart);
 }

 /// Create a timer with a given clock
@@ -109,7 +111,8 @@ create_timer(
  rclcpp::Clock::SharedPtr clock,
  rclcpp::Duration period,
  CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::CallbackGroup::SharedPtr group = nullptr,
+  bool autostart = true)
 {
  return create_timer(
    clock,
@@ -117,7 +120,8 @@ create_timer(
    std::forward<CallbackT>(callback),
    group,
    rclcpp::node_interfaces::get_node_base_interface(node).get(),
-    rclcpp::node_interfaces::get_node_timers_interface(node).get());
+    rclcpp::node_interfaces::get_node_timers_interface(node).get(),
+    autostart);
 }

 /// Convenience method to create a general timer with node resources.
@@ -132,6 +136,7 @@ create_timer(
 * \param group callback group
 * \param node_base node base interface
 * \param node_timers node timer interface
+ * \param autostart defines if the timer should start it's countdown on initialization or not.
 * \return shared pointer to a generic timer
 * \throws std::invalid_argument if either clock, node_base or node_timers
 * are nullptr, or period is negative or too large
@@ -144,7 +149,8 @@ create_timer(
  CallbackT callback,
  rclcpp::CallbackGroup::SharedPtr group,
  node_interfaces::NodeBaseInterface * node_base,
-  node_interfaces::NodeTimersInterface * node_timers)
+  node_interfaces::NodeTimersInterface * node_timers,
+  bool autostart = true)
 {
  if (clock == nullptr) {
    throw std::invalid_argument{"clock cannot be null"};
@@ -160,7 +166,7 @@ create_timer(

  // Add a new generic timer.
  auto timer = rclcpp::GenericTimer<CallbackT>::make_shared(
-    std::move(clock), period_ns, std::move(callback), node_base->get_context());
+    std::move(clock), period_ns, std::move(callback), node_base->get_context(), autostart);
  node_timers->add_timer(timer, group);
  return timer;
 }
@@ -187,7 +193,8 @@ create_wall_timer(
  CallbackT callback,
  rclcpp::CallbackGroup::SharedPtr group,
  node_interfaces::NodeBaseInterface * node_base,
-  node_interfaces::NodeTimersInterface * node_timers)
+  node_interfaces::NodeTimersInterface * node_timers,
+  bool autostart = true)
 {
  if (node_base == nullptr) {
    throw std::invalid_argument{"input node_base cannot be null"};
@@ -201,7 +208,7 @@ create_wall_timer(

  // Add a new wall timer.
  auto timer = rclcpp::WallTimer<CallbackT>::make_shared(
-    period_ns, std::move(callback), node_base->get_context());
+    period_ns, std::move(callback), node_base->get_context(), autostart);
  node_timers->add_timer(timer, group);
  return timer;
 }
--- a/rclcpp/include/rclcpp/detail/resolve_intra_process_buffer_type.hpp
+++ b/rclcpp/include/rclcpp/detail/resolve_intra_process_buffer_type.hpp
@@ -47,6 +47,11 @@ resolve_intra_process_buffer_type(
  return resolved_buffer_type;
 }

+RCLCPP_PUBLIC
+rclcpp::IntraProcessBufferType
+resolve_intra_process_buffer_type(
+  const rclcpp::IntraProcessBufferType buffer_type);
+
 }  // namespace detail

 }  // namespace rclcpp
--- a/rclcpp/include/rclcpp/event_handler.hpp
+++ b/rclcpp/include/rclcpp/event_handler.hpp
@@ -260,6 +260,16 @@ public:
    }
  }

+  ~EventHandler()
+  {
+    // Since the rmw event listener holds a reference to the
+    // "on ready" callback, we need to clear it on destruction of this class.
+    // This clearing is not needed for other rclcpp entities like pub/subs, since
+    // they do own the underlying rmw entities, which are destroyed
+    // on their rclcpp destructors, thus no risk of dangling pointers.
+    clear_on_ready_callback();
+  }
+
  /// Take data so that the callback cannot be scheduled again
  std::shared_ptr<void>
  take_data() override
--- a/rclcpp/include/rclcpp/exceptions/exceptions.hpp
+++ b/rclcpp/include/rclcpp/exceptions/exceptions.hpp
@@ -206,6 +206,14 @@ public:
  const std::vector<std::string> unknown_ros_args;
 };

+/// Thrown when an unknown type is passed
+class UnknownTypeError : public std::runtime_error
+{
+public:
+  explicit UnknownTypeError(const std::string & type)
+  : std::runtime_error("Unknown type: " + type) {}
+};
+
 /// Thrown when an invalid rclcpp::Event object or SharedPtr is encountered.
 class InvalidEventError : public std::runtime_error
 {
@@ -222,6 +230,14 @@ public:
  : std::runtime_error("event already registered") {}
 };

+/// Thrown when a callback group is missing from the node, when it wants to utilize the group.
+class MissingGroupNodeException : public std::runtime_error
+{
+public:
+  explicit MissingGroupNodeException(const std::string & obj_type)
+  : std::runtime_error("cannot create: " + obj_type + " , callback group not in node") {}
+};
+
 /// Thrown if passed parameters are inconsistent or invalid
 class InvalidParametersException : public std::runtime_error
 {
--- a/rclcpp/include/rclcpp/executor.hpp
+++ b/rclcpp/include/rclcpp/executor.hpp
@@ -51,6 +51,7 @@ typedef std::map<rclcpp::CallbackGroup::WeakPtr,

 // Forward declaration is used in convenience method signature.
 class Node;
+class ExecutorImplementation;

 /// Coordinate the order and timing of available communication tasks.
 /**
@@ -296,6 +297,21 @@ public:
  virtual void
  spin_some(std::chrono::nanoseconds max_duration = std::chrono::nanoseconds(0));

+  /// Add a node, complete all immediately available work exhaustively, and remove the node.
+  /**
+   * \param[in] node Shared pointer to the node to add.
+   */
+  RCLCPP_PUBLIC
+  void
+  spin_node_all(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node,
+    std::chrono::nanoseconds max_duration);
+
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  RCLCPP_PUBLIC
+  void
+  spin_node_all(std::shared_ptr<rclcpp::Node> node, std::chrono::nanoseconds max_duration);
+
  /// Collect and execute work repeatedly within a duration or until no more work is available.
  /**
   * This function can be overridden. The default implementation is suitable for a
@@ -697,6 +713,9 @@ protected:

  /// shutdown callback handle registered to Context
  rclcpp::OnShutdownCallbackHandle shutdown_callback_handle_;
+
+  /// Pointer to implementation
+  std::unique_ptr<ExecutorImplementation> impl_;
 };

 }  // namespace rclcpp
--- a/rclcpp/include/rclcpp/executors.hpp
+++ b/rclcpp/include/rclcpp/executors.hpp
@@ -29,6 +29,18 @@
 namespace rclcpp
 {

+/// Create a default single-threaded executor and execute all available work exhaustively.
+/** \param[in] node_ptr Shared pointer to the node to spin. */
+RCLCPP_PUBLIC
+void
+spin_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  std::chrono::nanoseconds max_duration);
+
+RCLCPP_PUBLIC
+void
+spin_all(rclcpp::Node::SharedPtr node_ptr, std::chrono::nanoseconds max_duration);
+
 /// Create a default single-threaded executor and execute any immediately available work.
 /** \param[in] node_ptr Shared pointer to the node to spin. */
 RCLCPP_PUBLIC
--- a/rclcpp/include/rclcpp/executors/executor_entities_collection.hpp
+++ b/rclcpp/include/rclcpp/executors/executor_entities_collection.hpp
@@ -198,14 +198,15 @@ build_entities_collection(
 *
 * \param[in] collection Collection of entities corresponding to the current wait set.
 * \param[in] wait_result Result of rclcpp::WaitSet::wait corresponding to the collection.
- * \return A queue of executables that have been marked ready by the waitset.
+ * \param[inout] queue of executables to append new ready executables to
+ * \return number of new ready executables
 */
-std::deque<rclcpp::AnyExecutable>
+size_t
 ready_executables(
  const ExecutorEntitiesCollection & collection,
-  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result
+  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
+  std::deque<rclcpp::AnyExecutable> & executables
 );
-
 }  // namespace executors
 }  // namespace rclcpp

--- a/rclcpp/include/rclcpp/experimental/buffers/buffer_implementation_base.hpp
+++ b/rclcpp/include/rclcpp/experimental/buffers/buffer_implementation_base.hpp
@@ -15,6 +15,8 @@
 #ifndef RCLCPP__EXPERIMENTAL__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
 #define RCLCPP__EXPERIMENTAL__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_

+#include <vector>
+
 namespace rclcpp
 {
 namespace experimental
@@ -31,8 +33,11 @@ public:
  virtual BufferT dequeue() = 0;
  virtual void enqueue(BufferT request) = 0;

+  virtual std::vector<BufferT> get_all_data() = 0;
+
  virtual void clear() = 0;
  virtual bool has_data() const = 0;
+  virtual size_t available_capacity() const = 0;
 };

 }  // namespace buffers
--- a/rclcpp/include/rclcpp/experimental/buffers/intra_process_buffer.hpp
+++ b/rclcpp/include/rclcpp/experimental/buffers/intra_process_buffer.hpp
@@ -19,6 +19,7 @@
 #include <stdexcept>
 #include <type_traits>
 #include <utility>
+#include <vector>

 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/allocator/allocator_deleter.hpp"
@@ -44,6 +45,7 @@ public:

  virtual bool has_data() const = 0;
  virtual bool use_take_shared_method() const = 0;
+  virtual size_t available_capacity() const = 0;
 };

 template<
@@ -65,6 +67,9 @@ public:

  virtual MessageSharedPtr consume_shared() = 0;
  virtual MessageUniquePtr consume_unique() = 0;
+
+  virtual std::vector<MessageSharedPtr> get_all_data_shared() = 0;
+  virtual std::vector<MessageUniquePtr> get_all_data_unique() = 0;
 };

 template<
@@ -95,7 +100,7 @@ public:

    buffer_ = std::move(buffer_impl);

-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_buffer_to_ipb,
      static_cast<const void *>(buffer_.get()),
      static_cast<const void *>(this));
@@ -128,6 +133,16 @@ public:
    return consume_unique_impl<BufferT>();
  }

+  std::vector<MessageSharedPtr> get_all_data_shared() override
+  {
+    return get_all_data_shared_impl();
+  }
+
+  std::vector<MessageUniquePtr> get_all_data_unique() override
+  {
+    return get_all_data_unique_impl();
+  }
+
  bool has_data() const override
  {
    return buffer_->has_data();
@@ -143,6 +158,11 @@ public:
    return std::is_same<BufferT, MessageSharedPtr>::value;
  }

+  size_t available_capacity() const override
+  {
+    return buffer_->available_capacity();
+  }
+
 private:
  std::unique_ptr<BufferImplementationBase<BufferT>> buffer_;

@@ -237,6 +257,71 @@ private:
  {
    return buffer_->dequeue();
  }
+
+  // MessageSharedPtr to MessageSharedPtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageSharedPtr>::value,
+    std::vector<MessageSharedPtr>
+  >::type
+  get_all_data_shared_impl()
+  {
+    return buffer_->get_all_data();
+  }
+
+  // MessageUniquePtr to MessageSharedPtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageUniquePtr>::value,
+    std::vector<MessageSharedPtr>
+  >::type
+  get_all_data_shared_impl()
+  {
+    std::vector<MessageSharedPtr> result;
+    auto uni_ptr_vec = buffer_->get_all_data();
+    result.reserve(uni_ptr_vec.size());
+    for (MessageUniquePtr & uni_ptr : uni_ptr_vec) {
+      result.emplace_back(std::move(uni_ptr));
+    }
+    return result;
+  }
+
+  // MessageSharedPtr to MessageUniquePtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageSharedPtr>::value,
+    std::vector<MessageUniquePtr>
+  >::type
+  get_all_data_unique_impl()
+  {
+    std::vector<MessageUniquePtr> result;
+    auto shared_ptr_vec = buffer_->get_all_data();
+    result.reserve(shared_ptr_vec.size());
+    for (MessageSharedPtr shared_msg : shared_ptr_vec) {
+      MessageUniquePtr unique_msg;
+      MessageDeleter * deleter = std::get_deleter<MessageDeleter, const MessageT>(shared_msg);
+      auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
+      MessageAllocTraits::construct(*message_allocator_.get(), ptr, *shared_msg);
+      if (deleter) {
+        unique_msg = MessageUniquePtr(ptr, *deleter);
+      } else {
+        unique_msg = MessageUniquePtr(ptr);
+      }
+      result.push_back(std::move(unique_msg));
+    }
+    return result;
+  }
+
+  // MessageUniquePtr to MessageUniquePtr
+  template<typename T = BufferT>
+  typename std::enable_if<
+    std::is_same<T, MessageUniquePtr>::value,
+    std::vector<MessageUniquePtr>
+  >::type
+  get_all_data_unique_impl()
+  {
+    return buffer_->get_all_data();
+  }
 };

 }  // namespace buffers
--- a/rclcpp/include/rclcpp/experimental/buffers/ring_buffer_implementation.hpp
+++ b/rclcpp/include/rclcpp/experimental/buffers/ring_buffer_implementation.hpp
@@ -15,6 +15,7 @@
 #ifndef RCLCPP__EXPERIMENTAL__BUFFERS__RING_BUFFER_IMPLEMENTATION_HPP_
 #define RCLCPP__EXPERIMENTAL__BUFFERS__RING_BUFFER_IMPLEMENTATION_HPP_

+#include <memory>
 #include <mutex>
 #include <stdexcept>
 #include <utility>
@@ -52,7 +53,10 @@ public:
    if (capacity == 0) {
      throw std::invalid_argument("capacity must be a positive, non-zero value");
    }
-    TRACEPOINT(rclcpp_construct_ring_buffer, static_cast<const void *>(this), capacity_);
+    TRACETOOLS_TRACEPOINT(
+      rclcpp_construct_ring_buffer,
+      static_cast<const void *>(this),
+      capacity_);
  }

  virtual ~RingBufferImplementation() {}
@@ -69,7 +73,7 @@ public:

    write_index_ = next_(write_index_);
    ring_buffer_[write_index_] = std::move(request);
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_ring_buffer_enqueue,
      static_cast<const void *>(this),
      write_index_,
@@ -98,7 +102,7 @@ public:
    }

    auto request = std::move(ring_buffer_[read_index_]);
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_ring_buffer_dequeue,
      static_cast<const void *>(this),
      read_index_,
@@ -110,6 +114,17 @@ public:
    return request;
  }

+  /// Get all the elements from the ring buffer
+  /**
+   * This member function is thread-safe.
+   *
+   * \return a vector containing all the elements from the ring buffer
+   */
+  std::vector<BufferT> get_all_data() override
+  {
+    return get_all_data_impl();
+  }
+
  /// Get the next index value for the ring buffer
  /**
   * This member function is thread-safe.
@@ -148,9 +163,21 @@ public:
    return is_full_();
  }

+  /// Get the remaining capacity to store messages
+  /**
+   * This member function is thread-safe.
+   *
+   * \return the number of free capacity for new messages
+   */
+  size_t available_capacity() const
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return available_capacity_();
+  }
+
  void clear()
  {
-    TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
+    TRACETOOLS_TRACEPOINT(rclcpp_ring_buffer_clear, static_cast<const void *>(this));
  }

 private:
@@ -189,6 +216,82 @@ private:
    return size_ == capacity_;
  }

+  /// Get the remaining capacity to store messages
+  /**
+   * This member function is not thread-safe.
+   *
+   * \return the number of free capacity for new messages
+   */
+  inline size_t available_capacity_() const
+  {
+    return capacity_ - size_;
+  }
+
+  /// Traits for checking if a type is std::unique_ptr
+  template<typename ...>
+  struct is_std_unique_ptr final : std::false_type {};
+  template<class T, typename ... Args>
+  struct is_std_unique_ptr<std::unique_ptr<T, Args...>> final : std::true_type
+  {
+    typedef T Ptr_type;
+  };
+
+  /// Get all the elements from the ring buffer
+  /**
+   * This member function is thread-safe.
+   * Two versions for the implementation of the function.
+   * One for buffer containing unique_ptr and the other for other types
+   *
+   * \return a vector containing all the elements from the ring buffer
+   */
+  template<typename T = BufferT, std::enable_if_t<is_std_unique_ptr<T>::value &&
+    std::is_copy_constructible<
+      typename is_std_unique_ptr<T>::Ptr_type
+    >::value,
+    void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    std::vector<BufferT> result_vtr;
+    result_vtr.reserve(size_);
+    for (size_t id = 0; id < size_; ++id) {
+      result_vtr.emplace_back(
+        new typename is_std_unique_ptr<T>::Ptr_type(
+          *(ring_buffer_[(read_index_ + id) % capacity_])));
+    }
+    return result_vtr;
+  }
+
+  template<typename T = BufferT, std::enable_if_t<
+      std::is_copy_constructible<T>::value, void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    std::vector<BufferT> result_vtr;
+    result_vtr.reserve(size_);
+    for (size_t id = 0; id < size_; ++id) {
+      result_vtr.emplace_back(ring_buffer_[(read_index_ + id) % capacity_]);
+    }
+    return result_vtr;
+  }
+
+  template<typename T = BufferT, std::enable_if_t<!is_std_unique_ptr<T>::value &&
+    !std::is_copy_constructible<T>::value, void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    throw std::logic_error("Underlined type results in invalid get_all_data_impl()");
+    return {};
+  }
+
+  template<typename T = BufferT, std::enable_if_t<is_std_unique_ptr<T>::value &&
+    !std::is_copy_constructible<typename is_std_unique_ptr<T>::Ptr_type>::value,
+    void> * = nullptr>
+  std::vector<BufferT> get_all_data_impl()
+  {
+    throw std::logic_error("Underlined type in unique_ptr results in invalid get_all_data_impl()");
+    return {};
+  }
+
  size_t capacity_;

  std::vector<BufferT> ring_buffer_;
--- a/rclcpp/include/rclcpp/experimental/executors/events_executor/events_executor.hpp
+++ b/rclcpp/include/rclcpp/experimental/executors/events_executor/events_executor.hpp
@@ -243,6 +243,11 @@ private:
  std::function<void(size_t, int)>
  create_waitable_callback(const rclcpp::Waitable * waitable_id);

+  /// Utility to add the notify waitable to an entities collection
+  void
+  add_notify_waitable_to_collection(
+    rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection);
+
  /// Searches for the provided entity_id in the collection and returns the entity if valid
  template<typename CollectionType>
  typename CollectionType::EntitySharedPtr
@@ -269,9 +274,12 @@ private:
  rclcpp::experimental::executors::EventsQueue::UniquePtr events_queue_;

  std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollector> entities_collector_;
-  std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollection> current_entities_collection_;
  std::shared_ptr<rclcpp::executors::ExecutorNotifyWaitable> notify_waitable_;

+  /// Mutex to protect the current_entities_collection_
+  std::recursive_mutex collection_mutex_;
+  std::shared_ptr<rclcpp::executors::ExecutorEntitiesCollection> current_entities_collection_;
+
  /// Flag used to reduce the number of unnecessary waitable events
  std::atomic<bool> notify_waitable_event_pushed_ {false};

--- a/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp
+++ b/rclcpp/include/rclcpp/experimental/intra_process_manager.hpp
@@ -28,6 +28,7 @@
 #include <typeinfo>

 #include "rclcpp/allocator/allocator_deleter.hpp"
+#include "rclcpp/experimental/buffers/intra_process_buffer.hpp"
 #include "rclcpp/experimental/ros_message_intra_process_buffer.hpp"
 #include "rclcpp/experimental/subscription_intra_process.hpp"
 #include "rclcpp/experimental/subscription_intra_process_base.hpp"
@@ -112,9 +113,40 @@ public:
   * \param subscription the SubscriptionIntraProcess to register.
   * \return an unsigned 64-bit integer which is the subscription's unique id.
   */
-  RCLCPP_PUBLIC
+  template<
+    typename ROSMessageType,
+    typename Alloc = std::allocator<ROSMessageType>
+  >
  uint64_t
-  add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription);
+  add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription)
+  {
+    std::unique_lock<std::shared_timed_mutex> lock(mutex_);
+
+    uint64_t sub_id = IntraProcessManager::get_next_unique_id();
+
+    subscriptions_[sub_id] = subscription;
+
+    // adds the subscription id to all the matchable publishers
+    for (auto & pair : publishers_) {
+      auto publisher = pair.second.lock();
+      if (!publisher) {
+        continue;
+      }
+      if (can_communicate(publisher, subscription)) {
+        uint64_t pub_id = pair.first;
+        insert_sub_id_for_pub(sub_id, pub_id, subscription->use_take_shared_method());
+        if (publisher->is_durability_transient_local() &&
+          subscription->is_durability_transient_local())
+        {
+          do_transient_local_publish<ROSMessageType, Alloc>(
+            pub_id, sub_id,
+            subscription->use_take_shared_method());
+        }
+      }
+    }
+
+    return sub_id;
+  }

  /// Unregister a subscription using the subscription's unique id.
  /**
@@ -131,14 +163,21 @@ public:
   * This method stores the publisher intra process object, together with
   * the information of its wrapped publisher (i.e. topic name and QoS).
   *
+   * If the publisher's durability is transient local, its buffer pointer should
+   * be passed and the method will store it as well.
+   *
   * In addition this generates a unique intra process id for the publisher.
   *
   * \param publisher publisher to be registered with the manager.
+   * \param buffer publisher's buffer to be stored if its duability is transient local.
   * \return an unsigned 64-bit integer which is the publisher's unique id.
   */
  RCLCPP_PUBLIC
  uint64_t
-  add_publisher(rclcpp::PublisherBase::SharedPtr publisher);
+  add_publisher(
+    rclcpp::PublisherBase::SharedPtr publisher,
+    rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr buffer =
+    rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr());

  /// Unregister a publisher using the publisher's unique id.
  /**
@@ -292,6 +331,34 @@ public:
    }
  }

+  template<
+    typename MessageT,
+    typename Alloc,
+    typename Deleter,
+    typename ROSMessageType>
+  void
+  add_shared_msg_to_buffer(
+    std::shared_ptr<const MessageT> message,
+    uint64_t subscription_id)
+  {
+    add_shared_msg_to_buffers<MessageT, Alloc, Deleter, ROSMessageType>(message, {subscription_id});
+  }
+
+  template<
+    typename MessageT,
+    typename Alloc,
+    typename Deleter,
+    typename ROSMessageType>
+  void
+  add_owned_msg_to_buffer(
+    std::unique_ptr<MessageT, Deleter> message,
+    uint64_t subscription_id,
+    typename allocator::AllocRebind<MessageT, Alloc>::allocator_type & allocator)
+  {
+    add_owned_msg_to_buffers<MessageT, Alloc, Deleter, ROSMessageType>(
+      std::move(message), {subscription_id}, allocator);
+  }
+
  /// Return true if the given rmw_gid_t matches any stored Publishers.
  RCLCPP_PUBLIC
  bool
@@ -306,6 +373,11 @@ public:
  rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr
  get_subscription_intra_process(uint64_t intra_process_subscription_id);

+  /// Return the lowest available capacity for all subscription buffers for a publisher id.
+  RCLCPP_PUBLIC
+  size_t
+  lowest_available_capacity(const uint64_t intra_process_publisher_id) const;
+
 private:
  struct SplittedSubscriptions
  {
@@ -319,6 +391,9 @@ private:
  using PublisherMap =
    std::unordered_map<uint64_t, rclcpp::PublisherBase::WeakPtr>;

+  using PublisherBufferMap =
+    std::unordered_map<uint64_t, rclcpp::experimental::buffers::IntraProcessBufferBase::WeakPtr>;
+
  using PublisherToSubscriptionIdsMap =
    std::unordered_map<uint64_t, SplittedSubscriptions>;

@@ -337,6 +412,54 @@ private:
    rclcpp::PublisherBase::SharedPtr pub,
    rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr sub) const;

+  template<
+    typename ROSMessageType,
+    typename Alloc = std::allocator<ROSMessageType>
+  >
+  void do_transient_local_publish(
+    const uint64_t pub_id, const uint64_t sub_id,
+    const bool use_take_shared_method)
+  {
+    using ROSMessageTypeAllocatorTraits = allocator::AllocRebind<ROSMessageType, Alloc>;
+    using ROSMessageTypeAllocator = typename ROSMessageTypeAllocatorTraits::allocator_type;
+    using ROSMessageTypeDeleter = allocator::Deleter<ROSMessageTypeAllocator, ROSMessageType>;
+
+    auto publisher_buffer = publisher_buffers_[pub_id].lock();
+    if (!publisher_buffer) {
+      throw std::runtime_error("publisher buffer has unexpectedly gone out of scope");
+    }
+    auto buffer = std::dynamic_pointer_cast<
+      rclcpp::experimental::buffers::IntraProcessBuffer<
+        ROSMessageType,
+        ROSMessageTypeAllocator,
+        ROSMessageTypeDeleter
+      >
+      >(publisher_buffer);
+    if (!buffer) {
+      throw std::runtime_error(
+              "failed to dynamic cast publisher's IntraProcessBufferBase to "
+              "IntraProcessBuffer<ROSMessageType,ROSMessageTypeAllocator,"
+              "ROSMessageTypeDeleter> which can happen when the publisher and "
+              "subscription use different allocator types, which is not supported");
+    }
+    if (use_take_shared_method) {
+      auto data_vec = buffer->get_all_data_shared();
+      for (auto shared_data : data_vec) {
+        this->template add_shared_msg_to_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter, ROSMessageType>(
+          shared_data, sub_id);
+      }
+    } else {
+      auto data_vec = buffer->get_all_data_unique();
+      for (auto & owned_data : data_vec) {
+        auto allocator = ROSMessageTypeAllocator();
+        this->template add_owned_msg_to_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter, ROSMessageType>(
+          std::move(owned_data), sub_id, allocator);
+      }
+    }
+  }
+
  template<
    typename MessageT,
    typename Alloc,
@@ -462,7 +585,7 @@ private:
          auto ptr = MessageAllocTraits::allocate(allocator, 1);
          MessageAllocTraits::construct(allocator, ptr, *message);

-          subscription->provide_intra_process_data(std::move(MessageUniquePtr(ptr, deleter)));
+          subscription->provide_intra_process_data(MessageUniquePtr(ptr, deleter));
        }

        continue;
@@ -481,13 +604,13 @@ private:
                "subscription use different allocator types, which is not supported");
      }

-      if constexpr (rclcpp::TypeAdapter<MessageT>::is_specialized::value) {
+      if constexpr (rclcpp::TypeAdapter<MessageT, ROSMessageType>::is_specialized::value) {
        ROSMessageTypeAllocator ros_message_alloc(allocator);
-        auto ptr = ros_message_alloc.allocate(1);
-        ros_message_alloc.construct(ptr);
+        auto ptr = ROSMessageTypeAllocatorTraits::allocate(ros_message_alloc, 1);
+        ROSMessageTypeAllocatorTraits::construct(ros_message_alloc, ptr);
        ROSMessageTypeDeleter deleter;
        allocator::set_allocator_for_deleter(&deleter, &allocator);
-        rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*message, *ptr);
+        rclcpp::TypeAdapter<MessageT, ROSMessageType>::convert_to_ros_message(*message, *ptr);
        auto ros_msg = std::unique_ptr<ROSMessageType, ROSMessageTypeDeleter>(ptr, deleter);
        ros_message_subscription->provide_intra_process_message(std::move(ros_msg));
      } else {
@@ -505,7 +628,7 @@ private:
            MessageAllocTraits::construct(allocator, ptr, *message);

            ros_message_subscription->provide_intra_process_message(
-              std::move(MessageUniquePtr(ptr, deleter)));
+              MessageUniquePtr(ptr, deleter));
          }
        }
      }
@@ -515,6 +638,7 @@ private:
  PublisherToSubscriptionIdsMap pub_to_subs_;
  SubscriptionMap subscriptions_;
  PublisherMap publishers_;
+  PublisherBufferMap publisher_buffers_;

  mutable std::shared_timed_mutex mutex_;
 };
--- a/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp
+++ b/rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp
@@ -87,7 +87,7 @@ public:
      buffer_type),
    any_callback_(callback)
  {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_subscription_callback_added,
      static_cast<const void *>(this),
      static_cast<const void *>(&any_callback_));
--- a/rclcpp/include/rclcpp/experimental/subscription_intra_process_base.hpp
+++ b/rclcpp/include/rclcpp/experimental/subscription_intra_process_base.hpp
@@ -62,6 +62,15 @@ public:
  void
  add_to_wait_set(rcl_wait_set_t * wait_set) override;

+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  available_capacity() const = 0;
+
+  RCLCPP_PUBLIC
+  bool
+  is_durability_transient_local() const;
+
  bool
  is_ready(rcl_wait_set_t * wait_set) override = 0;

--- a/rclcpp/include/rclcpp/experimental/subscription_intra_process_buffer.hpp
+++ b/rclcpp/include/rclcpp/experimental/subscription_intra_process_buffer.hpp
@@ -93,7 +93,7 @@ public:
      buffer_type,
      qos_profile,
      std::make_shared<Alloc>(subscribed_type_allocator_));
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_ipb_to_subscription,
      static_cast<const void *>(buffer_.get()),
      static_cast<const void *>(this));
@@ -169,6 +169,11 @@ public:
    return buffer_->use_take_shared_method();
  }

+  size_t available_capacity() const override
+  {
+    return buffer_->available_capacity();
+  }
+
 protected:
  void
  trigger_guard_condition() override
--- a/rclcpp/include/rclcpp/experimental/timers_manager.hpp
+++ b/rclcpp/include/rclcpp/experimental/timers_manager.hpp
@@ -527,7 +527,7 @@ private:
  void execute_ready_timers_unsafe();

  // Callback to be called when timer is ready
-  std::function<void(const rclcpp::TimerBase *)> on_ready_callback_ = nullptr;
+  std::function<void(const rclcpp::TimerBase *)> on_ready_callback_;

  // Thread used to run the timers execution task
  std::thread timers_thread_;
--- a/rclcpp/include/rclcpp/generic_publisher.hpp
+++ b/rclcpp/include/rclcpp/generic_publisher.hpp
@@ -77,7 +77,7 @@ public:
  : rclcpp::PublisherBase(
      node_base,
      topic_name,
-      *rclcpp::get_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
+      *rclcpp::get_message_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
      options.template to_rcl_publisher_options<rclcpp::SerializedMessage>(qos),
      // NOTE(methylDragon): Passing these args separately is necessary for event binding
      options.event_callbacks,
--- a/rclcpp/include/rclcpp/generic_subscription.hpp
+++ b/rclcpp/include/rclcpp/generic_subscription.hpp
@@ -74,18 +74,21 @@ public:
    const std::string & topic_name,
    const std::string & topic_type,
    const rclcpp::QoS & qos,
-    // TODO(nnmm): Add variant for callback with message info. See issue #1604.
-    std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+    AnySubscriptionCallback<rclcpp::SerializedMessage, AllocatorT> callback,
    const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options)
  : SubscriptionBase(
      node_base,
-      *rclcpp::get_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
+      *rclcpp::get_message_typesupport_handle(topic_type, "rosidl_typesupport_cpp", *ts_lib),
      topic_name,
      options.to_rcl_subscription_options(qos),
      options.event_callbacks,
      options.use_default_callbacks,
      DeliveredMessageKind::SERIALIZED_MESSAGE),
-    callback_(callback),
+    callback_([callback](
+        std::shared_ptr<const rclcpp::SerializedMessage> serialized_message,
+        const rclcpp::MessageInfo & message_info) mutable {
+        callback.dispatch(serialized_message, message_info);
+      }),
    ts_lib_(ts_lib)
  {}

@@ -151,7 +154,9 @@ public:
 private:
  RCLCPP_DISABLE_COPY(GenericSubscription)

-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback_;
+  std::function<void(
+      std::shared_ptr<const rclcpp::SerializedMessage>,
+      const rclcpp::MessageInfo)> callback_;
  // The type support library should stay loaded, so it is stored in the GenericSubscription
  std::shared_ptr<rcpputils::SharedLibrary> ts_lib_;
 };
--- a/rclcpp/include/rclcpp/guard_condition.hpp
+++ b/rclcpp/include/rclcpp/guard_condition.hpp
@@ -48,7 +48,7 @@ public:
   */
  RCLCPP_PUBLIC
  explicit GuardCondition(
-    rclcpp::Context::SharedPtr context =
+    const rclcpp::Context::SharedPtr & context =
    rclcpp::contexts::get_global_default_context(),
    rcl_guard_condition_options_t guard_condition_options =
    rcl_guard_condition_get_default_options());
@@ -57,11 +57,6 @@ public:
  virtual
  ~GuardCondition();

-  /// Return the context used when creating this guard condition.
-  RCLCPP_PUBLIC
-  rclcpp::Context::SharedPtr
-  get_context() const;
-
  /// Return the underlying rcl guard condition structure.
  RCLCPP_PUBLIC
  rcl_guard_condition_t &
@@ -128,7 +123,6 @@ public:
  set_on_trigger_callback(std::function<void(size_t)> callback);

 protected:
-  rclcpp::Context::SharedPtr context_;
  rcl_guard_condition_t rcl_guard_condition_;
  std::atomic<bool> in_use_by_wait_set_{false};
  std::recursive_mutex reentrant_mutex_;
--- a/rclcpp/include/rclcpp/logger.hpp
+++ b/rclcpp/include/rclcpp/logger.hpp
@@ -126,9 +126,6 @@ private:
  std::shared_ptr<std::pair<std::string, std::string>> logger_sublogger_pairname_ = nullptr;

 public:
-  RCLCPP_PUBLIC
-  Logger(const Logger &) = default;
-
  /// Get the name of this logger.
  /**
   * \return the full name of the logger including any prefixes, or
--- a/rclcpp/include/rclcpp/node.hpp
+++ b/rclcpp/include/rclcpp/node.hpp
@@ -56,6 +56,7 @@
 #include "rclcpp/node_interfaces/node_time_source_interface.hpp"
 #include "rclcpp/node_interfaces/node_timers_interface.hpp"
 #include "rclcpp/node_interfaces/node_topics_interface.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
 #include "rclcpp/node_interfaces/node_waitables_interface.hpp"
 #include "rclcpp/node_options.hpp"
 #include "rclcpp/parameter.hpp"
@@ -232,13 +233,15 @@ public:
   * \param[in] period Time interval between triggers of the callback.
   * \param[in] callback User-defined callback function.
   * \param[in] group Callback group to execute this timer's callback in.
+   * \param[in] autostart The state of the clock on initialization.
   */
  template<typename DurationRepT = int64_t, typename DurationT = std::milli, typename CallbackT>
  typename rclcpp::WallTimer<CallbackT>::SharedPtr
  create_wall_timer(
    std::chrono::duration<DurationRepT, DurationT> period,
    CallbackT callback,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::CallbackGroup::SharedPtr group = nullptr,
+    bool autostart = true);

  /// Create a timer that uses the node clock to drive the callback.
  /**
@@ -355,12 +358,14 @@ public:
 * `%callback_group`.
   * \return Shared pointer to the created generic subscription.
   */
-  template<typename AllocatorT = std::allocator<void>>
+  template<
+    typename CallbackT,
+    typename AllocatorT = std::allocator<void>>
  std::shared_ptr<rclcpp::GenericSubscription> create_generic_subscription(
    const std::string & topic_name,
    const std::string & topic_type,
    const rclcpp::QoS & qos,
-    std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+    CallbackT && callback,
    const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options = (
      rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>()
    )
@@ -969,7 +974,16 @@ public:

  /// Return a list of parameters with any of the given prefixes, up to the given depth.
  /**
-   * \todo: properly document and test this method.
+   * Parameters are separated into a hierarchy using the "." (dot) character.
+   * The "prefixes" argument is a way to select only particular parts of the hierarchy.
+   *
+   * \param[in] prefixes The list of prefixes that should be searched for within the
+   * current parameters. If this vector of prefixes is empty, then list_parameters
+   * will return all parameters.
+   * \param[in] depth An unsigned integer that represents the recursive depth to search.
+   * If this depth = 0, then all parameters that fit the prefixes will be returned.
+   * \returns A ListParametersResult message which contains both an array of unique prefixes
+   * and an array of names that were matched to the prefixes given.
   */
  RCLCPP_PUBLIC
  rcl_interfaces::msg::ListParametersResult
@@ -1302,6 +1316,26 @@ public:
  size_t
  count_subscribers(const std::string & topic_name) const;

+  /// Return the number of clients created for a given service.
+  /**
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   * \return number of clients that have been created for the given service.
+   * \throws std::runtime_error if clients could not be counted
+   */
+  RCLCPP_PUBLIC
+  size_t
+  count_clients(const std::string & service_name) const;
+
+  /// Return the number of services created for a given service.
+  /**
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   * \return number of services that have been created for the given service.
+   * \throws std::runtime_error if services could not be counted
+   */
+  RCLCPP_PUBLIC
+  size_t
+  count_services(const std::string & service_name) const;
+
  /// Return the topic endpoint information about publishers on a given topic.
  /**
   * The returned parameter is a list of topic endpoint information, where each item will contain
@@ -1454,6 +1488,11 @@ public:
  rclcpp::node_interfaces::NodeTimeSourceInterface::SharedPtr
  get_node_time_source_interface();

+  /// Return the Node's internal NodeTypeDescriptionsInterface implementation.
+  RCLCPP_PUBLIC
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
+  get_node_type_descriptions_interface();
+
  /// Return the sub-namespace, if this is a sub-node, otherwise an empty string.
  /**
   * The returned sub-namespace is either the accumulated sub-namespaces which
@@ -1586,11 +1625,18 @@ private:
  rclcpp::node_interfaces::NodeClockInterface::SharedPtr node_clock_;
  rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters_;
  rclcpp::node_interfaces::NodeTimeSourceInterface::SharedPtr node_time_source_;
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr node_type_descriptions_;
  rclcpp::node_interfaces::NodeWaitablesInterface::SharedPtr node_waitables_;

  const rclcpp::NodeOptions node_options_;
  const std::string sub_namespace_;
  const std::string effective_namespace_;
+
+  class NodeImpl;
+  // This member is meant to be a place to backport features into stable distributions,
+  // and new features targeting Rolling should not use this.
+  // See the comment in node.cpp for more information.
+  std::shared_ptr<NodeImpl> hidden_impl_{nullptr};
 };

 }  // namespace rclcpp
--- a/rclcpp/include/rclcpp/node_impl.hpp
+++ b/rclcpp/include/rclcpp/node_impl.hpp
@@ -110,14 +110,16 @@ typename rclcpp::WallTimer<CallbackT>::SharedPtr
 Node::create_wall_timer(
  std::chrono::duration<DurationRepT, DurationT> period,
  CallbackT callback,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::CallbackGroup::SharedPtr group,
+  bool autostart)
 {
  return rclcpp::create_wall_timer(
    period,
    std::move(callback),
    group,
    this->node_base_.get(),
-    this->node_timers_.get());
+    this->node_timers_.get(),
+    autostart);
 }

 template<typename DurationRepT, typename DurationT, typename CallbackT>
@@ -219,13 +221,13 @@ Node::create_generic_publisher(
  );
 }

-template<typename AllocatorT>
+template<typename CallbackT, typename AllocatorT>
 std::shared_ptr<rclcpp::GenericSubscription>
 Node::create_generic_subscription(
  const std::string & topic_name,
  const std::string & topic_type,
  const rclcpp::QoS & qos,
-  std::function<void(std::shared_ptr<rclcpp::SerializedMessage>)> callback,
+  CallbackT && callback,
  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options)
 {
  return rclcpp::create_generic_subscription(
@@ -233,7 +235,7 @@ Node::create_generic_subscription(
    extend_name_with_sub_namespace(topic_name, this->get_sub_namespace()),
    topic_type,
    qos,
-    std::move(callback),
+    std::forward<CallbackT>(callback),
    options
  );
 }
--- a/rclcpp/include/rclcpp/node_interfaces/node_graph.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_graph.hpp
@@ -113,6 +113,14 @@ public:
  size_t
  count_subscribers(const std::string & topic_name) const override;

+  RCLCPP_PUBLIC
+  size_t
+  count_clients(const std::string & service_name) const override;
+
+  RCLCPP_PUBLIC
+  size_t
+  count_services(const std::string & service_name) const override;
+
  RCLCPP_PUBLIC
  const rcl_guard_condition_t *
  get_graph_guard_condition() const override;
--- a/rclcpp/include/rclcpp/node_interfaces/node_graph_interface.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_graph_interface.hpp
@@ -305,6 +305,24 @@ public:
  size_t
  count_subscribers(const std::string & topic_name) const = 0;

+  /// Return the number of clients created for a given service.
+  /*
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   */
+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  count_clients(const std::string & service_name) const = 0;
+
+  /// Return the number of services created for a given service.
+  /*
+   * \param[in] service_name the actual service name used; it will not be automatically remapped.
+   */
+  RCLCPP_PUBLIC
+  virtual
+  size_t
+  count_services(const std::string & service_name) const = 0;
+
  /// Return the rcl guard condition which is triggered when the ROS graph changes.
  RCLCPP_PUBLIC
  virtual
--- a/rclcpp/include/rclcpp/node_interfaces/node_interfaces.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_interfaces.hpp
@@ -30,6 +30,7 @@
  rclcpp::node_interfaces::NodeTimeSourceInterface, \
  rclcpp::node_interfaces::NodeTimersInterface, \
  rclcpp::node_interfaces::NodeTopicsInterface, \
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface, \
  rclcpp::node_interfaces::NodeWaitablesInterface


@@ -118,6 +119,7 @@ public:
   *   - rclcpp::node_interfaces::NodeTimeSourceInterface
   *   - rclcpp::node_interfaces::NodeTimersInterface
   *   - rclcpp::node_interfaces::NodeTopicsInterface
+   *   - rclcpp::node_interfaces::NodeTypeDescriptionsInterface
   *   - rclcpp::node_interfaces::NodeWaitablesInterface
   *
   * Or you use custom interfaces as long as you make a template specialization
--- a/rclcpp/include/rclcpp/node_interfaces/node_logging.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_logging.hpp
@@ -23,6 +23,9 @@
 #include "rclcpp/node_interfaces/node_logging_interface.hpp"
 #include "rclcpp/visibility_control.hpp"

+#include "rcl_interfaces/srv/get_logger_levels.hpp"
+#include "rcl_interfaces/srv/set_logger_levels.hpp"
+
 namespace rclcpp
 {
 namespace node_interfaces
@@ -35,7 +38,7 @@ public:
  RCLCPP_SMART_PTR_ALIASES_ONLY(NodeLoggingInterface)

  RCLCPP_PUBLIC
-  explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base);
+  explicit NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base);

  RCLCPP_PUBLIC
  virtual
@@ -49,13 +52,21 @@ public:
  const char *
  get_logger_name() const override;

+  RCLCPP_PUBLIC
+  void
+  create_logger_services(
+    node_interfaces::NodeServicesInterface::SharedPtr node_services) override;
+
 private:
  RCLCPP_DISABLE_COPY(NodeLogging)

  /// Handle to the NodeBaseInterface given in the constructor.
-  rclcpp::node_interfaces::NodeBaseInterface * node_base_;
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;

  rclcpp::Logger logger_;
+
+  rclcpp::Service<rcl_interfaces::srv::GetLoggerLevels>::SharedPtr get_loggers_service_;
+  rclcpp::Service<rcl_interfaces::srv::SetLoggerLevels>::SharedPtr set_loggers_service_;
 };

 }  // namespace node_interfaces
--- a/rclcpp/include/rclcpp/node_interfaces/node_logging_interface.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_logging_interface.hpp
@@ -19,6 +19,7 @@

 #include "rclcpp/logger.hpp"
 #include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"
 #include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
 #include "rclcpp/visibility_control.hpp"

@@ -54,6 +55,13 @@ public:
  virtual
  const char *
  get_logger_name() const = 0;
+
+  /// create logger services
+  RCLCPP_PUBLIC
+  virtual
+  void
+  create_logger_services(
+    node_interfaces::NodeServicesInterface::SharedPtr node_services) = 0;
 };

 }  // namespace node_interfaces
--- a/rclcpp/include/rclcpp/node_interfaces/node_type_descriptions.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_type_descriptions.hpp
@@ -0,0 +1,63 @@
+// Copyright 2023 Open Source Robotics Foundation, Inc.
+//
+// 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 RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
+#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
+
+#include <memory>
+
+#include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/node_base_interface.hpp"
+#include "rclcpp/node_interfaces/node_logging_interface.hpp"
+#include "rclcpp/node_interfaces/node_parameters_interface.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"
+#include "rclcpp/node_interfaces/node_topics_interface.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions_interface.hpp"
+#include "rclcpp/visibility_control.hpp"
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+/// Implementation of the NodeTypeDescriptions part of the Node API.
+class NodeTypeDescriptions : public NodeTypeDescriptionsInterface
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptions)
+
+  RCLCPP_PUBLIC
+  explicit NodeTypeDescriptions(
+    NodeBaseInterface::SharedPtr node_base,
+    NodeLoggingInterface::SharedPtr node_logging,
+    NodeParametersInterface::SharedPtr node_parameters,
+    NodeServicesInterface::SharedPtr node_services);
+
+  RCLCPP_PUBLIC
+  virtual
+  ~NodeTypeDescriptions();
+
+private:
+  RCLCPP_DISABLE_COPY(NodeTypeDescriptions)
+
+  // Pimpl hides helper types and functions used for wrapping a C service, which would be
+  // awkward to expose in this header.
+  class NodeTypeDescriptionsImpl;
+  std::unique_ptr<NodeTypeDescriptionsImpl> impl_;
+};
+
+}  // namespace node_interfaces
+}  // namespace rclcpp
+
+#endif  // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_HPP_
--- a/rclcpp/include/rclcpp/node_interfaces/node_type_descriptions_interface.hpp
+++ b/rclcpp/include/rclcpp/node_interfaces/node_type_descriptions_interface.hpp
@@ -0,0 +1,44 @@
+// Copyright 2023 Open Source Robotics Foundation, Inc.
+//
+// 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 RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
+#define RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
+
+#include "rclcpp/macros.hpp"
+#include "rclcpp/node_interfaces/detail/node_interfaces_helpers.hpp"
+#include "rclcpp/visibility_control.hpp"
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+/// Pure virtual interface class for the NodeTypeDescriptions part of the Node API.
+class NodeTypeDescriptionsInterface
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(NodeTypeDescriptionsInterface)
+
+  RCLCPP_PUBLIC
+  virtual
+  ~NodeTypeDescriptionsInterface() = default;
+};
+
+}  // namespace node_interfaces
+}  // namespace rclcpp
+
+RCLCPP_NODE_INTERFACE_HELPERS_SUPPORT(
+  rclcpp::node_interfaces::NodeTypeDescriptionsInterface, type_descriptions)
+
+#endif  // RCLCPP__NODE_INTERFACES__NODE_TYPE_DESCRIPTIONS_INTERFACE_HPP_
--- a/rclcpp/include/rclcpp/node_options.hpp
+++ b/rclcpp/include/rclcpp/node_options.hpp
@@ -43,6 +43,7 @@ public:
   *   - arguments = {}
   *   - parameter_overrides = {}
   *   - use_global_arguments = true
+   *   - enable_rosout = true
   *   - use_intra_process_comms = false
   *   - enable_topic_statistics = false
   *   - start_parameter_services = true
@@ -50,6 +51,7 @@ public:
   *   - clock_type = RCL_ROS_TIME
   *   - clock_qos = rclcpp::ClockQoS()
   *   - use_clock_thread = true
+   *   - enable_logger_service = false
   *   - rosout_qos = rclcpp::RosoutQoS()
   *   - parameter_event_qos = rclcpp::ParameterEventQoS
   *     - with history setting and depth from rmw_qos_profile_parameter_events
@@ -232,6 +234,24 @@ public:
  NodeOptions &
  start_parameter_services(bool start_parameter_services);

+  /// Return the enable_logger_service flag.
+  RCLCPP_PUBLIC
+  bool
+  enable_logger_service() const;
+
+  /// Set the enable_logger_service flag, return this for logger idiom.
+  /**
+   * If true, ROS services are created to allow external nodes to get
+   * and set logger levels of this node.
+   *
+   * If false, loggers will still be configured and set logger levels locally,
+   * but logger levels cannot be changed remotely .
+   *
+   */
+  RCLCPP_PUBLIC
+  NodeOptions &
+  enable_logger_service(bool enable_log_service);
+
  /// Return the start_parameter_event_publisher flag.
  RCLCPP_PUBLIC
  bool
@@ -421,6 +441,8 @@ private:

  bool use_clock_thread_ {true};

+  bool enable_logger_service_ {false};
+
  rclcpp::QoS parameter_event_qos_ = rclcpp::ParameterEventsQoS(
    rclcpp::QoSInitialization::from_rmw(rmw_qos_profile_parameter_events)
  );
--- a/rclcpp/include/rclcpp/parameter_events_filter.hpp
+++ b/rclcpp/include/rclcpp/parameter_events_filter.hpp
@@ -48,7 +48,7 @@ public:
   *
   * Example Usage:
   *
-   * If you have recieved a parameter event and are only interested in parameters foo and
+   * If you have received a parameter event and are only interested in parameters foo and
   * bar being added or changed but don't care about deletion.
   *
   * ```cpp
--- a/rclcpp/include/rclcpp/parameter_value.hpp
+++ b/rclcpp/include/rclcpp/parameter_value.hpp
@@ -24,6 +24,7 @@

 #include "rcl_interfaces/msg/parameter_type.hpp"
 #include "rcl_interfaces/msg/parameter_value.hpp"
+#include "rclcpp/exceptions/exceptions.hpp"
 #include "rclcpp/visibility_control.hpp"

 namespace rclcpp
--- a/rclcpp/include/rclcpp/publisher.hpp
+++ b/rclcpp/include/rclcpp/publisher.hpp
@@ -32,6 +32,9 @@
 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/allocator/allocator_deleter.hpp"
 #include "rclcpp/detail/resolve_use_intra_process.hpp"
+#include "rclcpp/detail/resolve_intra_process_buffer_type.hpp"
+#include "rclcpp/experimental/buffers/intra_process_buffer.hpp"
+#include "rclcpp/experimental/create_intra_process_buffer.hpp"
 #include "rclcpp/experimental/intra_process_manager.hpp"
 #include "rclcpp/get_message_type_support_handle.hpp"
 #include "rclcpp/is_ros_compatible_type.hpp"
@@ -109,6 +112,12 @@ public:
  [[deprecated("use std::shared_ptr<const PublishedType>")]] =
    std::shared_ptr<const PublishedType>;

+  using BufferSharedPtr = typename rclcpp::experimental::buffers::IntraProcessBuffer<
+    ROSMessageType,
+    ROSMessageTypeAllocator,
+    ROSMessageTypeDeleter
+    >::SharedPtr;
+
  RCLCPP_SMART_PTR_DEFINITIONS(Publisher<MessageT, AllocatorT>)

  /// Default constructor.
@@ -171,11 +180,14 @@ public:
        throw std::invalid_argument(
                "intraprocess communication is not allowed with a zero qos history depth value");
      }
-      if (qos.durability() != rclcpp::DurabilityPolicy::Volatile) {
-        throw std::invalid_argument(
-                "intraprocess communication allowed only with volatile durability");
+      if (qos.durability() == rclcpp::DurabilityPolicy::TransientLocal) {
+        buffer_ = rclcpp::experimental::create_intra_process_buffer<
+          ROSMessageType, ROSMessageTypeAllocator, ROSMessageTypeDeleter>(
+          rclcpp::detail::resolve_intra_process_buffer_type(options_.intra_process_buffer_type),
+          qos,
+          std::make_shared<ROSMessageTypeAllocator>(ros_message_type_allocator_));
      }
-      uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this());
+      uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this(), buffer_);
      this->setup_intra_process(
        intra_process_publisher_id,
        ipm);
@@ -242,9 +254,18 @@ public:
    if (inter_process_publish_needed) {
      auto shared_msg =
        this->do_intra_process_ros_message_publish_and_return_shared(std::move(msg));
+      if (buffer_) {
+        buffer_->add_shared(shared_msg);
+      }
      this->do_inter_process_publish(*shared_msg);
    } else {
-      this->do_intra_process_ros_message_publish(std::move(msg));
+      if (buffer_) {
+        auto shared_msg =
+          this->do_intra_process_ros_message_publish_and_return_shared(std::move(msg));
+        buffer_->add_shared(shared_msg);
+      } else {
+        this->do_intra_process_ros_message_publish(std::move(msg));
+      }
    }
  }

@@ -309,14 +330,22 @@ public:
      get_subscription_count() > get_intra_process_subscription_count();

    if (inter_process_publish_needed) {
-      ROSMessageType ros_msg;
+      auto ros_msg_ptr = std::make_shared<ROSMessageType>();
      // TODO(clalancette): This is unnecessarily doing an additional conversion
      // that may have already been done in do_intra_process_publish_and_return_shared().
      // We should just reuse that effort.
-      rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, ros_msg);
+      rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, *ros_msg_ptr);
      this->do_intra_process_publish(std::move(msg));
-      this->do_inter_process_publish(ros_msg);
+      this->do_inter_process_publish(*ros_msg_ptr);
+      if (buffer_) {
+        buffer_->add_shared(ros_msg_ptr);
+      }
    } else {
+      if (buffer_) {
+        auto ros_msg_ptr = std::make_shared<ROSMessageType>();
+        rclcpp::TypeAdapter<MessageT>::convert_to_ros_message(*msg, *ros_msg_ptr);
+        buffer_->add_shared(ros_msg_ptr);
+      }
      this->do_intra_process_publish(std::move(msg));
    }
  }
@@ -390,7 +419,7 @@ public:
    if (this->can_loan_messages()) {
      // we release the ownership from the rclpp::LoanedMessage instance
      // and let the middleware clean up the memory.
-      this->do_loaned_message_publish(std::move(loaned_msg.release()));
+      this->do_loaned_message_publish(loaned_msg.release());
    } else {
      // we don't release the ownership, let the middleware copy the ros message
      // and thus the destructor of rclcpp::LoanedMessage cleans up the memory.
@@ -421,7 +450,7 @@ protected:
  void
  do_inter_process_publish(const ROSMessageType & msg)
  {
-    TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(&msg));
+    TRACETOOLS_TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(&msg));
    auto status = rcl_publish(publisher_handle_.get(), &msg, nullptr);

    if (RCL_RET_PUBLISHER_INVALID == status) {
@@ -456,6 +485,7 @@ protected:
  do_loaned_message_publish(
    std::unique_ptr<ROSMessageType, std::function<void(ROSMessageType *)>> msg)
  {
+    TRACETOOLS_TRACEPOINT(rclcpp_publish, nullptr, static_cast<const void *>(msg.get()));
    auto status = rcl_publish_loaned_message(publisher_handle_.get(), msg.get(), nullptr);

    if (RCL_RET_PUBLISHER_INVALID == status) {
@@ -484,7 +514,7 @@ protected:
    if (!msg) {
      throw std::runtime_error("cannot publish msg which is a null pointer");
    }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_intra_publish,
      static_cast<const void *>(publisher_handle_.get()),
      msg.get());
@@ -506,7 +536,7 @@ protected:
    if (!msg) {
      throw std::runtime_error("cannot publish msg which is a null pointer");
    }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_intra_publish,
      static_cast<const void *>(publisher_handle_.get()),
      msg.get());
@@ -529,7 +559,7 @@ protected:
    if (!msg) {
      throw std::runtime_error("cannot publish msg which is a null pointer");
    }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_intra_publish,
      static_cast<const void *>(publisher_handle_.get()),
      msg.get());
@@ -580,6 +610,8 @@ protected:
  PublishedTypeDeleter published_type_deleter_;
  ROSMessageTypeAllocator ros_message_type_allocator_;
  ROSMessageTypeDeleter ros_message_type_deleter_;
+
+  BufferSharedPtr buffer_{nullptr};
 };

 }  // namespace rclcpp
--- a/rclcpp/include/rclcpp/publisher_base.hpp
+++ b/rclcpp/include/rclcpp/publisher_base.hpp
@@ -139,6 +139,12 @@ public:
  size_t
  get_intra_process_subscription_count() const;

+  /// Get if durability is transient local
+  /** \return If durability is transient local*/
+  RCLCPP_PUBLIC
+  bool
+  is_durability_transient_local() const;
+
  /// Manually assert that this Publisher is alive (for RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC).
  /**
   * If the rmw Liveliness policy is set to RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC, the creator
@@ -215,6 +221,17 @@ public:
  std::vector<rclcpp::NetworkFlowEndpoint>
  get_network_flow_endpoints() const;

+  /// Return the lowest available capacity for all subscription buffers.
+  /**
+   * For intraprocess communication return the lowest buffer capacity for all subscriptions.
+   * If intraprocess is disabled or no intraprocess subscriptions present, return maximum of size_t.
+   * On failure return 0.
+   * \return lowest buffer capacity for all subscriptions
+   */
+  RCLCPP_PUBLIC
+  size_t
+  lowest_available_ipm_capacity() const;
+
  /// Wait until all published messages are acknowledged or until the specified timeout elapses.
  /**
   * This method waits until all published messages are acknowledged by all matching
--- a/rclcpp/include/rclcpp/publisher_options.hpp
+++ b/rclcpp/include/rclcpp/publisher_options.hpp
@@ -24,6 +24,7 @@

 #include "rclcpp/allocator/allocator_common.hpp"
 #include "rclcpp/detail/rmw_implementation_specific_publisher_payload.hpp"
+#include "rclcpp/intra_process_buffer_type.hpp"
 #include "rclcpp/intra_process_setting.hpp"
 #include "rclcpp/qos.hpp"
 #include "rclcpp/event_handler.hpp"
@@ -40,6 +41,9 @@ struct PublisherOptionsBase
  /// Setting to explicitly set intraprocess communications.
  IntraProcessSetting use_intra_process_comm = IntraProcessSetting::NodeDefault;

+  /// Setting the data-type stored in the intraprocess buffer
+  IntraProcessBufferType intra_process_buffer_type = IntraProcessBufferType::SharedPtr;
+
  /// Callbacks for various events related to publishers.
  PublisherEventCallbacks event_callbacks;

--- a/rclcpp/include/rclcpp/rate.hpp
+++ b/rclcpp/include/rclcpp/rate.hpp
@@ -19,6 +19,8 @@
 #include <memory>
 #include <thread>

+#include "rclcpp/clock.hpp"
+#include "rclcpp/duration.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/utilities.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -31,9 +33,20 @@ class RateBase
 public:
  RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(RateBase)

+  RCLCPP_PUBLIC
  virtual ~RateBase() {}
+
+  RCLCPP_PUBLIC
  virtual bool sleep() = 0;
+
+  [[deprecated("use get_type() instead")]]
+  RCLCPP_PUBLIC
  virtual bool is_steady() const = 0;
+
+  RCLCPP_PUBLIC
+  virtual rcl_clock_type_t get_type() const = 0;
+
+  RCLCPP_PUBLIC
  virtual void reset() = 0;
 };

@@ -42,14 +55,13 @@ using std::chrono::duration_cast;
 using std::chrono::nanoseconds;

 template<class Clock = std::chrono::high_resolution_clock>
-class GenericRate : public RateBase
+class [[deprecated("use rclcpp::Rate class instead of GenericRate")]] GenericRate : public RateBase
 {
 public:
  RCLCPP_SMART_PTR_DEFINITIONS(GenericRate)

  explicit GenericRate(double rate)
-  : GenericRate<Clock>(
-      duration_cast<nanoseconds>(duration<double>(1.0 / rate)))
+  : period_(duration_cast<nanoseconds>(duration<double>(1.0 / rate))), last_interval_(Clock::now())
  {}
  explicit GenericRate(std::chrono::nanoseconds period)
  : period_(period), last_interval_(Clock::now())
@@ -87,12 +99,18 @@ public:
    return true;
  }

+  [[deprecated("use get_type() instead")]]
  virtual bool
  is_steady() const
  {
    return Clock::is_steady;
  }

+  virtual rcl_clock_type_t get_type() const
+  {
+    return Clock::is_steady ? RCL_STEADY_TIME : RCL_SYSTEM_TIME;
+  }
+
  virtual void
  reset()
  {
@@ -112,8 +130,59 @@ private:
  std::chrono::time_point<Clock, ClockDurationNano> last_interval_;
 };

-using Rate = GenericRate<std::chrono::system_clock>;
-using WallRate = GenericRate<std::chrono::steady_clock>;
+class Rate : public RateBase
+{
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(Rate)
+
+  RCLCPP_PUBLIC
+  explicit Rate(
+    const double rate,
+    Clock::SharedPtr clock = std::make_shared<Clock>(RCL_SYSTEM_TIME));
+
+  RCLCPP_PUBLIC
+  explicit Rate(
+    const Duration & period,
+    Clock::SharedPtr clock = std::make_shared<Clock>(RCL_SYSTEM_TIME));
+
+  RCLCPP_PUBLIC
+  virtual bool
+  sleep();
+
+  [[deprecated("use get_type() instead")]]
+  RCLCPP_PUBLIC
+  virtual bool
+  is_steady() const;
+
+  RCLCPP_PUBLIC
+  virtual rcl_clock_type_t
+  get_type() const;
+
+  RCLCPP_PUBLIC
+  virtual void
+  reset();
+
+  RCLCPP_PUBLIC
+  std::chrono::nanoseconds
+  period() const;
+
+private:
+  RCLCPP_DISABLE_COPY(Rate)
+
+  Clock::SharedPtr clock_;
+  Duration period_;
+  Time last_interval_;
+};
+
+class WallRate : public Rate
+{
+public:
+  RCLCPP_PUBLIC
+  explicit WallRate(const double rate);
+
+  RCLCPP_PUBLIC
+  explicit WallRate(const Duration & period);
+};

 }  // namespace rclcpp

--- a/rclcpp/include/rclcpp/rclcpp.hpp
+++ b/rclcpp/include/rclcpp/rclcpp.hpp
@@ -54,6 +54,7 @@
 *   - rclcpp::ParameterValue
 *   - rclcpp::AsyncParametersClient
 *   - rclcpp::SyncParametersClient
+ *   - rclcpp::copy_all_parameter_values()
 *   - rclcpp/parameter.hpp
 *   - rclcpp/parameter_value.hpp
 *   - rclcpp/parameter_client.hpp
@@ -95,6 +96,9 @@
 * - Get the number of publishers or subscribers on a topic:
 *   - rclcpp::Node::count_publishers()
 *   - rclcpp::Node::count_subscribers()
+ * - Get the number of clients or servers on a service:
+ *   - rclcpp::Node::count_clients()
+ *   - rclcpp::Node::count_services()
 *
 * And components related to logging:
 *
@@ -164,6 +168,7 @@
 #include <csignal>
 #include <memory>

+#include "rclcpp/copy_all_parameter_values.hpp"
 #include "rclcpp/executors.hpp"
 #include "rclcpp/guard_condition.hpp"
 #include "rclcpp/logging.hpp"
--- a/rclcpp/include/rclcpp/service.hpp
+++ b/rclcpp/include/rclcpp/service.hpp
@@ -265,15 +265,19 @@ protected:

  std::shared_ptr<rcl_node_t> node_handle_;

+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_request_callback_ before
+  // service_handle_, so on destruction the service is
+  // destroyed first. Otherwise, the rmw service callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_request_callback_{nullptr};
+  // Declare service_handle_ after callback
  std::shared_ptr<rcl_service_t> service_handle_;
  bool owns_rcl_handle_ = true;

  rclcpp::Logger node_logger_;

  std::atomic<bool> in_use_by_wait_set_{false};
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_request_callback_{nullptr};
 };

 template<typename ServiceT>
@@ -348,7 +352,7 @@ public:

      rclcpp::exceptions::throw_from_rcl_error(ret, "could not create service");
    }
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_service_callback_added,
      static_cast<const void *>(get_service_handle().get()),
      static_cast<const void *>(&any_callback_));
@@ -383,7 +387,7 @@ public:
    }

    service_handle_ = service_handle;
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_service_callback_added,
      static_cast<const void *>(get_service_handle().get()),
      static_cast<const void *>(&any_callback_));
@@ -420,7 +424,7 @@ public:
    // In this case, rcl owns the service handle memory
    service_handle_ = std::shared_ptr<rcl_service_t>(new rcl_service_t);
    service_handle_->impl = service_handle->impl;
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_service_callback_added,
      static_cast<const void *>(get_service_handle().get()),
      static_cast<const void *>(&any_callback_));
@@ -482,6 +486,14 @@ public:
  {
    rcl_ret_t ret = rcl_send_response(get_service_handle().get(), &req_id, &response);

+    if (ret == RCL_RET_TIMEOUT) {
+      RCLCPP_WARN(
+        node_logger_.get_child("rclcpp"),
+        "failed to send response to %s (timeout): %s",
+        this->get_service_name(), rcl_get_error_string().str);
+      rcl_reset_error();
+      return;
+    }
    if (ret != RCL_RET_OK) {
      rclcpp::exceptions::throw_from_rcl_error(ret, "failed to send response");
    }
--- a/rclcpp/include/rclcpp/strategies/message_pool_memory_strategy.hpp
+++ b/rclcpp/include/rclcpp/strategies/message_pool_memory_strategy.hpp
@@ -15,10 +15,17 @@
 #ifndef RCLCPP__STRATEGIES__MESSAGE_POOL_MEMORY_STRATEGY_HPP_
 #define RCLCPP__STRATEGIES__MESSAGE_POOL_MEMORY_STRATEGY_HPP_

+#include <array>
+#include <cstring>
 #include <memory>
+#include <mutex>
+#include <stdexcept>
+#include <type_traits>

 #include "rosidl_runtime_cpp/traits.hpp"

+#include "rclcpp/logger.hpp"
+#include "rclcpp/logging.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/message_memory_strategy.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -50,13 +57,24 @@ class MessagePoolMemoryStrategy
 public:
  RCLCPP_SMART_PTR_DEFINITIONS(MessagePoolMemoryStrategy)

-  /// Default constructor
  MessagePoolMemoryStrategy()
-  : next_array_index_(0)
  {
+    pool_mutex_ = std::make_shared<std::mutex>();
+
+    pool_ = std::shared_ptr<std::array<MessageT *, Size>>(
+      new std::array<MessageT *, Size>,
+      [](std::array<MessageT *, Size> * arr) {
+        for (size_t i = 0; i < Size; ++i) {
+          free((*arr)[i]);
+        }
+        delete arr;
+      });
+
+    free_list_ = std::make_shared<CircularArray<Size>>();
+
    for (size_t i = 0; i < Size; ++i) {
-      pool_[i].msg_ptr_ = std::make_shared<MessageT>();
-      pool_[i].used = false;
+      (*pool_)[i] = static_cast<MessageT *>(malloc(sizeof(MessageT)));
+      free_list_->push_back(i);
    }
  }

@@ -68,43 +86,85 @@ public:
   */
  std::shared_ptr<MessageT> borrow_message()
  {
-    size_t current_index = next_array_index_;
-    next_array_index_ = (next_array_index_ + 1) % Size;
-    if (pool_[current_index].used) {
-      throw std::runtime_error("Tried to access message that was still in use! Abort.");
+    std::lock_guard<std::mutex> lock(*pool_mutex_);
+    if (free_list_->size() == 0) {
+      throw std::runtime_error("No more free slots in the pool");
    }
-    pool_[current_index].msg_ptr_->~MessageT();
-    new (pool_[current_index].msg_ptr_.get())MessageT;

-    pool_[current_index].used = true;
-    return pool_[current_index].msg_ptr_;
+    size_t current_index = free_list_->pop_front();
+
+    return std::shared_ptr<MessageT>(
+      new((*pool_)[current_index]) MessageT(),
+      [pool = this->pool_, pool_mutex = this->pool_mutex_,
+      free_list = this->free_list_](MessageT * p) {
+        std::lock_guard<std::mutex> lock(*pool_mutex);
+        for (size_t i = 0; i < Size; ++i) {
+          if ((*pool)[i] == p) {
+            p->~MessageT();
+            free_list->push_back(i);
+            break;
+          }
+        }
+      });
  }

  /// Return a message to the message pool.
  /**
-   * Manage metadata in the message pool ring buffer to release the message.
+   * This does nothing since the message isn't returned to the pool until the user has dropped
+   * all references.
   * \param[in] msg Shared pointer to the message to return.
   */
  void return_message(std::shared_ptr<MessageT> & msg)
  {
-    for (size_t i = 0; i < Size; ++i) {
-      if (pool_[i].msg_ptr_ == msg) {
-        pool_[i].used = false;
-        return;
-      }
-    }
-    throw std::runtime_error("Unrecognized message ptr in return_message.");
+    (void)msg;
  }

 protected:
-  struct PoolMember
+  template<size_t N>
+  class CircularArray
  {
-    std::shared_ptr<MessageT> msg_ptr_;
-    bool used;
+public:
+    void push_back(const size_t v)
+    {
+      if (size_ + 1 > N) {
+        throw std::runtime_error("Tried to push too many items into the array");
+      }
+      array_[(front_ + size_) % N] = v;
+      ++size_;
+    }
+
+    size_t pop_front()
+    {
+      if (size_ < 1) {
+        throw std::runtime_error("Tried to pop item from empty array");
+      }
+
+      size_t val = array_[front_];
+
+      front_ = (front_ + 1) % N;
+      --size_;
+
+      return val;
+    }
+
+    size_t size() const
+    {
+      return size_;
+    }
+
+private:
+    size_t front_ = 0;
+    size_t size_ = 0;
+    std::array<size_t, N> array_;
  };

-  std::array<PoolMember, Size> pool_;
-  size_t next_array_index_;
+  // It's very important that these are shared_ptrs, since users of this class might hold a
+  // reference to a pool item longer than the lifetime of the class.  In that scenario, the
+  // shared_ptr ensures that the lifetime of these variables outlives this class, and hence ensures
+  // the custom destructor for each pool item can successfully run.
+  std::shared_ptr<std::mutex> pool_mutex_;
+  std::shared_ptr<std::array<MessageT *, Size>> pool_;
+  std::shared_ptr<CircularArray<Size>> free_list_;
 };

 }  // namespace message_pool_memory_strategy
--- a/rclcpp/include/rclcpp/subscription.hpp
+++ b/rclcpp/include/rclcpp/subscription.hpp
@@ -104,7 +104,7 @@ public:

 private:
  using SubscriptionTopicStatisticsSharedPtr =
-    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>;
+    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>;

 public:
  RCLCPP_SMART_PTR_DEFINITIONS(Subscription)
@@ -163,10 +163,6 @@ public:
        throw std::invalid_argument(
                "intraprocess communication is not allowed with 0 depth qos policy");
      }
-      if (qos_profile.durability() != rclcpp::DurabilityPolicy::Volatile) {
-        throw std::invalid_argument(
-                "intraprocess communication allowed only with volatile durability");
-      }

      using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<
        MessageT,
@@ -185,7 +181,7 @@ public:
        this->get_topic_name(),  // important to get like this, as it has the fully-qualified name
        qos_profile,
        resolve_intra_process_buffer_type(options_.intra_process_buffer_type, callback));
-      TRACEPOINT(
+      TRACETOOLS_TRACEPOINT(
        rclcpp_subscription_init,
        static_cast<const void *>(get_subscription_handle().get()),
        static_cast<const void *>(subscription_intra_process_.get()));
@@ -193,7 +189,8 @@ public:
      // Add it to the intra process manager.
      using rclcpp::experimental::IntraProcessManager;
      auto ipm = context->get_sub_context<IntraProcessManager>();
-      uint64_t intra_process_subscription_id = ipm->add_subscription(subscription_intra_process_);
+      uint64_t intra_process_subscription_id = ipm->template add_subscription<
+        ROSMessageType, ROSMessageTypeAllocator>(subscription_intra_process_);
      this->setup_intra_process(intra_process_subscription_id, ipm);
    }

@@ -201,11 +198,11 @@ public:
      this->subscription_topic_statistics_ = std::move(subscription_topic_statistics);
    }

-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_subscription_init,
      static_cast<const void *>(get_subscription_handle().get()),
      static_cast<const void *>(this));
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_subscription_callback_added,
      static_cast<const void *>(this),
      static_cast<const void *>(&any_callback_));
@@ -316,7 +313,7 @@ public:
    if (subscription_topic_statistics_) {
      const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
      const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
    }
  }

@@ -325,8 +322,20 @@ public:
    const std::shared_ptr<rclcpp::SerializedMessage> & serialized_message,
    const rclcpp::MessageInfo & message_info) override
  {
-    // TODO(wjwwood): enable topic statistics for serialized messages
+    std::chrono::time_point<std::chrono::system_clock> now;
+    if (subscription_topic_statistics_) {
+      // get current time before executing callback to
+      // exclude callback duration from topic statistics result.
+      now = std::chrono::system_clock::now();
+    }
+
    any_callback_.dispatch(serialized_message, message_info);
+
+    if (subscription_topic_statistics_) {
+      const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
+      const auto time = rclcpp::Time(nanos.time_since_epoch().count());
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
+    }
  }

  void
@@ -357,7 +366,7 @@ public:
    if (subscription_topic_statistics_) {
      const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
      const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
    }
  }

--- a/rclcpp/include/rclcpp/subscription_base.hpp
+++ b/rclcpp/include/rclcpp/subscription_base.hpp
@@ -260,13 +260,13 @@ public:
  bool
  is_serialized() const;

-  /// Return the type of the subscription.
+  /// Return the delivered message kind.
  /**
   * \return `DeliveredMessageKind`, which adjusts how messages are received and delivered.
   */
  RCLCPP_PUBLIC
  DeliveredMessageKind
-  get_subscription_type() const;
+  get_delivered_message_kind() const;

  /// Get matching publisher count.
  /** \return The number of publishers on this topic. */
@@ -645,6 +645,14 @@ protected:
  rclcpp::node_interfaces::NodeBaseInterface * const node_base_;

  std::shared_ptr<rcl_node_t> node_handle_;
+
+  std::recursive_mutex callback_mutex_;
+  // It is important to declare on_new_message_callback_ before
+  // subscription_handle_, so on destruction the subscription is
+  // destroyed first. Otherwise, the rmw subscription callback
+  // would point briefly to a destroyed function.
+  std::function<void(size_t)> on_new_message_callback_{nullptr};
+  // Declare subscription_handle_ after callback
  std::shared_ptr<rcl_subscription_t> subscription_handle_;
  std::shared_ptr<rcl_subscription_t> intra_process_subscription_handle_;
  rclcpp::Logger node_logger_;
@@ -663,15 +671,12 @@ private:
  RCLCPP_DISABLE_COPY(SubscriptionBase)

  rosidl_message_type_support_t type_support_;
-  DeliveredMessageKind delivered_message_type_;
+  DeliveredMessageKind delivered_message_kind_;

  std::atomic<bool> subscription_in_use_by_wait_set_{false};
  std::atomic<bool> intra_process_subscription_waitable_in_use_by_wait_set_{false};
  std::unordered_map<rclcpp::EventHandlerBase *,
    std::atomic<bool>> qos_events_in_use_by_wait_set_;
-
-  std::recursive_mutex callback_mutex_;
-  std::function<void(size_t)> on_new_message_callback_{nullptr};
 };

 }  // namespace rclcpp
--- a/rclcpp/include/rclcpp/subscription_factory.hpp
+++ b/rclcpp/include/rclcpp/subscription_factory.hpp
@@ -75,15 +75,14 @@ template<
  typename CallbackT,
  typename AllocatorT,
  typename SubscriptionT = rclcpp::Subscription<MessageT, AllocatorT>,
-  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType
+  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType
 >
 SubscriptionFactory
 create_subscription_factory(
  CallbackT && callback,
  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options,
  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat,
-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
  subscription_topic_stats = nullptr
 )
 {
--- a/rclcpp/include/rclcpp/subscription_options.hpp
+++ b/rclcpp/include/rclcpp/subscription_options.hpp
@@ -77,6 +77,10 @@ struct SubscriptionOptionsBase
    // Topic statistics publication period in ms. Defaults to one second.
    // Only values greater than zero are allowed.
    std::chrono::milliseconds publish_period{std::chrono::seconds(1)};
+
+    // An optional QoS which can provide topic_statistics with a stable QoS separate from
+    // the subscription's current QoS settings which could be unstable.
+    rclcpp::QoS qos = SystemDefaultsQoS();
  };

  TopicStatisticsOptions topic_stats_options;
--- a/rclcpp/include/rclcpp/time.hpp
+++ b/rclcpp/include/rclcpp/time.hpp
@@ -189,7 +189,7 @@ public:
   */
  RCLCPP_PUBLIC
  static Time
-  max();
+  max(rcl_clock_type_t clock_type = RCL_SYSTEM_TIME);  // NOLINT

  /// Get the seconds since epoch
  /**
--- a/rclcpp/include/rclcpp/timer.hpp
+++ b/rclcpp/include/rclcpp/timer.hpp
@@ -53,12 +53,17 @@ public:
   * \param clock A clock to use for time and sleeping
   * \param period The interval at which the timer fires
   * \param context node context
+   * \param autostart timer state on initialization
+   *
+   * In order to activate a timer that is not started on initialization,
+   * user should call the reset() method.
   */
  RCLCPP_PUBLIC
  explicit TimerBase(
    Clock::SharedPtr clock,
    std::chrono::nanoseconds period,
-    rclcpp::Context::SharedPtr context);
+    rclcpp::Context::SharedPtr context,
+    bool autostart = true);

  /// TimerBase destructor
  RCLCPP_PUBLIC
@@ -216,21 +221,22 @@ public:
   * \param[in] period The interval at which the timer fires.
   * \param[in] callback User-specified callback function.
   * \param[in] context custom context to be used.
+   * \param autostart timer state on initialization
   */
  explicit GenericTimer(
    Clock::SharedPtr clock, std::chrono::nanoseconds period, FunctorT && callback,
-    rclcpp::Context::SharedPtr context
+    rclcpp::Context::SharedPtr context, bool autostart = true
  )
-  : TimerBase(clock, period, context), callback_(std::forward<FunctorT>(callback))
+  : TimerBase(clock, period, context, autostart), callback_(std::forward<FunctorT>(callback))
  {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_timer_callback_added,
      static_cast<const void *>(get_timer_handle().get()),
      reinterpret_cast<const void *>(&callback_));
 #ifndef TRACETOOLS_DISABLED
-    if (TRACEPOINT_ENABLED(rclcpp_callback_register)) {
+    if (TRACETOOLS_TRACEPOINT_ENABLED(rclcpp_callback_register)) {
      char * symbol = tracetools::get_symbol(callback_);
-      DO_TRACEPOINT(
+      TRACETOOLS_DO_TRACEPOINT(
        rclcpp_callback_register,
        reinterpret_cast<const void *>(&callback_),
        symbol);
@@ -269,9 +275,9 @@ public:
  void
  execute_callback() override
  {
-    TRACEPOINT(callback_start, reinterpret_cast<const void *>(&callback_), false);
+    TRACETOOLS_TRACEPOINT(callback_start, reinterpret_cast<const void *>(&callback_), false);
    execute_callback_delegate<>();
-    TRACEPOINT(callback_end, reinterpret_cast<const void *>(&callback_));
+    TRACETOOLS_TRACEPOINT(callback_end, reinterpret_cast<const void *>(&callback_));
  }

  // void specialization
@@ -330,13 +336,15 @@ public:
   * \param period The interval at which the timer fires
   * \param callback The callback function to execute every interval
   * \param context node context
+   * \param autostart timer state on initialization
   */
  WallTimer(
    std::chrono::nanoseconds period,
    FunctorT && callback,
-    rclcpp::Context::SharedPtr context)
+    rclcpp::Context::SharedPtr context,
+    bool autostart = true)
  : GenericTimer<FunctorT>(
-      std::make_shared<Clock>(RCL_STEADY_TIME), period, std::move(callback), context)
+      std::make_shared<Clock>(RCL_STEADY_TIME), period, std::move(callback), context, autostart)
  {}

 protected:
--- a/rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp
+++ b/rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp
@@ -48,21 +48,12 @@ using libstatistics_collector::moving_average_statistics::StatisticData;
 /**
 * Class used to collect, measure, and publish topic statistics data. Current statistics
 * supported for subscribers are received message age and received message period.
- *
- * \tparam CallbackMessageT the subscribed message type
- */
-template<typename CallbackMessageT>
+  */
 class SubscriptionTopicStatistics
 {
-  using TopicStatsCollector =
-    libstatistics_collector::topic_statistics_collector::TopicStatisticsCollector<
-    CallbackMessageT>;
-  using ReceivedMessageAge =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessageAgeCollector<
-    CallbackMessageT>;
-  using ReceivedMessagePeriod =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessagePeriodCollector<
-    CallbackMessageT>;
+  using TopicStatsCollector = libstatistics_collector::TopicStatisticsCollector;
+  using ReceivedMessageAge = libstatistics_collector::ReceivedMessageAgeCollector;
+  using ReceivedMessagePeriod = libstatistics_collector::ReceivedMessagePeriodCollector;

 public:
  /// Construct a SubscriptionTopicStatistics object.
@@ -101,16 +92,16 @@ public:
  /**
   * This method acquires a lock to prevent race conditions to collectors list.
   *
-   * \param received_message the message received by the subscription
+   * \param message_info the message info corresponding to the received message
   * \param now_nanoseconds current time in nanoseconds
   */
  virtual void handle_message(
-    const CallbackMessageT & received_message,
+    const rmw_message_info_t & message_info,
    const rclcpp::Time now_nanoseconds) const
  {
    std::lock_guard<std::mutex> lock(mutex_);
    for (const auto & collector : subscriber_statistics_collectors_) {
-      collector->OnMessageReceived(received_message, now_nanoseconds.nanoseconds());
+      collector->OnMessageReceived(message_info, now_nanoseconds.nanoseconds());
    }
  }

--- a/rclcpp/include/rclcpp/typesupport_helpers.hpp
+++ b/rclcpp/include/rclcpp/typesupport_helpers.hpp
@@ -22,6 +22,7 @@

 #include "rcpputils/shared_library.hpp"
 #include "rosidl_runtime_cpp/message_type_support_decl.hpp"
+#include "rosidl_runtime_cpp/service_type_support_decl.hpp"

 #include "rclcpp/visibility_control.hpp"

@@ -40,11 +41,15 @@ get_typesupport_library(const std::string & type, const std::string & typesuppor
 /// Extract the type support handle from the library.
 /**
 * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \deprecated Use get_message_typesupport_handle() instead
+ *
 * \param[in] type The topic type, e.g. "std_msgs/msg/String"
 * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
 * \param[in] library The shared type support library
 * \return A type support handle
 */
+[[deprecated("Use `get_message_typesupport_handle` instead")]]
 RCLCPP_PUBLIC
 const rosidl_message_type_support_t *
 get_typesupport_handle(
@@ -52,6 +57,40 @@ get_typesupport_handle(
  const std::string & typesupport_identifier,
  rcpputils::SharedLibrary & library);

+/// Extract the message type support handle from the library.
+/**
+ * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \param[in] type The topic type, e.g. "std_msgs/msg/String"
+ * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
+ * \param[in] library The shared type support library
+ * \throws std::runtime_error if the symbol of type not found in the library.
+ * \return A message type support handle
+ */
+RCLCPP_PUBLIC
+const rosidl_message_type_support_t *
+get_message_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library);
+
+/// Extract the service type support handle from the library.
+/**
+ * The library needs to match the topic type. The shared library must stay loaded for the lifetime of the result.
+ *
+ * \param[in] type The service type, e.g. "std_srvs/srv/Empty"
+ * \param[in] typesupport_identifier Type support identifier, typically "rosidl_typesupport_cpp"
+ * \param[in] library The shared type support library
+ * \throws std::runtime_error if the symbol of type not found in the library.
+ * \return A service type support handle
+ */
+RCLCPP_PUBLIC
+const rosidl_service_type_support_t *
+get_service_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library);
+
 }  // namespace rclcpp

 #endif  // RCLCPP__TYPESUPPORT_HELPERS_HPP_
--- a/rclcpp/package.xml
+++ b/rclcpp/package.xml
@@ -2,7 +2,7 @@
 <?xml-model href="http://download.ros.org/schema/package_format2.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
 <package format="2">
  <name>rclcpp</name>
-  <version>20.0.0</version>
+  <version>26.0.0</version>
  <description>The ROS client library in C++.</description>

  <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>
@@ -22,6 +22,7 @@
  <build_depend>builtin_interfaces</build_depend>
  <build_depend>rcl_interfaces</build_depend>
  <build_depend>rosgraph_msgs</build_depend>
+  <build_depend>rosidl_runtime_c</build_depend>
  <build_depend>rosidl_runtime_cpp</build_depend>
  <build_depend>rosidl_typesupport_c</build_depend>
  <build_depend>rosidl_typesupport_cpp</build_depend>
@@ -29,12 +30,14 @@
  <build_export_depend>builtin_interfaces</build_export_depend>
  <build_export_depend>rcl_interfaces</build_export_depend>
  <build_export_depend>rosgraph_msgs</build_export_depend>
+  <build_export_depend>rosidl_runtime_c</build_export_depend>
  <build_export_depend>rosidl_runtime_cpp</build_export_depend>
  <build_export_depend>rosidl_typesupport_c</build_export_depend>
  <build_export_depend>rosidl_typesupport_cpp</build_export_depend>

  <depend>libstatistics_collector</depend>
  <depend>rcl</depend>
+  <depend>rcl_logging_interface</depend>
  <depend>rcl_yaml_param_parser</depend>
  <depend>rcpputils</depend>
  <depend>rcutils</depend>
--- a/rclcpp/src/rclcpp/callback_group.cpp
+++ b/rclcpp/src/rclcpp/callback_group.cpp
@@ -31,12 +31,12 @@ using rclcpp::CallbackGroupType;

 CallbackGroup::CallbackGroup(
  CallbackGroupType group_type,
-  std::function<rclcpp::Context::SharedPtr(void)> get_context,
+  rclcpp::Context::WeakPtr context,
  bool automatically_add_to_executor_with_node)
 : type_(group_type), associated_with_executor_(false),
  can_be_taken_from_(true),
  automatically_add_to_executor_with_node_(automatically_add_to_executor_with_node),
-  get_context_(get_context)
+  context_(context)
 {}

 CallbackGroup::~CallbackGroup()
@@ -56,6 +56,16 @@ CallbackGroup::type() const
  return type_;
 }

+size_t
+CallbackGroup::size() const
+{
+  return
+    subscription_ptrs_.size() +
+    service_ptrs_.size() +
+    client_ptrs_.size() +
+    timer_ptrs_.size() +
+    waitable_ptrs_.size();
+}
 void CallbackGroup::collect_all_ptrs(
  std::function<void(const rclcpp::SubscriptionBase::SharedPtr &)> sub_func,
  std::function<void(const rclcpp::ServiceBase::SharedPtr &)> service_func,
@@ -113,40 +123,12 @@ CallbackGroup::automatically_add_to_executor_with_node() const
  return automatically_add_to_executor_with_node_;
 }

-// \TODO(mjcarroll) Deprecated, remove on tock
-rclcpp::GuardCondition::SharedPtr
-CallbackGroup::get_notify_guard_condition(const rclcpp::Context::SharedPtr context_ptr)
-{
-  std::lock_guard<std::recursive_mutex> lock(notify_guard_condition_mutex_);
-  if (notify_guard_condition_ && context_ptr != notify_guard_condition_->get_context()) {
-    if (associated_with_executor_) {
-      trigger_notify_guard_condition();
-    }
-    notify_guard_condition_ = nullptr;
-  }
-
-  if (!notify_guard_condition_) {
-    notify_guard_condition_ = std::make_shared<rclcpp::GuardCondition>(context_ptr);
-  }
-
-  return notify_guard_condition_;
-}
-
 rclcpp::GuardCondition::SharedPtr
 CallbackGroup::get_notify_guard_condition()
 {
  std::lock_guard<std::recursive_mutex> lock(notify_guard_condition_mutex_);
-  if (!this->get_context_) {
-    throw std::runtime_error("Callback group was created without context and not passed context");
-  }
-  auto context_ptr = this->get_context_();
+  rclcpp::Context::SharedPtr context_ptr = context_.lock();
  if (context_ptr && context_ptr->is_valid()) {
-    if (notify_guard_condition_ && context_ptr != notify_guard_condition_->get_context()) {
-      if (associated_with_executor_) {
-        trigger_notify_guard_condition();
-      }
-      notify_guard_condition_ = nullptr;
-    }
    if (!notify_guard_condition_) {
      notify_guard_condition_ = std::make_shared<rclcpp::GuardCondition>(context_ptr);
    }
--- a/rclcpp/src/rclcpp/client.cpp
+++ b/rclcpp/src/rclcpp/client.cpp
@@ -125,7 +125,6 @@ bool
 ClientBase::wait_for_service_nanoseconds(std::chrono::nanoseconds timeout)
 {
  auto start = std::chrono::steady_clock::now();
-  // make an event to reuse, rather than create a new one each time
  auto node_ptr = node_graph_.lock();
  if (!node_ptr) {
    throw InvalidNodeError();
@@ -138,6 +137,7 @@ ClientBase::wait_for_service_nanoseconds(std::chrono::nanoseconds timeout)
    // check was non-blocking, return immediately
    return false;
  }
+  // make an event to reuse, rather than create a new one each time
  auto event = node_ptr->get_graph_event();
  // update the time even on the first loop to account for time spent in the first call
  // to this->server_is_ready()
--- a/rclcpp/src/rclcpp/context.cpp
+++ b/rclcpp/src/rclcpp/context.cpp
@@ -462,7 +462,7 @@ template<Context::ShutdownType shutdown_type>
 std::vector<rclcpp::Context::ShutdownCallback>
 Context::get_shutdown_callback() const
 {
-  const auto get_callback_vector = [this](auto & mutex, auto & callback_set) {
+  const auto get_callback_vector = [](auto & mutex, auto & callback_set) {
      const std::lock_guard<std::mutex> lock(mutex);
      std::vector<rclcpp::Context::ShutdownCallback> callbacks;
      for (auto & callback : callback_set) {
@@ -496,7 +496,7 @@ Context::sleep_for(const std::chrono::nanoseconds & nanoseconds)
      std::unique_lock<std::mutex> lock(interrupt_mutex_);
      auto start = std::chrono::steady_clock::now();
      // this will release the lock while waiting
-      interrupt_condition_variable_.wait_for(lock, nanoseconds);
+      interrupt_condition_variable_.wait_for(lock, time_left);
      time_left -= std::chrono::steady_clock::now() - start;
    }
  } while (time_left > std::chrono::nanoseconds::zero() && this->is_valid());
--- a/rclcpp/src/rclcpp/detail/resolve_intra_process_buffer_type.cpp
+++ b/rclcpp/src/rclcpp/detail/resolve_intra_process_buffer_type.cpp
@@ -0,0 +1,37 @@
+// Copyright 2023 Open Source Robotics Foundation, Inc.
+//
+// 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 <rclcpp/detail/resolve_intra_process_buffer_type.hpp>
+
+namespace rclcpp
+{
+
+namespace detail
+{
+rclcpp::IntraProcessBufferType
+resolve_intra_process_buffer_type(
+  const rclcpp::IntraProcessBufferType buffer_type)
+{
+  if (buffer_type == IntraProcessBufferType::CallbackDefault) {
+    throw std::invalid_argument(
+            "IntraProcessBufferType::CallbackDefault is not allowed "
+            "when there is no callback function");
+  }
+
+  return buffer_type;
+}
+
+}  // namespace detail
+
+}  // namespace rclcpp
--- a/rclcpp/src/rclcpp/event_handler.cpp
+++ b/rclcpp/src/rclcpp/event_handler.cpp
@@ -39,13 +39,6 @@ UnsupportedEventTypeException::UnsupportedEventTypeException(

 EventHandlerBase::~EventHandlerBase()
 {
-  // Since the rmw event listener holds a reference to
-  // this callback, we need to clear it on destruction of this class.
-  // This clearing is not needed for other rclcpp entities like pub/subs, since
-  // they do own the underlying rmw entities, which are destroyed
-  // on their rclcpp destructors, thus no risk of dangling pointers.
-  clear_on_ready_callback();
-
  if (rcl_event_fini(&event_handle_) != RCL_RET_OK) {
    RCUTILS_LOG_ERROR_NAMED(
      "rclcpp",
--- a/rclcpp/src/rclcpp/executor.cpp
+++ b/rclcpp/src/rclcpp/executor.cpp
@@ -41,11 +41,14 @@ using namespace std::chrono_literals;
 using rclcpp::exceptions::throw_from_rcl_error;
 using rclcpp::Executor;

+class rclcpp::ExecutorImplementation {};
+
 Executor::Executor(const rclcpp::ExecutorOptions & options)
 : spinning(false),
  interrupt_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
  shutdown_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
-  memory_strategy_(options.memory_strategy)
+  memory_strategy_(options.memory_strategy),
+  impl_(std::make_unique<rclcpp::ExecutorImplementation>())
 {
  // Store the context for later use.
  context_ = options.context;
@@ -428,6 +431,22 @@ void Executor::spin_some(std::chrono::nanoseconds max_duration)
  return this->spin_some_impl(max_duration, false);
 }

+void
+Executor::spin_node_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node,
+  std::chrono::nanoseconds max_duration)
+{
+  this->add_node(node, false);
+  spin_all(max_duration);
+  this->remove_node(node, false);
+}
+
+void
+Executor::spin_node_all(std::shared_ptr<rclcpp::Node> node, std::chrono::nanoseconds max_duration)
+{
+  this->spin_node_all(node->get_node_base_interface(), max_duration);
+}
+
 void Executor::spin_all(std::chrono::nanoseconds max_duration)
 {
  if (max_duration < 0ns) {
@@ -522,13 +541,13 @@ Executor::execute_any_executable(AnyExecutable & any_exec)
    return;
  }
  if (any_exec.timer) {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_executor_execute,
      static_cast<const void *>(any_exec.timer->get_timer_handle().get()));
    execute_timer(any_exec.timer);
  }
  if (any_exec.subscription) {
-    TRACEPOINT(
+    TRACETOOLS_TRACEPOINT(
      rclcpp_executor_execute,
      static_cast<const void *>(any_exec.subscription->get_subscription_handle().get()));
    execute_subscription(any_exec.subscription);
@@ -600,7 +619,7 @@ Executor::execute_subscription(rclcpp::SubscriptionBase::SharedPtr subscription)
  rclcpp::MessageInfo message_info;
  message_info.get_rmw_message_info().from_intra_process = false;

-  switch (subscription->get_subscription_type()) {
+  switch (subscription->get_delivered_message_kind()) {
    // Deliver ROS message
    case rclcpp::DeliveredMessageKind::ROS_MESSAGE:
      {
@@ -650,6 +669,11 @@ Executor::execute_subscription(rclcpp::SubscriptionBase::SharedPtr subscription)
            subscription->get_topic_name(),
            [&]() {return subscription->take_type_erased(message.get(), message_info);},
            [&]() {subscription->handle_message(message, message_info);});
+          // TODO(clalancette): In the case that the user is using the MessageMemoryPool,
+          // and they take a shared_ptr reference to the message in the callback, this can
+          // inadvertently return the message to the pool when the user is still using it.
+          // This is a bug that needs to be fixed in the pool, and we should probably have
+          // a custom deleter for the message that actually does the return_message().
          subscription->return_message(message);
        }
        break;
@@ -723,7 +747,7 @@ Executor::execute_client(
 void
 Executor::wait_for_work(std::chrono::nanoseconds timeout)
 {
-  TRACEPOINT(rclcpp_executor_wait_for_work, timeout.count());
+  TRACETOOLS_TRACEPOINT(rclcpp_executor_wait_for_work, timeout.count());
  {
    std::lock_guard<std::mutex> guard(mutex_);

@@ -879,7 +903,7 @@ Executor::get_next_ready_executable_from_map(
  const rclcpp::memory_strategy::MemoryStrategy::WeakCallbackGroupsToNodesMap &
  weak_groups_to_nodes)
 {
-  TRACEPOINT(rclcpp_executor_get_next_ready);
+  TRACETOOLS_TRACEPOINT(rclcpp_executor_get_next_ready);
  bool success = false;
  std::lock_guard<std::mutex> guard{mutex_};
  // Check the timers to see if there are any that are ready
--- a/rclcpp/src/rclcpp/executors.cpp
+++ b/rclcpp/src/rclcpp/executors.cpp
@@ -14,6 +14,21 @@

 #include "rclcpp/executors.hpp"

+void
+rclcpp::spin_all(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
+  std::chrono::nanoseconds max_duration)
+{
+  rclcpp::executors::SingleThreadedExecutor exec;
+  exec.spin_node_all(node_ptr, max_duration);
+}
+
+void
+rclcpp::spin_all(rclcpp::Node::SharedPtr node_ptr, std::chrono::nanoseconds max_duration)
+{
+  rclcpp::spin_all(node_ptr->get_node_base_interface(), max_duration);
+}
+
 void
 rclcpp::spin_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr)
 {
--- a/rclcpp/src/rclcpp/executors/executor_entities_collection.cpp
+++ b/rclcpp/src/rclcpp/executors/executor_entities_collection.cpp
@@ -20,12 +20,13 @@ namespace executors
 {
 bool ExecutorEntitiesCollection::empty() const
 {
-  return subscriptions.empty() &&
-         timers.empty() &&
-         guard_conditions.empty() &&
-         clients.empty() &&
-         services.empty() &&
-         waitables.empty();
+  return
+    subscriptions.empty() &&
+    timers.empty() &&
+    guard_conditions.empty() &&
+    clients.empty() &&
+    services.empty() &&
+    waitables.empty();
 }

 void ExecutorEntitiesCollection::clear()
@@ -38,7 +39,6 @@ void ExecutorEntitiesCollection::clear()
  waitables.clear();
 }

-
 void
 build_entities_collection(
  const std::vector<rclcpp::CallbackGroup::WeakPtr> & callback_groups,
@@ -94,109 +94,136 @@ build_entities_collection(
  }
 }

-template<typename EntityCollectionType>
-void check_ready(
-  EntityCollectionType & collection,
-  std::deque<rclcpp::AnyExecutable> & executables,
-  size_t size_of_waited_entities,
-  typename EntityCollectionType::Key * waited_entities,
-  std::function<bool(rclcpp::AnyExecutable &,
-  typename EntityCollectionType::EntitySharedPtr &)> fill_executable)
+size_t
+ready_executables(
+  const ExecutorEntitiesCollection & collection,
+  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result,
+  std::deque<rclcpp::AnyExecutable> & executables
+)
 {
-  for (size_t ii = 0; ii < size_of_waited_entities; ++ii) {
-    if (!waited_entities[ii]) {continue;}
-    auto entity_iter = collection.find(waited_entities[ii]);
-    if (entity_iter != collection.end()) {
+  size_t added = 0;
+  if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
+    return added;
+  }
+  auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
+
+  // Cache shared pointers to groups to avoid extra work re-locking them
+  std::map<rclcpp::CallbackGroup::WeakPtr,
+    rclcpp::CallbackGroup::SharedPtr,
+    std::owner_less<rclcpp::CallbackGroup::WeakPtr>> group_map;
+
+  auto group_cache = [&group_map](const rclcpp::CallbackGroup::WeakPtr & weak_cbg_ptr)
+    {
+      if (group_map.count(weak_cbg_ptr) == 0) {
+        group_map.insert({weak_cbg_ptr, weak_cbg_ptr.lock()});
+      }
+      return group_map.find(weak_cbg_ptr)->second;
+    };
+
+  for (size_t ii = 0; ii < rcl_wait_set.size_of_timers; ++ii) {
+    if (nullptr == rcl_wait_set.timers[ii]) {continue;}
+    auto entity_iter = collection.timers.find(rcl_wait_set.timers[ii]);
+    if (entity_iter != collection.timers.end()) {
      auto entity = entity_iter->second.entity.lock();
      if (!entity) {
        continue;
      }
-
-      auto callback_group = entity_iter->second.callback_group.lock();
-      if (callback_group && !callback_group->can_be_taken_from().load()) {
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      if (!entity->call()) {
        continue;
      }
      rclcpp::AnyExecutable exec;
-
-      exec.callback_group = callback_group;
-      if (fill_executable(exec, entity)) {
-        executables.push_back(exec);
-      }
+      exec.timer = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
    }
  }
-}

-std::deque<rclcpp::AnyExecutable>
-ready_executables(
-  const ExecutorEntitiesCollection & collection,
-  rclcpp::WaitResult<rclcpp::WaitSet> & wait_result
-)
-{
-  std::deque<rclcpp::AnyExecutable> ret;
-
-  if (wait_result.kind() != rclcpp::WaitResultKind::Ready) {
-    return ret;
-  }
-  auto rcl_wait_set = wait_result.get_wait_set().get_rcl_wait_set();
-  check_ready(
-    collection.timers,
-    ret,
-    rcl_wait_set.size_of_timers,
-    rcl_wait_set.timers,
-    [](rclcpp::AnyExecutable & exec, auto timer) {
-      if (!timer->call()) {
-        return false;
+  for (size_t ii = 0; ii < rcl_wait_set.size_of_subscriptions; ++ii) {
+    if (nullptr == rcl_wait_set.subscriptions[ii]) {continue;}
+    auto entity_iter = collection.subscriptions.find(rcl_wait_set.subscriptions[ii]);
+    if (entity_iter != collection.subscriptions.end()) {
+      auto entity = entity_iter->second.entity.lock();
+      if (!entity) {
+        continue;
      }
-      exec.timer = timer;
-      return true;
-    });
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.subscription = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }

-  check_ready(
-    collection.subscriptions,
-    ret,
-    rcl_wait_set.size_of_subscriptions,
-    rcl_wait_set.subscriptions,
-    [](rclcpp::AnyExecutable & exec, auto subscription) {
-      exec.subscription = subscription;
-      return true;
-    });
+  for (size_t ii = 0; ii < rcl_wait_set.size_of_services; ++ii) {
+    if (nullptr == rcl_wait_set.services[ii]) {continue;}
+    auto entity_iter = collection.services.find(rcl_wait_set.services[ii]);
+    if (entity_iter != collection.services.end()) {
+      auto entity = entity_iter->second.entity.lock();
+      if (!entity) {
+        continue;
+      }
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.service = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }

-
-  check_ready(
-    collection.services,
-    ret,
-    rcl_wait_set.size_of_services,
-    rcl_wait_set.services,
-    [](rclcpp::AnyExecutable & exec, auto service) {
-      exec.service = service;
-      return true;
-    });
-
-  check_ready(
-    collection.clients,
-    ret,
-    rcl_wait_set.size_of_clients,
-    rcl_wait_set.clients,
-    [](rclcpp::AnyExecutable & exec, auto client) {
-      exec.client = client;
-      return true;
-    });
+  for (size_t ii = 0; ii < rcl_wait_set.size_of_clients; ++ii) {
+    if (nullptr == rcl_wait_set.clients[ii]) {continue;}
+    auto entity_iter = collection.clients.find(rcl_wait_set.clients[ii]);
+    if (entity_iter != collection.clients.end()) {
+      auto entity = entity_iter->second.entity.lock();
+      if (!entity) {
+        continue;
+      }
+      auto group_info = group_cache(entity_iter->second.callback_group);
+      if (group_info && !group_info->can_be_taken_from().load()) {
+        continue;
+      }
+      rclcpp::AnyExecutable exec;
+      exec.client = entity;
+      exec.callback_group = group_info;
+      executables.push_back(exec);
+      added++;
+    }
+  }

  for (auto & [handle, entry] : collection.waitables) {
    auto waitable = entry.entity.lock();
-    if (waitable && waitable->is_ready(&rcl_wait_set)) {
-      auto group = entry.callback_group.lock();
-      if (group && !group->can_be_taken_from().load()) {
-        continue;
-      }
-
-      rclcpp::AnyExecutable exec;
-      exec.waitable = waitable;
-      exec.callback_group = group;
-      ret.push_back(exec);
+    if (!waitable) {
+      continue;
    }
+    if (!waitable->is_ready(&rcl_wait_set)) {
+      continue;
+    }
+    auto group_info = group_cache(entry.callback_group);
+    if (group_info && !group_info->can_be_taken_from().load()) {
+      continue;
+    }
+    rclcpp::AnyExecutable exec;
+    exec.waitable = waitable;
+    exec.callback_group = group_info;
+    exec.data = waitable->take_data();
+    executables.push_back(exec);
+    added++;
  }
-  return ret;
+
+  return added;
 }

 }  // namespace executors
--- a/rclcpp/src/rclcpp/experimental/executors/events_executor/events_executor.cpp
+++ b/rclcpp/src/rclcpp/experimental/executors/events_executor/events_executor.cpp
@@ -50,6 +50,9 @@ EventsExecutor::EventsExecutor(
  timers_manager_ =
    std::make_shared<rclcpp::experimental::TimersManager>(context_, timer_on_ready_cb);

+  this->current_entities_collection_ =
+    std::make_shared<rclcpp::executors::ExecutorEntitiesCollection>();
+
  notify_waitable_ = std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(
    [this]() {
      // This callback is invoked when:
@@ -61,6 +64,10 @@ EventsExecutor::EventsExecutor(
      this->refresh_current_collection_from_callback_groups();
    });

+  // Make sure that the notify waitable is immediately added to the collection
+  // to avoid missing events
+  this->add_notify_waitable_to_collection(current_entities_collection_->waitables);
+
  notify_waitable_->add_guard_condition(interrupt_guard_condition_);
  notify_waitable_->add_guard_condition(shutdown_guard_condition_);

@@ -87,9 +94,6 @@ EventsExecutor::EventsExecutor(

  this->entities_collector_ =
    std::make_shared<rclcpp::executors::ExecutorEntitiesCollector>(notify_waitable_);
-
-  this->current_entities_collection_ =
-    std::make_shared<rclcpp::executors::ExecutorEntitiesCollection>();
 }

 EventsExecutor::~EventsExecutor()
@@ -269,10 +273,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
  switch (event.type) {
    case ExecutorEventType::CLIENT_EVENT:
      {
-        auto client = this->retrieve_entity(
-          static_cast<const rcl_client_t *>(event.entity_key),
-          current_entities_collection_->clients);
-
+        rclcpp::ClientBase::SharedPtr client;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          client = this->retrieve_entity(
+            static_cast<const rcl_client_t *>(event.entity_key),
+            current_entities_collection_->clients);
+        }
        if (client) {
          for (size_t i = 0; i < event.num_events; i++) {
            execute_client(client);
@@ -283,9 +290,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
      }
    case ExecutorEventType::SUBSCRIPTION_EVENT:
      {
-        auto subscription = this->retrieve_entity(
-          static_cast<const rcl_subscription_t *>(event.entity_key),
-          current_entities_collection_->subscriptions);
+        rclcpp::SubscriptionBase::SharedPtr subscription;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          subscription = this->retrieve_entity(
+            static_cast<const rcl_subscription_t *>(event.entity_key),
+            current_entities_collection_->subscriptions);
+        }
        if (subscription) {
          for (size_t i = 0; i < event.num_events; i++) {
            execute_subscription(subscription);
@@ -295,10 +306,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
      }
    case ExecutorEventType::SERVICE_EVENT:
      {
-        auto service = this->retrieve_entity(
-          static_cast<const rcl_service_t *>(event.entity_key),
-          current_entities_collection_->services);
-
+        rclcpp::ServiceBase::SharedPtr service;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          service = this->retrieve_entity(
+            static_cast<const rcl_service_t *>(event.entity_key),
+            current_entities_collection_->services);
+        }
        if (service) {
          for (size_t i = 0; i < event.num_events; i++) {
            execute_service(service);
@@ -315,9 +329,13 @@ EventsExecutor::execute_event(const ExecutorEvent & event)
      }
    case ExecutorEventType::WAITABLE_EVENT:
      {
-        auto waitable = this->retrieve_entity(
-          static_cast<const rclcpp::Waitable *>(event.entity_key),
-          current_entities_collection_->waitables);
+        rclcpp::Waitable::SharedPtr waitable;
+        {
+          std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+          waitable = this->retrieve_entity(
+            static_cast<const rclcpp::Waitable *>(event.entity_key),
+            current_entities_collection_->waitables);
+        }
        if (waitable) {
          for (size_t i = 0; i < event.num_events; i++) {
            auto data = waitable->take_data_by_entity_id(event.waitable_data);
@@ -382,6 +400,7 @@ EventsExecutor::get_automatically_added_callback_groups_from_nodes()
 void
 EventsExecutor::refresh_current_collection_from_callback_groups()
 {
+  // Build the new collection
  this->entities_collector_->update_collections();
  auto callback_groups = this->entities_collector_->get_all_callback_groups();
  rclcpp::executors::ExecutorEntitiesCollection new_collection;
@@ -395,18 +414,11 @@ EventsExecutor::refresh_current_collection_from_callback_groups()
  // retrieved in the "standard" way.
  // To do it, we need to add the notify waitable as an entry in both the new and
  // current collections such that it's neither added or removed.
-  rclcpp::CallbackGroup::WeakPtr weak_group_ptr;
-  new_collection.waitables.insert(
-  {
-    this->notify_waitable_.get(),
-    {this->notify_waitable_, weak_group_ptr}
-  });
+  this->add_notify_waitable_to_collection(new_collection.waitables);

-  this->current_entities_collection_->waitables.insert(
-  {
-    this->notify_waitable_.get(),
-    {this->notify_waitable_, weak_group_ptr}
-  });
+  // Acquire lock before modifying the current collection
+  std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+  this->add_notify_waitable_to_collection(current_entities_collection_->waitables);

  this->refresh_current_collection(new_collection);
 }
@@ -415,6 +427,9 @@ void
 EventsExecutor::refresh_current_collection(
  const rclcpp::executors::ExecutorEntitiesCollection & new_collection)
 {
+  // Acquire lock before modifying the current collection
+  std::lock_guard<std::recursive_mutex> lock(collection_mutex_);
+
  current_entities_collection_->timers.update(
    new_collection.timers,
    [this](rclcpp::TimerBase::SharedPtr timer) {timers_manager_->add_timer(timer);},
@@ -486,3 +501,16 @@ EventsExecutor::create_waitable_callback(const rclcpp::Waitable * entity_key)
    };
  return callback;
 }
+
+void
+EventsExecutor::add_notify_waitable_to_collection(
+  rclcpp::executors::ExecutorEntitiesCollection::WaitableCollection & collection)
+{
+  // The notify waitable is not associated to any group, so use an invalid one
+  rclcpp::CallbackGroup::WeakPtr weak_group_ptr;
+  collection.insert(
+  {
+    this->notify_waitable_.get(),
+    {this->notify_waitable_, weak_group_ptr}
+  });
+}
--- a/rclcpp/src/rclcpp/experimental/timers_manager.cpp
+++ b/rclcpp/src/rclcpp/experimental/timers_manager.cpp
@@ -28,9 +28,9 @@ using rclcpp::experimental::TimersManager;
 TimersManager::TimersManager(
  std::shared_ptr<rclcpp::Context> context,
  std::function<void(const rclcpp::TimerBase *)> on_ready_callback)
+: on_ready_callback_(on_ready_callback),
+  context_(context)
 {
-  context_ = context;
-  on_ready_callback_ = on_ready_callback;
 }

 TimersManager::~TimersManager()
--- a/rclcpp/src/rclcpp/generic_subscription.cpp
+++ b/rclcpp/src/rclcpp/generic_subscription.cpp
@@ -49,9 +49,9 @@ GenericSubscription::handle_message(
 void
 GenericSubscription::handle_serialized_message(
  const std::shared_ptr<rclcpp::SerializedMessage> & message,
-  const rclcpp::MessageInfo &)
+  const rclcpp::MessageInfo & message_info)
 {
-  callback_(message);
+  callback_(message, message_info);
 }

 void
--- a/rclcpp/src/rclcpp/guard_condition.cpp
+++ b/rclcpp/src/rclcpp/guard_condition.cpp
@@ -23,16 +23,17 @@ namespace rclcpp
 {

 GuardCondition::GuardCondition(
-  rclcpp::Context::SharedPtr context,
+  const rclcpp::Context::SharedPtr & context,
  rcl_guard_condition_options_t guard_condition_options)
-: context_(context), rcl_guard_condition_{rcl_get_zero_initialized_guard_condition()}
+: rcl_guard_condition_{rcl_get_zero_initialized_guard_condition()}
 {
-  if (!context_) {
+  if (!context) {
    throw std::invalid_argument("context argument unexpectedly nullptr");
  }
+
  rcl_ret_t ret = rcl_guard_condition_init(
    &this->rcl_guard_condition_,
-    context_->get_rcl_context().get(),
+    context->get_rcl_context().get(),
    guard_condition_options);
  if (RCL_RET_OK != ret) {
    rclcpp::exceptions::throw_from_rcl_error(ret, "failed to create guard condition");
@@ -53,12 +54,6 @@ GuardCondition::~GuardCondition()
  }
 }

-rclcpp::Context::SharedPtr
-GuardCondition::get_context() const
-{
-  return context_;
-}
-
 rcl_guard_condition_t &
 GuardCondition::get_rcl_guard_condition()
 {
--- a/rclcpp/src/rclcpp/intra_process_manager.cpp
+++ b/rclcpp/src/rclcpp/intra_process_manager.cpp
@@ -32,13 +32,24 @@ IntraProcessManager::~IntraProcessManager()
 {}

 uint64_t
-IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
+IntraProcessManager::add_publisher(
+  rclcpp::PublisherBase::SharedPtr publisher,
+  rclcpp::experimental::buffers::IntraProcessBufferBase::SharedPtr buffer)
 {
  std::unique_lock<std::shared_timed_mutex> lock(mutex_);

  uint64_t pub_id = IntraProcessManager::get_next_unique_id();

  publishers_[pub_id] = publisher;
+  if (publisher->is_durability_transient_local()) {
+    if (buffer) {
+      publisher_buffers_[pub_id] = buffer;
+    } else {
+      throw std::runtime_error(
+              "transient_local publisher needs to pass"
+              "a valid publisher buffer ptr when calling add_publisher()");
+    }
+  }

  // Initialize the subscriptions storage for this publisher.
  pub_to_subs_[pub_id] = SplittedSubscriptions();
@@ -58,30 +69,6 @@ IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
  return pub_id;
 }

-uint64_t
-IntraProcessManager::add_subscription(SubscriptionIntraProcessBase::SharedPtr subscription)
-{
-  std::unique_lock<std::shared_timed_mutex> lock(mutex_);
-
-  uint64_t sub_id = IntraProcessManager::get_next_unique_id();
-
-  subscriptions_[sub_id] = subscription;
-
-  // adds the subscription id to all the matchable publishers
-  for (auto & pair : publishers_) {
-    auto publisher = pair.second.lock();
-    if (!publisher) {
-      continue;
-    }
-    if (can_communicate(publisher, subscription)) {
-      uint64_t pub_id = pair.first;
-      insert_sub_id_for_pub(sub_id, pub_id, subscription->use_take_shared_method());
-    }
-  }
-
-  return sub_id;
-}
-
 void
 IntraProcessManager::remove_subscription(uint64_t intra_process_subscription_id)
 {
@@ -112,6 +99,7 @@ IntraProcessManager::remove_publisher(uint64_t intra_process_publisher_id)
  std::unique_lock<std::shared_timed_mutex> lock(mutex_);

  publishers_.erase(intra_process_publisher_id);
+  publisher_buffers_.erase(intra_process_publisher_id);
  pub_to_subs_.erase(intra_process_publisher_id);
 }

@@ -225,5 +213,52 @@ IntraProcessManager::can_communicate(
  return true;
 }

+size_t
+IntraProcessManager::lowest_available_capacity(const uint64_t intra_process_publisher_id) const
+{
+  size_t capacity = std::numeric_limits<size_t>::max();
+
+  auto publisher_it = pub_to_subs_.find(intra_process_publisher_id);
+  if (publisher_it == pub_to_subs_.end()) {
+    // Publisher is either invalid or no longer exists.
+    RCLCPP_WARN(
+      rclcpp::get_logger("rclcpp"),
+      "Calling lowest_available_capacity for invalid or no longer existing publisher id");
+    return 0u;
+  }
+
+  if (publisher_it->second.take_shared_subscriptions.empty() &&
+    publisher_it->second.take_ownership_subscriptions.empty())
+  {
+    // no subscriptions available
+    return 0u;
+  }
+
+  auto available_capacity = [this, &capacity](const uint64_t intra_process_subscription_id)
+    {
+      auto subscription_it = subscriptions_.find(intra_process_subscription_id);
+      if (subscription_it != subscriptions_.end()) {
+        auto subscription = subscription_it->second.lock();
+        if (subscription) {
+          capacity = std::min(capacity, subscription->available_capacity());
+        }
+      } else {
+        // Subscription is either invalid or no longer exists.
+        RCLCPP_WARN(
+          rclcpp::get_logger("rclcpp"),
+          "Calling available_capacity for invalid or no longer existing subscription id");
+      }
+    };
+
+  for (const auto sub_id : publisher_it->second.take_shared_subscriptions) {
+    available_capacity(sub_id);
+  }
+
+  for (const auto sub_id : publisher_it->second.take_ownership_subscriptions) {
+    available_capacity(sub_id);
+  }
+
+  return capacity;
+}
 }  // namespace experimental
 }  // namespace rclcpp
--- a/rclcpp/src/rclcpp/logger.cpp
+++ b/rclcpp/src/rclcpp/logger.cpp
@@ -17,6 +17,7 @@
 #include <utility>

 #include "rcl_logging_interface/rcl_logging_interface.h"
+#include "rcl/error_handling.h"
 #include "rcl/logging_rosout.h"

 #include "rclcpp/exceptions.hpp"
@@ -80,10 +81,12 @@ Logger::get_child(const std::string & suffix)
  {
    std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
    rcl_ret = rcl_logging_rosout_add_sublogger((*name_).c_str(), suffix.c_str());
-    if (RCL_RET_OK != rcl_ret) {
+    if (RCL_RET_NOT_FOUND == rcl_ret) {
+      rcl_reset_error();
+    } else if (RCL_RET_OK != rcl_ret) {
      exceptions::throw_from_rcl_error(
        rcl_ret, "failed to call rcl_logging_rosout_add_sublogger",
-        rcutils_get_error_state(), rcutils_reset_error);
+        rcl_get_error_state(), rcl_reset_error);
    }
  }

@@ -98,9 +101,7 @@ Logger::get_child(const std::string & suffix)
          logger_sublogger_pairname_ptr->second.c_str());
        delete logger_sublogger_pairname_ptr;
        if (RCL_RET_OK != rcl_ret) {
-          exceptions::throw_from_rcl_error(
-            rcl_ret, "failed to call rcl_logging_rosout_remove_sublogger",
-            rcutils_get_error_state(), rcutils_reset_error);
+          rcl_reset_error();
        }
      });
  }
--- a/rclcpp/src/rclcpp/logging_mutex.cpp
+++ b/rclcpp/src/rclcpp/logging_mutex.cpp
@@ -14,6 +14,7 @@

 #include <memory>
 #include <mutex>
+#include <stdexcept>

 #include "rcutils/macros.h"

--- a/rclcpp/src/rclcpp/node.cpp
+++ b/rclcpp/src/rclcpp/node.cpp
@@ -36,6 +36,7 @@
 #include "rclcpp/node_interfaces/node_time_source.hpp"
 #include "rclcpp/node_interfaces/node_timers.hpp"
 #include "rclcpp/node_interfaces/node_topics.hpp"
+#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
 #include "rclcpp/node_interfaces/node_waitables.hpp"
 #include "rclcpp/qos_overriding_options.hpp"

@@ -109,6 +110,22 @@ create_effective_namespace(const std::string & node_namespace, const std::string

 }  // namespace

+/// Internal implementation to provide hidden and API/ABI stable changes to the Node.
+/**
+ * This class is intended to be an "escape hatch" within a stable distribution, so that certain
+ * smaller features and bugfixes can be backported, having a place to put new members, while
+ * maintaining the ABI.
+ *
+ * This is not intended to be a parking place for new features, it should be used for backports
+ * only, left empty and unallocated in Rolling.
+ */
+class Node::NodeImpl
+{
+public:
+  NodeImpl() = default;
+  ~NodeImpl() = default;
+};
+
 Node::Node(
  const std::string & node_name,
  const NodeOptions & options)
@@ -167,7 +184,7 @@ Node::Node(
      options.use_intra_process_comms(),
      options.enable_topic_statistics())),
  node_graph_(new rclcpp::node_interfaces::NodeGraph(node_base_.get())),
-  node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_.get())),
+  node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_)),
  node_timers_(new rclcpp::node_interfaces::NodeTimers(node_base_.get())),
  node_topics_(new rclcpp::node_interfaces::NodeTopics(node_base_.get(), node_timers_.get())),
  node_services_(new rclcpp::node_interfaces::NodeServices(node_base_.get())),
@@ -206,6 +223,12 @@ Node::Node(
      options.clock_qos(),
      options.use_clock_thread()
    )),
+  node_type_descriptions_(new rclcpp::node_interfaces::NodeTypeDescriptions(
+      node_base_,
+      node_logging_,
+      node_parameters_,
+      node_services_
+    )),
  node_waitables_(new rclcpp::node_interfaces::NodeWaitables(node_base_.get())),
  node_options_(options),
  sub_namespace_(""),
@@ -225,6 +248,10 @@ Node::Node(
    node_topics_->resolve_topic_name("/parameter_events"),
    options.parameter_event_qos(),
    rclcpp::detail::PublisherQosParametersTraits{});
+
+  if (options.enable_logger_service()) {
+    node_logging_->create_logger_services(node_services_);
+  }
 }

 Node::Node(
@@ -242,7 +269,8 @@ Node::Node(
  node_waitables_(other.node_waitables_),
  node_options_(other.node_options_),
  sub_namespace_(extend_sub_namespace(other.get_sub_namespace(), sub_namespace)),
-  effective_namespace_(create_effective_namespace(other.get_namespace(), sub_namespace_))
+  effective_namespace_(create_effective_namespace(other.get_namespace(), sub_namespace_)),
+  hidden_impl_(other.hidden_impl_)
 {
  // Validate new effective namespace.
  int validation_result;
@@ -494,6 +522,18 @@ Node::count_subscribers(const std::string & topic_name) const
  return node_graph_->count_subscribers(topic_name);
 }

+size_t
+Node::count_clients(const std::string & service_name) const
+{
+  return node_graph_->count_clients(service_name);
+}
+
+size_t
+Node::count_services(const std::string & service_name) const
+{
+  return node_graph_->count_services(service_name);
+}
+
 std::vector<rclcpp::TopicEndpointInfo>
 Node::get_publishers_info_by_topic(const std::string & topic_name, bool no_mangle) const
 {
@@ -587,6 +627,12 @@ Node::get_node_topics_interface()
  return node_topics_;
 }

+rclcpp::node_interfaces::NodeTypeDescriptionsInterface::SharedPtr
+Node::get_node_type_descriptions_interface()
+{
+  return node_type_descriptions_;
+}
+
 rclcpp::node_interfaces::NodeServicesInterface::SharedPtr
 Node::get_node_services_interface()
 {
--- a/rclcpp/src/rclcpp/node_interfaces/node_base.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_base.cpp
@@ -20,6 +20,9 @@
 #include "rclcpp/node_interfaces/node_base.hpp"

 #include "rcl/arguments.h"
+#include "rcl/node_type_cache.h"
+#include "rcl/logging.h"
+#include "rcl/logging_rosout.h"
 #include "rclcpp/exceptions.hpp"
 #include "rcutils/logging_macros.h"
 #include "rmw/validate_namespace.h"
@@ -54,17 +57,12 @@ NodeBase::NodeBase(
  std::shared_ptr<std::recursive_mutex> logging_mutex = get_global_logging_mutex();

  rcl_ret_t ret;
-  {
-    std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
-    // TODO(ivanpauno): /rosout Qos should be reconfigurable.
-    // TODO(ivanpauno): Instead of mutually excluding rcl_node_init with the global logger mutex,
-    // rcl_logging_rosout_init_publisher_for_node could be decoupled from there and be called
-    // here directly.
-    ret = rcl_node_init(
-      rcl_node.get(),
-      node_name.c_str(), namespace_.c_str(),
-      context_->get_rcl_context().get(), &rcl_node_options);
-  }
+
+  // TODO(ivanpauno): /rosout Qos should be reconfigurable.
+  ret = rcl_node_init(
+    rcl_node.get(),
+    node_name.c_str(), namespace_.c_str(),
+    context_->get_rcl_context().get(), &rcl_node_options);
  if (ret != RCL_RET_OK) {
    if (ret == RCL_RET_NODE_INVALID_NAME) {
      rcl_reset_error();  // discard rcl_node_init error
@@ -114,13 +112,29 @@ NodeBase::NodeBase(
    throw_from_rcl_error(ret, "failed to initialize rcl node");
  }

+  // The initialization for the rosout publisher
+  if (rcl_logging_rosout_enabled() && rcl_node_options.enable_rosout) {
+    std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
+    ret = rcl_logging_rosout_init_publisher_for_node(rcl_node.get());
+    if (ret != RCL_RET_OK) {
+      throw_from_rcl_error(ret, "failed to initialize rosout publisher");
+    }
+  }
+
  node_handle_.reset(
    rcl_node.release(),
-    [logging_mutex](rcl_node_t * node) -> void {
-      std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
-      // TODO(ivanpauno): Instead of mutually excluding rcl_node_fini with the global logger mutex,
-      // rcl_logging_rosout_fini_publisher_for_node could be decoupled from there and be called
-      // here directly.
+    [logging_mutex, rcl_node_options](rcl_node_t * node) -> void {
+      {
+        std::lock_guard<std::recursive_mutex> guard(*logging_mutex);
+        if (rcl_logging_rosout_enabled() && rcl_node_options.enable_rosout) {
+          rcl_ret_t ret = rcl_logging_rosout_fini_publisher_for_node(node);
+          if (ret != RCL_RET_OK) {
+            RCUTILS_LOG_ERROR_NAMED(
+              "rclcpp",
+              "Error in destruction of rosout publisher: %s", rcl_get_error_string().str);
+          }
+        }
+      }
      if (rcl_node_fini(node) != RCL_RET_OK) {
        RCUTILS_LOG_ERROR_NAMED(
          "rclcpp",
@@ -204,14 +218,9 @@ NodeBase::create_callback_group(
  rclcpp::CallbackGroupType group_type,
  bool automatically_add_to_executor_with_node)
 {
-  auto weak_context = this->get_context()->weak_from_this();
-  auto get_node_context = [weak_context]() -> rclcpp::Context::SharedPtr {
-      return weak_context.lock();
-    };
-
  auto group = std::make_shared<rclcpp::CallbackGroup>(
    group_type,
-    get_node_context,
+    context_->weak_from_this(),
    automatically_add_to_executor_with_node);
  std::lock_guard<std::mutex> lock(callback_groups_mutex_);
  callback_groups_.push_back(group);
--- a/rclcpp/src/rclcpp/node_interfaces/node_graph.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_graph.cpp
@@ -498,6 +498,50 @@ NodeGraph::count_subscribers(const std::string & topic_name) const
  return count;
 }

+size_t
+NodeGraph::count_clients(const std::string & service_name) const
+{
+  auto rcl_node_handle = node_base_->get_rcl_node_handle();
+
+  auto fqdn = rclcpp::expand_topic_or_service_name(
+    service_name,
+    rcl_node_get_name(rcl_node_handle),
+    rcl_node_get_namespace(rcl_node_handle),
+    true);
+
+  size_t count;
+  auto ret = rcl_count_clients(rcl_node_handle, fqdn.c_str(), &count);
+  if (ret != RMW_RET_OK) {
+    // *INDENT-OFF*
+    throw std::runtime_error(
+      std::string("could not count clients: ") + rmw_get_error_string().str);
+    // *INDENT-ON*
+  }
+  return count;
+}
+
+size_t
+NodeGraph::count_services(const std::string & service_name) const
+{
+  auto rcl_node_handle = node_base_->get_rcl_node_handle();
+
+  auto fqdn = rclcpp::expand_topic_or_service_name(
+    service_name,
+    rcl_node_get_name(rcl_node_handle),
+    rcl_node_get_namespace(rcl_node_handle),
+    true);
+
+  size_t count;
+  auto ret = rcl_count_services(rcl_node_handle, fqdn.c_str(), &count);
+  if (ret != RMW_RET_OK) {
+    // *INDENT-OFF*
+    throw std::runtime_error(
+      std::string("could not count services: ") + rmw_get_error_string().str);
+    // *INDENT-ON*
+  }
+  return count;
+}
+
 const rcl_guard_condition_t *
 NodeGraph::get_graph_guard_condition() const
 {
--- a/rclcpp/src/rclcpp/node_interfaces/node_logging.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_logging.cpp
@@ -12,11 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+#include "rclcpp/node_impl.hpp"
 #include "rclcpp/node_interfaces/node_logging.hpp"
+#include "rclcpp/node_interfaces/node_services_interface.hpp"

 using rclcpp::node_interfaces::NodeLogging;

-NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface * node_base)
+NodeLogging::NodeLogging(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base)
 : node_base_(node_base)
 {
  logger_ = rclcpp::get_logger(NodeLogging::get_logger_name());
@@ -37,3 +39,55 @@ NodeLogging::get_logger_name() const
 {
  return rcl_node_get_logger_name(node_base_->get_rcl_node_handle());
 }
+
+void NodeLogging::create_logger_services(
+  node_interfaces::NodeServicesInterface::SharedPtr node_services)
+{
+  rclcpp::ServicesQoS qos_profile;
+  const std::string node_name = node_base_->get_name();
+  auto callback_group = node_base_->get_default_callback_group();
+
+  get_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::GetLoggerLevels>(
+    node_base_, node_services,
+    node_name + "/get_logger_levels",
+    [](
+      const std::shared_ptr<rmw_request_id_t>,
+      const std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Request> request,
+      std::shared_ptr<rcl_interfaces::srv::GetLoggerLevels::Response> response)
+    {
+      for (auto & name : request->names) {
+        rcl_interfaces::msg::LoggerLevel logger_level;
+        logger_level.name = name;
+        auto ret = rcutils_logging_get_logger_level(name.c_str());
+        if (ret < 0) {
+          logger_level.level = 0;
+        } else {
+          logger_level.level = static_cast<uint8_t>(ret);
+        }
+        response->levels.push_back(std::move(logger_level));
+      }
+    },
+    qos_profile, callback_group);
+
+  set_loggers_service_ = rclcpp::create_service<rcl_interfaces::srv::SetLoggerLevels>(
+    node_base_, node_services,
+    node_name + "/set_logger_levels",
+    [](
+      const std::shared_ptr<rmw_request_id_t>,
+      const std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Request> request,
+      std::shared_ptr<rcl_interfaces::srv::SetLoggerLevels::Response> response)
+    {
+      rcl_interfaces::msg::SetLoggerLevelsResult result;
+      for (auto & level : request->levels) {
+        auto ret = rcutils_logging_set_logger_level(level.name.c_str(), level.level);
+        if (ret != RCUTILS_RET_OK) {
+          result.successful = false;
+          result.reason = rcutils_get_error_string().str;
+        } else {
+          result.successful = true;
+        }
+        response->results.push_back(std::move(result));
+      }
+    },
+    qos_profile, callback_group);
+}
--- a/rclcpp/src/rclcpp/node_interfaces/node_parameters.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_parameters.cpp
@@ -1038,37 +1038,50 @@ NodeParameters::list_parameters(const std::vector<std::string> & prefixes, uint6
  // TODO(mikaelarguedas) define parameter separator different from "/" to avoid ambiguity
  // using "." for now
  const char * separator = ".";
-  for (auto & kv : parameters_) {
-    bool get_all = (prefixes.size() == 0) &&
-      ((depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) ||
-      (static_cast<uint64_t>(std::count(kv.first.begin(), kv.first.end(), *separator)) < depth));
-    bool prefix_matches = std::any_of(
-      prefixes.cbegin(), prefixes.cend(),
-      [&kv, &depth, &separator](const std::string & prefix) {
-        if (kv.first == prefix) {
-          return true;
-        } else if (kv.first.find(prefix + separator) == 0) {
-          size_t length = prefix.length();
-          std::string substr = kv.first.substr(length);
-          // Cast as unsigned integer to avoid warning
-          return (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) ||
-          (static_cast<uint64_t>(std::count(substr.begin(), substr.end(), *separator)) < depth);
-        }
-        return false;
-      });
-    if (get_all || prefix_matches) {
-      result.names.push_back(kv.first);
-      size_t last_separator = kv.first.find_last_of(separator);
-      if (std::string::npos != last_separator) {
-        std::string prefix = kv.first.substr(0, last_separator);
-        if (
-          std::find(result.prefixes.cbegin(), result.prefixes.cend(), prefix) ==
-          result.prefixes.cend())
-        {
-          result.prefixes.push_back(prefix);
+
+  auto separators_less_than_depth = [&depth, &separator](const std::string & str) -> bool {
+      return static_cast<uint64_t>(std::count(str.begin(), str.end(), *separator)) < depth;
+    };
+
+  bool recursive = (prefixes.size() == 0) &&
+    (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE);
+
+  for (const std::pair<const std::string, ParameterInfo> & kv : parameters_) {
+    if (!recursive) {
+      bool get_all = (prefixes.size() == 0) && separators_less_than_depth(kv.first);
+      if (!get_all) {
+        bool prefix_matches = std::any_of(
+          prefixes.cbegin(), prefixes.cend(),
+          [&kv, &depth, &separator, &separators_less_than_depth](const std::string & prefix) {
+            if (kv.first == prefix) {
+              return true;
+            } else if (kv.first.find(prefix + separator) == 0) {
+              if (depth == rcl_interfaces::srv::ListParameters::Request::DEPTH_RECURSIVE) {
+                return true;
+              }
+              std::string substr = kv.first.substr(prefix.length() + 1);
+              return separators_less_than_depth(substr);
+            }
+            return false;
+          });
+
+        if (!prefix_matches) {
+          continue;
        }
      }
    }
+
+    result.names.push_back(kv.first);
+    size_t last_separator = kv.first.find_last_of(separator);
+    if (std::string::npos != last_separator) {
+      std::string prefix = kv.first.substr(0, last_separator);
+      if (
+        std::find(result.prefixes.cbegin(), result.prefixes.cend(), prefix) ==
+        result.prefixes.cend())
+      {
+        result.prefixes.push_back(prefix);
+      }
+    }
  }
  return result;
 }
--- a/rclcpp/src/rclcpp/node_interfaces/node_services.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_services.cpp
@@ -32,8 +32,7 @@ NodeServices::add_service(
 {
  if (group) {
    if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create service, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("service");
    }
  } else {
    group = node_base_->get_default_callback_group();
@@ -58,8 +57,7 @@ NodeServices::add_client(
 {
  if (group) {
    if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create client, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("client");
    }
  } else {
    group = node_base_->get_default_callback_group();
--- a/rclcpp/src/rclcpp/node_interfaces/node_timers.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_timers.cpp
@@ -34,8 +34,7 @@ NodeTimers::add_timer(
 {
  if (callback_group) {
    if (!node_base_->callback_group_in_node(callback_group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create timer, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("timer");
    }
  } else {
    callback_group = node_base_->get_default_callback_group();
@@ -50,7 +49,7 @@ NodeTimers::add_timer(
            std::string("failed to notify wait set on timer creation: ") + ex.what());
  }

-  TRACEPOINT(
+  TRACETOOLS_TRACEPOINT(
    rclcpp_timer_link_node,
    static_cast<const void *>(timer->get_timer_handle().get()),
    static_cast<const void *>(node_base_->get_rcl_node_handle()));
--- a/rclcpp/src/rclcpp/node_interfaces/node_topics.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_topics.cpp
@@ -58,7 +58,7 @@ NodeTopics::add_publisher(
  // Assign to a group.
  if (callback_group) {
    if (!node_base_->callback_group_in_node(callback_group)) {
-      throw std::runtime_error("Cannot create publisher, callback group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("publisher");
    }
  } else {
    callback_group = node_base_->get_default_callback_group();
@@ -97,8 +97,7 @@ NodeTopics::add_subscription(
  // Assign to a group.
  if (callback_group) {
    if (!node_base_->callback_group_in_node(callback_group)) {
-      // TODO(jacquelinekay): use custom exception
-      throw std::runtime_error("Cannot create subscription, callback group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("subscription");
    }
  } else {
    callback_group = node_base_->get_default_callback_group();
--- a/rclcpp/src/rclcpp/node_interfaces/node_type_descriptions.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_type_descriptions.cpp
@@ -0,0 +1,153 @@
+// Copyright 2023 Open Source Robotics Foundation, Inc.
+//
+// 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 <memory>
+#include <string>
+#include <thread>
+
+#include "rclcpp/node_interfaces/node_type_descriptions.hpp"
+#include "rclcpp/parameter_client.hpp"
+
+#include "type_description_interfaces/srv/get_type_description.h"
+
+namespace
+{
+// Helper wrapper for rclcpp::Service to access ::Request and ::Response types for allocation.
+struct GetTypeDescription__C
+{
+  using Request = type_description_interfaces__srv__GetTypeDescription_Request;
+  using Response = type_description_interfaces__srv__GetTypeDescription_Response;
+  using Event = type_description_interfaces__srv__GetTypeDescription_Event;
+};
+}  // namespace
+
+// Helper function for C typesupport.
+namespace rosidl_typesupport_cpp
+{
+template<>
+rosidl_service_type_support_t const *
+get_service_type_support_handle<GetTypeDescription__C>()
+{
+  return ROSIDL_GET_SRV_TYPE_SUPPORT(type_description_interfaces, srv, GetTypeDescription);
+}
+}  // namespace rosidl_typesupport_cpp
+
+namespace rclcpp
+{
+namespace node_interfaces
+{
+
+class NodeTypeDescriptions::NodeTypeDescriptionsImpl
+{
+public:
+  using ServiceT = GetTypeDescription__C;
+
+  rclcpp::Logger logger_;
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
+  rclcpp::Service<ServiceT>::SharedPtr type_description_srv_;
+
+  NodeTypeDescriptionsImpl(
+    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
+    rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
+    rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
+    rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
+  : logger_(node_logging->get_logger()),
+    node_base_(node_base)
+  {
+    const std::string enable_param_name = "start_type_description_service";
+
+    bool enabled = false;
+    try {
+      auto enable_param = node_parameters->declare_parameter(
+        enable_param_name,
+        rclcpp::ParameterValue(true),
+        rcl_interfaces::msg::ParameterDescriptor()
+        .set__name(enable_param_name)
+        .set__type(rclcpp::PARAMETER_BOOL)
+        .set__description("Start the ~/get_type_description service for this node.")
+        .set__read_only(true));
+      enabled = enable_param.get<bool>();
+    } catch (const rclcpp::exceptions::InvalidParameterTypeException & exc) {
+      RCLCPP_ERROR(logger_, "%s", exc.what());
+      throw;
+    }
+
+    if (enabled) {
+      auto * rcl_node = node_base->get_rcl_node_handle();
+      std::shared_ptr<rcl_service_t> rcl_srv(
+        new rcl_service_t,
+        [rcl_node, logger = this->logger_](rcl_service_t * service)
+        {
+          if (rcl_service_fini(service, rcl_node) != RCL_RET_OK) {
+            RCLCPP_ERROR(
+              logger,
+              "Error in destruction of rcl service handle [~/get_type_description]: %s",
+              rcl_get_error_string().str);
+            rcl_reset_error();
+          }
+          delete service;
+        });
+      *rcl_srv = rcl_get_zero_initialized_service();
+      rcl_ret_t rcl_ret = rcl_node_type_description_service_init(rcl_srv.get(), rcl_node);
+
+      if (rcl_ret != RCL_RET_OK) {
+        RCLCPP_ERROR(
+          logger_, "Failed to initialize ~/get_type_description service: %s",
+          rcl_get_error_string().str);
+        throw std::runtime_error(
+                "Failed to initialize ~/get_type_description service.");
+      }
+
+      rclcpp::AnyServiceCallback<ServiceT> cb;
+      cb.set(
+        [this](
+          std::shared_ptr<rmw_request_id_t> header,
+          std::shared_ptr<ServiceT::Request> request,
+          std::shared_ptr<ServiceT::Response> response
+        ) {
+          rcl_node_type_description_service_handle_request(
+            node_base_->get_rcl_node_handle(),
+            header.get(),
+            request.get(),
+            response.get());
+        });
+
+      type_description_srv_ = std::make_shared<Service<ServiceT>>(
+        node_base_->get_shared_rcl_node_handle(),
+        rcl_srv,
+        cb);
+      node_services->add_service(
+        std::dynamic_pointer_cast<ServiceBase>(type_description_srv_),
+        nullptr);
+    }
+  }
+};
+
+NodeTypeDescriptions::NodeTypeDescriptions(
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base,
+  rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging,
+  rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters,
+  rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services)
+: impl_(new NodeTypeDescriptionsImpl(
+      node_base,
+      node_logging,
+      node_parameters,
+      node_services))
+{}
+
+NodeTypeDescriptions::~NodeTypeDescriptions()
+{}
+
+}  // namespace node_interfaces
+}  // namespace rclcpp
--- a/rclcpp/src/rclcpp/node_interfaces/node_waitables.cpp
+++ b/rclcpp/src/rclcpp/node_interfaces/node_waitables.cpp
@@ -32,8 +32,7 @@ NodeWaitables::add_waitable(
 {
  if (group) {
    if (!node_base_->callback_group_in_node(group)) {
-      // TODO(jacobperron): use custom exception
-      throw std::runtime_error("Cannot create waitable, group not in node.");
+      throw rclcpp::exceptions::MissingGroupNodeException("waitable");
    }
  } else {
    group = node_base_->get_default_callback_group();
--- a/rclcpp/src/rclcpp/node_options.cpp
+++ b/rclcpp/src/rclcpp/node_options.cpp
@@ -248,6 +248,19 @@ NodeOptions::start_parameter_services(bool start_parameter_services)
  return *this;
 }

+bool
+NodeOptions::enable_logger_service() const
+{
+  return this->enable_logger_service_;
+}
+
+NodeOptions &
+NodeOptions::enable_logger_service(bool enable_logger_service)
+{
+  this->enable_logger_service_ = enable_logger_service;
+  return *this;
+}
+
 bool
 NodeOptions::start_parameter_event_publisher() const
 {
--- a/rclcpp/src/rclcpp/parameter_value.cpp
+++ b/rclcpp/src/rclcpp/parameter_value.cpp
@@ -129,8 +129,7 @@ ParameterValue::ParameterValue(const rcl_interfaces::msg::ParameterValue & value
    case PARAMETER_NOT_SET:
      break;
    default:
-      // TODO(wjwwood): use custom exception
-      throw std::runtime_error("Unknown type: " + std::to_string(value.type));
+      throw rclcpp::exceptions::UnknownTypeError(std::to_string(value.type));
  }
 }

--- a/rclcpp/src/rclcpp/publisher_base.cpp
+++ b/rclcpp/src/rclcpp/publisher_base.cpp
@@ -270,6 +270,13 @@ PublisherBase::get_intra_process_subscription_count() const
  return ipm->get_subscription_count(intra_process_publisher_id_);
 }

+bool
+PublisherBase::is_durability_transient_local() const
+{
+  return rcl_publisher_get_actual_qos(publisher_handle_.get())->durability ==
+         RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL;
+}
+
 rclcpp::QoS
 PublisherBase::get_actual_qos() const
 {
@@ -384,3 +391,22 @@ std::vector<rclcpp::NetworkFlowEndpoint> PublisherBase::get_network_flow_endpoin

  return network_flow_endpoint_vector;
 }
+
+size_t PublisherBase::lowest_available_ipm_capacity() const
+{
+  if (!intra_process_is_enabled_) {
+    return 0u;
+  }
+
+  auto ipm = weak_ipm_.lock();
+
+  if (!ipm) {
+    // TODO(ivanpauno): should this raise an error?
+    RCLCPP_WARN(
+      rclcpp::get_logger("rclcpp"),
+      "Intra process manager died for a publisher.");
+    return 0u;
+  }
+
+  return ipm->lowest_available_capacity(intra_process_publisher_id_);
+}
--- a/rclcpp/src/rclcpp/rate.cpp
+++ b/rclcpp/src/rclcpp/rate.cpp
@@ -0,0 +1,106 @@
+// Copyright 2023 Open Source Robotics Foundation, Inc.
+//
+// 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 "rclcpp/rate.hpp"
+
+#include <chrono>
+#include <stdexcept>
+
+namespace rclcpp
+{
+
+Rate::Rate(
+  const double rate, Clock::SharedPtr clock)
+: clock_(clock), period_(0, 0), last_interval_(clock_->now())
+{
+  if (rate <= 0.0) {
+    throw std::invalid_argument{"rate must be greater than 0"};
+  }
+  period_ = Duration::from_seconds(1.0 / rate);
+}
+
+Rate::Rate(
+  const Duration & period, Clock::SharedPtr clock)
+: clock_(clock), period_(period), last_interval_(clock_->now())
+{
+  if (period <= Duration(0, 0)) {
+    throw std::invalid_argument{"period must be greater than 0"};
+  }
+}
+
+bool
+Rate::sleep()
+{
+  // Time coming into sleep
+  auto now = clock_->now();
+  // Time of next interval
+  auto next_interval = last_interval_ + period_;
+  // Detect backwards time flow
+  if (now < last_interval_) {
+    // Best thing to do is to set the next_interval to now + period
+    next_interval = now + period_;
+  }
+  // Update the interval
+  last_interval_ += period_;
+  // If the time_to_sleep is negative or zero, don't sleep
+  if (next_interval <= now) {
+    // If an entire cycle was missed then reset next interval.
+    // This might happen if the loop took more than a cycle.
+    // Or if time jumps forward.
+    if (now > next_interval + period_) {
+      last_interval_ = now + period_;
+    }
+    // Either way do not sleep and return false
+    return false;
+  }
+  // Calculate the time to sleep
+  auto time_to_sleep = next_interval - now;
+  // Sleep (will get interrupted by ctrl-c, may not sleep full time)
+  clock_->sleep_for(time_to_sleep);
+  return true;
+}
+
+bool
+Rate::is_steady() const
+{
+  return clock_->get_clock_type() == RCL_STEADY_TIME;
+}
+
+rcl_clock_type_t
+Rate::get_type() const
+{
+  return clock_->get_clock_type();
+}
+
+void
+Rate::reset()
+{
+  last_interval_ = clock_->now();
+}
+
+std::chrono::nanoseconds
+Rate::period() const
+{
+  return std::chrono::nanoseconds(period_.nanoseconds());
+}
+
+WallRate::WallRate(const double rate)
+: Rate(rate, std::make_shared<Clock>(RCL_STEADY_TIME))
+{}
+
+WallRate::WallRate(const Duration & period)
+: Rate(period, std::make_shared<Clock>(RCL_STEADY_TIME))
+{}
+
+}  // namespace rclcpp
--- a/rclcpp/src/rclcpp/signal_handler.cpp
+++ b/rclcpp/src/rclcpp/signal_handler.cpp
@@ -113,7 +113,7 @@ SignalHandler::get_logger()
 SignalHandler &
 SignalHandler::get_global_signal_handler()
 {
-  static SignalHandler signal_handler;
+  static SignalHandler & signal_handler = *new SignalHandler();
  return signal_handler;
 }

--- a/rclcpp/src/rclcpp/signal_handler.hpp
+++ b/rclcpp/src/rclcpp/signal_handler.hpp
@@ -75,7 +75,7 @@ public:
  bool
  install(SignalHandlerOptions signal_handler_options = SignalHandlerOptions::All);

-  /// Uninstall the signal handler for SIGINT/SIGTERM and join the dedicated singal handling
+  /// Uninstall the signal handler for SIGINT/SIGTERM and join the dedicated signal handling
  /// thread.
  /**
   * Also restores the previous signal handler.
@@ -189,7 +189,7 @@ private:

  // Whether or not a signal has been received.
  std::atomic_bool signal_received_ = false;
-  // A thread to which singal handling tasks are deferred.
+  // A thread to which signal handling tasks are deferred.
  std::thread signal_handler_thread_;

  // A mutex used to synchronize the install() and uninstall() methods.
--- a/rclcpp/src/rclcpp/subscription_base.cpp
+++ b/rclcpp/src/rclcpp/subscription_base.cpp
@@ -52,7 +52,7 @@ SubscriptionBase::SubscriptionBase(
  intra_process_subscription_id_(0),
  event_callbacks_(event_callbacks),
  type_support_(type_support_handle),
-  delivered_message_type_(delivered_message_kind)
+  delivered_message_kind_(delivered_message_kind)
 {
  auto custom_deletor = [node_handle = this->node_handle_](rcl_subscription_t * rcl_subs)
    {
@@ -218,7 +218,7 @@ SubscriptionBase::take_type_erased(void * message_out, rclcpp::MessageInfo & mes
    &message_info_out.get_rmw_message_info(),
    nullptr  // rmw_subscription_allocation_t is unused here
  );
-  TRACEPOINT(rclcpp_take, static_cast<const void *>(message_out));
+  TRACETOOLS_TRACEPOINT(rclcpp_take, static_cast<const void *>(message_out));
  if (RCL_RET_SUBSCRIPTION_TAKE_FAILED == ret) {
    return false;
  } else if (RCL_RET_OK != ret) {
@@ -261,13 +261,13 @@ SubscriptionBase::get_message_type_support_handle() const
 bool
 SubscriptionBase::is_serialized() const
 {
-  return delivered_message_type_ == rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE;
+  return delivered_message_kind_ == rclcpp::DeliveredMessageKind::SERIALIZED_MESSAGE;
 }

 rclcpp::DeliveredMessageKind
-SubscriptionBase::get_subscription_type() const
+SubscriptionBase::get_delivered_message_kind() const
 {
-  return delivered_message_type_;
+  return delivered_message_kind_;
 }

 size_t
@@ -298,7 +298,20 @@ SubscriptionBase::setup_intra_process(
 bool
 SubscriptionBase::can_loan_messages() const
 {
-  return rcl_subscription_can_loan_messages(subscription_handle_.get());
+  bool retval = rcl_subscription_can_loan_messages(subscription_handle_.get());
+  if (retval) {
+    // TODO(clalancette): The loaned message interface is currently not safe to use with
+    // shared_ptr callbacks.  If a user takes a copy of the shared_ptr, it can get freed from
+    // underneath them via rcl_return_loaned_message_from_subscription().  The correct solution is
+    // to return the loaned message in a custom deleter, but that needs to be carefully handled
+    // with locking.  Warn the user about this until we fix it.
+    RCLCPP_WARN_ONCE(
+      this->node_logger_,
+      "Loaned messages are only safe with const ref subscription callbacks. "
+      "If you are using any other kind of subscriptions, "
+      "set the ROS_DISABLE_LOANED_MESSAGES environment variable to 1 (the default).");
+  }
+  return retval;
 }

 rclcpp::Waitable::SharedPtr
--- a/rclcpp/src/rclcpp/subscription_intra_process_base.cpp
+++ b/rclcpp/src/rclcpp/subscription_intra_process_base.cpp
@@ -34,3 +34,9 @@ SubscriptionIntraProcessBase::get_actual_qos() const
 {
  return qos_profile_;
 }
+
+bool
+SubscriptionIntraProcessBase::is_durability_transient_local() const
+{
+  return qos_profile_.durability() == rclcpp::DurabilityPolicy::TransientLocal;
+}
--- a/rclcpp/src/rclcpp/time.cpp
+++ b/rclcpp/src/rclcpp/time.cpp
@@ -276,9 +276,9 @@ Time::operator-=(const rclcpp::Duration & rhs)
 }

 Time
-Time::max()
+Time::max(rcl_clock_type_t clock_type)
 {
-  return Time(std::numeric_limits<int32_t>::max(), 999999999);
+  return Time(std::numeric_limits<int32_t>::max(), 999999999, clock_type);
 }


--- a/rclcpp/src/rclcpp/time_source.cpp
+++ b/rclcpp/src/rclcpp/time_source.cpp
@@ -14,6 +14,7 @@

 #include <memory>
 #include <string>
+#include <unordered_set>
 #include <utility>
 #include <vector>

@@ -54,9 +55,7 @@ public:
    ros_time_active_ = true;

    // Update all attached clocks to zero or last recorded time
-    for (auto it = associated_clocks_.begin(); it != associated_clocks_.end(); ++it) {
-      set_clock(last_time_msg_, true, *it);
-    }
+    set_all_clocks(last_time_msg_, true);
  }

  // An internal method to use in the clock callback that iterates and disables all clocks
@@ -71,11 +70,8 @@ public:
    ros_time_active_ = false;

    // Update all attached clocks
-    std::lock_guard<std::mutex> guard(clock_list_lock_);
-    for (auto it = associated_clocks_.begin(); it != associated_clocks_.end(); ++it) {
-      auto msg = std::make_shared<builtin_interfaces::msg::Time>();
-      set_clock(msg, false, *it);
-    }
+    auto msg = std::make_shared<builtin_interfaces::msg::Time>();
+    set_all_clocks(msg, false);
  }

  // Check if ROS time is active
@@ -95,7 +91,7 @@ public:
      }
    }
    std::lock_guard<std::mutex> guard(clock_list_lock_);
-    associated_clocks_.push_back(clock);
+    associated_clocks_.insert(clock);
    // Set the clock to zero unless there's a recently received message
    set_clock(last_time_msg_, ros_time_active_, clock);
  }
@@ -104,10 +100,8 @@ public:
  void detachClock(rclcpp::Clock::SharedPtr clock)
  {
    std::lock_guard<std::mutex> guard(clock_list_lock_);
-    auto result = std::find(associated_clocks_.begin(), associated_clocks_.end(), clock);
-    if (result != associated_clocks_.end()) {
-      associated_clocks_.erase(result);
-    } else {
+    auto removed = associated_clocks_.erase(clock);
+    if (removed == 0) {
      RCLCPP_ERROR(logger_, "failed to remove clock");
    }
  }
@@ -184,8 +178,8 @@ private:

  // A lock to protect iterating the associated_clocks_ field.
  std::mutex clock_list_lock_;
-  // A vector to store references to associated clocks.
-  std::vector<rclcpp::Clock::SharedPtr> associated_clocks_;
+  // An unordered_set to store references to associated clocks.
+  std::unordered_set<rclcpp::Clock::SharedPtr> associated_clocks_;

  // Local storage of validity of ROS time
  // This is needed when new clocks are added.
@@ -242,6 +236,7 @@ public:
    rclcpp::node_interfaces::NodeClockInterface::SharedPtr node_clock_interface,
    rclcpp::node_interfaces::NodeParametersInterface::SharedPtr node_parameters_interface)
  {
+    std::lock_guard<std::mutex> guard(node_base_lock_);
    node_base_ = node_base_interface;
    node_topics_ = node_topics_interface;
    node_graph_ = node_graph_interface;
@@ -286,17 +281,14 @@ public:
    parameter_subscription_ = rclcpp::AsyncParametersClient::on_parameter_event(
      node_topics_,
      [this](std::shared_ptr<const rcl_interfaces::msg::ParameterEvent> event) {
-        if (node_base_ != nullptr) {
-          this->on_parameter_event(event);
-        }
-        // Do nothing if node_base_ is nullptr because it means the TimeSource is now
-        // without an attached node
+        this->on_parameter_event(event);
      });
  }

  // Detach the attached node
  void detachNode()
  {
+    std::lock_guard<std::mutex> guard(node_base_lock_);
    // destroy_clock_sub() *must* be first here, to ensure that the executor
    // can't possibly call any of the callbacks as we are cleaning up.
    destroy_clock_sub();
@@ -333,6 +325,7 @@ private:
  std::thread clock_executor_thread_;

  // Preserve the node reference
+  std::mutex node_base_lock_;
  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_{nullptr};
  rclcpp::node_interfaces::NodeTopicsInterface::SharedPtr node_topics_{nullptr};
  rclcpp::node_interfaces::NodeGraphInterface::SharedPtr node_graph_{nullptr};
@@ -470,6 +463,14 @@ private:
  // Callback for parameter updates
  void on_parameter_event(std::shared_ptr<const rcl_interfaces::msg::ParameterEvent> event)
  {
+    std::lock_guard<std::mutex> guard(node_base_lock_);
+
+    if (node_base_ == nullptr) {
+      // Do nothing if node_base_ is nullptr because it means the TimeSource is now
+      // without an attached node
+      return;
+    }
+
    // Filter out events on 'use_sim_time' parameter instances in other nodes.
    if (event->node != node_base_->get_fully_qualified_name()) {
      return;
--- a/rclcpp/src/rclcpp/timer.cpp
+++ b/rclcpp/src/rclcpp/timer.cpp
@@ -32,7 +32,8 @@ using rclcpp::TimerBase;
 TimerBase::TimerBase(
  rclcpp::Clock::SharedPtr clock,
  std::chrono::nanoseconds period,
-  rclcpp::Context::SharedPtr context)
+  rclcpp::Context::SharedPtr context,
+  bool autostart)
 : clock_(clock), timer_handle_(nullptr)
 {
  if (nullptr == context) {
@@ -64,9 +65,9 @@ TimerBase::TimerBase(
  rcl_clock_t * clock_handle = clock_->get_clock_handle();
  {
    std::lock_guard<std::mutex> clock_guard(clock_->get_clock_mutex());
-    rcl_ret_t ret = rcl_timer_init(
-      timer_handle_.get(), clock_handle, rcl_context.get(), period.count(), nullptr,
-      rcl_get_default_allocator());
+    rcl_ret_t ret = rcl_timer_init2(
+      timer_handle_.get(), clock_handle, rcl_context.get(), period.count(),
+      nullptr, rcl_get_default_allocator(), autostart);
    if (ret != RCL_RET_OK) {
      rclcpp::exceptions::throw_from_rcl_error(ret, "Couldn't initialize rcl timer handle");
    }
--- a/rclcpp/src/rclcpp/typesupport_helpers.cpp
+++ b/rclcpp/src/rclcpp/typesupport_helpers.cpp
@@ -91,6 +91,44 @@ extract_type_identifier(const std::string & full_type)
  return std::make_tuple(package_name, middle_module, type_name);
 }

+const void * get_typesupport_handle_impl(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  const std::string & typesupport_name,
+  const std::string & symbol_part_name,
+  const std::string & middle_module_additional,
+  rcpputils::SharedLibrary & library)
+{
+  std::string package_name;
+  std::string middle_module;
+  std::string type_name;
+  std::tie(package_name, middle_module, type_name) = extract_type_identifier(type);
+
+  if (middle_module.empty()) {
+    middle_module = middle_module_additional;
+  }
+
+  auto mk_error = [&package_name, &type_name, &typesupport_name](auto reason) {
+      std::stringstream rcutils_dynamic_loading_error;
+      rcutils_dynamic_loading_error <<
+        "Something went wrong loading the typesupport library for " <<
+        typesupport_name << " type " << package_name <<
+        "/" << type_name << ". " << reason;
+      return rcutils_dynamic_loading_error.str();
+    };
+
+  try {
+    std::string symbol_name = typesupport_identifier + symbol_part_name +
+      package_name + "__" + middle_module + "__" + type_name;
+    const void * (* get_ts)() = nullptr;
+    // This will throw runtime_error if the symbol was not found.
+    get_ts = reinterpret_cast<decltype(get_ts)>(library.get_symbol(symbol_name));
+    return get_ts();
+  } catch (std::runtime_error &) {
+    throw std::runtime_error{mk_error("Library could not be found.")};
+  }
+}
+
 }  // anonymous namespace

 std::shared_ptr<rcpputils::SharedLibrary>
@@ -101,36 +139,42 @@ get_typesupport_library(const std::string & type, const std::string & typesuppor
  return std::make_shared<rcpputils::SharedLibrary>(library_path);
 }

-const rosidl_message_type_support_t *
-get_typesupport_handle(
+const rosidl_message_type_support_t * get_typesupport_handle(
  const std::string & type,
  const std::string & typesupport_identifier,
  rcpputils::SharedLibrary & library)
 {
-  std::string package_name;
-  std::string middle_module;
-  std::string type_name;
-  std::tie(package_name, middle_module, type_name) = extract_type_identifier(type);
+  return get_message_typesupport_handle(type, typesupport_identifier, library);
+}

-  auto mk_error = [&package_name, &type_name](auto reason) {
-      std::stringstream rcutils_dynamic_loading_error;
-      rcutils_dynamic_loading_error <<
-        "Something went wrong loading the typesupport library for message type " << package_name <<
-        "/" << type_name << ". " << reason;
-      return rcutils_dynamic_loading_error.str();
-    };
+const rosidl_message_type_support_t * get_message_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library)
+{
+  static const std::string typesupport_name = "message";
+  static const std::string symbol_part_name = "__get_message_type_support_handle__";
+  static const std::string middle_module_additional = "msg";

-  try {
-    std::string symbol_name = typesupport_identifier + "__get_message_type_support_handle__" +
-      package_name + "__" + (middle_module.empty() ? "msg" : middle_module) + "__" + type_name;
+  return static_cast<const rosidl_message_type_support_t *>(get_typesupport_handle_impl(
+           type, typesupport_identifier, typesupport_name, symbol_part_name,
+           middle_module_additional, library
+  ));
+}

-    const rosidl_message_type_support_t * (* get_ts)() = nullptr;
-    // This will throw runtime_error if the symbol was not found.
-    get_ts = reinterpret_cast<decltype(get_ts)>(library.get_symbol(symbol_name));
-    return get_ts();
-  } catch (std::runtime_error &) {
-    throw std::runtime_error{mk_error("Library could not be found.")};
-  }
+const rosidl_service_type_support_t * get_service_typesupport_handle(
+  const std::string & type,
+  const std::string & typesupport_identifier,
+  rcpputils::SharedLibrary & library)
+{
+  static const std::string typesupport_name = "service";
+  static const std::string symbol_part_name = "__get_service_type_support_handle__";
+  static const std::string middle_module_additional = "srv";
+
+  return static_cast<const rosidl_service_type_support_t *>(get_typesupport_handle_impl(
+           type, typesupport_identifier, typesupport_name, symbol_part_name,
+           middle_module_additional, library
+  ));
 }

 }  // namespace rclcpp
--- a/rclcpp/test/benchmark/CMakeLists.txt
+++ b/rclcpp/test/benchmark/CMakeLists.txt
@@ -7,14 +7,12 @@ find_package(performance_test_fixture REQUIRED)

 add_performance_test(benchmark_client benchmark_client.cpp)
 if(TARGET benchmark_client)
-  target_link_libraries(benchmark_client ${PROJECT_NAME})
-  ament_target_dependencies(benchmark_client test_msgs rcl_interfaces)
+  target_link_libraries(benchmark_client ${PROJECT_NAME} ${rcl_interfaces_TARGETS} ${test_msgs_TARGETS})
 endif()

 add_performance_test(benchmark_executor benchmark_executor.cpp)
 if(TARGET benchmark_executor)
-  target_link_libraries(benchmark_executor ${PROJECT_NAME})
-  ament_target_dependencies(benchmark_executor test_msgs)
+  target_link_libraries(benchmark_executor ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()

 add_performance_test(benchmark_init_shutdown benchmark_init_shutdown.cpp)
@@ -39,6 +37,5 @@ endif()

 add_performance_test(benchmark_service benchmark_service.cpp)
 if(TARGET benchmark_service)
-  target_link_libraries(benchmark_service ${PROJECT_NAME})
-  ament_target_dependencies(benchmark_service test_msgs rcl_interfaces)
+  target_link_libraries(benchmark_service ${PROJECT_NAME} ${rcl_interfaces_TARGETS} ${test_msgs_TARGETS})
 endif()
--- a/rclcpp/test/rclcpp/CMakeLists.txt
+++ b/rclcpp/test/rclcpp/CMakeLists.txt
@@ -34,71 +34,38 @@ if(TARGET test_exceptions)
  target_link_libraries(test_exceptions ${PROJECT_NAME} mimick)
 endif()

-# Increasing timeout because connext can take a long time to destroy nodes
-# TODO(brawner) remove when destroying Node for Connext is resolved. See:
-# https://github.com/ros2/rclcpp/issues/1250
-ament_add_gtest(
-  test_allocator_memory_strategy
-  strategies/test_allocator_memory_strategy.cpp
-  TIMEOUT 360)
+ament_add_gtest(test_allocator_memory_strategy strategies/test_allocator_memory_strategy.cpp)
 if(TARGET test_allocator_memory_strategy)
-  ament_target_dependencies(test_allocator_memory_strategy
-    "rcl"
-    "test_msgs"
-  )
-  target_link_libraries(test_allocator_memory_strategy ${PROJECT_NAME})
+  target_link_libraries(test_allocator_memory_strategy ${PROJECT_NAME} rcpputils::rcpputils ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_message_pool_memory_strategy strategies/test_message_pool_memory_strategy.cpp)
 if(TARGET test_message_pool_memory_strategy)
-  ament_target_dependencies(test_message_pool_memory_strategy
-    "rcl"
-    "test_msgs"
-  )
-  target_link_libraries(test_message_pool_memory_strategy ${PROJECT_NAME})
+  target_link_libraries(test_message_pool_memory_strategy ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_any_service_callback test_any_service_callback.cpp)
 if(TARGET test_any_service_callback)
-  ament_target_dependencies(test_any_service_callback
-    "test_msgs"
-  )
-  target_link_libraries(test_any_service_callback ${PROJECT_NAME})
+  target_link_libraries(test_any_service_callback ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_any_subscription_callback test_any_subscription_callback.cpp)
 if(TARGET test_any_subscription_callback)
-  ament_target_dependencies(test_any_subscription_callback
-    "test_msgs"
-  )
-  target_link_libraries(test_any_subscription_callback ${PROJECT_NAME})
+  target_link_libraries(test_any_subscription_callback ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_client test_client.cpp)
 if(TARGET test_client)
-  ament_target_dependencies(test_client
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_client ${PROJECT_NAME} mimick)
+  target_link_libraries(test_client ${PROJECT_NAME} mimick ${rcl_interfaces_TARGETS} ${test_msgs_TARGETS})
+endif()
+ament_add_gtest(test_copy_all_parameter_values test_copy_all_parameter_values.cpp)
+if(TARGET test_copy_all_parameter_values)
+  target_link_libraries(test_copy_all_parameter_values ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_create_timer test_create_timer.cpp)
 if(TARGET test_create_timer)
-  ament_target_dependencies(test_create_timer
-    "rcl_interfaces"
-    "rmw"
-    "rcl"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_create_timer ${PROJECT_NAME})
  target_include_directories(test_create_timer PRIVATE ./)
 endif()
 ament_add_gtest(test_create_subscription test_create_subscription.cpp)
 if(TARGET test_create_subscription)
-  target_link_libraries(test_create_subscription ${PROJECT_NAME})
-  ament_target_dependencies(test_create_subscription
-    "test_msgs"
-  )
+  target_link_libraries(test_create_subscription ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 function(test_add_callback_groups_to_executor_for_rmw_implementation)
  set(rmw_implementation_env_var RMW_IMPLEMENTATION=${rmw_implementation})
@@ -107,32 +74,17 @@ function(test_add_callback_groups_to_executor_for_rmw_implementation)
    TIMEOUT 120
  )
  if(TARGET test_add_callback_groups_to_executor${target_suffix})
-    target_link_libraries(test_add_callback_groups_to_executor${target_suffix} ${PROJECT_NAME})
-    ament_target_dependencies(test_add_callback_groups_to_executor${target_suffix}
-      "test_msgs"
-    )
+    target_link_libraries(test_add_callback_groups_to_executor${target_suffix} ${PROJECT_NAME} ${test_msgs_TARGETS})
  endif()
 endfunction()
 call_for_each_rmw_implementation(test_add_callback_groups_to_executor_for_rmw_implementation)
 ament_add_gtest(test_expand_topic_or_service_name test_expand_topic_or_service_name.cpp)
 if(TARGET test_expand_topic_or_service_name)
-  ament_target_dependencies(test_expand_topic_or_service_name
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
-  target_link_libraries(test_expand_topic_or_service_name ${PROJECT_NAME} mimick)
+  target_link_libraries(test_expand_topic_or_service_name ${PROJECT_NAME} mimick rcl::rcl rmw::rmw)
 endif()
 ament_add_gtest(test_function_traits test_function_traits.cpp)
 if(TARGET test_function_traits)
-  target_include_directories(test_function_traits PUBLIC ../../include)
-  ament_target_dependencies(test_function_traits
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
+  target_link_libraries(test_function_traits ${PROJECT_NAME})
 endif()
 ament_add_gtest(
  test_future_return_code
@@ -142,72 +94,33 @@ if(TARGET test_future_return_code)
 endif()
 ament_add_gmock(test_intra_process_manager test_intra_process_manager.cpp)
 if(TARGET test_intra_process_manager)
-  ament_target_dependencies(test_intra_process_manager
-    "rcl"
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
-  target_link_libraries(test_intra_process_manager ${PROJECT_NAME})
+  target_link_libraries(test_intra_process_manager ${PROJECT_NAME} ${rcl_interfaces_TARGETS} rmw::rmw)
 endif()
 ament_add_gmock(test_intra_process_manager_with_allocators test_intra_process_manager_with_allocators.cpp)
 if(TARGET test_intra_process_manager_with_allocators)
-  ament_target_dependencies(test_intra_process_manager_with_allocators
-    "test_msgs"
-  )
-  target_link_libraries(test_intra_process_manager_with_allocators ${PROJECT_NAME})
+  target_link_libraries(test_intra_process_manager_with_allocators ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_ring_buffer_implementation test_ring_buffer_implementation.cpp)
 if(TARGET test_ring_buffer_implementation)
-  ament_target_dependencies(test_ring_buffer_implementation
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_ring_buffer_implementation ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_intra_process_buffer test_intra_process_buffer.cpp)
 if(TARGET test_intra_process_buffer)
-  ament_target_dependencies(test_intra_process_buffer
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_intra_process_buffer ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_loaned_message test_loaned_message.cpp)
-ament_target_dependencies(test_loaned_message
-  "test_msgs"
-)
-target_link_libraries(test_loaned_message ${PROJECT_NAME} mimick)
+target_link_libraries(test_loaned_message ${PROJECT_NAME} mimick ${test_msgs_TARGETS})

 ament_add_gtest(test_memory_strategy test_memory_strategy.cpp)
-ament_target_dependencies(test_memory_strategy
-  "test_msgs"
-)
-target_link_libraries(test_memory_strategy ${PROJECT_NAME})
+target_link_libraries(test_memory_strategy ${PROJECT_NAME} ${test_msgs_TARGETS})

 ament_add_gtest(test_message_memory_strategy test_message_memory_strategy.cpp)
-ament_target_dependencies(test_message_memory_strategy
-  "test_msgs"
-)
-target_link_libraries(test_message_memory_strategy ${PROJECT_NAME})
+target_link_libraries(test_message_memory_strategy ${PROJECT_NAME} ${test_msgs_TARGETS})

 ament_add_gtest(test_node test_node.cpp TIMEOUT 240)
 if(TARGET test_node)
-  ament_target_dependencies(test_node
-    "rcl_interfaces"
-    "rcpputils"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_node ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node ${PROJECT_NAME} mimick rcpputils::rcpputils rmw::rmw ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_node_interfaces__get_node_interfaces
@@ -218,7 +131,7 @@ endif()
 ament_add_gtest(test_node_interfaces__node_base
  node_interfaces/test_node_base.cpp)
 if(TARGET test_node_interfaces__node_base)
-  target_link_libraries(test_node_interfaces__node_base ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_base ${PROJECT_NAME} mimick rcl::rcl rmw::rmw)
 endif()
 ament_add_gtest(test_node_interfaces__node_clock
  node_interfaces/test_node_clock.cpp)
@@ -229,43 +142,42 @@ ament_add_gtest(test_node_interfaces__node_graph
  node_interfaces/test_node_graph.cpp
  TIMEOUT 120)
 if(TARGET test_node_interfaces__node_graph)
-  ament_target_dependencies(
-    test_node_interfaces__node_graph
-    "test_msgs")
-  target_link_libraries(test_node_interfaces__node_graph ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_graph ${PROJECT_NAME} mimick rcl::rcl ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_node_interfaces__node_interfaces
  node_interfaces/test_node_interfaces.cpp)
 if(TARGET test_node_interfaces__node_interfaces)
-  target_link_libraries(test_node_interfaces__node_interfaces ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_interfaces ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_node_interfaces__node_parameters
  node_interfaces/test_node_parameters.cpp)
 if(TARGET test_node_interfaces__node_parameters)
-  target_link_libraries(test_node_interfaces__node_parameters ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_parameters ${PROJECT_NAME} mimick rcpputils::rcpputils)
 endif()
 ament_add_gtest(test_node_interfaces__node_services
  node_interfaces/test_node_services.cpp)
 if(TARGET test_node_interfaces__node_services)
-  target_link_libraries(test_node_interfaces__node_services ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_services ${PROJECT_NAME} mimick rcl::rcl)
 endif()
 ament_add_gtest(test_node_interfaces__node_timers
  node_interfaces/test_node_timers.cpp)
 if(TARGET test_node_interfaces__node_timers)
-  target_link_libraries(test_node_interfaces__node_timers ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_timers ${PROJECT_NAME} mimick rcl::rcl)
 endif()
 ament_add_gtest(test_node_interfaces__node_topics
  node_interfaces/test_node_topics.cpp)
 if(TARGET test_node_interfaces__node_topics)
-  ament_target_dependencies(
-    test_node_interfaces__node_topics
-    "test_msgs")
-  target_link_libraries(test_node_interfaces__node_topics ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_topics ${PROJECT_NAME} mimick rcl::rcl ${test_msgs_TARGETS})
+endif()
+ament_add_gtest(test_node_interfaces__node_type_descriptions
+  node_interfaces/test_node_type_descriptions.cpp)
+if(TARGET test_node_interfaces__node_type_descriptions)
+  target_link_libraries(test_node_interfaces__node_type_descriptions ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_node_interfaces__node_waitables
  node_interfaces/test_node_waitables.cpp)
 if(TARGET test_node_interfaces__node_waitables)
-  target_link_libraries(test_node_interfaces__node_waitables ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_interfaces__node_waitables ${PROJECT_NAME} mimick rcl::rcl)
 endif()
 ament_add_gtest(test_node_interfaces__test_template_utils  # Compile time test
  node_interfaces/detail/test_template_utils.cpp)
@@ -296,82 +208,43 @@ endif()

 ament_add_gtest(test_node_global_args test_node_global_args.cpp)
 if(TARGET test_node_global_args)
-  ament_target_dependencies(test_node_global_args
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_node_global_args ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_node_options test_node_options.cpp)
 if(TARGET test_node_options)
-  ament_target_dependencies(test_node_options "rcl")
-  target_link_libraries(test_node_options ${PROJECT_NAME} mimick)
+  target_link_libraries(test_node_options ${PROJECT_NAME} mimick rcl::rcl)
 endif()
 ament_add_gtest(test_init_options test_init_options.cpp)
 if(TARGET test_init_options)
-  ament_target_dependencies(test_init_options "rcl")
-  target_link_libraries(test_init_options ${PROJECT_NAME} mimick)
+  target_link_libraries(test_init_options ${PROJECT_NAME} mimick rcl::rcl)
 endif()
 ament_add_gmock(test_parameter_client test_parameter_client.cpp)
 if(TARGET test_parameter_client)
-  ament_target_dependencies(test_parameter_client
-    "rcl_interfaces"
-  )
-  target_link_libraries(test_parameter_client ${PROJECT_NAME})
+  target_link_libraries(test_parameter_client ${PROJECT_NAME} ${rcl_interfaces_TARGETS})
 endif()
 ament_add_gtest(test_parameter_service test_parameter_service.cpp)
 if(TARGET test_parameter_service)
-  ament_target_dependencies(test_parameter_service
-    "rcl_interfaces"
-  )
  target_link_libraries(test_parameter_service ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_parameter_events_filter test_parameter_events_filter.cpp)
 if(TARGET test_parameter_events_filter)
-  ament_target_dependencies(test_parameter_events_filter
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
-  target_link_libraries(test_parameter_events_filter ${PROJECT_NAME})
+  target_link_libraries(test_parameter_events_filter ${PROJECT_NAME} ${rcl_interfaces_TARGETS})
 endif()
 ament_add_gtest(test_parameter test_parameter.cpp)
 if(TARGET test_parameter)
-  ament_target_dependencies(test_parameter
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_parameter ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_parameter_event_handler test_parameter_event_handler.cpp)
 if(TARGET test_parameter_event_handler)
-  ament_target_dependencies(test_parameter_event_handler
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_generator_cpp"
-    "rosidl_typesupport_cpp"
-  )
  target_link_libraries(test_parameter_event_handler ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_parameter_map test_parameter_map.cpp)
 if(TARGET test_parameter_map)
-  target_link_libraries(test_parameter_map ${PROJECT_NAME})
+  target_link_libraries(test_parameter_map ${PROJECT_NAME} rcl::rcl rcl_yaml_param_parser::rcl_yaml_param_parser rcutils::rcutils)
 endif()
 ament_add_gtest(test_publisher test_publisher.cpp TIMEOUT 120)
 if(TARGET test_publisher)
-  ament_target_dependencies(test_publisher
-    "rcl"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_publisher ${PROJECT_NAME} mimick)
+  target_link_libraries(test_publisher ${PROJECT_NAME} mimick rcl::rcl rcutils::rcutils ${test_msgs_TARGETS})
 endif()

 set(append_library_dirs "${CMAKE_CURRENT_BINARY_DIR}")
@@ -401,32 +274,22 @@ ament_add_gtest(test_subscription_publisher_with_same_type_adapter test_subscrip
  APPEND_LIBRARY_DIRS "${append_library_dirs}"
 )
 if(TARGET test_subscription_publisher_with_same_type_adapter)
-  ament_target_dependencies(test_subscription_publisher_with_same_type_adapter
-    "statistics_msgs"
-  )
  target_link_libraries(test_subscription_publisher_with_same_type_adapter
    ${PROJECT_NAME}
-    ${cpp_typesupport_target})
+    ${cpp_typesupport_target}
+    ${statistics_msgs_TARGETS}
+  )
 endif()

 ament_add_gtest(test_publisher_subscription_count_api test_publisher_subscription_count_api.cpp)
 if(TARGET test_publisher_subscription_count_api)
-  ament_target_dependencies(test_publisher_subscription_count_api
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_publisher_subscription_count_api ${PROJECT_NAME})
+  target_link_libraries(test_publisher_subscription_count_api ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_qos test_qos.cpp)
 if(TARGET test_qos)
-  ament_target_dependencies(test_qos
-    "rmw"
-  )
  target_link_libraries(test_qos
    ${PROJECT_NAME}
+    rmw::rmw
  )
 endif()
 function(test_generic_pubsub_for_rmw_implementation)
@@ -435,12 +298,7 @@ function(test_generic_pubsub_for_rmw_implementation)
    ENV ${rmw_implementation_env_var}
  )
  if(TARGET test_generic_pubsub${target_suffix})
-    target_link_libraries(test_generic_pubsub${target_suffix} ${PROJECT_NAME})
-    ament_target_dependencies(test_generic_pubsub${target_suffix}
-      "rcpputils"
-      "rosidl_typesupport_cpp"
-      "test_msgs"
-    )
+    target_link_libraries(test_generic_pubsub${target_suffix} ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})
  endif()
 endfunction()
 call_for_each_rmw_implementation(test_generic_pubsub_for_rmw_implementation)
@@ -451,153 +309,79 @@ function(test_qos_event_for_rmw_implementation)
    ENV ${rmw_implementation_env_var}
  )
  if(TARGET test_qos_event${target_suffix})
-    target_link_libraries(test_qos_event${target_suffix} ${PROJECT_NAME} mimick)
-    ament_target_dependencies(test_qos_event${target_suffix}
-      "rmw"
-      "rosidl_typesupport_cpp"
-      "test_msgs"
-    )
+    target_link_libraries(test_qos_event${target_suffix} ${PROJECT_NAME} mimick rcutils::rcutils rmw::rmw ${test_msgs_TARGETS})
  endif()
 endfunction()
 call_for_each_rmw_implementation(test_qos_event_for_rmw_implementation)

 ament_add_gmock(test_qos_overriding_options test_qos_overriding_options.cpp)
 if(TARGET test_qos_overriding_options)
-  target_link_libraries(test_qos_overriding_options
-    ${PROJECT_NAME}
-  )
+  target_link_libraries(test_qos_overriding_options ${PROJECT_NAME})
 endif()
 ament_add_gmock(test_qos_parameters test_qos_parameters.cpp)
 if(TARGET test_qos_parameters)
-  target_link_libraries(test_qos_parameters
-    ${PROJECT_NAME}
-  )
+  target_link_libraries(test_qos_parameters ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_rate test_rate.cpp)
 if(TARGET test_rate)
-  ament_target_dependencies(test_rate
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-  )
-  target_link_libraries(test_rate
-    ${PROJECT_NAME}
-  )
+  target_link_libraries(test_rate ${PROJECT_NAME})
 endif()
 ament_add_gtest(test_serialized_message_allocator test_serialized_message_allocator.cpp)
 if(TARGET test_serialized_message_allocator)
-  ament_target_dependencies(test_serialized_message_allocator
-    test_msgs
-  )
-  target_link_libraries(test_serialized_message_allocator
-    ${PROJECT_NAME}
-  )
+  target_link_libraries(test_serialized_message_allocator ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_serialized_message test_serialized_message.cpp)
 if(TARGET test_serialized_message)
-  ament_target_dependencies(test_serialized_message
-    test_msgs
-  )
-  target_link_libraries(test_serialized_message
-    ${PROJECT_NAME}
-  )
+  target_link_libraries(test_serialized_message ${PROJECT_NAME} rcpputils::rcpputils ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_service test_service.cpp)
 if(TARGET test_service)
-  ament_target_dependencies(test_service
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_service ${PROJECT_NAME} mimick)
+  target_link_libraries(test_service ${PROJECT_NAME} mimick ${rcl_interfaces_TARGES} ${test_msgs_TARGETS})
 endif()
 ament_add_gmock(test_service_introspection test_service_introspection.cpp)
-if(TARGET test_service)
-  ament_target_dependencies(test_service_introspection
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_service_introspection ${PROJECT_NAME} mimick)
+if(TARGET test_service_introspection)
+  target_link_libraries(test_service_introspection ${PROJECT_NAME} mimick ${service_msgs_TARGETS} ${test_msgs_TARGETS})
 endif()
 # Creating and destroying nodes is slow with Connext, so this needs larger timeout.
 ament_add_gtest(test_subscription test_subscription.cpp TIMEOUT 120)
 if(TARGET test_subscription)
-  ament_target_dependencies(test_subscription
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_subscription ${PROJECT_NAME} mimick)
+  target_link_libraries(test_subscription ${PROJECT_NAME} mimick ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_subscription_publisher_count_api test_subscription_publisher_count_api.cpp)
 if(TARGET test_subscription_publisher_count_api)
-  ament_target_dependencies(test_subscription_publisher_count_api
-    "rcl_interfaces"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_subscription_publisher_count_api ${PROJECT_NAME})
+  target_link_libraries(test_subscription_publisher_count_api ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_subscription_traits test_subscription_traits.cpp)
 if(TARGET test_subscription_traits)
-  ament_target_dependencies(test_subscription_traits
-    "rcl"
-    "test_msgs"
-  )
-  target_link_libraries(test_subscription_traits ${PROJECT_NAME})
+  target_link_libraries(test_subscription_traits ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})
 endif()
 ament_add_gtest(test_type_support test_type_support.cpp)
 if(TARGET test_type_support)
-  ament_target_dependencies(test_type_support
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "test_msgs"
-  )
-  target_link_libraries(test_type_support ${PROJECT_NAME})
+  target_link_libraries(test_type_support ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 ament_add_gmock(test_typesupport_helpers test_typesupport_helpers.cpp)
 if(TARGET test_typesupport_helpers)
-  target_link_libraries(test_typesupport_helpers ${PROJECT_NAME})
+  target_link_libraries(test_typesupport_helpers ${PROJECT_NAME} rcpputils::rcpputils)
 endif()
 ament_add_gtest(test_find_weak_nodes test_find_weak_nodes.cpp)
 if(TARGET test_find_weak_nodes)
-  ament_target_dependencies(test_find_weak_nodes
-    "rcl"
-  )
  target_link_libraries(test_find_weak_nodes ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_externally_defined_services test_externally_defined_services.cpp)
-ament_target_dependencies(test_externally_defined_services
-  "rcl"
-  "test_msgs"
-)
-target_link_libraries(test_externally_defined_services ${PROJECT_NAME})
+target_link_libraries(test_externally_defined_services ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})

 ament_add_gtest(test_duration test_duration.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_duration)
-  ament_target_dependencies(test_duration
-    "rcl")
-  target_link_libraries(test_duration ${PROJECT_NAME})
+  target_link_libraries(test_duration ${PROJECT_NAME} rcl::rcl)
 endif()

 ament_add_gtest(test_logger test_logger.cpp)
-target_link_libraries(test_logger ${PROJECT_NAME})
+target_link_libraries(test_logger ${PROJECT_NAME} rcutils::rcutils)

 ament_add_gmock(test_logging test_logging.cpp)
-target_link_libraries(test_logging ${PROJECT_NAME})
+target_link_libraries(test_logging ${PROJECT_NAME} rcutils::rcutils)

 ament_add_gmock(test_context test_context.cpp)
 target_link_libraries(test_context ${PROJECT_NAME})
@@ -605,127 +389,96 @@ target_link_libraries(test_context ${PROJECT_NAME})
 ament_add_gtest(test_time test_time.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_time)
-  ament_target_dependencies(test_time
-    "rcl")
-  target_link_libraries(test_time ${PROJECT_NAME})
+  target_link_libraries(test_time ${PROJECT_NAME} rcl::rcl rcutils::rcutils)
 endif()

 ament_add_gtest(test_timer test_timer.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_timer)
-  ament_target_dependencies(test_timer
-    "rcl")
-  target_link_libraries(test_timer ${PROJECT_NAME} mimick)
+  target_link_libraries(test_timer ${PROJECT_NAME} mimick rcl::rcl)
 endif()

 ament_add_gtest(test_timers_manager test_timers_manager.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_timers_manager)
-  ament_target_dependencies(test_timers_manager
-    "rcl")
-  target_link_libraries(test_timers_manager ${PROJECT_NAME} mimick)
+  target_link_libraries(test_timers_manager ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_time_source test_time_source.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_time_source)
-  ament_target_dependencies(test_time_source
-    "rcl")
-  target_link_libraries(test_time_source ${PROJECT_NAME})
+  target_link_libraries(test_time_source ${PROJECT_NAME} rcl::rcl)
 endif()

 ament_add_gtest(test_utilities test_utilities.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_utilities)
-  ament_target_dependencies(test_utilities
-    "rcl")
-  target_link_libraries(test_utilities ${PROJECT_NAME} mimick)
+  target_link_libraries(test_utilities ${PROJECT_NAME} mimick rcl::rcl)
 endif()

 ament_add_gtest(test_wait_for_message test_wait_for_message.cpp)
 if(TARGET test_wait_for_message)
-  ament_target_dependencies(test_wait_for_message
-    "test_msgs")
-  target_link_libraries(test_wait_for_message ${PROJECT_NAME})
+  target_link_libraries(test_wait_for_message ${PROJECT_NAME} ${test_msgs_TARGETS})
+endif()
+
+ament_add_gtest(test_logger_service test_logger_service.cpp)
+if(TARGET test_logger_service)
+  target_link_libraries(test_logger_service ${PROJECT_NAME} ${rcl_interfaces_TARGETS})
 endif()

 ament_add_gtest(test_interface_traits test_interface_traits.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_interface_traits)
-  ament_target_dependencies(test_interface_traits
-    "rcl")
  target_link_libraries(test_interface_traits ${PROJECT_NAME})
 endif()

-# TODO(brawner) remove when destroying Node for Connext is resolved. See:
-# https://github.com/ros2/rclcpp/issues/1250
 ament_add_gtest(
  test_executors
  executors/test_executors.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}"
  TIMEOUT 180)
 if(TARGET test_executors)
-  ament_target_dependencies(test_executors
-    "rcl"
-    "test_msgs")
-  target_link_libraries(test_executors ${PROJECT_NAME})
+  target_link_libraries(test_executors ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_static_single_threaded_executor executors/test_static_single_threaded_executor.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_static_single_threaded_executor)
-  ament_target_dependencies(test_static_single_threaded_executor
-    "test_msgs")
-  target_link_libraries(test_static_single_threaded_executor ${PROJECT_NAME} mimick)
+  target_link_libraries(test_static_single_threaded_executor ${PROJECT_NAME} mimick ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_multi_threaded_executor executors/test_multi_threaded_executor.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_multi_threaded_executor)
-  ament_target_dependencies(test_multi_threaded_executor
-    "rcl")
  target_link_libraries(test_multi_threaded_executor ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_static_executor_entities_collector executors/test_static_executor_entities_collector.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}" TIMEOUT 120)
 if(TARGET test_static_executor_entities_collector)
-  ament_target_dependencies(test_static_executor_entities_collector
-    "rcl"
-    "test_msgs")
-  target_link_libraries(test_static_executor_entities_collector ${PROJECT_NAME} mimick)
+  target_link_libraries(test_static_executor_entities_collector ${PROJECT_NAME} mimick rcpputils::rcpputils ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_entities_collector executors/test_entities_collector.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}" TIMEOUT 120)
 if(TARGET test_entities_collector)
-  ament_target_dependencies(test_entities_collector
-    "rcl"
-    "test_msgs")
-  target_link_libraries(test_entities_collector ${PROJECT_NAME} mimick)
+  target_link_libraries(test_entities_collector ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_executor_notify_waitable executors/test_executor_notify_waitable.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}" TIMEOUT 120)
 if(TARGET test_executor_notify_waitable)
-  ament_target_dependencies(test_executor_notify_waitable
-    "rcl"
-    "test_msgs")
-  target_link_libraries(test_executor_notify_waitable ${PROJECT_NAME} mimick)
+  target_link_libraries(test_executor_notify_waitable ${PROJECT_NAME} mimick rcpputils::rcpputils)
 endif()

 ament_add_gtest(test_events_executor executors/test_events_executor.cpp TIMEOUT 5)
 if(TARGET test_events_executor)
-  ament_target_dependencies(test_events_executor
-    "test_msgs")
-  target_link_libraries(test_events_executor ${PROJECT_NAME})
+  target_link_libraries(test_events_executor ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_events_queue executors/test_events_queue.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_events_queue)
-  ament_target_dependencies(test_events_queue
-    "test_msgs")
  target_link_libraries(test_events_queue ${PROJECT_NAME})
 endif()

@@ -738,76 +491,60 @@ endif()
 ament_add_gtest(test_wait_set test_wait_set.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}")
 if(TARGET test_wait_set)
-  ament_target_dependencies(test_wait_set "test_msgs")
-  target_link_libraries(test_wait_set ${PROJECT_NAME})
+  target_link_libraries(test_wait_set ${PROJECT_NAME} ${rcl_interfaces_TARGETS} ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_subscription_topic_statistics topic_statistics/test_subscription_topic_statistics.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}"
 )
 if(TARGET test_subscription_topic_statistics)
-  ament_target_dependencies(test_subscription_topic_statistics
-    "builtin_interfaces"
-    "libstatistics_collector"
-    "rcl_interfaces"
-    "rcutils"
-    "rmw"
-    "rosidl_runtime_cpp"
-    "rosidl_typesupport_cpp"
-    "statistics_msgs"
-    "test_msgs")
  target_link_libraries(test_subscription_topic_statistics
    ${PROJECT_NAME}
-    ${cpp_typesupport_target})
+    libstatistics_collector::libstatistics_collector
+    ${statistics_msgs_TARGETS}
+    ${test_msgs_TARGETS}
+  )
 endif()

 ament_add_gtest(test_subscription_options test_subscription_options.cpp)
 if(TARGET test_subscription_options)
-  ament_target_dependencies(test_subscription_options "rcl")
  target_link_libraries(test_subscription_options ${PROJECT_NAME})
 endif()

 ament_add_gtest(test_dynamic_storage wait_set_policies/test_dynamic_storage.cpp)
 if(TARGET test_dynamic_storage)
-  ament_target_dependencies(test_dynamic_storage "rcl" "test_msgs")
-  target_link_libraries(test_dynamic_storage ${PROJECT_NAME})
+  target_link_libraries(test_dynamic_storage ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_storage_policy_common wait_set_policies/test_storage_policy_common.cpp)
 if(TARGET test_storage_policy_common)
-  ament_target_dependencies(test_storage_policy_common "rcl" "test_msgs")
-  target_link_libraries(test_storage_policy_common ${PROJECT_NAME} mimick)
+  target_link_libraries(test_storage_policy_common ${PROJECT_NAME} mimick ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_static_storage wait_set_policies/test_static_storage.cpp)
 if(TARGET test_static_storage)
-  ament_target_dependencies(test_static_storage "rcl" "test_msgs")
-  target_link_libraries(test_static_storage ${PROJECT_NAME})
+  target_link_libraries(test_static_storage ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_thread_safe_synchronization wait_set_policies/test_thread_safe_synchronization.cpp)
 if(TARGET test_thread_safe_synchronization)
-  ament_target_dependencies(test_thread_safe_synchronization "rcl" "test_msgs")
-  target_link_libraries(test_thread_safe_synchronization ${PROJECT_NAME})
+  target_link_libraries(test_thread_safe_synchronization ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()

 ament_add_gtest(test_rosout_qos test_rosout_qos.cpp)
 if(TARGET test_rosout_qos)
-  ament_target_dependencies(test_rosout_qos "rcl")
-  target_link_libraries(test_rosout_qos ${PROJECT_NAME})
+  target_link_libraries(test_rosout_qos ${PROJECT_NAME} rcl::rcl rmw::rmw)
 endif()

 ament_add_gtest(test_rosout_subscription test_rosout_subscription.cpp)
 if(TARGET test_rosout_subscription)
-  ament_target_dependencies(test_rosout_subscription "rcl")
-  target_link_libraries(test_rosout_subscription ${PROJECT_NAME})
+  target_link_libraries(test_rosout_subscription ${PROJECT_NAME} ${rcl_interfaces_TARGETS})
 endif()

 ament_add_gtest(test_executor test_executor.cpp
  APPEND_LIBRARY_DIRS "${append_library_dirs}"
  TIMEOUT 120)
 if(TARGET test_executor)
-  ament_target_dependencies(test_executor "rcl")
  target_link_libraries(test_executor ${PROJECT_NAME} mimick)
 endif()

@@ -824,12 +561,7 @@ function(test_subscription_content_filter_for_rmw_implementation)
    TIMEOUT 120
  )
  if(TARGET test_subscription_content_filter${target_suffix})
-    target_link_libraries(test_subscription_content_filter${target_suffix} ${PROJECT_NAME} mimick)
-    ament_target_dependencies(test_subscription_content_filter${target_suffix}
-      "rcpputils"
-      "rosidl_typesupport_cpp"
-      "test_msgs"
-    )
+    target_link_libraries(test_subscription_content_filter${target_suffix} ${PROJECT_NAME} mimick ${test_msgs_TARGETS})
  endif()
 endfunction()
 call_for_each_rmw_implementation(test_subscription_content_filter_for_rmw_implementation)
--- a/rclcpp/test/rclcpp/executors/test_events_executor.cpp
+++ b/rclcpp/test/rclcpp/executors/test_events_executor.cpp
@@ -43,11 +43,6 @@ public:

 TEST_F(TestEventsExecutor, run_pub_sub)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  bool msg_received = false;
@@ -65,7 +60,7 @@ TEST_F(TestEventsExecutor, run_pub_sub)
  executor.add_node(node);

  bool spin_exited = false;
-  std::thread spinner([&spin_exited, &executor, this]() {
+  std::thread spinner([&spin_exited, &executor]() {
      executor.spin();
      spin_exited = true;
    });
@@ -80,8 +75,6 @@ TEST_F(TestEventsExecutor, run_pub_sub)
    !spin_exited &&
    (std::chrono::high_resolution_clock::now() - start < 1s))
  {
-    auto time = std::chrono::high_resolution_clock::now() - start;
-    auto time_msec = std::chrono::duration_cast<std::chrono::milliseconds>(time);
    std::this_thread::sleep_for(25ms);
  }

@@ -95,11 +88,6 @@ TEST_F(TestEventsExecutor, run_pub_sub)

 TEST_F(TestEventsExecutor, run_clients_servers)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  bool request_received = false;
@@ -119,7 +107,7 @@ TEST_F(TestEventsExecutor, run_clients_servers)
  executor.add_node(node);

  bool spin_exited = false;
-  std::thread spinner([&spin_exited, &executor, this]() {
+  std::thread spinner([&spin_exited, &executor]() {
      executor.spin();
      spin_exited = true;
    });
@@ -153,11 +141,6 @@ TEST_F(TestEventsExecutor, run_clients_servers)

 TEST_F(TestEventsExecutor, spin_once_max_duration_timeout)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  EventsExecutor executor;
@@ -190,11 +173,6 @@ TEST_F(TestEventsExecutor, spin_once_max_duration_timeout)

 TEST_F(TestEventsExecutor, spin_once_max_duration_timer)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  EventsExecutor executor;
@@ -226,11 +204,6 @@ TEST_F(TestEventsExecutor, spin_once_max_duration_timer)

 TEST_F(TestEventsExecutor, spin_some_max_duration)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  {
    auto node = std::make_shared<rclcpp::Node>("node");

@@ -277,11 +250,6 @@ TEST_F(TestEventsExecutor, spin_some_max_duration)

 TEST_F(TestEventsExecutor, spin_some_zero_duration)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  size_t t_runs = 0;
@@ -303,11 +271,6 @@ TEST_F(TestEventsExecutor, spin_some_zero_duration)

 TEST_F(TestEventsExecutor, spin_all_max_duration)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  {
    auto node = std::make_shared<rclcpp::Node>("node");

@@ -358,11 +321,6 @@ TEST_F(TestEventsExecutor, spin_all_max_duration)

 TEST_F(TestEventsExecutor, cancel_while_timers_running)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  EventsExecutor executor;
@@ -388,7 +346,7 @@ TEST_F(TestEventsExecutor, cancel_while_timers_running)
    });


-  std::thread spinner([&executor, this]() {executor.spin();});
+  std::thread spinner([&executor]() {executor.spin();});

  std::this_thread::sleep_for(10ms);
  // Call cancel while t1 callback is still being executed
@@ -402,11 +360,6 @@ TEST_F(TestEventsExecutor, cancel_while_timers_running)

 TEST_F(TestEventsExecutor, cancel_while_timers_waiting)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");

  size_t t1_runs = 0;
@@ -420,7 +373,7 @@ TEST_F(TestEventsExecutor, cancel_while_timers_waiting)
  executor.add_node(node);

  auto start = std::chrono::steady_clock::now();
-  std::thread spinner([&executor, this]() {executor.spin();});
+  std::thread spinner([&executor]() {executor.spin();});

  std::this_thread::sleep_for(10ms);
  executor.cancel();
@@ -435,11 +388,6 @@ TEST_F(TestEventsExecutor, destroy_entities)
  // This test fails on Windows! We skip it for now
  GTEST_SKIP();

-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  // Create a publisher node and start publishing messages
  auto node_pub = std::make_shared<rclcpp::Node>("node_pub");
  auto publisher = node_pub->create_publisher<test_msgs::msg::Empty>("topic", rclcpp::QoS(10));
@@ -447,7 +395,7 @@ TEST_F(TestEventsExecutor, destroy_entities)
    2ms, [&]() {publisher->publish(std::make_unique<test_msgs::msg::Empty>());});
  EventsExecutor executor_pub;
  executor_pub.add_node(node_pub);
-  std::thread spinner([&executor_pub, this]() {executor_pub.spin();});
+  std::thread spinner([&executor_pub]() {executor_pub.spin();});

  // Create a node with two different subscriptions to the topic
  auto node_sub = std::make_shared<rclcpp::Node>("node_sub");
@@ -485,11 +433,6 @@ std::string * g_sub_log_msg;
 std::promise<void> * g_log_msgs_promise;
 TEST_F(TestEventsExecutor, test_default_incompatible_qos_callbacks)
 {
-  // rmw_connextdds doesn't support events-executor
-  if (std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0) {
-    GTEST_SKIP();
-  }
-
  auto node = std::make_shared<rclcpp::Node>("node");
  rcutils_logging_output_handler_t original_output_handler = rcutils_logging_get_output_handler();

--- a/rclcpp/test/rclcpp/executors/test_executors.cpp
+++ b/rclcpp/test/rclcpp/executors/test_executors.cpp
@@ -20,12 +20,14 @@
 #include <gtest/gtest.h>

 #include <algorithm>
+#include <atomic>
 #include <chrono>
 #include <limits>
 #include <memory>
 #include <string>
 #include <thread>
 #include <utility>
+#include <vector>

 #include "rcl/error_handling.h"
 #include "rcl/time.h"
@@ -43,18 +45,10 @@ template<typename T>
 class TestExecutors : public ::testing::Test
 {
 public:
-  static void SetUpTestCase()
-  {
-    rclcpp::init(0, nullptr);
-  }
-
-  static void TearDownTestCase()
-  {
-    rclcpp::shutdown();
-  }
-
  void SetUp()
  {
+    rclcpp::init(0, nullptr);
+
    const auto test_info = ::testing::UnitTest::GetInstance()->current_test_info();
    std::stringstream test_name;
    test_name << test_info->test_case_name() << "_" << test_info->name();
@@ -75,6 +69,8 @@ public:
    publisher.reset();
    subscription.reset();
    node.reset();
+
+    rclcpp::shutdown();
  }

  rclcpp::Node::SharedPtr node;
@@ -139,14 +135,6 @@ TYPED_TEST_SUITE(TestExecutorsStable, StandardExecutors, ExecutorTypeNames);
 TYPED_TEST(TestExecutors, detachOnDestruction)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  {
    ExecutorType executor;
    executor.add_node(this->node);
@@ -163,14 +151,6 @@ TYPED_TEST(TestExecutors, detachOnDestruction)
 TYPED_TEST(TestExecutorsStable, addTemporaryNode)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;

  {
@@ -191,14 +171,6 @@ TYPED_TEST(TestExecutorsStable, addTemporaryNode)
 TYPED_TEST(TestExecutors, emptyExecutor)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  std::thread spinner([&]() {EXPECT_NO_THROW(executor.spin());});
  std::this_thread::sleep_for(50ms);
@@ -210,14 +182,6 @@ TYPED_TEST(TestExecutors, emptyExecutor)
 TYPED_TEST(TestExecutors, addNodeTwoExecutors)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor1;
  ExecutorType executor2;
  EXPECT_NO_THROW(executor1.add_node(this->node));
@@ -229,14 +193,6 @@ TYPED_TEST(TestExecutors, addNodeTwoExecutors)
 TYPED_TEST(TestExecutors, spinWithTimer)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;

  bool timer_completed = false;
@@ -260,14 +216,6 @@ TYPED_TEST(TestExecutors, spinWithTimer)
 TYPED_TEST(TestExecutors, spinWhileAlreadySpinning)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -295,14 +243,6 @@ TYPED_TEST(TestExecutors, spinWhileAlreadySpinning)
 TYPED_TEST(TestExecutors, testSpinUntilFutureComplete)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -326,14 +266,6 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureComplete)
 TYPED_TEST(TestExecutors, testSpinUntilSharedFutureComplete)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -358,14 +290,6 @@ TYPED_TEST(TestExecutors, testSpinUntilSharedFutureComplete)
 TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteNoTimeout)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -413,14 +337,6 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteNoTimeout)
 TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteWithTimeout)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -533,14 +449,6 @@ private:
 TYPED_TEST(TestExecutors, spinAll)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  auto waitable_interfaces = this->node->get_node_waitables_interface();
  auto my_waitable = std::make_shared<TestWaitable>();
@@ -583,14 +491,6 @@ TYPED_TEST(TestExecutors, spinAll)
 TYPED_TEST(TestExecutors, spinSome)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  auto waitable_interfaces = this->node->get_node_waitables_interface();
  auto my_waitable = std::make_shared<TestWaitable>();
@@ -633,14 +533,6 @@ TYPED_TEST(TestExecutors, spinSome)
 TYPED_TEST(TestExecutors, testSpinNodeUntilFutureCompleteNodeBasePtr)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;

  std::promise<bool> promise;
@@ -657,14 +549,6 @@ TYPED_TEST(TestExecutors, testSpinNodeUntilFutureCompleteNodeBasePtr)
 TYPED_TEST(TestExecutors, testSpinNodeUntilFutureCompleteNodePtr)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;

  std::promise<bool> promise;
@@ -681,14 +565,6 @@ TYPED_TEST(TestExecutors, testSpinNodeUntilFutureCompleteNodePtr)
 TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteInterrupted)
 {
  using ExecutorType = TypeParam;
-  // rmw_connextdds doesn't support events-executor
-  if (
-    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
-    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
-  {
-    GTEST_SKIP();
-  }
-
  ExecutorType executor;
  executor.add_node(this->node);

@@ -729,6 +605,78 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteInterrupted)
  spinner.join();
 }

+// This test verifies that the add_node operation is robust wrt race conditions.
+// It's mostly meant to prevent regressions in the events-executor, but the operation should be
+// thread-safe in all executor implementations.
+// The initial implementation of the events-executor contained a bug where the executor
+// would end up in an inconsistent state and stop processing interrupt/shutdown notifications.
+// Manually adding a node to the executor results in a) producing a notify waitable event
+// and b) refreshing the executor collections.
+// The inconsistent state would happen if the event was processed before the collections were
+// finished to be refreshed: the executor would pick up the event but be unable to process it.
+// This would leave the `notify_waitable_event_pushed_` flag to true, preventing additional
+// notify waitable events to be pushed.
+// The behavior is observable only under heavy load, so this test spawns several worker
+// threads. Due to the nature of the bug, this test may still succeed even if the
+// bug is present. However repeated runs will show its flakiness nature and indicate
+// an eventual regression.
+TYPED_TEST(TestExecutors, testRaceConditionAddNode)
+{
+  using ExecutorType = TypeParam;
+  // rmw_connextdds doesn't support events-executor
+  if (
+    std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>() &&
+    std::string(rmw_get_implementation_identifier()).find("rmw_connextdds") == 0)
+  {
+    GTEST_SKIP();
+  }
+
+  // Spawn some threads to do some heavy work
+  std::atomic<bool> should_cancel = false;
+  std::vector<std::thread> stress_threads;
+  for (size_t i = 0; i < 5 * std::thread::hardware_concurrency(); i++) {
+    stress_threads.emplace_back(
+      [&should_cancel, i]() {
+        // This is just some arbitrary heavy work
+        volatile size_t total = 0;
+        for (size_t k = 0; k < 549528914167; k++) {
+          if (should_cancel) {
+            break;
+          }
+          total += k * (i + 42);
+          (void)total;
+        }
+      });
+  }
+
+  // Create an executor
+  auto executor = std::make_shared<ExecutorType>();
+  // Start spinning
+  auto executor_thread = std::thread(
+    [executor]() {
+      executor->spin();
+    });
+  // Add a node to the executor
+  executor->add_node(this->node);
+
+  // Cancel the executor (make sure that it's already spinning first)
+  while (!executor->is_spinning() && rclcpp::ok()) {
+    continue;
+  }
+  executor->cancel();
+
+  // Try to join the thread after cancelling the executor
+  // This is the "test". We want to make sure that we can still cancel the executor
+  // regardless of the presence of race conditions
+  executor_thread.join();
+
+  // The test is now completed: we can join the stress threads
+  should_cancel = true;
+  for (auto & t : stress_threads) {
+    t.join();
+  }
+}
+
 // Check spin_until_future_complete with node base pointer (instantiates its own executor)
 TEST(TestExecutors, testSpinUntilFutureCompleteNodeBasePtr)
 {
@@ -791,7 +739,7 @@ public:
    test_name << test_info->test_case_name() << "_" << test_info->name();
    node = std::make_shared<rclcpp::Node>("node", test_name.str());

-    callback_count = 0;
+    callback_count = 0u;

    const std::string topic_name = std::string("topic_") + test_name.str();

@@ -800,7 +748,7 @@ public:
    publisher = node->create_publisher<test_msgs::msg::Empty>(topic_name, rclcpp::QoS(1), po);

    auto callback = [this](test_msgs::msg::Empty::ConstSharedPtr) {
-        this->callback_count.fetch_add(1);
+        this->callback_count.fetch_add(1u);
      };

    rclcpp::SubscriptionOptions so;
@@ -822,7 +770,7 @@ public:
  rclcpp::Node::SharedPtr node;
  rclcpp::Publisher<test_msgs::msg::Empty>::SharedPtr publisher;
  rclcpp::Subscription<test_msgs::msg::Empty>::SharedPtr subscription;
-  std::atomic_int callback_count;
+  std::atomic_size_t callback_count;
 };

 TYPED_TEST_SUITE(TestIntraprocessExecutors, ExecutorTypes, ExecutorTypeNames);
@@ -832,13 +780,13 @@ TYPED_TEST(TestIntraprocessExecutors, testIntraprocessRetrigger) {
  // that publishers aren't continuing to publish.
  // This was previously broken in that intraprocess guard conditions were only triggered on
  // publish and the test was added to prevent future regressions.
-  const size_t kNumMessages = 100;
+  static constexpr size_t kNumMessages = 100;

  using ExecutorType = TypeParam;
  ExecutorType executor;
  executor.add_node(this->node);

-  EXPECT_EQ(0, this->callback_count.load());
+  EXPECT_EQ(0u, this->callback_count.load());
  this->publisher->publish(test_msgs::msg::Empty());

  // Wait for up to 5 seconds for the first message to come available.
@@ -852,7 +800,7 @@ TYPED_TEST(TestIntraprocessExecutors, testIntraprocessRetrigger) {
  EXPECT_EQ(1u, this->callback_count.load());

  // reset counter
-  this->callback_count.store(0);
+  this->callback_count.store(0u);

  for (size_t ii = 0; ii < kNumMessages; ++ii) {
    this->publisher->publish(test_msgs::msg::Empty());
@@ -861,11 +809,9 @@ TYPED_TEST(TestIntraprocessExecutors, testIntraprocessRetrigger) {
  // Fire a timer every 10ms up to 5 seconds waiting for subscriptions to be read.
  loops = 0;
  auto timer = this->node->create_wall_timer(
-    std::chrono::milliseconds(10), [this, &executor, &loops, &kNumMessages]() {
+    std::chrono::milliseconds(10), [this, &executor, &loops]() {
      loops++;
-      if (kNumMessages == this->callback_count.load() ||
-      loops == 500)
-      {
+      if (kNumMessages == this->callback_count.load() || loops == 500) {
        executor.cancel();
      }
    });
--- a/Show More
+++ b/Show More