28.1.10

Signed-off-by: Alejandro Hernandez Cordero <ahcorde@gmail.com>

.github/workflows/mirror-rolling-to-master.yaml

@@ -1,13 +0,0 @@
-name: Mirror rolling to master
-
-on:
-  push:
-    branches: [ rolling ]
-
-jobs:
-  mirror-to-master:
-    runs-on: ubuntu-latest
-    steps:
-    - uses: zofrex/mirror-branch@v1
-      with:
-        target-branch: master

CODEOWNERS

@@ -1,2 +0,0 @@
-# This file was generated by https://github.com/audrow/update-ros2-repos
-* @ivanpauno @hidmic @wjwwood

rclcpp/CHANGELOG.rst

@@ -2,6 +2,216 @@
 Changelog for package rclcpp
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+28.1.10 (2025-06-23)
+--------------------
+* Fix for memory leaks in rclcpp::SerializedMessage (`#2861 <https://github.com/ros2/rclcpp/issues/2861>`_) (`#2864 <https://github.com/ros2/rclcpp/issues/2864>`_)
+* Added missing chrono includes (`#2854 <https://github.com/ros2/rclcpp/issues/2854>`_) (`#2856 <https://github.com/ros2/rclcpp/issues/2856>`_)
+* get_all_data_impl() does not handle null pointers properly, causing segmentation fault (backport `#2840 <https://github.com/ros2/rclcpp/issues/2840>`_) (`#2851 <https://github.com/ros2/rclcpp/issues/2851>`_)
+* QoSInitialization::from_rmw does not validate invalid history policy values, leading to silent failures (`#2841 <https://github.com/ros2/rclcpp/issues/2841>`_) (`#2845 <https://github.com/ros2/rclcpp/issues/2845>`_)
+* throws std::invalid_argument if ParameterEvent is NULL. (`#2814 <https://github.com/ros2/rclcpp/issues/2814>`_) (`#2825 <https://github.com/ros2/rclcpp/issues/2825>`_)
+* Merge pull request `#2821 <https://github.com/ros2/rclcpp/issues/2821>`_ from ros2/mergify/bp/jazzy/pr-2819
+* Fix race condition (`#2819 <https://github.com/ros2/rclcpp/issues/2819>`_)
+* Contributors: Michael Orlov, Pedro de Azeredo, mergify[bot]
+
+28.1.9 (2025-04-23)
+-------------------
+* remove redundant typesupport check in serialization module (`#2808 <https://github.com/ros2/rclcpp/issues/2808>`_) (`#2815 <https://github.com/ros2/rclcpp/issues/2815>`_)
+  (cherry picked from commit f78ed952b27acc63ef8022d78cb816c309a9ca3d)
+  Co-authored-by: Tanishq Chaudhary <tanishqchaudhary101010@gmail.com>
+* Contributors: mergify[bot]
+
+28.1.8 (2025-04-02)
+-------------------
+
+28.1.7 (2025-03-26)
+-------------------
+* fix(ClockConditionalVariable): Fixed potential crash on shutdown (`#2762 <https://github.com/ros2/rclcpp/issues/2762>`_)
+* doc: Added warning to not instantiate Clock directly with RCL_ROS_TIME (`#2769 <https://github.com/ros2/rclcpp/issues/2769>`_)
+  * Backports: `#2768 <https://github.com/ros2/rclcpp/issues/2768>`_
+* Use rmw_event_type_is_supported in test_qos_event (`#2766 <https://github.com/ros2/rclcpp/issues/2766>`_)
+  * Backports: `#2761 <https://github.com/ros2/rclcpp/issues/2761>`_
+* fix: Fixed expiring of goals if events executor is used (`#2674 <https://github.com/ros2/rclcpp/issues/2674>`_)
+* Executor strong reference fix (`#2754 <https://github.com/ros2/rclcpp/issues/2754>`_)
+  * Backports: `#2745 <https://github.com/ros2/rclcpp/issues/2745>`_
+* Double gc executor fix (`#2753 <https://github.com/ros2/rclcpp/issues/2753>`_)
+* Fix typo in doc section for get_service_typesupport_handle (`#2752 <https://github.com/ros2/rclcpp/issues/2752>`_)
+  * Backports: `#2751 <https://github.com/ros2/rclcpp/issues/2751>`_
+* Test case and fix for for https://github.com/ros2/rclcpp/issues/2652 (`#2740 <https://github.com/ros2/rclcpp/issues/2740>`_)
+  * Backports: `#2713 <https://github.com/ros2/rclcpp/issues/2713>`_
+* fix(timer): Delete node, after executor thread terminated (`#2738 <https://github.com/ros2/rclcpp/issues/2738>`_)
+  * Backports: `#2737 <https://github.com/ros2/rclcpp/issues/2737>`_
+* fix(Executor): Fixed entities not beeing executed after just beeing added (`#2729 <https://github.com/ros2/rclcpp/issues/2729>`_)
+  * Backports: `#2737 <https://github.com/ros2/rclcpp/issues/2724>`_
+* Fix transient local IPC publish  (`#2722 <https://github.com/ros2/rclcpp/issues/2722>`_)
+  * Backports: `#2708 <https://github.com/ros2/rclcpp/issues/2708>`_
+* Contributors: Janosch Machowinski, Jeffery Hsu, Tomoya Fujita, Francisco Martín Rico
+
+28.1.6 (2024-12-18)
+-------------------
+* apply actual QoS from rmw to the IPC publisher. (`#2707 <https://github.com/ros2/rclcpp/issues/2707>`_) (`#2712 <https://github.com/ros2/rclcpp/issues/2712>`_)
+  * apply actual QoS from rmw to the IPC publisher.
+  * address uncrustify warning.
+  ---------
+  (cherry picked from commit 016cfeac99e4b67f58abdf247e57f05b85c09ec4)
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Adding in topic name to logging on IPC issues (`#2706 <https://github.com/ros2/rclcpp/issues/2706>`_) (`#2710 <https://github.com/ros2/rclcpp/issues/2710>`_)
+  * Adding in topic name to logging on IPC issues
+  * Update test matching output logging
+  * adding in single quotes
+  ---------
+  (cherry picked from commit a13e16e2cbaeacb14ff31272d01cbb21bd8ac037)
+  Co-authored-by: Steve Macenski <stevenmacenski@gmail.com>
+* enable testRaceConditionAddNode for rmw_connextdds. (`#2698 <https://github.com/ros2/rclcpp/issues/2698>`_)
+* Re-enable executor test on rmw_connextdds. (`#2693 <https://github.com/ros2/rclcpp/issues/2693>`_) (`#2695 <https://github.com/ros2/rclcpp/issues/2695>`_)
+  It supports the events executor now, so re-enable the test.
+  (cherry picked from commit d7245365ed867db9b309ed3efbfb0391bda09bd5)
+  Co-authored-by: Chris Lalancette <clalancette@gmail.com>
+* Fix warnings on Windows. (backport `#2692 <https://github.com/ros2/rclcpp/issues/2692>`_) (`#2694 <https://github.com/ros2/rclcpp/issues/2694>`_)
+  * Fix warnings on Windows. (`#2692 <https://github.com/ros2/rclcpp/issues/2692>`_)
+  For reasons I admit I do not understand, the deprecation
+  warnings for StaticSingleThreadedExecutor on Windows
+  happen when we construct a shared_ptr for it in the tests.
+  If we construct a regular object, then it is fine.  Luckily
+  this test does not require a shared_ptr, so just make it
+  a regular object here, which rixes the warning.
+  While we are in here, make all of the tests camel case to
+  be consistent.
+  (cherry picked from commit 3310f9eaed967e0c18d17bb2f82d2def838bb7a5)
+  # Conflicts:
+  #	rclcpp/test/rclcpp/executors/test_executors.cpp
+  * resolve backport conflict.
+  ---------
+  Co-authored-by: Chris Lalancette <clalancette@gmail.com>
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Omnibus fixes for running tests with Connext. (backport `#2684 <https://github.com/ros2/rclcpp/issues/2684>`_) (`#2690 <https://github.com/ros2/rclcpp/issues/2690>`_)
+  * Omnibus fixes for running tests with Connext. (`#2684 <https://github.com/ros2/rclcpp/issues/2684>`_)
+  * Omnibus fixes for running tests with Connext.
+  When running the tests with RTI Connext as the default
+  RMW, some of the tests are failing.  There are three
+  different failures fixed here:
+  1.  Setting the liveliness duration to a value smaller than
+  a microsecond causes Connext to throw an error.  Set it to
+  a millisecond.
+  2.  Using the SystemDefaultsQoS sets the QoS to KEEP_LAST 1.
+  Connext is somewhat slow in this regard, so it can be the case
+  that we are overwriting a previous service introspection event
+  with the next one.  Switch to the ServicesDefaultQoS in the test,
+  which ensures we will not lose events.
+  3.  Connext is slow to match publishers and subscriptions.  Thus,
+  when creating a subscription "on-the-fly", we should wait for the
+  publisher to match it before expecting the subscription to actually
+  receive data from it.
+  With these fixes in place, the test_client_common, test_generic_service,
+  test_service_introspection, and test_executors tests all pass for
+  me with rmw_connextdds.
+  * Fixes for executors.
+  * One more fix for services.
+  * More fixes for service_introspection.
+  * More fixes for introspection.
+  ---------
+  (cherry picked from commit 9984197c292d6c5ae0e7661aaea245ffb0fea057)
+  # Conflicts:
+  #	rclcpp/test/rclcpp/executors/test_executors.cpp
+  #	rclcpp/test/rclcpp/test_generic_service.cpp
+  * address backport merge conflicts.
+  ---------
+  Co-authored-by: Chris Lalancette <clalancette@gmail.com>
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* fix(Executor): Fix segfault if callback group is deleted during rmw_wait (`#2682 <https://github.com/ros2/rclcpp/issues/2682>`_)
+* Fix NodeOptions assignment operator (`#2656 <https://github.com/ros2/rclcpp/issues/2656>`_) (`#2660 <https://github.com/ros2/rclcpp/issues/2660>`_)
+  * Fix NodeOptions assignment operator
+  Also copy the enable_logger_service\_ member during the assignment operation
+  * Add more checks for NodeOptions copy test
+  * Set non default values by avoiding the copy-assignement
+  Co-authored-by: Christophe Bedard <bedard.christophe@gmail.com>
+  (cherry picked from commit 9b654942f99f17850e0e95480958abdbb508bc00)
+  Co-authored-by: Romain DESILLE <r.desille@gmail.com>
+* set QoS History KEEP_ALL explicitly for statistics publisher. (`#2650 <https://github.com/ros2/rclcpp/issues/2650>`_) (`#2657 <https://github.com/ros2/rclcpp/issues/2657>`_)
+  * set QoS History KEEP_ALL explicitly for statistics publisher.
+  * test_subscription_options adjustment.
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Contributors: Cristóbal Arroyo, Tomoya Fujita, jmachowinski, mergify[bot]
+
+28.1.5 (2024-09-19)
+-------------------
+* backport fix events-executor warm-up bug and add unit-tests (`#2591 <https://github.com/ros2/rclcpp/issues/2591>`_) (`#2628 <https://github.com/ros2/rclcpp/issues/2628>`_)
+* Contributors: Alberto Soragna
+
+28.1.4 (2024-09-06)
+-------------------
+* Split test_executors.cpp even further. (`#2572 <https://github.com/ros2/rclcpp/issues/2572>`_) (`#2619 <https://github.com/ros2/rclcpp/issues/2619>`_)
+  That's because it is too large for Windows Debug to compile,
+  so split into smaller bits.
+  Even with this split, the file is too big; that's likely
+  because we are using TYPED_TEST here, which generates multiple
+  symbols per test case.  To deal with this, without further
+  breaking up the file, also add in the /bigobj flag when
+  compiling on Windows Debug.
+  (cherry picked from commit c743c173e68d92af872cf163e10721a8dbe51dd0)
+  Co-authored-by: Chris Lalancette <clalancette@gmail.com>
+* Correct node name in service test code (`#2615 <https://github.com/ros2/rclcpp/issues/2615>`_) (`#2616 <https://github.com/ros2/rclcpp/issues/2616>`_)
+  (cherry picked from commit e846f56224a39b93f1c609e7ee03fff0662b7453)
+  Co-authored-by: Barry Xu <barry.xu@sony.com>
+* Release ownership of entities after spinning cancelled (backport `#2556 <https://github.com/ros2/rclcpp/issues/2556>`_) (`#2580 <https://github.com/ros2/rclcpp/issues/2580>`_)
+  * Release ownership of entities after spinning cancelled (`#2556 <https://github.com/ros2/rclcpp/issues/2556>`_)
+  * Release ownership of entities after spinning cancelled
+  * Move release action to every exit point in different spin functions
+  * Move wait_result\_.reset() before setting spinning to false
+  * Update test code
+  * Move test code to test_executors.cpp
+  ---------
+  (cherry picked from commit 069a0018935b33a14632a1cdf4074984a1cf80fe)
+  # Conflicts:
+  #	rclcpp/test/rclcpp/executors/test_executors.cpp
+  * Fix backport issue (`#2581 <https://github.com/ros2/rclcpp/issues/2581>`_)
+  ---------
+  Co-authored-by: Barry Xu <barry.xu@sony.com>
+* Contributors: mergify[bot]
+
+28.1.3 (2024-06-27)
+-------------------
+* Add test creating two content filter topics with the same topic name (`#2546 <https://github.com/ros2/rclcpp/issues/2546>`_) (`#2549 <https://github.com/ros2/rclcpp/issues/2549>`_) (`#2552 <https://github.com/ros2/rclcpp/issues/2552>`_)
+  Co-authored-by: Mario Domínguez López <116071334+Mario-DL@users.noreply.github.com>
+  (cherry picked from commit 7c096888caf92aa7557e1d3efc5448b56d8ce81c)
+  Co-authored-by: Alejandro Hernández Cordero <ahcorde@gmail.com>
+* Contributors: mergify[bot]
+
+28.1.2 (2024-05-13)
+-------------------
+* add impl pointer for ExecutorOptions (`#2523 <https://github.com/ros2/rclcpp/issues/2523>`_) (`#2525 <https://github.com/ros2/rclcpp/issues/2525>`_)
+  * add impl pointer for ExecutorOptions
+  (cherry picked from commit 343b29b617b163ad72b9fe3f6441dd4ed3d3af09)
+  Co-authored-by: William Woodall <william@osrfoundation.org>
+* Fixup Executor::spin_all() regression fix (`#2517 <https://github.com/ros2/rclcpp/issues/2517>`_) (`#2521 <https://github.com/ros2/rclcpp/issues/2521>`_)
+  * test(Executors): Added tests for busy waiting
+  Checks if executors are busy waiting while they should block
+  in spin_some or spin_all.
+  * fix: Reworked spinAll test
+  This test was strange. It looked like, it assumed that spin_all did
+  not return instantly. Also it was racy, as the thread could terminate
+  instantly.
+  * fix(Executor): Fixed spin_all not returning instantly is no work was available
+  * Update rclcpp/test/rclcpp/executors/test_executors.cpp
+  * test(executors): Added test for busy waiting while calling spin
+  * fix(executor): Reset wait_result on every call to spin_some_impl
+  Before, the method would not recollect available work in case of
+  spin_some, spin_all. This would lead to the method behaving differently
+  than to what the documentation states.
+  * restore previous test logic for now
+  * refactor spin_some_impl's logic and improve busy wait tests
+  * added some more comments about the implementation
+  ---------
+  Co-authored-by: Janosch Machowinski <J.Machowinski@cellumation.com>
+  Co-authored-by: jmachowinski <jmachowinski@users.noreply.github.com>
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+  Co-authored-by: William Woodall <william@osrfoundation.org>
+* Contributors: mergify[bot]
+
+28.1.1 (2024-04-24)
+-------------------
+* Revise the description of service configure_introspection() (`#2511 <https://github.com/ros2/rclcpp/issues/2511>`_) (`#2513 <https://github.com/ros2/rclcpp/issues/2513>`_)
+* Contributors: mergify[bot]
+
 28.1.0 (2024-04-16)
 -------------------
 * Remove references to index.ros.org. (`#2504 <https://github.com/ros2/rclcpp/issues/2504>`_)

rclcpp/CMakeLists.txt

@@ -63,6 +63,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/exceptions/exceptions.cpp
   src/rclcpp/executable_list.cpp
   src/rclcpp/executor.cpp
+  src/rclcpp/executor_options.cpp
   src/rclcpp/executors.cpp
   src/rclcpp/executors/executor_entities_collection.cpp
   src/rclcpp/executors/executor_entities_collector.cpp

rclcpp/include/rclcpp/clock.hpp

@@ -60,6 +60,13 @@ public:
   /**
    * Initializes the clock instance with the given clock_type.
    *
+   * WARNING Don't instantiate a clock using RCL_ROS_TIME directly,
+   * unless you really know what you are doing. By default no TimeSource
+   * is attached to a new clock. This will lead to the unexpected behavior,
+   * that your RCL_ROS_TIME will run always on system time. If you want
+   * a RCL_ROS_TIME use Node::get_clock(), or make sure to attach a
+   * TimeSource yourself.
+   *
    * \param clock_type type of the clock.
    * \throws anything rclcpp::exceptions::throw_from_rcl_error can throw.
    */
@@ -260,6 +267,117 @@ private:
   std::shared_ptr<Impl> impl_;
 };
 
+/**
+ * A synchronization primitive, equal to std::conditional_variable,
+ * that works with the rclcpp::Clock.
+ *
+ * For more information on the API see https://en.cppreference.com/w/cpp/thread/condition_variable.
+ *
+ * Note, this class does not handle shutdowns, if you want to
+ * haven them handles as well, use ClockConditionalVariable.
+ */
+class ClockWaiter
+{
+private:
+  class ClockWaiterImpl;
+  std::unique_ptr<ClockWaiterImpl> impl_;
+
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(ClockWaiter)
+
+  RCLCPP_PUBLIC
+  explicit ClockWaiter(const rclcpp::Clock::SharedPtr & clock);
+
+  RCLCPP_PUBLIC
+  ~ClockWaiter();
+
+  /**
+   * Calling this function will block the current thread, until abs_time is reached,
+   * or pred returns true.
+   * @param lock A locked lock. The lock must be locked at call time, or this method will throw.
+   *             The lock will be atomically released and this thread will blocked.
+   * @param abs_time The time until which this thread shall be blocked.
+   * @param pred may be called in cased of spurious wakeups, but must be called every time
+   *             notify_one() was called. During the call to pred, the given lock will be locked.
+   *             This method will return, if pred returns true.
+   */
+  RCLCPP_PUBLIC
+  bool
+  wait_until(
+    std::unique_lock<std::mutex> & lock,
+    const rclcpp::Time & abs_time, const std::function<bool ()> & pred);
+
+  /**
+   * Notify the blocked thread, that it should reevaluate the wakeup condition.
+   * The given pred function in wait_until will be reevaluated and wait_until
+   * will return if it evaluates to true.
+   */
+  RCLCPP_PUBLIC
+  void
+  notify_one();
+};
+
+
+/**
+ * A synchronization primitive, similar to std::conditional_variable,
+ * that works with the rclcpp::Clock.
+ *
+ * For more information on the API see https://en.cppreference.com/w/cpp/thread/condition_variable.
+ *
+ * This primitive will wake up if the context was shut down.
+ */
+class ClockConditionalVariable
+{
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(ClockConditionalVariable)
+
+  RCLCPP_PUBLIC
+  ClockConditionalVariable(
+    const rclcpp::Clock::SharedPtr & clock,
+    rclcpp::Context::SharedPtr context = rclcpp::contexts::get_global_default_context());
+  RCLCPP_PUBLIC
+  ~ClockConditionalVariable();
+
+  /**
+   * Calling this function will block the current thread, until abs_time is reached,
+   * or pred returns true.
+   * @param lock A locked lock. The lock must be locked at call time, or this method will throw.
+   *             The lock will be atomically released and this thread will blocked.
+   *             The given lock must be created using the mutex returned my mutex().
+   * @param abs_time The time until which this thread shall be blocked.
+   * @param pred may be called in cased of spurious wakeups, but must be called every time
+   *             notify_one() was called. During the call to pred, the given lock will be locked.
+   *             This method will return, if pred returns true.
+   *
+   * @return true if until was reached.
+   */
+  RCLCPP_PUBLIC
+  bool
+  wait_until(
+    std::unique_lock<std::mutex> & lock, rclcpp::Time until,
+    const std::function<bool ()> & pred);
+
+  /**
+   * Notify the blocked thread, that is should reevaluate the wakeup condition.
+   * E.g. the given pred function in wait_until shall be reevaluated.
+   */
+  RCLCPP_PUBLIC
+  void
+  notify_one();
+
+  /**
+   * Returns the internal mutex. In order to be race free with the context shutdown,
+   * this mutex must be used for the wait_until call.
+   */
+  RCLCPP_PUBLIC
+  std::mutex &
+  mutex();
+};
+
+
 }  // namespace rclcpp
 
 #endif  // RCLCPP__CLOCK_HPP_

rclcpp/include/rclcpp/executor_options.hpp

@@ -15,6 +15,8 @@
 #ifndef RCLCPP__EXECUTOR_OPTIONS_HPP_
 #define RCLCPP__EXECUTOR_OPTIONS_HPP_
 
+#include <memory>
+
 #include "rclcpp/context.hpp"
 #include "rclcpp/contexts/default_context.hpp"
 #include "rclcpp/memory_strategies.hpp"
@@ -24,18 +26,30 @@
 namespace rclcpp
 {
 
+class ExecutorOptionsImplementation;
+
 /// Options to be passed to the executor constructor.
 struct ExecutorOptions
 {
-  ExecutorOptions()
-  : memory_strategy(rclcpp::memory_strategies::create_default_strategy()),
-    context(rclcpp::contexts::get_global_default_context()),
-    max_conditions(0)
-  {}
+  RCLCPP_PUBLIC
+  ExecutorOptions();
+
+  RCLCPP_PUBLIC
+  virtual ~ExecutorOptions();
+
+  RCLCPP_PUBLIC
+  ExecutorOptions(const ExecutorOptions &);
+
+  RCLCPP_PUBLIC
+  ExecutorOptions & operator=(const ExecutorOptions &);
 
   rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy;
   rclcpp::Context::SharedPtr context;
   size_t max_conditions;
+
+private:
+  /// Pointer to implementation
+  std::unique_ptr<ExecutorOptionsImplementation> impl_;
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/experimental/buffers/ring_buffer_implementation.hpp

@@ -255,9 +255,13 @@ private:
     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_])));
+      const auto & elem(ring_buffer_[(read_index_ + id) % capacity_]);
+      if (elem != nullptr) {
+        result_vtr.emplace_back(new typename is_std_unique_ptr<T>::Ptr_type(
+          *elem));
+      } else {
+        result_vtr.emplace_back(nullptr);
+      }
     }
     return result_vtr;
   }

rclcpp/include/rclcpp/parameter_events_filter.hpp

@@ -45,6 +45,7 @@ public:
    * \param[in] names A list of parameter names of interest.
    * \param[in] types A list of the types of parameter events of iterest.
    *    EventType NEW, DELETED, or CHANGED
+   * \throws std::invalid_argument if event is NULL.
    *
    * Example Usage:
    *

rclcpp/include/rclcpp/publisher.hpp

@@ -162,8 +162,7 @@ public:
     const rclcpp::QoS & qos,
     const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options)
   {
-    // Topic is unused for now.
-    (void)topic;
+    (void)qos;
     (void)options;
 
     // If needed, setup intra process communication.
@@ -171,22 +170,26 @@ public:
       auto context = node_base->get_context();
       // Get the intra process manager instance for this context.
       auto ipm = context->get_sub_context<rclcpp::experimental::IntraProcessManager>();
-      // Register the publisher with the intra process manager.
-      if (qos.history() != rclcpp::HistoryPolicy::KeepLast) {
+      // Check if the QoS is compatible with intra-process.
+      auto qos_profile = get_actual_qos();
+      if (qos_profile.history() != rclcpp::HistoryPolicy::KeepLast) {
         throw std::invalid_argument(
-                "intraprocess communication allowed only with keep last history qos policy");
+                "intraprocess communication on topic '" + topic +
+                "' allowed only with keep last history qos policy");
       }
-      if (qos.depth() == 0) {
+      if (qos_profile.depth() == 0) {
         throw std::invalid_argument(
-                "intraprocess communication is not allowed with a zero qos history depth value");
+                "intraprocess communication on topic '" + topic +
+                "' is not allowed with a zero qos history depth value");
       }
-      if (qos.durability() == rclcpp::DurabilityPolicy::TransientLocal) {
+      if (qos_profile.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,
+          qos_profile,
           std::make_shared<ROSMessageTypeAllocator>(ros_message_type_allocator_));
       }
+      // Register the publisher with the intra process manager.
       uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this(), buffer_);
       this->setup_intra_process(
         intra_process_publisher_id,
@@ -248,8 +251,12 @@ public:
     // interprocess publish, resulting in lower publish-to-subscribe latency.
     // It's not possible to do that with an unique_ptr,
     // as do_intra_process_publish takes the ownership of the message.
+
+    // When durability is set to TransientLocal (i.e. there is a buffer),
+    // inter process publish should always take place to ensure
+    // late joiners receive past data.
     bool inter_process_publish_needed =
-      get_subscription_count() > get_intra_process_subscription_count();
+      get_subscription_count() > get_intra_process_subscription_count() || buffer_;
 
     if (inter_process_publish_needed) {
       auto shared_msg =
@@ -326,8 +333,11 @@ public:
       return;
     }
 
+    // When durability is set to TransientLocal (i.e. there is a buffer),
+    // inter process publish should always take place to ensure
+    // late joiners receive past data.
     bool inter_process_publish_needed =
-      get_subscription_count() > get_intra_process_subscription_count();
+      get_subscription_count() > get_intra_process_subscription_count() || buffer_;
 
     if (inter_process_publish_needed) {
       auto ros_msg_ptr = std::make_shared<ROSMessageType>();

rclcpp/include/rclcpp/service.hpp

@@ -499,7 +499,7 @@ public:
     }
   }
 
-  /// Configure client introspection.
+  /// Configure service introspection.
   /**
    * \param[in] clock clock to use to generate introspection timestamps
    * \param[in] qos_service_event_pub QoS settings to use when creating the introspection publisher

rclcpp/include/rclcpp/subscription.hpp

@@ -159,11 +159,13 @@ public:
       auto qos_profile = get_actual_qos();
       if (qos_profile.history() != rclcpp::HistoryPolicy::KeepLast) {
         throw std::invalid_argument(
-                "intraprocess communication allowed only with keep last history qos policy");
+                "intraprocess communication on topic '" + topic_name +
+                "' allowed only with keep last history qos policy");
       }
       if (qos_profile.depth() == 0) {
         throw std::invalid_argument(
-                "intraprocess communication is not allowed with 0 depth qos policy");
+                "intraprocess communication on topic '" + topic_name +
+                "' is not allowed with 0 depth qos policy");
       }
 
       using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<

rclcpp/include/rclcpp/subscription_options.hpp

@@ -80,7 +80,8 @@ struct SubscriptionOptionsBase
 
     // 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();
+    // Explicitly set the enough depth to avoid missing the statistics messages.
+    rclcpp::QoS qos = SystemDefaultsQoS().keep_last(10);
   };
 
   TopicStatisticsOptions topic_stats_options;

rclcpp/include/rclcpp/typesupport_helpers.hpp

@@ -76,7 +76,7 @@ get_message_typesupport_handle(
 
 /// 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.
+ * The library needs to match the service 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"

rclcpp/include/rclcpp/wait_result.hpp

@@ -176,7 +176,9 @@ public:
       for (; ii < wait_set.size_of_timers(); ++ii) {
         if (rcl_wait_set.timers[ii] != nullptr) {
           ret = wait_set.timers(ii);
-          break;
+          if (ret) {
+            break;
+          }
         }
       }
     }
@@ -217,7 +219,9 @@ public:
         if (rcl_wait_set.subscriptions[ii] != nullptr) {
           ret = wait_set.subscriptions(ii);
           rcl_wait_set.subscriptions[ii] = nullptr;
-          break;
+          if (ret) {
+            break;
+          }
         }
       }
     }
@@ -237,7 +241,9 @@ public:
         if (rcl_wait_set.services[ii] != nullptr) {
           ret = wait_set.services(ii);
           rcl_wait_set.services[ii] = nullptr;
-          break;
+          if (ret) {
+            break;
+          }
         }
       }
     }
@@ -257,7 +263,9 @@ public:
         if (rcl_wait_set.clients[ii] != nullptr) {
           ret = wait_set.clients(ii);
           rcl_wait_set.clients[ii] = nullptr;
-          break;
+          if (ret) {
+            break;
+          }
         }
       }
     }

rclcpp/include/rclcpp/wait_set_policies/dynamic_storage.hpp

@@ -216,6 +216,11 @@ public:
       shared_waitables_
     );
 
+    if (this->needs_pruning_) {
+      this->storage_prune_deleted_entities();
+      this->needs_pruning_ = false;
+    }
+
     this->storage_release_ownerships();
   }
 
@@ -455,59 +460,60 @@ public:
 
   size_t size_of_subscriptions() const
   {
-    return shared_subscriptions_.size();
+    return subscriptions_.size();
   }
 
   size_t size_of_timers() const
   {
-    return shared_timers_.size();
+    return timers_.size();
   }
 
   size_t size_of_clients() const
   {
-    return shared_clients_.size();
+    return clients_.size();
   }
 
   size_t size_of_services() const
   {
-    return shared_services_.size();
+    return services_.size();
   }
 
   size_t size_of_waitables() const
   {
-    return shared_waitables_.size();
+    return waitables_.size();
   }
 
   std::shared_ptr<rclcpp::SubscriptionBase>
   subscriptions(size_t ii) const
   {
-    return shared_subscriptions_[ii].subscription;
+    return subscriptions_[ii].lock();
   }
 
   std::shared_ptr<rclcpp::TimerBase>
   timers(size_t ii) const
   {
-    return shared_timers_[ii];
+    return timers_[ii].lock();
   }
 
   std::shared_ptr<rclcpp::ClientBase>
   clients(size_t ii) const
   {
-    return shared_clients_[ii];
+    return clients_[ii].lock();
   }
 
   std::shared_ptr<rclcpp::ServiceBase>
   services(size_t ii) const
   {
-    return shared_services_[ii];
+    return services_[ii].lock();
   }
 
   std::shared_ptr<rclcpp::Waitable>
   waitables(size_t ii) const
   {
-    return shared_waitables_[ii].waitable;
+    return waitables_[ii].lock();
   }
 
+private:
   size_t ownership_reference_counter_ = 0;
 
   SequenceOfWeakSubscriptions subscriptions_;

rclcpp/include/rclcpp/wait_set_policies/static_storage.hpp

@@ -160,6 +160,15 @@ public:
       services_,
       waitables_
     );
+
+    if(this->needs_pruning_) {
+      // we need to throw here, as the indexing of the rcl_waitset is broken,
+      // in case of invalid entries
+
+      throw std::runtime_error(
+            "StaticStorage::storage_rebuild_rcl_wait_set(): entity weak_ptr "
+            "unexpectedly expired in static entity storage");
+    }
   }
 
   // storage_add_subscription() explicitly not declared here

rclcpp/include/rclcpp/wait_set_template.hpp

@@ -734,8 +734,6 @@ private:
     wait_result_dirty_ = false;
     // this method comes from the SynchronizationPolicy
     this->sync_wait_result_acquire();
-    // this method comes from the StoragePolicy
-    this->storage_acquire_ownerships();
   }
 
   /// Called by the WaitResult's destructor to release resources.
@@ -752,8 +750,6 @@ private:
     }
     wait_result_holding_ = false;
     wait_result_dirty_ = false;
-    // this method comes from the StoragePolicy
-    this->storage_release_ownerships();
     // this method comes from the SynchronizationPolicy
     this->sync_wait_result_release();
   }

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>28.1.0</version>
+  <version>28.1.10</version>
   <description>The ROS client library in C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp/src/rclcpp/clock.cpp

@@ -118,7 +118,7 @@ Clock::sleep_until(
     });
   // No longer need the shutdown callback when this function exits
   auto callback_remover = rcpputils::scope_exit(
-    [context, &shutdown_cb_handle]() {
+    [&context, &shutdown_cb_handle]() {
       context->remove_on_shutdown_callback(shutdown_cb_handle);
     });
 
@@ -367,4 +367,245 @@ Clock::create_jump_callback(
   // *INDENT-ON*
 }
 
+class ClockWaiter::ClockWaiterImpl
+{
+private:
+  std::condition_variable cv_;
+
+  rclcpp::Clock::SharedPtr clock_;
+  bool time_source_changed_ = false;
+  std::function<void(const rcl_time_jump_t &)> post_time_jump_callback;
+
+  bool
+  wait_until_system_time(
+    std::unique_lock<std::mutex> & lock,
+    const rclcpp::Time & abs_time, const std::function<bool ()> & pred)
+  {
+    auto system_time = std::chrono::system_clock::time_point(
+    // Cast because system clock resolution is too big for nanoseconds on some systems
+    std::chrono::duration_cast<std::chrono::system_clock::duration>(
+        std::chrono::nanoseconds(abs_time.nanoseconds())));
+
+    return cv_.wait_until(lock, system_time, pred);
+  }
+
+  bool
+  wait_until_steady_time(
+    std::unique_lock<std::mutex> & lock,
+    const rclcpp::Time & abs_time, const std::function<bool ()> & pred)
+  {
+    // Synchronize because RCL steady clock epoch might differ from chrono::steady_clock epoch
+    const rclcpp::Time rcl_entry = clock_->now();
+    const std::chrono::steady_clock::time_point chrono_entry = std::chrono::steady_clock::now();
+    const rclcpp::Duration delta_t = abs_time - rcl_entry;
+    const std::chrono::steady_clock::time_point chrono_until =
+      chrono_entry + std::chrono::nanoseconds(delta_t.nanoseconds());
+
+    return cv_.wait_until(lock, chrono_until, pred);
+  }
+
+
+  bool
+  wait_until_ros_time(
+    std::unique_lock<std::mutex> & lock,
+    const rclcpp::Time & abs_time, const std::function<bool ()> & pred)
+  {
+    // Install jump handler for any amount of time change, for two purposes:
+    // - if ROS time is active, check if time reached on each new clock sample
+    // - Trigger via on_clock_change to detect if time source changes, to invalidate sleep
+    rcl_jump_threshold_t threshold;
+    threshold.on_clock_change = true;
+    // 0 is disable, so -1 and 1 are smallest possible time changes
+    threshold.min_backward.nanoseconds = -1;
+    threshold.min_forward.nanoseconds = 1;
+
+    time_source_changed_ = false;
+
+    post_time_jump_callback = [this, &lock] (const rcl_time_jump_t & jump)
+      {
+        if (jump.clock_change != RCL_ROS_TIME_NO_CHANGE) {
+          std::lock_guard<std::mutex> lk(*lock.mutex());
+          time_source_changed_ = true;
+        }
+        cv_.notify_one();
+      };
+
+    // Note this is a trade-off. Adding the callback for every call
+    // is expensive for high frequency calls. For low frequency waits
+    // its more overhead to have the callback being called all the time.
+    // As we expect the use case to be low frequency calls to wait_until
+    // with relative big pauses between the calls, we install it on demand.
+    auto clock_handler = clock_->create_jump_callback(
+      nullptr,
+      post_time_jump_callback,
+      threshold);
+
+    if (!clock_->ros_time_is_active()) {
+      auto system_time = std::chrono::system_clock::time_point(
+        // Cast because system clock resolution is too big for nanoseconds on some systems
+        std::chrono::duration_cast<std::chrono::system_clock::duration>(
+          std::chrono::nanoseconds(abs_time.nanoseconds())));
+
+      return cv_.wait_until(lock, system_time, [this, &pred] () {
+                 return time_source_changed_ || pred();
+      });
+    }
+
+    // RCL_ROS_TIME with ros_time_is_active.
+    // Just wait without "until" because installed
+    // jump callbacks wake the cv on every new sample.
+    cv_.wait(lock, [this, &pred, &abs_time] () {
+        return clock_->now() >= abs_time || time_source_changed_ || pred();
+    });
+
+    return clock_->now() < abs_time;
+  }
+
+public:
+  explicit ClockWaiterImpl(const rclcpp::Clock::SharedPtr & clock)
+  :clock_(clock)
+  {
+  }
+
+  bool
+  wait_until(
+    std::unique_lock<std::mutex> & lock,
+    const rclcpp::Time & abs_time, const std::function<bool ()> & pred)
+  {
+    switch(clock_->get_clock_type()) {
+      case RCL_CLOCK_UNINITIALIZED:
+        throw std::runtime_error("Error, wait on uninitialized clock called");
+      case RCL_ROS_TIME:
+        return wait_until_ros_time(lock, abs_time, pred);
+        break;
+      case RCL_STEADY_TIME:
+        return wait_until_steady_time(lock, abs_time, pred);
+        break;
+      case RCL_SYSTEM_TIME:
+        return wait_until_system_time(lock, abs_time, pred);
+        break;
+    }
+
+    return false;
+  }
+
+  void
+  notify_one()
+  {
+    cv_.notify_one();
+  }
+};
+
+ClockWaiter::ClockWaiter(const rclcpp::Clock::SharedPtr & clock)
+:impl_(std::make_unique<ClockWaiterImpl>(clock))
+{
+}
+
+ClockWaiter::~ClockWaiter() = default;
+
+bool
+ClockWaiter::wait_until(
+  std::unique_lock<std::mutex> & lock,
+  const rclcpp::Time & abs_time, const std::function<bool ()> & pred)
+{
+  return impl_->wait_until(lock, abs_time, pred);
+}
+
+void
+ClockWaiter::notify_one()
+{
+  impl_->notify_one();
+}
+
+class ClockConditionalVariable::Impl
+{
+  std::mutex pred_mutex_;
+  bool shutdown_ = false;
+  rclcpp::Context::SharedPtr context_;
+  rclcpp::OnShutdownCallbackHandle shutdown_cb_handle_;
+  ClockWaiter::UniquePtr clock_;
+
+public:
+  Impl(const rclcpp::Clock::SharedPtr & clock, rclcpp::Context::SharedPtr context)
+  :context_(context),
+    clock_(std::make_unique<ClockWaiter>(clock))
+  {
+    if (!context_ || !context_->is_valid()) {
+      throw std::runtime_error("context cannot be slept with because it's invalid");
+    }
+    // Wake this thread if the context is shutdown
+    shutdown_cb_handle_ = context_->add_on_shutdown_callback(
+      [this]() {
+        {
+          std::unique_lock lock(pred_mutex_);
+          shutdown_ = true;
+        }
+        clock_->notify_one();
+    });
+  }
+
+  ~Impl()
+  {
+    context_->remove_on_shutdown_callback(shutdown_cb_handle_);
+  }
+
+  bool
+  wait_until(
+    std::unique_lock<std::mutex> & lock, rclcpp::Time until,
+    const std::function<bool ()> & pred)
+  {
+    if(lock.mutex() != &pred_mutex_) {
+      throw std::runtime_error(
+          "ClockConditionalVariable::wait_until: Internal error, given lock does not use"
+          " mutex returned by this->mutex()");
+    }
+
+    clock_->wait_until(lock, until, [this, &pred] () -> bool {
+        return shutdown_ || pred();
+      });
+    return true;
+  }
+
+  void
+  notify_one()
+  {
+    clock_->notify_one();
+  }
+
+  std::mutex &
+  mutex()
+  {
+    return pred_mutex_;
+  }
+};
+
+ClockConditionalVariable::ClockConditionalVariable(
+  const rclcpp::Clock::SharedPtr & clock,
+  rclcpp::Context::SharedPtr context)
+:impl_(std::make_unique<Impl>(clock, context))
+{
+}
+
+ClockConditionalVariable::~ClockConditionalVariable() = default;
+
+void
+ClockConditionalVariable::notify_one()
+{
+  impl_->notify_one();
+}
+
+bool
+ClockConditionalVariable::wait_until(
+  std::unique_lock<std::mutex> & lock, rclcpp::Time until,
+  const std::function<bool ()> & pred)
+{
+  return impl_->wait_until(lock, until, pred);
+}
+
+std::mutex &
+ClockConditionalVariable::mutex()
+{
+  return impl_->mutex();
+}
+
 }  // namespace rclcpp

rclcpp/src/rclcpp/executor.cpp

@@ -84,7 +84,9 @@ Executor::Executor(const rclcpp::ExecutorOptions & options)
   notify_waitable_->add_guard_condition(interrupt_guard_condition_);
   notify_waitable_->add_guard_condition(shutdown_guard_condition_);
 
-  wait_set_.add_waitable(notify_waitable_);
+  // we need to initially rebuild the collection,
+  // so that the notify_waitable_ is added
+  collect_entities();
 }
 
 Executor::~Executor()
@@ -275,7 +277,7 @@ Executor::spin_until_future_complete_impl(
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_until_future_complete() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
   while (rclcpp::ok(this->context_) && spinning.load()) {
     // Do one item of work.
     spin_once_impl(timeout_left);
@@ -364,26 +366,69 @@ Executor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_some() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
 
+  // clear the wait result and wait for work without blocking to collect the work
+  // for the first time
+  // both spin_some and spin_all wait for work at the beginning
+  wait_result_.reset();
+  wait_for_work(std::chrono::milliseconds(0));
+  bool entity_states_fully_polled = true;
+
+  if (entities_need_rebuild_) {
+    // if the last wait triggered a collection rebuild, we need to call
+    // wait_for_work once more, in order to do a collection rebuild and collect
+    // events from the just added entities
+    entity_states_fully_polled = false;
+  }
+
+  // The logic of this while loop is as follows:
+  //
+  // - while not shutdown, and spinning (not canceled), and not max duration reached...
+  // - try to get an executable item to execute, and execute it if available
+  // - otherwise, reset the wait result, and ...
+  // - if there was no work available just after waiting, break the loop unconditionally
+  //   - this is appropriate for both spin_some and spin_all which use this function
+  // - else if exhaustive = true, then wait for work again
+  //   - this is only used for spin_all and not spin_some
+  // - else break
+  //   - this only occurs with spin_some
+  //
+  // The logic of this loop is subtle and should be carefully changed if at all.
+  // See also:
+  //   https://github.com/ros2/rclcpp/issues/2508
+  //   https://github.com/ros2/rclcpp/pull/2517
   while (rclcpp::ok(context_) && spinning.load() && max_duration_not_elapsed()) {
-    if (!wait_result_.has_value()) {
-      wait_for_work(std::chrono::milliseconds(0));
-    }
-
     AnyExecutable any_exec;
     if (get_next_ready_executable(any_exec)) {
       execute_any_executable(any_exec);
+      // during the execution some entity might got ready therefore we need
+      // to repoll the states of all entities
+      entity_states_fully_polled = false;
     } else {
-      // If nothing is ready, reset the result to signal we are
-      // ready to wait again
+      // if nothing is ready, reset the result to clear it
       wait_result_.reset();
-    }
 
-    if (!wait_result_.has_value() && !exhaustive) {
-      // In the case of spin some, then we can exit
-      // In the case of spin all, then we will allow ourselves to wait again.
-      break;
+      if (entity_states_fully_polled) {
+        // there was no work after just waiting, always exit in this case
+        // before the exhaustive condition can be checked
+        break;
+      }
+
+      if (exhaustive) {
+        // if exhaustive, wait for work again
+        // this only happens for spin_all; spin_some only waits at the start
+        wait_for_work(std::chrono::milliseconds(0));
+        entity_states_fully_polled = true;
+        if (entities_need_rebuild_) {
+          // if the last wait triggered a collection rebuild, we need to call
+          // wait_for_work once more, in order to do a collection rebuild and
+          // collect events from the just added entities
+          entity_states_fully_polled = false;
+        }
+      } else {
+        break;
+      }
     }
   }
 }
@@ -403,7 +448,7 @@ Executor::spin_once(std::chrono::nanoseconds timeout)
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_once() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
   spin_once_impl(timeout);
 }
 
@@ -701,13 +746,33 @@ Executor::wait_for_work(std::chrono::nanoseconds timeout)
   // Clear any previous wait result
   this->wait_result_.reset();
 
+  // we need to make sure that callback groups don't get out of scope
+  // during the wait. As in jazzy, they are not covered by the DynamicStorage,
+  // we explicitly hold them here as a bugfix
+  std::vector<rclcpp::CallbackGroup::SharedPtr> cbgs;
+
   {
     std::lock_guard<std::mutex> guard(mutex_);
+
     if (this->entities_need_rebuild_.exchange(false) || current_collection_.empty()) {
       this->collect_entities();
     }
+
+    auto callback_groups = this->collector_.get_all_callback_groups();
+    cbgs.resize(callback_groups.size());
+    for(const auto & w_ptr : callback_groups) {
+      auto shr_ptr = w_ptr.lock();
+      if(shr_ptr) {
+        cbgs.push_back(std::move(shr_ptr));
+      }
+    }
   }
+
   this->wait_result_.emplace(wait_set_.wait(timeout));
+
+  // drop references to the callback groups, before trying to execute anything
+  cbgs.clear();
+
   if (!this->wait_result_ || this->wait_result_->kind() == WaitResultKind::Empty) {
     RCUTILS_LOG_WARN_NAMED(
       "rclcpp",

rclcpp/src/rclcpp/executor_options.cpp

@@ -0,0 +1,55 @@
+// Copyright 2024 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/executor_options.hpp"
+
+using rclcpp::ExecutorOptions;
+
+namespace rclcpp
+{
+
+class ExecutorOptionsImplementation {};
+
+}  // namespace rclcpp
+
+ExecutorOptions::ExecutorOptions()
+: memory_strategy(rclcpp::memory_strategies::create_default_strategy()),
+  context(rclcpp::contexts::get_global_default_context()),
+  max_conditions(0),
+  impl_(nullptr)
+{}
+
+ExecutorOptions::~ExecutorOptions()
+{}
+
+ExecutorOptions::ExecutorOptions(const ExecutorOptions & other)
+{
+  *this = other;
+}
+
+ExecutorOptions & ExecutorOptions::operator=(const ExecutorOptions & other)
+{
+  if (this == &other) {
+    return *this;
+  }
+
+  this->memory_strategy = other.memory_strategy;
+  this->context = other.context;
+  this->max_conditions = other.max_conditions;
+  if (nullptr != other.impl_) {
+    this->impl_ = std::make_unique<ExecutorOptionsImplementation>(*other.impl_);
+  }
+
+  return *this;
+}

rclcpp/src/rclcpp/executors/multi_threaded_executor.cpp

@@ -55,7 +55,7 @@ MultiThreadedExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false););
+  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
   std::vector<std::thread> threads;
   size_t thread_id = 0;
   {

rclcpp/src/rclcpp/executors/single_threaded_executor.cpp

@@ -30,7 +30,7 @@ SingleThreadedExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
 
   // Clear any previous result and rebuild the waitset
   this->wait_result_.reset();

rclcpp/src/rclcpp/executors/static_single_threaded_executor.cpp

@@ -110,10 +110,29 @@ StaticSingleThreadedExecutor::spin_once_impl(std::chrono::nanoseconds timeout)
 std::optional<rclcpp::WaitResult<rclcpp::WaitSet>>
 StaticSingleThreadedExecutor::collect_and_wait(std::chrono::nanoseconds timeout)
 {
+  // we need to make sure that callback groups don't get out of scope
+  // during the wait. As in jazzy, they are not covered by the DynamicStorage,
+  // we explicitly hold them here as a bugfix
+  std::vector<rclcpp::CallbackGroup::SharedPtr> cbgs;
+
   if (this->entities_need_rebuild_.exchange(false) || current_collection_.empty()) {
     this->collect_entities();
   }
+
+  auto callback_groups = this->collector_.get_all_callback_groups();
+  cbgs.resize(callback_groups.size());
+  for(const auto & w_ptr : callback_groups) {
+    auto shr_ptr = w_ptr.lock();
+    if(shr_ptr) {
+      cbgs.push_back(std::move(shr_ptr));
+    }
+  }
+
   auto wait_result = wait_set_.wait(std::chrono::nanoseconds(timeout));
+
+  // drop references to the callback groups, before trying to execute anything
+  cbgs.clear();
+
   if (wait_result.kind() == WaitResultKind::Empty) {
     RCUTILS_LOG_WARN_NAMED(
       "rclcpp",

rclcpp/src/rclcpp/experimental/executors/events_executor/events_executor.cpp

@@ -62,7 +62,6 @@ EventsExecutor::EventsExecutor(
       //    ---> we need to wake up the executor so that it can terminate
       // - a node or callback group guard condition is triggered:
       //    ---> the entities collection is changed, we need to update callbacks
-      notify_waitable_event_pushed_ = false;
       this->refresh_current_collection_from_callback_groups();
     });
 
@@ -164,6 +163,14 @@ EventsExecutor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhau
       return false;
     };
 
+  // If this spin is not exhaustive (e.g. spin_some), we need to explicitly check
+  // if entities need to be rebuilt here rather than letting the notify waitable event do it.
+  // A non-exhaustive spin would not check for work a second time, thus delaying the execution
+  // of some entities to the next invocation of spin.
+  if (!exhaustive) {
+    this->refresh_current_collection_from_callback_groups();
+  }
+
   // Get the number of events and timers ready at start
   const size_t ready_events_at_start = events_queue_->size();
   size_t executed_events = 0;
@@ -403,6 +410,16 @@ EventsExecutor::get_automatically_added_callback_groups_from_nodes()
 void
 EventsExecutor::refresh_current_collection_from_callback_groups()
 {
+  // Do not rebuild if we don't need to.
+  // A rebuild event could be generated, but then
+  // this function could end up being called from somewhere else
+  // before that event gets processed, for example if
+  // a node or callback group is manually added to the executor.
+  const bool notify_waitable_triggered = notify_waitable_event_pushed_.exchange(false);
+  if (!notify_waitable_triggered && !this->entities_collector_->has_pending()) {
+    return;
+  }
+
   // Build the new collection
   this->entities_collector_->update_collections();
   auto callback_groups = this->entities_collector_->get_all_callback_groups();

rclcpp/src/rclcpp/node_options.cpp

@@ -80,6 +80,7 @@ NodeOptions::operator=(const NodeOptions & other)
     this->clock_type_ = other.clock_type_;
     this->clock_qos_ = other.clock_qos_;
     this->use_clock_thread_ = other.use_clock_thread_;
+    this->enable_logger_service_ = other.enable_logger_service_;
     this->parameter_event_qos_ = other.parameter_event_qos_;
     this->rosout_qos_ = other.rosout_qos_;
     this->parameter_event_publisher_options_ = other.parameter_event_publisher_options_;

rclcpp/src/rclcpp/parameter_events_filter.cpp

@@ -28,6 +28,9 @@ ParameterEventsFilter::ParameterEventsFilter(
   const std::vector<EventType> & types)
 : event_(event)
 {
+  if (!event) {
+    throw std::invalid_argument("event cannot be null");
+  }
   if (std::find(types.begin(), types.end(), EventType::NEW) != types.end()) {
     for (auto & new_parameter : event->new_parameters) {
       if (std::find(names.begin(), names.end(), new_parameter.name) != names.end()) {

rclcpp/src/rclcpp/qos.cpp

@@ -69,8 +69,10 @@ QoSInitialization::from_rmw(const rmw_qos_profile_t & rmw_qos)
       return KeepLast(rmw_qos.depth, false);
     case RMW_QOS_POLICY_HISTORY_KEEP_LAST:
     case RMW_QOS_POLICY_HISTORY_UNKNOWN:
-    default:
       return KeepLast(rmw_qos.depth);
+    default:
+      throw std::invalid_argument(
+        "Invalid history policy enum value passed to QoSInitialization::from_rmw");
   }
 }
 

rclcpp/src/rclcpp/serialization.cpp

@@ -36,7 +36,6 @@ SerializationBase::SerializationBase(const rosidl_message_type_support_t * type_
 void SerializationBase::serialize_message(
   const void * ros_message, SerializedMessage * serialized_message) const
 {
-  rcpputils::check_true(nullptr != type_support_, "Typesupport is nullpointer.");
   rcpputils::check_true(nullptr != ros_message, "ROS message is nullpointer.");
   rcpputils::check_true(nullptr != serialized_message, "Serialized message is nullpointer.");
 
@@ -52,7 +51,6 @@ void SerializationBase::serialize_message(
 void SerializationBase::deserialize_message(
   const SerializedMessage * serialized_message, void * ros_message) const
 {
-  rcpputils::check_true(nullptr != type_support_, "Typesupport is nullpointer.");
   rcpputils::check_true(nullptr != serialized_message, "Serialized message is nullpointer.");
   rcpputils::check_true(
     0u != serialized_message->capacity(),

rclcpp/src/rclcpp/serialized_message.cpp

@@ -26,8 +26,13 @@ namespace rclcpp
 
 inline void copy_rcl_message(const rcl_serialized_message_t & from, rcl_serialized_message_t & to)
 {
-  const auto ret = rmw_serialized_message_init(
-    &to, from.buffer_capacity, &from.allocator);
+  auto ret = RCL_RET_ERROR;
+  if (nullptr == to.buffer) {
+    ret = rmw_serialized_message_init(&to, from.buffer_capacity, &from.allocator);
+  } else {
+    ret = rmw_serialized_message_resize(&to, from.buffer_capacity);
+  }
+
   if (RCL_RET_OK != ret) {
     rclcpp::exceptions::throw_from_rcl_error(ret);
   }
@@ -78,7 +83,6 @@ SerializedMessage::SerializedMessage(rcl_serialized_message_t && other)
 SerializedMessage & SerializedMessage::operator=(const SerializedMessage & other)
 {
   if (this != &other) {
-    serialized_message_ = rmw_get_zero_initialized_serialized_message();
     copy_rcl_message(other.serialized_message_, serialized_message_);
   }
 
@@ -88,7 +92,6 @@ SerializedMessage & SerializedMessage::operator=(const SerializedMessage & other
 SerializedMessage & SerializedMessage::operator=(const rcl_serialized_message_t & other)
 {
   if (&serialized_message_ != &other) {
-    serialized_message_ = rmw_get_zero_initialized_serialized_message();
     copy_rcl_message(other, serialized_message_);
   }
 
@@ -98,6 +101,14 @@ SerializedMessage & SerializedMessage::operator=(const rcl_serialized_message_t
 SerializedMessage & SerializedMessage::operator=(SerializedMessage && other)
 {
   if (this != &other) {
+    if (nullptr != serialized_message_.buffer) {
+      const auto fini_ret = rmw_serialized_message_fini(&serialized_message_);
+      if (RCL_RET_OK != fini_ret) {
+        RCLCPP_ERROR(
+          get_logger("rclcpp"),
+          "Failed to destroy serialized message: %s", rcl_get_error_string().str);
+      }
+    }
     serialized_message_ =
       std::exchange(other.serialized_message_, rmw_get_zero_initialized_serialized_message());
   }
@@ -108,6 +119,14 @@ SerializedMessage & SerializedMessage::operator=(SerializedMessage && other)
 SerializedMessage & SerializedMessage::operator=(rcl_serialized_message_t && other)
 {
   if (&serialized_message_ != &other) {
+    if (nullptr != serialized_message_.buffer) {
+      const auto fini_ret = rmw_serialized_message_fini(&serialized_message_);
+      if (RCL_RET_OK != fini_ret) {
+        RCLCPP_ERROR(
+          get_logger("rclcpp"),
+          "Failed to destroy serialized message: %s", rcl_get_error_string().str);
+      }
+    }
     serialized_message_ =
       std::exchange(other, rmw_get_zero_initialized_serialized_message());
   }

rclcpp/test/benchmark/benchmark_executor.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <utility>

rclcpp/test/rclcpp/CMakeLists.txt

@@ -59,6 +59,11 @@ ament_add_gtest(test_clock test_clock.cpp)
 if(TARGET test_clock)
   target_link_libraries(test_clock ${PROJECT_NAME} mimick ${rcl_interfaces_TARGETS} ${test_msgs_TARGETS})
 endif()
+ament_add_gtest(test_clock_conditional test_clock_conditional.cpp)
+ament_add_test_label(test_clock_conditional mimick)
+if(TARGET test_clock_conditional)
+  target_link_libraries(test_clock_conditional ${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})
@@ -462,8 +467,12 @@ endif()
 ament_add_gtest(
   test_executors
   executors/test_executors.cpp
+  executors/test_waitable.cpp
   APPEND_LIBRARY_DIRS "${append_library_dirs}"
   TIMEOUT 180)
+if(CMAKE_BUILD_TYPE STREQUAL "Debug" AND MSVC)
+  target_compile_options(test_executors PRIVATE "/bigobj")
+endif()
 if(TARGET test_executors)
   target_link_libraries(test_executors ${PROJECT_NAME} rcl::rcl ${test_msgs_TARGETS})
 endif()
@@ -495,6 +504,26 @@ if(TARGET test_executors)
   target_link_libraries(test_executors_intraprocess ${PROJECT_NAME} ${test_msgs_TARGETS})
 endif()
 
+ament_add_gtest(
+  test_executors_busy_waiting
+  executors/test_executors_busy_waiting.cpp
+  executors/test_waitable.cpp
+  APPEND_LIBRARY_DIRS "${append_library_dirs}"
+  TIMEOUT 180)
+if(TARGET test_executors_busy_waiting)
+  target_link_libraries(test_executors_busy_waiting ${PROJECT_NAME})
+endif()
+
+ament_add_gtest(
+  test_executors_warmup
+  executors/test_executors_warmup.cpp
+  executors/test_waitable.cpp
+  APPEND_LIBRARY_DIRS "${append_library_dirs}"
+  TIMEOUT 180)
+if(TARGET test_executors_warmup)
+  target_link_libraries(test_executors_warmup ${PROJECT_NAME} ${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)
@@ -598,7 +627,7 @@ ament_add_gtest(test_executor test_executor.cpp
   APPEND_LIBRARY_DIRS "${append_library_dirs}"
   TIMEOUT 120)
 if(TARGET test_executor)
-  target_link_libraries(test_executor ${PROJECT_NAME} mimick)
+  target_link_libraries(test_executor ${PROJECT_NAME} mimick ${test_msgs_TARGETS})
 endif()
 
 ament_add_gtest(test_graph_listener test_graph_listener.cpp)

rclcpp/test/rclcpp/executors/test_events_queue.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 
 #include "rclcpp/experimental/executors/events_executor/events_executor_event_types.hpp"

rclcpp/test/rclcpp/executors/test_executor_notify_waitable.cpp

@@ -76,8 +76,7 @@ TEST_F(TestExecutorNotifyWaitable, wait) {
     std::make_shared<rclcpp::executors::ExecutorNotifyWaitable>(on_execute_callback);
 
   auto node = std::make_shared<rclcpp::Node>("my_node", "/ns");
-  auto notify_guard_condition =
-    node->get_node_base_interface()->get_shared_notify_guard_condition();
+  auto notify_guard_condition = std::make_shared<rclcpp::GuardCondition>();
   EXPECT_NO_THROW(waitable->add_guard_condition(notify_guard_condition));
 
   auto default_cbg = node->get_node_base_interface()->get_default_callback_group();
@@ -86,12 +85,15 @@ TEST_F(TestExecutorNotifyWaitable, wait) {
   auto waitables = node->get_node_waitables_interface();
   waitables->add_waitable(std::static_pointer_cast<rclcpp::Waitable>(waitable), default_cbg);
 
+  // notify the guard condition, this should trigger the on_execute_callback
+  notify_guard_condition->trigger();
+
   rclcpp::executors::SingleThreadedExecutor executor;
   executor.add_node(node);
   executor.spin_all(std::chrono::seconds(1));
   EXPECT_EQ(1u, on_execute_calls);
 
-  // on_execute_callback doesn't change if the topology doesn't change
+  // no further trigger, therefore no further callback
   executor.spin_all(std::chrono::seconds(1));
   EXPECT_EQ(1u, on_execute_calls);
 }

rclcpp/test/rclcpp/executors/test_executors.cpp

@@ -39,8 +39,10 @@
 #include "rclcpp/time_source.hpp"
 
 #include "test_msgs/msg/empty.hpp"
+#include "test_msgs/srv/empty.hpp"
 
 #include "./executor_types.hpp"
+#include "./test_waitable.hpp"
 
 using namespace std::chrono_literals;
 
@@ -331,106 +333,6 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteWithTimeout)
   spinner.join();
 }
 
-class TestWaitable : public rclcpp::Waitable
-{
-public:
-  TestWaitable() = default;
-
-  void
-  add_to_wait_set(rcl_wait_set_t & wait_set) override
-  {
-    if (trigger_count_ > 0) {
-      // Keep the gc triggered until the trigger count is reduced back to zero.
-      // This is necessary if trigger() results in the wait set waking, but not
-      // executing this waitable, in which case it needs to be re-triggered.
-      gc_.trigger();
-    }
-    rclcpp::detail::add_guard_condition_to_rcl_wait_set(wait_set, gc_);
-  }
-
-  void trigger()
-  {
-    trigger_count_++;
-    gc_.trigger();
-  }
-
-  bool
-  is_ready(const rcl_wait_set_t & wait_set) override
-  {
-    for (size_t i = 0; i < wait_set.size_of_guard_conditions; ++i) {
-      auto rcl_guard_condition = wait_set.guard_conditions[i];
-      if (&gc_.get_rcl_guard_condition() == rcl_guard_condition) {
-        return true;
-      }
-    }
-    return false;
-  }
-
-  std::shared_ptr<void>
-  take_data() override
-  {
-    return nullptr;
-  }
-
-  std::shared_ptr<void>
-  take_data_by_entity_id(size_t id) override
-  {
-    (void) id;
-    return nullptr;
-  }
-
-  void
-  execute(const std::shared_ptr<void> &) override
-  {
-    trigger_count_--;
-    count_++;
-    if (nullptr != on_execute_callback_) {
-      on_execute_callback_();
-    } else {
-      // TODO(wjwwood): I don't know why this was here, but probably it should
-      //   not be there, or test cases where that is important should use the
-      //   on_execute_callback?
-      std::this_thread::sleep_for(3ms);
-    }
-  }
-
-  void
-  set_on_execute_callback(std::function<void()> on_execute_callback)
-  {
-    on_execute_callback_ = on_execute_callback;
-  }
-
-  void
-  set_on_ready_callback(std::function<void(size_t, int)> callback) override
-  {
-    auto gc_callback = [callback](size_t count) {
-        callback(count, 0);
-      };
-    gc_.set_on_trigger_callback(gc_callback);
-  }
-
-  void
-  clear_on_ready_callback() override
-  {
-    gc_.set_on_trigger_callback(nullptr);
-  }
-
-  size_t
-  get_number_of_ready_guard_conditions() override {return 1;}
-
-  size_t
-  get_count() const
-  {
-    return count_;
-  }
-
-private:
-  std::atomic<size_t> trigger_count_ = 0;
-  std::atomic<size_t> count_ = 0;
-  rclcpp::GuardCondition gc_;
-  std::function<void()> on_execute_callback_ = nullptr;
-};
-
 TYPED_TEST(TestExecutors, spinAll)
 {
   using ExecutorType = TypeParam;
@@ -486,7 +388,7 @@ to_nanoseconds_helper(DurationT duration)
 //   - works nominally (it can execute entities)
 //   - it can execute multiple items at once
 //   - it does not wait for work to be available before returning
-TYPED_TEST(TestExecutors, spin_some)
+TYPED_TEST(TestExecutors, spinSome)
 {
   using ExecutorType = TypeParam;
 
@@ -578,7 +480,7 @@ TYPED_TEST(TestExecutors, spin_some)
 
 // The purpose of this test is to check that the ExecutorT.spin_some() method:
 //   - does not continue executing after max_duration has elapsed
-TYPED_TEST(TestExecutors, spin_some_max_duration)
+TYPED_TEST(TestExecutors, spinSomeMaxDuration)
 {
   using ExecutorType = TypeParam;
 
@@ -745,13 +647,6 @@ TYPED_TEST(TestExecutors, testSpinUntilFutureCompleteInterrupted)
 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;
@@ -869,3 +764,333 @@ TEST(TestExecutors, testSpinWithNonDefaultContext)
 
   rclcpp::shutdown(non_default_context);
 }
+
+TYPED_TEST(TestExecutors, releaseOwnershipEntityAfterSpinningCancel)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  auto future = std::async(std::launch::async, [&executor] {executor.spin();});
+
+  auto node = std::make_shared<rclcpp::Node>("test_node");
+  auto callback = [](
+    const test_msgs::srv::Empty::Request::SharedPtr, test_msgs::srv::Empty::Response::SharedPtr) {
+    };
+  auto server = node->create_service<test_msgs::srv::Empty>("test_service", callback);
+  while (!executor.is_spinning()) {
+    std::this_thread::sleep_for(50ms);
+  }
+  executor.add_node(node);
+  std::this_thread::sleep_for(50ms);
+  executor.cancel();
+  std::future_status future_status = future.wait_for(1s);
+  EXPECT_EQ(future_status, std::future_status::ready);
+
+  EXPECT_EQ(server.use_count(), 1);
+}
+
+TYPED_TEST(TestExecutors, testRaceDropCallbackGroupFromSecondThread)
+{
+  using ExecutorType = TypeParam;
+
+  // Create an executor
+  ExecutorType executor;
+  executor.add_node(this->node);
+
+  // Start spinning
+  auto executor_thread = std::thread(
+    [&executor]() {
+      executor.spin();
+    });
+
+  // As the problem is a race, we do this multiple times,
+  // to raise our chances of hitting the problem
+  for (size_t i = 0; i < 10; i++) {
+    auto cg = this->node->create_callback_group(
+      rclcpp::CallbackGroupType::MutuallyExclusive);
+    auto timer = this->node->create_timer(1s, [] {}, cg);
+    // sleep a bit, so that the spin thread can pick up the callback group
+    // and add it to the executor
+    std::this_thread::sleep_for(5ms);
+
+    // At this point the callbackgroup should be used within the waitset of the executor
+    // as we leave the scope, the reference to cg will be dropped.
+    // If the executor has a race, we will experience a segfault at this point.
+  }
+
+  executor.cancel();
+  executor_thread.join();
+}
+
+TYPED_TEST(TestExecutors, dropSomeTimer)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  auto node = std::make_shared<rclcpp::Node>("test_node");
+
+  bool timer1_works = false;
+  bool timer2_works = false;
+
+  auto timer1 = node->create_timer(std::chrono::milliseconds(10), [&timer1_works]() {
+        timer1_works = true;
+  });
+  auto timer2 = node->create_timer(std::chrono::milliseconds(10), [&timer2_works]() {
+        timer2_works = true;
+  });
+
+  executor.add_node(node);
+
+  // first let's make sure that both timers work
+  auto max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!timer1_works || !timer2_works) {
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  // delete timer 2. Note, the executor uses an unordered map internally, to order
+  // the entities added to the rcl waitset therefore the order is kind of undefined,
+  // and this test may be flaky. In case it triggers, something is most likely
+  // really broken.
+  timer2.reset();
+
+  timer1_works = false;
+  timer2_works = false;
+  max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!timer1_works && !timer2_works) {
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  ASSERT_TRUE(timer1_works || timer2_works);
+}
+
+TYPED_TEST(TestExecutors, dropSomeNodeWithTimer)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  auto node1 = std::make_shared<rclcpp::Node>("test_node_1");
+  auto node2 = std::make_shared<rclcpp::Node>("test_node_2");
+
+  bool timer1_works = false;
+  bool timer2_works = false;
+
+  auto timer1 = node1->create_timer(std::chrono::milliseconds(10), [&timer1_works]() {
+        timer1_works = true;
+  });
+  auto timer2 = node2->create_timer(std::chrono::milliseconds(10), [&timer2_works]() {
+        timer2_works = true;
+  });
+
+  executor.add_node(node1);
+  executor.add_node(node2);
+
+  // first let's make sure that both timers work
+  auto max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!timer1_works || !timer2_works) {
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  // delete node 1.
+  node1 = nullptr;
+
+  timer2_works = false;
+  max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!timer2_works) {
+    // let the executor pick up the node and the timer
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  ASSERT_TRUE(timer2_works);
+}
+
+TYPED_TEST(TestExecutors, dropSomeSubscription)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  auto node = std::make_shared<rclcpp::Node>("test_node");
+
+  bool sub1_works = false;
+  bool sub2_works = false;
+
+  auto sub1 = node->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub1_works](const test_msgs::msg::Empty &) {
+        sub1_works = true;
+  });
+  auto sub2 = node->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub2_works](const test_msgs::msg::Empty &) {
+        sub2_works = true;
+  });
+
+  auto pub = node->create_publisher<test_msgs::msg::Empty>("/test_drop", 10);
+
+  executor.add_node(node);
+
+  // first let's make sure that both timers work
+  auto max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!sub1_works || !sub2_works) {
+    pub->publish(test_msgs::msg::Empty());
+
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  // delete subscription 2. Note, the executor uses an unordered map internally, to order
+  // the entities added to the rcl waitset therefore the order is kind of undefined,
+  // and this test may be flaky. In case it triggers, something is most likely
+  // really broken.
+  sub2.reset();
+
+  sub1_works = false;
+  sub2_works = false;
+  max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!sub1_works && !sub2_works) {
+    pub->publish(test_msgs::msg::Empty());
+
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  ASSERT_TRUE(sub1_works || sub2_works);
+}
+
+TYPED_TEST(TestExecutors, dropSomeNodesWithSubscription)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  auto node = std::make_shared<rclcpp::Node>("test_node");
+  auto node1 = std::make_shared<rclcpp::Node>("test_node_1");
+  auto node2 = std::make_shared<rclcpp::Node>("test_node_2");
+
+  bool sub1_works = false;
+  bool sub2_works = false;
+
+  auto sub1 = node1->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub1_works](const test_msgs::msg::Empty &) {
+        sub1_works = true;
+  });
+  auto sub2 = node2->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub2_works](const test_msgs::msg::Empty &) {
+        sub2_works = true;
+  });
+
+  auto pub = node->create_publisher<test_msgs::msg::Empty>("/test_drop", 10);
+
+  executor.add_node(node);
+  executor.add_node(node1);
+  executor.add_node(node2);
+
+  // first let's make sure that both subscribers work
+  auto max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!sub1_works || !sub2_works) {
+    pub->publish(test_msgs::msg::Empty());
+
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  // delete node 2.
+  node2 = nullptr;
+
+  sub1_works = false;
+  max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while(!sub1_works) {
+    pub->publish(test_msgs::msg::Empty());
+
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  ASSERT_TRUE(sub1_works);
+}
+
+TYPED_TEST(TestExecutors, dropSubscriptionDuringCallback)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+
+  auto node = std::make_shared<rclcpp::Node>("test_node");
+
+  bool sub1_works = false;
+  bool sub2_works = false;
+
+  auto cbg = node->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive, true);
+  rclcpp::SubscriptionOptions sub_ops;
+  sub_ops.callback_group = cbg;
+
+  rclcpp::SubscriptionBase::SharedPtr sub1;
+  rclcpp::SubscriptionBase::SharedPtr sub2;
+
+  // Note, the executor uses an unordered map internally, to order
+  // the entities added to the rcl waitset therefore the order of the subscriptions
+  // is kind of undefined. Therefore each sub deletes the other one.
+  sub1 = node->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub1_works, &sub2](const test_msgs::msg::Empty &) {
+        sub1_works = true;
+        // delete the other subscriber
+        sub2.reset();
+  }, sub_ops);
+  sub2 = node->create_subscription<test_msgs::msg::Empty>("/test_drop", 10,
+      [&sub2_works, &sub1](const test_msgs::msg::Empty &) {
+        sub2_works = true;
+        // delete the other subscriber
+        sub1.reset();
+  }, sub_ops);
+
+  auto pub = node->create_publisher<test_msgs::msg::Empty>("/test_drop", 10);
+
+  // wait for both subs to be connected
+  auto max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(1500);
+  while ((sub1->get_publisher_count() == 0) || (sub2->get_publisher_count() == 0)) {
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  executor.add_node(node);
+
+  // publish some messages, until one subscriber fired. As both subscribers are
+  // connected to the same topic, they should fire in the same wait.
+  max_end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(100);
+  while (!sub1_works && !sub2_works) {
+    pub->publish(test_msgs::msg::Empty());
+
+    // let the executor pick up the node and the timers
+    executor.spin_all(std::chrono::milliseconds(10));
+
+    const auto cur_time = std::chrono::steady_clock::now();
+    ASSERT_LT(cur_time, max_end_time);
+  }
+
+  // only one subscriber must have worked, as the other
+  // one was deleted during the callback
+  ASSERT_TRUE(!sub1_works || !sub2_works);
+}

rclcpp/test/rclcpp/executors/test_executors_busy_waiting.cpp

@@ -0,0 +1,181 @@
+// Copyright 2024 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 <gtest/gtest.h>
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <sstream>
+#include <thread>
+
+#include "rclcpp/rclcpp.hpp"
+
+#include "./executor_types.hpp"
+#include "./test_waitable.hpp"
+
+using namespace std::chrono_literals;
+
+template<typename T>
+class TestBusyWaiting : public ::testing::Test
+{
+public:
+  void SetUp() override
+  {
+    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();
+    node = std::make_shared<rclcpp::Node>("node", test_name.str());
+    callback_group = node->create_callback_group(
+      rclcpp::CallbackGroupType::MutuallyExclusive,
+      /* automatically_add_to_executor_with_node =*/ false);
+
+    auto waitable_interfaces = node->get_node_waitables_interface();
+    waitable = std::make_shared<TestWaitable>();
+    waitable_interfaces->add_waitable(waitable, callback_group);
+
+    executor = std::make_shared<T>();
+    executor->add_callback_group(callback_group, node->get_node_base_interface());
+  }
+
+  void TearDown() override
+  {
+    rclcpp::shutdown();
+  }
+
+  void
+  set_up_and_trigger_waitable(std::function<void()> extra_callback = nullptr)
+  {
+    this->has_executed = false;
+    this->waitable->set_on_execute_callback([this, extra_callback]() {
+        if (!this->has_executed) {
+        // trigger once to see if the second trigger is handled or not
+        // this follow up trigger simulates new entities becoming ready while
+        // the executor is executing something else, e.g. subscription got data
+        // or a timer expired, etc.
+        // spin_some would not handle this second trigger, since it collects
+        // work only once, whereas spin_all should handle it since it
+        // collects work multiple times
+          this->waitable->trigger();
+          this->has_executed = true;
+        }
+        if (nullptr != extra_callback) {
+          extra_callback();
+        }
+    });
+    this->waitable->trigger();
+  }
+
+  void
+  check_for_busy_waits(std::chrono::steady_clock::time_point start_time)
+  {
+    // rough time based check, since the work to be done was very small it
+    // should be safe to check that we didn't use more than half the
+    // max duration, which itself is much larger than necessary
+    // however, it could still produce a false-positive
+    EXPECT_LT(
+      std::chrono::steady_clock::now() - start_time,
+      max_duration / 2)
+      << "executor took a long time to execute when it should have done "
+      << "nothing and should not have blocked either, but this could be a "
+      << "false negative if the computer is really slow";
+
+    // this check is making some assumptions about the implementation of the
+    // executors, but it should be safe to say that a busy wait may result in
+    // hundreds or thousands of calls to is_ready(), but "normal" executor
+    // behavior should be within an order of magnitude of the number of
+    // times that the waitable was executed
+    ASSERT_LT(waitable->get_is_ready_call_count(), 10u * this->waitable->get_count());
+  }
+
+  static constexpr auto max_duration = 10s;
+
+  rclcpp::Node::SharedPtr node;
+  rclcpp::CallbackGroup::SharedPtr callback_group;
+  std::shared_ptr<TestWaitable> waitable;
+  std::chrono::steady_clock::time_point start_time;
+  std::shared_ptr<T> executor;
+  bool has_executed;
+};
+
+TYPED_TEST_SUITE(TestBusyWaiting, ExecutorTypes, ExecutorTypeNames);
+
+TYPED_TEST(TestBusyWaiting, test_spin_all)
+{
+  this->set_up_and_trigger_waitable();
+
+  auto start_time = std::chrono::steady_clock::now();
+  this->executor->spin_all(this->max_duration);
+  this->check_for_busy_waits(start_time);
+  // this should get the initial trigger, and the follow up from in the callback
+  ASSERT_EQ(this->waitable->get_count(), 2u);
+}
+
+TYPED_TEST(TestBusyWaiting, test_spin_some)
+{
+  this->set_up_and_trigger_waitable();
+
+  auto start_time = std::chrono::steady_clock::now();
+  this->executor->spin_some(this->max_duration);
+  this->check_for_busy_waits(start_time);
+  // this should get the inital trigger, but not the follow up in the callback
+  ASSERT_EQ(this->waitable->get_count(), 1u);
+}
+
+TYPED_TEST(TestBusyWaiting, test_spin)
+{
+  std::condition_variable cv;
+  std::mutex cv_m;
+  bool first_check_passed = false;
+
+  this->set_up_and_trigger_waitable([&cv, &cv_m, &first_check_passed]() {
+      cv.notify_one();
+      if (!first_check_passed) {
+        std::unique_lock<std::mutex> lk(cv_m);
+        cv.wait_for(lk, 1s, [&]() {return first_check_passed;});
+      }
+  });
+
+  auto start_time = std::chrono::steady_clock::now();
+  std::thread t([this]() {
+      this->executor->spin();
+    });
+
+  // wait until thread has started (first execute of waitable)
+  {
+    std::unique_lock<std::mutex> lk(cv_m);
+    cv.wait_for(lk, 10s);
+  }
+  EXPECT_GT(this->waitable->get_count(), 0u);
+
+  first_check_passed = true;
+  cv.notify_one();
+
+  // wait until the executor has finished (second execute of waitable)
+  {
+    std::unique_lock<std::mutex> lk(cv_m);
+    cv.wait_for(lk, 10s);
+  }
+  EXPECT_EQ(this->waitable->get_count(), 2u);
+
+  this->executor->cancel();
+  t.join();
+
+  this->check_for_busy_waits(start_time);
+  // this should get the initial trigger, and the follow up from in the callback
+  ASSERT_EQ(this->waitable->get_count(), 2u);
+}

rclcpp/test/rclcpp/executors/test_executors_timer_cancel_behavior.cpp

@@ -267,12 +267,15 @@ public:
 
   void TearDown()
   {
-    node.reset();
+    executor.cancel();
 
     // Clean up thread object
     if (standalone_thread.joinable()) {
       standalone_thread.join();
     }
+
+    node.reset();
+    sim_clock_node.reset();
   }
 
   std::shared_ptr<TimerNode> node;
@@ -302,7 +305,6 @@ TYPED_TEST(TestTimerCancelBehavior, testTimer1CancelledWithExecutorSpin) {
   this->sim_clock_node->sleep_for(50ms);
   this->node->CancelTimer1();
   this->sim_clock_node->sleep_for(150ms);
-  this->executor.cancel();
 
   int t1_runs = this->node->GetTimer1Cnt();
   int t2_runs = this->node->GetTimer2Cnt();
@@ -320,7 +322,6 @@ TYPED_TEST(TestTimerCancelBehavior, testTimer2CancelledWithExecutorSpin) {
   this->sim_clock_node->sleep_for(50ms);
   this->node->CancelTimer2();
   this->sim_clock_node->sleep_for(150ms);
-  this->executor.cancel();
 
   int t1_runs = this->node->GetTimer1Cnt();
   int t2_runs = this->node->GetTimer2Cnt();
@@ -347,8 +348,6 @@ TYPED_TEST(TestTimerCancelBehavior, testHeadTimerCancelThenResetBehavior) {
   int t1_runs_final = this->node->GetTimer1Cnt();
   int t2_runs_final = this->node->GetTimer2Cnt();
 
-  this->executor.cancel();
-
   // T1 should have been restarted, and execute about 15 additional times.
   // Check 10 greater than initial, to account for some timing jitter.
   EXPECT_LT(t1_runs_initial + 50, t1_runs_final);
@@ -376,8 +375,6 @@ TYPED_TEST(TestTimerCancelBehavior, testBackTimerCancelThenResetBehavior) {
   int t1_runs_final = this->node->GetTimer1Cnt();
   int t2_runs_final = this->node->GetTimer2Cnt();
 
-  this->executor.cancel();
-
   // T2 should have been restarted, and execute about 15 additional times.
   // Check 10 greater than initial, to account for some timing jitter.
   EXPECT_LT(t2_runs_initial + 50, t2_runs_final);
@@ -411,8 +408,6 @@ TYPED_TEST(TestTimerCancelBehavior, testBothTimerCancelThenResetT1Behavior) {
   int t1_runs_final = this->node->GetTimer1Cnt();
   int t2_runs_final = this->node->GetTimer2Cnt();
 
-  this->executor.cancel();
-
   // T1 and T2 should have the same initial count.
   EXPECT_LE(std::abs(t1_runs_initial - t2_runs_initial), 1);
 
@@ -450,8 +445,6 @@ TYPED_TEST(TestTimerCancelBehavior, testBothTimerCancelThenResetT2Behavior) {
   int t1_runs_final = this->node->GetTimer1Cnt();
   int t2_runs_final = this->node->GetTimer2Cnt();
 
-  this->executor.cancel();
-
   // T1 and T2 should have the same initial count.
   EXPECT_LE(std::abs(t1_runs_initial - t2_runs_initial), 1);
 

rclcpp/test/rclcpp/executors/test_executors_warmup.cpp

@@ -0,0 +1,236 @@
+// Copyright 2024 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.
+
+/**
+ * This test checks all implementations of rclcpp::executor to check they pass they basic API
+ * tests. Anything specific to any executor in particular should go in a separate test file.
+ */
+
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <type_traits>
+#include <utility>
+
+#include "rclcpp/rclcpp.hpp"
+
+#include "test_msgs/msg/empty.hpp"
+
+#include "./executor_types.hpp"
+
+using namespace std::chrono_literals;
+
+template<typename T>
+class TestExecutorsWarmup : public ::testing::Test
+{
+public:
+  void SetUp()
+  {
+    rclcpp::init(0, nullptr);
+  }
+
+  void TearDown()
+  {
+    rclcpp::shutdown();
+  }
+};
+
+TYPED_TEST_SUITE(TestExecutorsWarmup, ExecutorTypes, ExecutorTypeNames);
+
+// This test verifies that spin_all is correctly collecting work multiple times
+// even when one of the items of work is a notifier waitable event and thus results in
+// rebuilding the entities collection.
+// When spin_all goes back to collect more work, it should see the ready items from
+// the new added entities
+TYPED_TEST(TestExecutorsWarmup, spin_all_doesnt_require_warmup)
+{
+  using ExecutorType = TypeParam;
+  ExecutorType executor;
+
+  // Enable intra-process to guarantee deterministic and synchronous delivery of the message / event
+  auto node_options = rclcpp::NodeOptions().use_intra_process_comms(true);
+  auto node = std::make_shared<rclcpp::Node>("test_node", node_options);
+
+  // Add node to the executor before creating the entities
+  executor.add_node(node);
+
+  // Create entities, this will produce a notifier waitable event, telling the executor to refresh
+  // the entities collection
+  auto publisher = node->create_publisher<test_msgs::msg::Empty>("test_topic", rclcpp::QoS(10));
+  size_t callback_count = 0;
+  auto callback = [&callback_count](test_msgs::msg::Empty::ConstSharedPtr) {callback_count++;};
+  auto subscription =
+    node->create_subscription<test_msgs::msg::Empty>(
+    "test_topic", rclcpp::QoS(10), std::move(callback));
+
+  ASSERT_EQ(callback_count, 0u);
+
+  // Publish a message so that the new entities (i.e. the subscriber) already have work to do
+  publisher->publish(test_msgs::msg::Empty());
+
+  // We need to select a duration that is greater than
+  // the time taken to refresh the entities collection and rebuild the waitset.
+  // spin-all is expected to process the notifier waitable event, rebuild the collection,
+  // and then collect more work, finding the subscription message event.
+  // This duration has been selected empirically.
+  executor.spin_all(std::chrono::milliseconds(500));
+
+  // Verify that the callback is called as part of the spin above
+  EXPECT_EQ(callback_count, 1u);
+}
+
+// Same test as `spin_all_doesnt_require_warmup`, but uses a callback group
+// This test reproduces the bug reported by https://github.com/ros2/rclcpp/issues/2589
+TYPED_TEST(TestExecutorsWarmup, spin_all_doesnt_require_warmup_with_cbgroup)
+{
+  using ExecutorType = TypeParam;
+
+  ExecutorType executor;
+
+  // Enable intra-process to guarantee deterministic and synchronous delivery of the message / event
+  auto node_options = rclcpp::NodeOptions().use_intra_process_comms(true);
+  auto node = std::make_shared<rclcpp::Node>("test_node", node_options);
+
+  auto callback_group = node->create_callback_group(
+    rclcpp::CallbackGroupType::MutuallyExclusive,
+    false);
+
+  // Add callback group to the executor before creating the entities
+  executor.add_callback_group(callback_group, node->get_node_base_interface());
+
+  // Create entities, this will produce a notifier waitable event, telling the executor to refresh
+  // the entities collection
+  auto publisher = node->create_publisher<test_msgs::msg::Empty>("test_topic", rclcpp::QoS(10));
+  size_t callback_count = 0;
+  auto callback = [&callback_count](test_msgs::msg::Empty::ConstSharedPtr) {callback_count++;};
+  rclcpp::SubscriptionOptions sub_options;
+  sub_options.callback_group = callback_group;
+  auto subscription =
+    node->create_subscription<test_msgs::msg::Empty>(
+    "test_topic", rclcpp::QoS(10), std::move(callback), sub_options);
+
+  ASSERT_EQ(callback_count, 0u);
+
+  // Publish a message so that the new entities (i.e. the subscriber) already have work to do
+  publisher->publish(test_msgs::msg::Empty());
+
+  // We need to select a duration that is greater than
+  // the time taken to refresh the entities collection and rebuild the waitset.
+  // spin-all is expected to process the notifier waitable event, rebuild the collection,
+  // and then collect more work, finding the subscription message event.
+  // This duration has been selected empirically.
+  executor.spin_all(std::chrono::milliseconds(500));
+
+  // Verify that the callback is called as part of the spin above
+  EXPECT_EQ(callback_count, 1u);
+}
+
+TYPED_TEST(TestExecutorsWarmup, spin_some_doesnt_require_warmup)
+{
+  using ExecutorType = TypeParam;
+
+  // TODO(alsora): currently only the events-executor passes this test.
+  // Enable single-threaded and multi-threaded executors
+  // when https://github.com/ros2/rclcpp/pull/2595 gets merged
+  if (
+    !std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>())
+  {
+    GTEST_SKIP();
+  }
+
+  ExecutorType executor;
+
+  // Enable intra-process to guarantee deterministic and synchronous delivery of the message / event
+  auto node_options = rclcpp::NodeOptions().use_intra_process_comms(true);
+  auto node = std::make_shared<rclcpp::Node>("test_node", node_options);
+
+  // Add node to the executor before creating the entities
+  executor.add_node(node);
+
+  // Create entities, this will produce a notifier waitable event, telling the executor to refresh
+  // the entities collection
+  auto publisher = node->create_publisher<test_msgs::msg::Empty>("test_topic", rclcpp::QoS(10));
+  size_t callback_count = 0;
+  auto callback = [&callback_count](test_msgs::msg::Empty::ConstSharedPtr) {callback_count++;};
+  auto subscription =
+    node->create_subscription<test_msgs::msg::Empty>(
+    "test_topic", rclcpp::QoS(10), std::move(callback));
+
+  ASSERT_EQ(callback_count, 0u);
+
+  // Publish a message so that the new entities (i.e. the subscriber) already have work to do
+  publisher->publish(test_msgs::msg::Empty());
+
+  // NOTE: intra-process communication is enabled, so the subscription will immediately see
+  // the new message, no risk of race conditions where spin_some gets called before the
+  // message has been delivered.
+  executor.spin_some();
+
+  // Verify that the callback is called as part of the spin above
+  EXPECT_EQ(callback_count, 1u);
+}
+
+TYPED_TEST(TestExecutorsWarmup, spin_some_doesnt_require_warmup_with_cbgroup)
+{
+  using ExecutorType = TypeParam;
+
+  // TODO(alsora): currently only the events-executor passes this test.
+  // Enable single-threaded and multi-threaded executors
+  // when https://github.com/ros2/rclcpp/pull/2595 gets merged
+  if (
+    !std::is_same<ExecutorType, rclcpp::experimental::executors::EventsExecutor>())
+  {
+    GTEST_SKIP();
+  }
+
+  ExecutorType executor;
+
+  // Enable intra-process to guarantee deterministic and synchronous delivery of the message / event
+  auto node_options = rclcpp::NodeOptions().use_intra_process_comms(true);
+  auto node = std::make_shared<rclcpp::Node>("test_node", node_options);
+
+  auto callback_group = node->create_callback_group(
+    rclcpp::CallbackGroupType::MutuallyExclusive,
+    false);
+
+  // Add callback group to the executor before creating the entities
+  executor.add_callback_group(callback_group, node->get_node_base_interface());
+
+  // Create entities, this will produce a notifier waitable event, telling the executor to refresh
+  // the entities collection
+  auto publisher = node->create_publisher<test_msgs::msg::Empty>("test_topic", rclcpp::QoS(10));
+  size_t callback_count = 0;
+  auto callback = [&callback_count](test_msgs::msg::Empty::ConstSharedPtr) {callback_count++;};
+  rclcpp::SubscriptionOptions sub_options;
+  sub_options.callback_group = callback_group;
+  auto subscription =
+    node->create_subscription<test_msgs::msg::Empty>(
+    "test_topic", rclcpp::QoS(10), std::move(callback), sub_options);
+
+  ASSERT_EQ(callback_count, 0u);
+
+  // Publish a message so that the new entities (i.e. the subscriber) already have work to do
+  publisher->publish(test_msgs::msg::Empty());
+
+  // NOTE: intra-process communication is enabled, so the subscription will immediately see
+  // the new message, no risk of race conditions where spin_some gets called before the
+  // message has been delivered.
+  executor.spin_some();
+
+  // Verify that the callback is called as part of the spin above
+  EXPECT_EQ(callback_count, 1u);
+}

rclcpp/test/rclcpp/executors/test_waitable.cpp

@@ -0,0 +1,126 @@
+// Copyright 2024 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 <atomic>
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <thread>
+
+#include "rclcpp/detail/add_guard_condition_to_rcl_wait_set.hpp"
+#include "rclcpp/waitable.hpp"
+
+#include "rcl/wait.h"
+
+#include "test_waitable.hpp"
+
+using namespace std::chrono_literals;
+
+void
+TestWaitable::add_to_wait_set(rcl_wait_set_t & wait_set)
+{
+  if (trigger_count_ > 0) {
+    // Keep the gc triggered until the trigger count is reduced back to zero.
+    // This is necessary if trigger() results in the wait set waking, but not
+    // executing this waitable, in which case it needs to be re-triggered.
+    gc_.trigger();
+  }
+  rclcpp::detail::add_guard_condition_to_rcl_wait_set(wait_set, gc_);
+}
+
+void TestWaitable::trigger()
+{
+  trigger_count_++;
+  gc_.trigger();
+}
+
+bool
+TestWaitable::is_ready(const rcl_wait_set_t & wait_set)
+{
+  is_ready_count_++;
+  for (size_t i = 0; i < wait_set.size_of_guard_conditions; ++i) {
+    auto rcl_guard_condition = wait_set.guard_conditions[i];
+    if (&gc_.get_rcl_guard_condition() == rcl_guard_condition) {
+      return true;
+    }
+  }
+  return false;
+}
+
+std::shared_ptr<void>
+TestWaitable::take_data()
+{
+  return nullptr;
+}
+
+std::shared_ptr<void>
+TestWaitable::take_data_by_entity_id(size_t id)
+{
+  (void) id;
+  return nullptr;
+}
+
+void
+TestWaitable::execute(const std::shared_ptr<void> &)
+{
+  trigger_count_--;
+  count_++;
+  if (nullptr != on_execute_callback_) {
+    on_execute_callback_();
+  } else {
+    // TODO(wjwwood): I don't know why this was here, but probably it should
+    //   not be there, or test cases where that is important should use the
+    //   on_execute_callback?
+    std::this_thread::sleep_for(3ms);
+  }
+}
+
+void
+TestWaitable::set_on_execute_callback(std::function<void()> on_execute_callback)
+{
+  on_execute_callback_ = on_execute_callback;
+}
+
+void
+TestWaitable::set_on_ready_callback(std::function<void(size_t, int)> callback)
+{
+  auto gc_callback = [callback](size_t count) {
+      callback(count, 0);
+    };
+  gc_.set_on_trigger_callback(gc_callback);
+}
+
+void
+TestWaitable::clear_on_ready_callback()
+{
+  gc_.set_on_trigger_callback(nullptr);
+}
+
+size_t
+TestWaitable::get_number_of_ready_guard_conditions()
+{
+  return 1;
+}
+
+size_t
+TestWaitable::get_count() const
+{
+  return count_;
+}
+
+size_t
+TestWaitable::get_is_ready_call_count() const
+{
+  return is_ready_count_;
+}

rclcpp/test/rclcpp/executors/test_waitable.hpp

@@ -0,0 +1,75 @@
+// Copyright 2024 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__EXECUTORS__TEST_WAITABLE_HPP_
+#define RCLCPP__EXECUTORS__TEST_WAITABLE_HPP_
+
+#include <atomic>
+#include <functional>
+#include <memory>
+
+#include "rclcpp/waitable.hpp"
+#include "rclcpp/guard_condition.hpp"
+
+#include "rcl/wait.h"
+
+class TestWaitable : public rclcpp::Waitable
+{
+public:
+  TestWaitable() = default;
+
+  void
+  add_to_wait_set(rcl_wait_set_t & wait_set) override;
+
+  void trigger();
+
+  bool
+  is_ready(const rcl_wait_set_t & wait_set) override;
+
+  std::shared_ptr<void>
+  take_data() override;
+
+  std::shared_ptr<void>
+  take_data_by_entity_id(size_t id) override;
+
+  void
+  execute(const std::shared_ptr<void> &) override;
+
+  void
+  set_on_execute_callback(std::function<void()> on_execute_callback);
+
+  void
+  set_on_ready_callback(std::function<void(size_t, int)> callback) override;
+
+  void
+  clear_on_ready_callback() override;
+
+  size_t
+  get_number_of_ready_guard_conditions() override;
+
+  size_t
+  get_count() const;
+
+  size_t
+  get_is_ready_call_count() const;
+
+private:
+  std::atomic<size_t> trigger_count_ = 0;
+  std::atomic<size_t> is_ready_count_ = 0;
+  std::atomic<size_t> count_ = 0;
+  rclcpp::GuardCondition gc_;
+  std::function<void()> on_execute_callback_ = nullptr;
+};
+
+#endif  // RCLCPP__EXECUTORS__TEST_WAITABLE_HPP_

rclcpp/test/rclcpp/test_client_common.cpp

@@ -431,7 +431,7 @@ TYPED_TEST(TestAllClientTypesWithServer, client_qos)
 
   rclcpp::ServicesQoS qos_profile;
   qos_profile.liveliness(rclcpp::LivelinessPolicy::Automatic);
-  rclcpp::Duration duration(std::chrono::nanoseconds(1));
+  rclcpp::Duration duration(std::chrono::milliseconds(1));
   qos_profile.deadline(duration);
   qos_profile.lifespan(duration);
   qos_profile.liveliness_lease_duration(duration);

rclcpp/test/rclcpp/test_clock.cpp

@@ -178,7 +178,10 @@ TEST_F(TestClockWakeup, no_wakeup_on_sim_time) {
 TEST_F(TestClockWakeup, multiple_threads_wait_on_one_clock) {
   auto clock = std::make_shared<rclcpp::Clock>(RCL_ROS_TIME);
 
-  std::vector<bool> thread_finished(10, false);
+  std::vector<std::atomic_bool> thread_finished(10);
+  for (std::atomic_bool & finished : thread_finished) {
+    finished = false;
+  }
 
   std::vector<std::thread> threads;
 
@@ -196,7 +199,7 @@ TEST_F(TestClockWakeup, multiple_threads_wait_on_one_clock) {
   // wait a bit so all threads can execute the sleep_until
   std::this_thread::sleep_for(std::chrono::milliseconds(500));
 
-  for (const bool & finished : thread_finished) {
+  for (const std::atomic_bool & finished : thread_finished) {
     EXPECT_FALSE(finished);
   }
 
@@ -207,7 +210,7 @@ TEST_F(TestClockWakeup, multiple_threads_wait_on_one_clock) {
   bool threads_finished = false;
   while (!threads_finished && start_time + std::chrono::seconds(1) > cur_time) {
     threads_finished = true;
-    for (const bool finished : thread_finished) {
+    for (const std::atomic_bool & finished : thread_finished) {
       if (!finished) {
         threads_finished = false;
       }
@@ -217,7 +220,7 @@ TEST_F(TestClockWakeup, multiple_threads_wait_on_one_clock) {
     cur_time = std::chrono::steady_clock::now();
   }
 
-  for (const bool finished : thread_finished) {
+  for (const std::atomic_bool & finished : thread_finished) {
     EXPECT_TRUE(finished);
   }
 

rclcpp/test/rclcpp/test_clock_conditional.cpp

@@ -0,0 +1,246 @@
+// Copyright 2025 Cellumation GmbH
+//
+// 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 <gtest/gtest.h>
+
+#include <chrono>
+
+#include "rcl/error_handling.h"
+#include "rcl/time.h"
+#include "rclcpp/clock.hpp"
+#include "rclcpp/rclcpp.hpp"
+#include "rclcpp/time_source.hpp"
+
+#include "../utils/rclcpp_gtest_macros.hpp"
+
+using namespace std::chrono_literals;
+
+class TestClockWakeup : public ::testing::TestWithParam<rcl_clock_type_e>
+{
+public:
+  void test_wakeup_before_sleep(const rclcpp::Clock::SharedPtr & clock)
+  {
+    std::atomic_bool thread_finished = false;
+    rclcpp::ClockConditionalVariable cond(clock);
+
+    bool stopSleeping = false;
+
+    std::thread wait_thread = std::thread(
+      [&cond, &clock, &stopSleeping, &thread_finished]()
+      {
+        // make sure the thread starts sleeping late
+        std::this_thread::sleep_for(std::chrono::milliseconds(100));
+        std::unique_lock lk(cond.mutex());
+        cond.wait_until(lk, clock->now() + std::chrono::seconds(3),
+        [&stopSleeping] () {return stopSleeping;});
+        thread_finished = true;
+      });
+
+    {
+      std::lock_guard lk(cond.mutex());
+      // stop sleeping after next notification
+      stopSleeping = true;
+    }
+    // notify the conditional, to recheck it pred
+    cond.notify_one();
+
+    auto start_time = std::chrono::steady_clock::now();
+    auto cur_time = start_time;
+    while (!thread_finished && start_time + std::chrono::seconds(1) > cur_time) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+      cur_time = std::chrono::steady_clock::now();
+    }
+
+    wait_thread.join();
+
+    EXPECT_TRUE(thread_finished);
+    EXPECT_LT(cur_time, start_time + std::chrono::seconds(1));
+  }
+
+  void test_wakeup_after_sleep(const rclcpp::Clock::SharedPtr & clock)
+  {
+    std::atomic_bool thread_finished = false;
+    rclcpp::ClockConditionalVariable cond(clock);
+
+    bool stopSleeping = false;
+
+    std::thread wait_thread = std::thread(
+      [&cond, &clock, &stopSleeping, &thread_finished]()
+      {
+        std::unique_lock lk(cond.mutex());
+        cond.wait_until(lk, clock->now() + std::chrono::seconds(3),
+        [&stopSleeping] () {return stopSleeping;});
+        thread_finished = true;
+      });
+
+    // make sure the thread is already sleeping before we send the cancel
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    {
+      std::lock_guard lk(cond.mutex());
+      // stop sleeping after next notification
+      stopSleeping = true;
+    }
+    // notify the conditional, to recheck it pred
+    cond.notify_one();
+
+    auto start_time = std::chrono::steady_clock::now();
+    auto cur_time = start_time;
+    while (!thread_finished && start_time + std::chrono::seconds(1) > cur_time) {
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+      cur_time = std::chrono::steady_clock::now();
+    }
+
+    wait_thread.join();
+
+    EXPECT_TRUE(thread_finished);
+    EXPECT_LT(cur_time, start_time + std::chrono::seconds(1));
+  }
+
+protected:
+  static void SetUpTestCase()
+  {
+    rclcpp::init(0, nullptr);
+  }
+
+  static void TearDownTestCase()
+  {
+    rclcpp::shutdown();
+  }
+
+  void SetUp()
+  {
+    node = std::make_shared<rclcpp::Node>("my_node");
+  }
+
+  void TearDown()
+  {
+    node.reset();
+  }
+
+  rclcpp::Node::SharedPtr node;
+};
+
+INSTANTIATE_TEST_SUITE_P(
+  ClockConditionalVariable,
+  TestClockWakeup,
+  ::testing::Values(
+    RCL_SYSTEM_TIME, RCL_ROS_TIME, RCL_STEADY_TIME
+));
+
+TEST_P(TestClockWakeup, wakeup_sleep) {
+  auto clock = std::make_shared<rclcpp::Clock>(GetParam());
+  test_wakeup_after_sleep(clock);
+  test_wakeup_before_sleep(clock);
+}
+
+TEST_F(TestClockWakeup, wakeup_sleep_ros_time_active) {
+  node->set_parameter({"use_sim_time", true});
+  auto clock = std::make_shared<rclcpp::Clock>(RCL_ROS_TIME);
+  rclcpp::TimeSource time_source(node);
+  time_source.attachClock(clock);
+
+  EXPECT_TRUE(clock->ros_time_is_active());
+
+  test_wakeup_after_sleep(clock);
+  test_wakeup_before_sleep(clock);
+}
+
+TEST_F(TestClockWakeup, no_wakeup_on_sim_time) {
+  node->set_parameter({"use_sim_time", true});
+  auto clock = std::make_shared<rclcpp::Clock>(RCL_ROS_TIME);
+  EXPECT_FALSE(clock->ros_time_is_active());
+
+  rclcpp::TimeSource time_source(node);
+  time_source.attachClock(clock);
+  EXPECT_TRUE(clock->ros_time_is_active());
+
+  std::atomic_bool thread_finished = false;
+  rclcpp::ClockConditionalVariable cond(clock);
+
+  bool stopSleeping = false;
+
+  std::thread wait_thread = std::thread(
+    [&cond, &clock, &stopSleeping, &thread_finished]()
+    {
+      std::unique_lock lk(cond.mutex());
+      // only sleep for an short period
+      cond.wait_until(lk, clock->now() + std::chrono::milliseconds(10),
+      [&stopSleeping] () {return stopSleeping;});
+      thread_finished = true;
+    });
+
+  // make sure, that the sim time clock does not wakeup, as no clock is provided
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+  EXPECT_FALSE(thread_finished);
+
+  {
+    std::lock_guard lk(cond.mutex());
+    // stop sleeping after next notification
+    stopSleeping = true;
+  }
+  // notify the conditional, to recheck it pred
+  cond.notify_one();
+
+  auto start_time = std::chrono::steady_clock::now();
+  auto cur_time = start_time;
+  while (!thread_finished && start_time + std::chrono::seconds(1) > cur_time) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    cur_time = std::chrono::steady_clock::now();
+  }
+
+  wait_thread.join();
+
+  EXPECT_TRUE(thread_finished);
+  EXPECT_LT(cur_time, start_time + std::chrono::seconds(1));
+}
+
+TEST_F(TestClockWakeup, wakeup_on_ros_shutdown) {
+  auto clock = std::make_shared<rclcpp::Clock>(RCL_ROS_TIME);
+
+  std::atomic_bool thread_finished = false;
+  rclcpp::ClockConditionalVariable cond(clock);
+
+  bool stopSleeping = false;
+
+  std::thread wait_thread = std::thread(
+    [&cond, &clock, &stopSleeping, &thread_finished]()
+    {
+      std::unique_lock lk(cond.mutex());
+      // only sleep for an short period
+      cond.wait_until(lk, clock->now() + std::chrono::seconds(10),
+      [&stopSleeping] () {return stopSleeping;});
+      thread_finished = true;
+    });
+
+  // wait a bit to be sure the thread is sleeping
+  std::this_thread::sleep_for(std::chrono::milliseconds(500));
+
+  EXPECT_FALSE(thread_finished);
+
+  rclcpp::shutdown();
+
+  auto start_time = std::chrono::steady_clock::now();
+  auto cur_time = start_time;
+  while (!thread_finished && start_time + std::chrono::seconds(1) > cur_time) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    cur_time = std::chrono::steady_clock::now();
+  }
+
+  EXPECT_TRUE(thread_finished);
+
+  wait_thread.join();
+
+  EXPECT_LT(cur_time, start_time + std::chrono::seconds(1));
+}

rclcpp/test/rclcpp/test_create_subscription.cpp

@@ -93,3 +93,19 @@ TEST_F(TestCreateSubscription, create_with_statistics) {
   ASSERT_NE(nullptr, subscription);
   EXPECT_STREQ("/ns/topic_name", subscription->get_topic_name());
 }
+
+TEST_F(TestCreateSubscription, create_with_intra_process_com) {
+  auto node = std::make_shared<rclcpp::Node>("my_node", "/ns");
+  auto options = rclcpp::SubscriptionOptions();
+  options.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
+
+  auto callback = [](test_msgs::msg::Empty::ConstSharedPtr) {};
+  rclcpp::Subscription<test_msgs::msg::Empty>::SharedPtr subscription;
+  ASSERT_NO_THROW(
+  {
+    subscription = rclcpp::create_subscription<test_msgs::msg::Empty>(
+      node, "topic_name", rclcpp::SystemDefaultsQoS(), callback, options);
+  });
+  ASSERT_NE(nullptr, subscription);
+  EXPECT_STREQ("/ns/topic_name", subscription->get_topic_name());
+}

rclcpp/test/rclcpp/test_generic_pubsub.cpp

@@ -15,6 +15,7 @@
 
 #include <gmock/gmock.h>
 
+#include <chrono>
 #include <future>
 #include <memory>
 #include <string>

rclcpp/test/rclcpp/test_logger_service.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <vector>

rclcpp/test/rclcpp/test_node_options.cpp

@@ -200,6 +200,57 @@ TEST(TestNodeOptions, copy) {
       rcl_arguments_get_count_unparsed(&other_rcl_options->arguments),
       rcl_arguments_get_count_unparsed(&rcl_options->arguments));
   }
+
+  {
+    // The following scope test is missing:
+    //    "arguments" because it is already tested in the above scopes
+    //    "parameter_event_publisher_options" because it can not be directly compared with EXPECT_EQ
+    //    "allocator" because it can not be directly compared with EXPECT_EQ
+
+    // We separate attribute modification from variable initialisation (copy assignment operator)
+    // to be sure the "non_default_options"'s properties are correctly set before testing the
+    // assignment operator.
+    auto non_default_options = rclcpp::NodeOptions();
+    non_default_options
+    .parameter_overrides({rclcpp::Parameter("foo", 0), rclcpp::Parameter("bar", "1")})
+    .use_global_arguments(false)
+    .enable_rosout(false)
+    .use_intra_process_comms(true)
+    .enable_topic_statistics(true)
+    .start_parameter_services(false)
+    .enable_logger_service(true)
+    .start_parameter_event_publisher(false)
+    .clock_type(RCL_SYSTEM_TIME)
+    .clock_qos(rclcpp::SensorDataQoS())
+    .use_clock_thread(false)
+    .parameter_event_qos(rclcpp::ClockQoS())
+    .rosout_qos(rclcpp::ParameterEventsQoS())
+    .allow_undeclared_parameters(true)
+    .automatically_declare_parameters_from_overrides(true);
+
+    auto copied_options = non_default_options;
+    EXPECT_EQ(non_default_options.parameter_overrides(), copied_options.parameter_overrides());
+    EXPECT_EQ(non_default_options.use_global_arguments(), copied_options.use_global_arguments());
+    EXPECT_EQ(non_default_options.enable_rosout(), copied_options.enable_rosout());
+    EXPECT_EQ(non_default_options.use_intra_process_comms(),
+      copied_options.use_intra_process_comms());
+    EXPECT_EQ(non_default_options.enable_topic_statistics(),
+      copied_options.enable_topic_statistics());
+    EXPECT_EQ(non_default_options.start_parameter_services(),
+      copied_options.start_parameter_services());
+    EXPECT_EQ(non_default_options.enable_logger_service(), copied_options.enable_logger_service());
+    EXPECT_EQ(non_default_options.start_parameter_event_publisher(),
+      copied_options.start_parameter_event_publisher());
+    EXPECT_EQ(non_default_options.clock_type(), copied_options.clock_type());
+    EXPECT_EQ(non_default_options.clock_qos(), copied_options.clock_qos());
+    EXPECT_EQ(non_default_options.use_clock_thread(), copied_options.use_clock_thread());
+    EXPECT_EQ(non_default_options.parameter_event_qos(), copied_options.parameter_event_qos());
+    EXPECT_EQ(non_default_options.rosout_qos(), copied_options.rosout_qos());
+    EXPECT_EQ(non_default_options.allow_undeclared_parameters(),
+      copied_options.allow_undeclared_parameters());
+    EXPECT_EQ(non_default_options.automatically_declare_parameters_from_overrides(),
+      copied_options.automatically_declare_parameters_from_overrides());
+  }
 }
 
 TEST(TestNodeOptions, append_parameter_override) {

rclcpp/test/rclcpp/test_parameter_events_filter.cpp

@@ -72,6 +72,13 @@ protected:
 /*
    Testing filters.
  */
+TEST_F(TestParameterEventFilter, invalide_arguments) {
+  EXPECT_THROW(
+    rclcpp::ParameterEventsFilter(nullptr, {"new"}, {nt}),
+    std::invalid_argument
+  );
+}
+
 TEST_F(TestParameterEventFilter, full_by_type) {
   auto res = rclcpp::ParameterEventsFilter(
     full,

rclcpp/test/rclcpp/test_parameter_service.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <utility>

rclcpp/test/rclcpp/test_publisher.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <future>
 #include <chrono>
 #include <memory>
 #include <string>
@@ -178,6 +179,21 @@ TEST_F(TestPublisher, various_creation_signatures) {
   }
 }
 
+/*
+   Testing publisher with intraprocess enabled and SystemDefaultQoS
+ */
+TEST_F(TestPublisher, test_publisher_with_system_default_qos) {
+  initialize(rclcpp::NodeOptions().use_intra_process_comms(false));
+  // explicitly enable intra-process comm with publisher option
+  auto options = rclcpp::PublisherOptions();
+  options.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
+  using test_msgs::msg::Empty;
+  ASSERT_NO_THROW(
+  {
+    auto publisher = node->create_publisher<Empty>("topic", rclcpp::SystemDefaultsQoS());
+  });
+}
+
 /*
    Testing publisher with intraprocess enabled and invalid QoS
  */
@@ -423,11 +439,10 @@ TEST_F(TestPublisher, intra_process_publish_failures) {
           publisher->get_publisher_handle().get(), msg.get()));
     }
   }
-  RCLCPP_EXPECT_THROW_EQ(
-    node->create_publisher<test_msgs::msg::Empty>(
-      "topic", rclcpp::QoS(0), options),
-    std::invalid_argument(
-      "intraprocess communication is not allowed with a zero qos history depth value"));
+  // a zero depth with KEEP_LAST doesn't make sense,
+  // this will be interpreted as SystemDefaultQoS by rclcpp.
+  EXPECT_NO_THROW(
+    node->create_publisher<test_msgs::msg::Empty>("topic", rclcpp::QoS(0), options));
 }
 
 TEST_F(TestPublisher, inter_process_publish_failures) {
@@ -702,11 +717,7 @@ TEST_F(TestPublisher, intra_process_transient_local) {
   auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {};
   struct IntraProcessCallback
   {
-    void callback_fun(size_t s)
-    {
-      (void) s;
-      called = true;
-    }
+    void callback_fun(size_t) {called = true;}
     bool called = false;
   };
   rclcpp::SubscriptionOptions sub_options_ipm_disabled;
@@ -763,3 +774,44 @@ TEST_F(TestPublisher, intra_process_transient_local) {
   EXPECT_FALSE(callback3.called);
   EXPECT_FALSE(callback4.called);
 }
+
+TEST_F(TestPublisher, intra_process_inter_process_mix_transient_local) {
+  constexpr auto history_depth = 10u;
+  initialize(rclcpp::NodeOptions().use_intra_process_comms(true));
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> pub_options_ipm_enabled;
+  pub_options_ipm_enabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Enable;
+
+  auto pub_ipm_enabled_transient_local_enabled = node->create_publisher<test_msgs::msg::Empty>(
+    "topic1",
+    rclcpp::QoS(rclcpp::KeepLast(history_depth)).transient_local(), pub_options_ipm_enabled);
+
+  test_msgs::msg::Empty msg;
+  pub_ipm_enabled_transient_local_enabled->publish(msg);
+
+  auto do_nothing = [](std::shared_ptr<const test_msgs::msg::Empty>) {};
+  struct SubscriptionCallback
+  {
+    void callback_fun(size_t) {called.set_value();}
+    std::promise<void> called;
+  };
+  rclcpp::SubscriptionOptions sub_options_ipm_disabled;
+  sub_options_ipm_disabled.use_intra_process_comm = rclcpp::IntraProcessSetting::Disable;
+  SubscriptionCallback intra_callback, inter_callback;
+  std::future<void> intra_callback_future = intra_callback.called.get_future(),
+    inter_callback_future = inter_callback.called.get_future();
+  auto sub_ipm_disabled_transient_local_enabled = node->create_subscription<test_msgs::msg::Empty>(
+    "topic1",
+    rclcpp::QoS(rclcpp::KeepLast(history_depth)).transient_local(),
+    do_nothing, sub_options_ipm_disabled);
+  sub_ipm_disabled_transient_local_enabled->set_on_new_intra_process_message_callback(
+    std::bind(&SubscriptionCallback::callback_fun, &intra_callback, std::placeholders::_1));
+  sub_ipm_disabled_transient_local_enabled->set_on_new_message_callback(
+    std::bind(&SubscriptionCallback::callback_fun, &inter_callback, std::placeholders::_1));
+  rclcpp::executors::SingleThreadedExecutor executor;
+  executor.add_node(node);
+  EXPECT_EQ(executor.spin_until_future_complete(inter_callback_future,
+    std::chrono::milliseconds(100)), rclcpp::FutureReturnCode::SUCCESS);
+  EXPECT_EQ(executor.spin_until_future_complete(intra_callback_future,
+    std::chrono::milliseconds(100)), rclcpp::FutureReturnCode::TIMEOUT);
+}

rclcpp/test/rclcpp/test_qos.cpp

@@ -259,3 +259,14 @@ TEST(TestQoS, qos_check_compatible)
     EXPECT_FALSE(ret.reason.empty());
   }
 }
+
+TEST(TestQoS, from_rmw_validity)
+{
+  rmw_qos_profile_t invalid_qos;
+  memset(&invalid_qos, 0, sizeof(invalid_qos));
+  reinterpret_cast<uint32_t &>(invalid_qos.history) = 999;
+
+  EXPECT_THROW({
+    rclcpp::QoSInitialization::from_rmw(invalid_qos);
+  }, std::invalid_argument);
+}

rclcpp/test/rclcpp/test_qos_event.cpp

@@ -70,27 +70,31 @@ TEST_F(TestQosEvent, test_publisher_constructor)
   auto publisher = node->create_publisher<test_msgs::msg::Empty>(
     topic_name, 10, options);
 
-  // options arg with one of the callbacks
-  options.event_callbacks.deadline_callback =
-    [node = node.get()](rclcpp::QOSDeadlineOfferedInfo & event) {
-      RCLCPP_INFO(
-        node->get_logger(),
-        "Offered deadline missed - total %d (delta %d)",
-        event.total_count, event.total_count_change);
-    };
-  publisher = node->create_publisher<test_msgs::msg::Empty>(
-    topic_name, 10, options);
+  if (rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED) &&
+    rmw_event_type_is_supported(RMW_EVENT_LIVELINESS_LOST))
+  {
+    // options arg with one of the callbacks
+    options.event_callbacks.deadline_callback =
+      [node = node.get()](rclcpp::QOSDeadlineOfferedInfo & event) {
+        RCLCPP_INFO(
+          node->get_logger(),
+          "Offered deadline missed - total %d (delta %d)",
+          event.total_count, event.total_count_change);
+      };
+    publisher = node->create_publisher<test_msgs::msg::Empty>(
+      topic_name, 10, options);
 
-  // options arg with two of the callbacks
-  options.event_callbacks.liveliness_callback =
-    [node = node.get()](rclcpp::QOSLivelinessLostInfo & event) {
-      RCLCPP_INFO(
-        node->get_logger(),
-        "Liveliness lost - total %d (delta %d)",
-        event.total_count, event.total_count_change);
-    };
-  publisher = node->create_publisher<test_msgs::msg::Empty>(
-    topic_name, 10, options);
+    // options arg with two of the callbacks
+    options.event_callbacks.liveliness_callback =
+      [node = node.get()](rclcpp::QOSLivelinessLostInfo & event) {
+        RCLCPP_INFO(
+          node->get_logger(),
+          "Liveliness lost - total %d (delta %d)",
+          event.total_count, event.total_count_change);
+      };
+    publisher = node->create_publisher<test_msgs::msg::Empty>(
+      topic_name, 10, options);
+  }
 
   // options arg with three of the callbacks
   options.event_callbacks.incompatible_qos_callback =
@@ -115,28 +119,32 @@ TEST_F(TestQosEvent, test_subscription_constructor)
   auto subscription = node->create_subscription<test_msgs::msg::Empty>(
     topic_name, 10, message_callback, options);
 
-  // options arg with one of the callbacks
-  options.event_callbacks.deadline_callback =
-    [node = node.get()](rclcpp::QOSDeadlineRequestedInfo & event) {
-      RCLCPP_INFO(
-        node->get_logger(),
-        "Requested deadline missed - total %d (delta %d)",
-        event.total_count, event.total_count_change);
-    };
-  subscription = node->create_subscription<test_msgs::msg::Empty>(
-    topic_name, 10, message_callback, options);
+  if (rmw_event_type_is_supported(RMW_EVENT_REQUESTED_DEADLINE_MISSED) &&
+    rmw_event_type_is_supported(RMW_EVENT_LIVELINESS_CHANGED))
+  {
+    // options arg with one of the callbacks
+    options.event_callbacks.deadline_callback =
+      [node = node.get()](rclcpp::QOSDeadlineRequestedInfo & event) {
+        RCLCPP_INFO(
+          node->get_logger(),
+          "Requested deadline missed - total %d (delta %d)",
+          event.total_count, event.total_count_change);
+      };
+    subscription = node->create_subscription<test_msgs::msg::Empty>(
+      topic_name, 10, message_callback, options);
 
-  // options arg with two of the callbacks
-  options.event_callbacks.liveliness_callback =
-    [node = node.get()](rclcpp::QOSLivelinessChangedInfo & event) {
-      RCLCPP_INFO(
-        node->get_logger(),
-        "Liveliness changed - alive %d (delta %d), not alive %d (delta %d)",
-        event.alive_count, event.alive_count_change,
-        event.not_alive_count, event.not_alive_count_change);
-    };
-  subscription = node->create_subscription<test_msgs::msg::Empty>(
-    topic_name, 10, message_callback, options);
+    // options arg with two of the callbacks
+    options.event_callbacks.liveliness_callback =
+      [node = node.get()](rclcpp::QOSLivelinessChangedInfo & event) {
+        RCLCPP_INFO(
+          node->get_logger(),
+          "Liveliness changed - alive %d (delta %d), not alive %d (delta %d)",
+          event.alive_count, event.alive_count_change,
+          event.not_alive_count, event.not_alive_count_change);
+      };
+    subscription = node->create_subscription<test_msgs::msg::Empty>(
+      topic_name, 10, message_callback, options);
+  }
 
   // options arg with three of the callbacks
   options.event_callbacks.incompatible_qos_callback =
@@ -204,14 +212,18 @@ TEST_F(TestQosEvent, test_default_incompatible_qos_callbacks)
   const auto timeout = std::chrono::seconds(10);
   ex.spin_until_future_complete(log_msgs_future, timeout);
 
-  EXPECT_EQ(
-    "New subscription discovered on topic '/ns/test_topic', requesting incompatible QoS. "
-    "No messages will be sent to it. Last incompatible policy: DURABILITY_QOS_POLICY",
-    pub_log_msg);
-  EXPECT_EQ(
-    "New publisher discovered on topic '/ns/test_topic', offering incompatible QoS. "
-    "No messages will be sent to it. Last incompatible policy: DURABILITY_QOS_POLICY",
-    sub_log_msg);
+  if (qos_check_compatible(qos_profile_publisher,
+    qos_profile_subscription).compatibility != rclcpp::QoSCompatibility::Ok)
+  {
+    EXPECT_EQ(
+      "New subscription discovered on topic '/ns/test_topic', requesting incompatible QoS. "
+      "No messages will be sent to it. Last incompatible policy: DURABILITY_QOS_POLICY",
+      pub_log_msg);
+    EXPECT_EQ(
+      "New publisher discovered on topic '/ns/test_topic', offering incompatible QoS. "
+      "No messages will be sent to it. Last incompatible policy: DURABILITY_QOS_POLICY",
+      sub_log_msg);
+  }
 
   rcutils_logging_set_output_handler(original_output_handler);
 }
@@ -223,7 +235,9 @@ TEST_F(TestQosEvent, construct_destruct_rcl_error) {
 
   // This callback requires some type of parameter, but it could be anything
   auto callback = [](int) {};
-  const rcl_publisher_event_type_t event_type = RCL_PUBLISHER_OFFERED_DEADLINE_MISSED;
+  const rcl_publisher_event_type_t event_type =
+    !rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED) ?
+    RCL_PUBLISHER_MATCHED : RCL_PUBLISHER_OFFERED_DEADLINE_MISSED;
 
   {
     // Logs error and returns
@@ -260,6 +274,10 @@ TEST_F(TestQosEvent, construct_destruct_rcl_error) {
 }
 
 TEST_F(TestQosEvent, execute) {
+  if (!rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED)) {
+    GTEST_SKIP();
+  }
+
   auto publisher = node->create_publisher<test_msgs::msg::Empty>(topic_name, 10);
   auto rcl_handle = publisher->get_publisher_handle();
 
@@ -292,7 +310,10 @@ TEST_F(TestQosEvent, add_to_wait_set) {
   // This callback requires some type of parameter, but it could be anything
   auto callback = [](int) {};
 
-  rcl_publisher_event_type_t event_type = RCL_PUBLISHER_OFFERED_DEADLINE_MISSED;
+  const rcl_publisher_event_type_t event_type =
+    !rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED) ?
+    RCL_PUBLISHER_MATCHED : RCL_PUBLISHER_OFFERED_DEADLINE_MISSED;
+
   rclcpp::EventHandler<decltype(callback), decltype(rcl_handle)> handler(
     callback, rcl_publisher_event_init, rcl_handle, event_type);
 
@@ -314,6 +335,12 @@ TEST_F(TestQosEvent, add_to_wait_set) {
 
 TEST_F(TestQosEvent, test_on_new_event_callback)
 {
+  if (!rmw_event_type_is_supported(RMW_EVENT_REQUESTED_DEADLINE_MISSED) ||
+    !rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED))
+  {
+    GTEST_SKIP();
+  }
+
   auto offered_deadline = rclcpp::Duration(std::chrono::milliseconds(1));
   auto requested_deadline = rclcpp::Duration(std::chrono::milliseconds(2));
 
@@ -359,17 +386,21 @@ TEST_F(TestQosEvent, test_invalid_on_new_event_callback)
   auto sub = node->create_subscription<test_msgs::msg::Empty>(topic_name, 10, message_callback);
   auto dummy_cb = [](size_t count_events) {(void)count_events;};
 
-  EXPECT_NO_THROW(
-    pub->set_on_new_qos_event_callback(dummy_cb, RCL_PUBLISHER_OFFERED_DEADLINE_MISSED));
+  if (rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED) &&
+    rmw_event_type_is_supported(RMW_EVENT_LIVELINESS_LOST))
+  {
+    EXPECT_NO_THROW(
+      pub->set_on_new_qos_event_callback(dummy_cb, RCL_PUBLISHER_OFFERED_DEADLINE_MISSED));
 
-  EXPECT_NO_THROW(
-    pub->clear_on_new_qos_event_callback(RCL_PUBLISHER_OFFERED_DEADLINE_MISSED));
+    EXPECT_NO_THROW(
+      pub->clear_on_new_qos_event_callback(RCL_PUBLISHER_OFFERED_DEADLINE_MISSED));
 
-  EXPECT_NO_THROW(
-    pub->set_on_new_qos_event_callback(dummy_cb, RCL_PUBLISHER_LIVELINESS_LOST));
+    EXPECT_NO_THROW(
+      pub->set_on_new_qos_event_callback(dummy_cb, RCL_PUBLISHER_LIVELINESS_LOST));
 
-  EXPECT_NO_THROW(
-    pub->clear_on_new_qos_event_callback(RCL_PUBLISHER_LIVELINESS_LOST));
+    EXPECT_NO_THROW(
+      pub->clear_on_new_qos_event_callback(RCL_PUBLISHER_LIVELINESS_LOST));
+  }
 
   EXPECT_NO_THROW(
     pub->set_on_new_qos_event_callback(dummy_cb, RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS));
@@ -383,11 +414,15 @@ TEST_F(TestQosEvent, test_invalid_on_new_event_callback)
   EXPECT_NO_THROW(
     pub->clear_on_new_qos_event_callback(RCL_PUBLISHER_MATCHED));
 
-  EXPECT_NO_THROW(
-    sub->set_on_new_qos_event_callback(dummy_cb, RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED));
+  if (rmw_event_type_is_supported(RMW_EVENT_REQUESTED_DEADLINE_MISSED) &&
+    rmw_event_type_is_supported(RMW_EVENT_LIVELINESS_CHANGED))
+  {
+    EXPECT_NO_THROW(
+      sub->set_on_new_qos_event_callback(dummy_cb, RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED));
 
-  EXPECT_NO_THROW(
-    sub->clear_on_new_qos_event_callback(RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED));
+    EXPECT_NO_THROW(
+      sub->clear_on_new_qos_event_callback(RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED));
+  }
 
   EXPECT_NO_THROW(
     sub->set_on_new_qos_event_callback(dummy_cb, RCL_SUBSCRIPTION_LIVELINESS_CHANGED));
@@ -407,24 +442,28 @@ TEST_F(TestQosEvent, test_invalid_on_new_event_callback)
   EXPECT_NO_THROW(
     sub->clear_on_new_qos_event_callback(RCL_SUBSCRIPTION_MATCHED));
 
-  std::function<void(size_t)> invalid_cb;
+  if (rmw_event_type_is_supported(RMW_EVENT_REQUESTED_DEADLINE_MISSED) &&
+    rmw_event_type_is_supported(RMW_EVENT_OFFERED_DEADLINE_MISSED))
+  {
+    std::function<void(size_t)> invalid_cb;
 
-  rclcpp::SubscriptionOptions sub_options;
-  sub_options.event_callbacks.deadline_callback = [](auto) {};
-  sub = node->create_subscription<test_msgs::msg::Empty>(
-    topic_name, 10, message_callback, sub_options);
+    rclcpp::SubscriptionOptions sub_options;
+    sub_options.event_callbacks.deadline_callback = [](auto) {};
+    sub = node->create_subscription<test_msgs::msg::Empty>(
+      topic_name, 10, message_callback, sub_options);
 
-  EXPECT_THROW(
-    sub->set_on_new_qos_event_callback(invalid_cb, RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED),
-    std::invalid_argument);
+    EXPECT_THROW(
+      sub->set_on_new_qos_event_callback(invalid_cb, RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED),
+      std::invalid_argument);
 
-  rclcpp::PublisherOptions pub_options;
-  pub_options.event_callbacks.deadline_callback = [](auto) {};
-  pub = node->create_publisher<test_msgs::msg::Empty>(topic_name, 10, pub_options);
+    rclcpp::PublisherOptions pub_options;
+    pub_options.event_callbacks.deadline_callback = [](auto) {};
+    pub = node->create_publisher<test_msgs::msg::Empty>(topic_name, 10, pub_options);
 
-  EXPECT_THROW(
-    pub->set_on_new_qos_event_callback(invalid_cb, RCL_PUBLISHER_OFFERED_DEADLINE_MISSED),
-    std::invalid_argument);
+    EXPECT_THROW(
+      pub->set_on_new_qos_event_callback(invalid_cb, RCL_PUBLISHER_OFFERED_DEADLINE_MISSED),
+      std::invalid_argument);
+  }
 }
 
 TEST_F(TestQosEvent, test_pub_matched_event_by_set_event_callback)

rclcpp/test/rclcpp/test_ring_buffer_implementation.cpp

@@ -139,3 +139,15 @@ TEST(TestRingBufferImplementation, basic_usage_unique_ptr) {
   EXPECT_EQ(false, rb.has_data());
   EXPECT_EQ(false, rb.is_full());
 }
+
+TEST(TestRingBufferImplementation, handle_nullptr_deletion) {
+  rclcpp::experimental::buffers::RingBufferImplementation<std::unique_ptr<int>> rb(3);
+  rb.enqueue(std::make_unique<int>(42));
+  rb.enqueue(nullptr);  // intentionally enqueuing nullptr
+  rb.enqueue(std::make_unique<int>(84));
+  auto all_data = rb.get_all_data();
+  EXPECT_EQ(3u, all_data.size());
+  EXPECT_EQ(42, *(all_data[0]));
+  EXPECT_EQ(nullptr, all_data[1]);
+  EXPECT_EQ(84, *(all_data[2]));
+}

rclcpp/test/rclcpp/test_service.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <string>
 #include <memory>
 #include <utility>
@@ -123,7 +124,7 @@ TEST_F(TestServiceSub, construction_and_destruction) {
   {
     ASSERT_THROW(
     {
-      auto service = node->create_service<rcl_interfaces::srv::ListParameters>(
+      auto service = subnode->create_service<rcl_interfaces::srv::ListParameters>(
         "invalid_service?", callback);
     }, rclcpp::exceptions::InvalidServiceNameError);
   }
@@ -335,7 +336,7 @@ TEST_F(TestService, rcl_service_request_subscription_get_actual_qos_error) {
 TEST_F(TestService, server_qos) {
   rclcpp::ServicesQoS qos_profile;
   qos_profile.liveliness(rclcpp::LivelinessPolicy::Automatic);
-  rclcpp::Duration duration(std::chrono::nanoseconds(1));
+  rclcpp::Duration duration(std::chrono::milliseconds(1));
   qos_profile.deadline(duration);
   qos_profile.lifespan(duration);
   qos_profile.liveliness_lease_duration(duration);

rclcpp/test/rclcpp/test_service_introspection.cpp

@@ -15,6 +15,7 @@
 #include <rcl/service_introspection.h>
 #include <rmw/rmw.h>
 
+#include <chrono>
 #include <map>
 #include <string>
 
@@ -92,9 +93,16 @@ TEST_F(TestServiceIntrospection, service_introspection_nominal)
   request->set__int64_value(42);
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+
+  // Wait for the introspection to attach to our subscription
+  size_t tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   auto future = client->async_send_request(request);
   ASSERT_EQ(
@@ -149,9 +157,11 @@ TEST_F(TestServiceIntrospection, service_introspection_nominal)
 TEST_F(TestServiceIntrospection, service_introspection_enable_disable_events)
 {
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_OFF);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+
+  ASSERT_EQ(sub->get_publisher_count(), 0);
 
   auto request = std::make_shared<BasicTypes::Request>();
   request->set__bool_value(true);
@@ -169,9 +179,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_events)
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+
+  // Wait for the introspection to attach to our subscription
+  size_t tries = 1000;
+  while (this->sub->get_publisher_count() < 1 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 1u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(
@@ -186,9 +203,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_events)
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_OFF);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_OFF);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+
+  // Wait for the introspection to attach to our subscription
+  tries = 1000;
+  while (this->sub->get_publisher_count() < 1 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 1u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(
@@ -203,9 +227,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_events)
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+
+  // Wait for the introspection to attach to our subscription
+  tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(
@@ -221,9 +252,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_events)
 TEST_F(TestServiceIntrospection, service_introspection_enable_disable_event_content)
 {
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+
+  // Wait for the introspection to attach to our subscription
+  size_t tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   auto request = std::make_shared<BasicTypes::Request>();
   request->set__bool_value(true);
@@ -245,9 +283,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_event_cont
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+
+  // Wait for the introspection to attach to our subscription
+  tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(
@@ -278,9 +323,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_event_cont
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_METADATA);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+
+  // Wait for the introspection to attach to our subscription
+  tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(
@@ -311,9 +363,16 @@ TEST_F(TestServiceIntrospection, service_introspection_enable_disable_event_cont
   events.clear();
 
   client->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
   service->configure_introspection(
-    node->get_clock(), rclcpp::SystemDefaultsQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+    node->get_clock(), rclcpp::ServicesQoS(), RCL_SERVICE_INTROSPECTION_CONTENTS);
+
+  // Wait for the introspection to attach to our subscription
+  tries = 1000;
+  while (this->sub->get_publisher_count() < 2 && tries-- > 0) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+  ASSERT_EQ(sub->get_publisher_count(), 2u);
 
   future = client->async_send_request(request);
   ASSERT_EQ(

rclcpp/test/rclcpp/test_subscription_content_filter.cpp

@@ -310,3 +310,25 @@ TEST_F(CLASSNAME(TestContentFilterSubscription, RMW_IMPLEMENTATION), content_fil
     }
   }
 }
+
+TEST_F(
+  CLASSNAME(
+    TestContentFilterSubscription,
+    RMW_IMPLEMENTATION), create_two_content_filters_with_same_topic_name_and_destroy) {
+
+  // Create another content filter
+  auto options = rclcpp::SubscriptionOptions();
+
+  std::string filter_expression = "int32_value > %0";
+  std::vector<std::string> expression_parameters = {"4"};
+
+  options.content_filter_options.filter_expression = filter_expression;
+  options.content_filter_options.expression_parameters = expression_parameters;
+
+  auto callback = [](std::shared_ptr<const test_msgs::msg::BasicTypes>) {};
+  auto sub_2 = node->create_subscription<test_msgs::msg::BasicTypes>(
+    "content_filter_topic", qos, callback, options);
+
+  EXPECT_NE(nullptr, sub_2);
+  sub_2.reset();
+}

rclcpp/test/rclcpp/test_subscription_options.cpp

@@ -60,7 +60,7 @@ TEST_F(TestSubscriptionOptions, topic_statistics_options_default_and_set) {
   EXPECT_EQ(options.topic_stats_options.state, rclcpp::TopicStatisticsState::NodeDefault);
   EXPECT_EQ(options.topic_stats_options.publish_topic, defaultPublishTopic);
   EXPECT_EQ(options.topic_stats_options.publish_period, 1s);
-  EXPECT_EQ(options.topic_stats_options.qos, rclcpp::SystemDefaultsQoS());
+  EXPECT_EQ(options.topic_stats_options.qos, rclcpp::SystemDefaultsQoS().keep_last(10));
 
   options.topic_stats_options.state = rclcpp::TopicStatisticsState::Enable;
   options.topic_stats_options.publish_topic = "topic_statistics";

rclcpp/test/rclcpp/test_wait_for_message.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <vector>

rclcpp_action/CHANGELOG.rst

@@ -3,6 +3,53 @@ Changelog for package rclcpp_action
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 
+28.1.10 (2025-06-23)
+--------------------
+* Replace std::default_random_engine with std::mt19937 (humble) (`#2847 <https://github.com/ros2/rclcpp/issues/2847>`_) (`#2867 <https://github.com/ros2/rclcpp/issues/2867>`_)
+* Added missing chrono includes (`#2854 <https://github.com/ros2/rclcpp/issues/2854>`_) (`#2856 <https://github.com/ros2/rclcpp/issues/2856>`_)
+* Contributors: mergify[bot]
+
+28.1.9 (2025-04-23)
+-------------------
+* fix(rclcpp_action): Fix sleep of expire thread in case of canceled timer (`#2800 <https://github.com/ros2/rclcpp/issues/2800>`_)
+  This fixes a bug, that the expire action thread would not sleep as,
+  the sleep duration was not computed correctly.
+  Co-authored-by: Janosch Machowinski <J.Machowinski@cellumation.com>
+* Contributors: Janosch Machowinski
+
+28.1.8 (2025-04-02)
+-------------------
+* Harden rclcpp_action::convert(). (`#2786 <https://github.com/ros2/rclcpp/issues/2786>`_) (`#2789 <https://github.com/ros2/rclcpp/issues/2789>`_)
+  * Harden rclcpp_action::convert().
+  * update docstring.
+  ---------
+  (cherry picked from commit ce86ef7e621d96ce50d6ec1b49e9e1cd4f0a828b)
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Contributors: mergify[bot]
+
+28.1.7 (2025-03-26)
+-------------------
+* fix: Fixed expiring of goals if events executor is used (`#2674 <https://github.com/ros2/rclcpp/issues/2674>`_)
+* Contributors: Janosch Machowinski
+
+28.1.6 (2024-12-18)
+-------------------
+
+28.1.5 (2024-09-19)
+-------------------
+
+28.1.4 (2024-09-06)
+-------------------
+
+28.1.3 (2024-06-27)
+-------------------
+
+28.1.2 (2024-05-13)
+-------------------
+
+28.1.1 (2024-04-24)
+-------------------
+
 28.1.0 (2024-04-16)
 -------------------
 * Remove references to index.ros.org. (`#2504 <https://github.com/ros2/rclcpp/issues/2504>`_)

rclcpp_action/include/rclcpp_action/types.hpp

@@ -39,12 +39,22 @@ RCLCPP_ACTION_PUBLIC
 std::string
 to_string(const GoalUUID & goal_id);
 
-// Convert C++ GoalID to rcl_action_goal_info_t
+/// Convert C++ GoalID to rcl_action_goal_info_t
+/**
+ * \param[in] goal_id C++ GoalUUID reference to be converted.
+ * \param[inout] info rcl_action_goal_info_t structure to be filled.
+ * \throws std::runtime_error if info is null.
+ */
 RCLCPP_ACTION_PUBLIC
 void
 convert(const GoalUUID & goal_id, rcl_action_goal_info_t * info);
 
-// Convert rcl_action_goal_info_t to C++ GoalID
+/// Convert rcl_action_goal_info_t to C++ GoalID
+/**
+ * \param[in] info rcl_action_goal_info_t reference to be converted.
+ * \param[inout] goal_id C++ GoalUUID structure to be filled.
+ * \throws std::runtime_error if goal_id is null.
+ */
 RCLCPP_ACTION_PUBLIC
 void
 convert(const rcl_action_goal_info_t & info, GoalUUID * goal_id);

rclcpp_action/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_action</name>
-  <version>28.1.0</version>
+  <version>28.1.10</version>
   <description>Adds action APIs for C++.</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_action/src/client.cpp

@@ -181,7 +181,7 @@ public:
   std::mutex cancel_requests_mutex;
 
   std::independent_bits_engine<
-    std::default_random_engine, 8, unsigned int> random_bytes_generator;
+    std::mt19937, 8, unsigned int> random_bytes_generator;
 };
 
 ClientBase::ClientBase(

rclcpp_action/src/server.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <inttypes.h>
 #include <memory>
 #include <mutex>
 #include <string>
@@ -81,6 +82,146 @@ public:
   {
   }
 
+  ~ServerBaseImpl()
+  {
+    stop_expire_thread();
+  }
+
+  std::recursive_mutex expire_thread_reentrant_mutex_;
+  std::thread expire_thread;
+
+  // must be a class member, so keep it in scope
+  std::function<void(size_t)> expire_reset_callback;
+
+  rclcpp::ClockConditionalVariable::SharedPtr expire_cv;
+  std::atomic<bool> expire_time_changed = false;
+  std::atomic<bool> terminate_expire_thread = true;
+
+  void
+  stop_expire_thread()
+  {
+    std::lock_guard<std::recursive_mutex> lock(expire_thread_reentrant_mutex_);
+    if(expire_cv) {
+      {
+        // Note, even though terminate_expire_thread is an atomic, we
+        // need to acquire the mutex, to avoid a race between the wait_until
+        // and the setting of this terminate_expire_thread
+        std::lock_guard<std::mutex> l(expire_cv->mutex());
+        terminate_expire_thread = true;
+      }
+      expire_cv->notify_one();
+      expire_thread.join();
+      expire_reset_callback = std::function<void(size_t)>();
+      expire_cv.reset();
+    }
+  }
+
+  void
+  start_expire_thread(std::function<void(size_t, int)> expire_ready_cb)
+  {
+    {
+      std::lock_guard<std::recursive_mutex> lock(expire_thread_reentrant_mutex_);
+      if(expire_cv) {
+        return;
+      }
+
+      terminate_expire_thread = false;
+      expire_cv = std::make_shared<rclcpp::ClockConditionalVariable>(clock_);
+    }
+
+    rclcpp::Context::SharedPtr context = rclcpp::contexts::get_global_default_context();
+    std::shared_ptr<rcl_context_t> rcl_context = context->get_rcl_context();
+
+    rcl_wait_set_t wait_set = rcl_get_zero_initialized_wait_set();
+    rcl_ret_t ret =
+      rcl_wait_set_init(&wait_set, num_subscriptions_, num_guard_conditions_, num_timers_,
+        num_clients_, num_services_, 0, rcl_context.get(), rcl_get_default_allocator());
+    if(ret != RCL_RET_OK) {
+      throw std::runtime_error("Internal error starting timer thread");
+    }
+
+    ret = rcl_action_wait_set_add_action_server(
+      &wait_set, action_server_.get(), NULL);
+    if (RCL_RET_OK != ret) {
+      rclcpp::exceptions::throw_from_rcl_error(ret, "ServerBase::add_to_wait_set() failed");
+    }
+
+    // extract the timer from the wait set
+    const rcl_timer_t *expire_timer = wait_set.timers[0];
+
+    // don't need the wait set any more
+    ret = rcl_wait_set_fini(&wait_set);
+
+    // the maybe_unused is needed here to satisfy cpplint
+    expire_reset_callback = [this] ([[maybe_unused]] size_t reset_calls)
+      {
+        {
+          std::lock_guard<std::mutex> l(expire_cv->mutex());
+          expire_time_changed = true;
+        }
+        expire_cv->notify_one();
+      };
+
+    rcl_event_callback_t rcl_reset_callback = rclcpp::detail::cpp_callback_trampoline<
+      decltype(expire_reset_callback), const void *, size_t>;
+
+    ret = rcl_timer_set_on_reset_callback(expire_timer, rcl_reset_callback,
+        static_cast<const void *>(&expire_reset_callback));
+    if (ret != RCL_RET_OK) {
+      rclcpp::exceptions::throw_from_rcl_error(ret, "Failed to set timer on reset callback");
+    }
+
+    expire_thread = std::thread([this, expire_timer, expire_ready_callback = expire_ready_cb]() {
+          rcl_clock_type_t clock_type = clock_->get_clock_type();
+
+          while(!terminate_expire_thread) {
+            {
+              std::unique_lock<std::mutex> l(expire_cv->mutex());
+
+              // reset to false, we are handling the time change right now
+              expire_time_changed = false;
+              int64_t time_until_call = 0;
+
+              auto ret2 = rcl_timer_get_time_until_next_call(expire_timer, &time_until_call);
+              if (ret2 == RCL_RET_TIMER_CANCELED) {
+                rclcpp::Duration endless(std::chrono::hours(100));
+                expire_cv->wait_until(l, clock_->now() + endless, [this] () -> bool {
+                  return expire_time_changed || terminate_expire_thread;
+                });
+                continue;
+              }
+
+              if(time_until_call > 0) {
+                rclcpp::Time next_wakeup(clock_->now().nanoseconds() + time_until_call, clock_type);
+                expire_cv->wait_until(l, next_wakeup, [this] () -> bool {
+                  return expire_time_changed || terminate_expire_thread;
+                });
+              }
+            }
+
+            // don't call expire callback if we are terminating
+            if(terminate_expire_thread) {
+              return;
+            }
+
+            bool is_ready = false;
+            if(rcl_timer_is_ready(expire_timer, &is_ready) == RCL_RET_OK && is_ready) {
+              // we need to cancel the timer here, to avoid a endless loop
+              // in case a new goal expires, the timer will be reset.
+              auto ret2 = rcl_timer_cancel(const_cast<rcl_timer_t *>(expire_timer));
+              if(ret2 != RCL_RET_OK) {
+                rclcpp::exceptions::throw_from_rcl_error(ret2, "Failed to cancel timer");
+              }
+
+              std::lock_guard<std::recursive_mutex> lock(expire_thread_reentrant_mutex_);
+              if(expire_ready_callback) {
+                expire_ready_callback(1, static_cast<int>(ServerBase::EntityType::Expired));
+              }
+            }
+          }
+      });
+  }
+
   // Lock for action_server_
   std::recursive_mutex action_server_reentrant_mutex_;
 
@@ -785,6 +926,8 @@ ServerBase::set_on_ready_callback(std::function<void(size_t, int)> callback)
             "is not callable.");
   }
 
+  pimpl_->start_expire_thread(callback);
+
   set_callback_to_entity(EntityType::GoalService, callback);
   set_callback_to_entity(EntityType::ResultService, callback);
   set_callback_to_entity(EntityType::CancelService, callback);
@@ -905,6 +1048,7 @@ ServerBase::clear_on_ready_callback()
     set_on_ready_callback(EntityType::GoalService, nullptr, nullptr);
     set_on_ready_callback(EntityType::ResultService, nullptr, nullptr);
     set_on_ready_callback(EntityType::CancelService, nullptr, nullptr);
+    pimpl_->stop_expire_thread();
     on_ready_callback_set_ = false;
   }
 

rclcpp_action/src/types.cpp

@@ -41,6 +41,9 @@ to_string(const GoalUUID & goal_id)
 void
 convert(const GoalUUID & goal_id, rcl_action_goal_info_t * info)
 {
+  if (info == nullptr) {
+    throw std::invalid_argument("info is nullptr");
+  }
   for (size_t i = 0; i < UUID_SIZE; ++i) {
     info->goal_id.uuid[i] = goal_id[i];
   }
@@ -49,6 +52,9 @@ convert(const GoalUUID & goal_id, rcl_action_goal_info_t * info)
 void
 convert(const rcl_action_goal_info_t & info, GoalUUID * goal_id)
 {
+  if (goal_id == nullptr) {
+    throw std::invalid_argument("goal_id is nullptr");
+  }
   for (size_t i = 0; i < UUID_SIZE; ++i) {
     (*goal_id)[i] = info.goal_id.uuid[i];
   }

rclcpp_action/test/test_client.cpp

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <chrono>
 #include <future>
 #include <map>
 #include <memory>
 #include <string>
 #include <utility>
 #include <thread>
-#include <chrono>
 
 #include "gtest/gtest.h"
 

rclcpp_action/test/test_types.cpp

@@ -41,6 +41,7 @@ TEST(TestActionTypes, goal_uuid_to_rcl_action_goal_info) {
   for (uint8_t i = 0; i < UUID_SIZE; ++i) {
     goal_id[i] = i;
   }
+  ASSERT_THROW(rclcpp_action::convert(goal_id, nullptr), std::invalid_argument);
   rcl_action_goal_info_t goal_info = rcl_action_get_zero_initialized_goal_info();
   rclcpp_action::convert(goal_id, &goal_info);
   for (uint8_t i = 0; i < UUID_SIZE; ++i) {
@@ -54,6 +55,7 @@ TEST(TestActionTypes, rcl_action_goal_info_to_goal_uuid) {
     goal_info.goal_id.uuid[i] = i;
   }
 
+  ASSERT_THROW(rclcpp_action::convert(goal_info, nullptr), std::invalid_argument);
   rclcpp_action::GoalUUID goal_id;
   rclcpp_action::convert(goal_info, &goal_id);
   for (uint8_t i = 0; i < UUID_SIZE; ++i) {

rclcpp_components/CHANGELOG.rst

@@ -2,6 +2,40 @@
 Changelog for package rclcpp_components
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
+28.1.10 (2025-06-23)
+--------------------
+* Added missing chrono includes (`#2854 <https://github.com/ros2/rclcpp/issues/2854>`_) (`#2856 <https://github.com/ros2/rclcpp/issues/2856>`_)
+* Contributors: mergify[bot]
+
+28.1.9 (2025-04-23)
+-------------------
+
+28.1.8 (2025-04-02)
+-------------------
+
+28.1.7 (2025-03-26)
+-------------------
+* add NO_UNDEFINED_SYMBOLS to rclcpp_components_register_node cmake macro (`#2746 <https://github.com/ros2/rclcpp/issues/2746>`_)
+* Contributors: Jonas Otto
+
+28.1.6 (2024-12-18)
+-------------------
+
+28.1.5 (2024-09-19)
+-------------------
+
+28.1.4 (2024-09-06)
+-------------------
+
+28.1.3 (2024-06-27)
+-------------------
+
+28.1.2 (2024-05-13)
+-------------------
+
+28.1.1 (2024-04-24)
+-------------------
+
 28.1.0 (2024-04-16)
 -------------------
 * Remove references to index.ros.org. (`#2504 <https://github.com/ros2/rclcpp/issues/2504>`_)

rclcpp_components/cmake/rclcpp_components_register_node.cmake

@@ -28,9 +28,11 @@
 # :type EXECUTOR: string
 # :param RESOURCE_INDEX: the ament resource index to register the components
 # :type RESOURCE_INDEX: string
+# :param NO_UNDEFINED_SYMBOLS: add linker flags to deny undefined symbols
+# :type NO_UNDEFINED_SYMBOLS: option
 #
 macro(rclcpp_components_register_node target)
-  cmake_parse_arguments(ARGS "" "PLUGIN;EXECUTABLE;EXECUTOR;RESOURCE_INDEX" "" ${ARGN})
+  cmake_parse_arguments(ARGS "NO_UNDEFINED_SYMBOLS" "PLUGIN;EXECUTABLE;EXECUTOR;RESOURCE_INDEX" "" ${ARGN})
   if(ARGS_UNPARSED_ARGUMENTS)
     message(FATAL_ERROR "rclcpp_components_register_node() called with unused "
       "arguments: ${ARGS_UNPARSED_ARGUMENTS}")
@@ -67,6 +69,26 @@ macro(rclcpp_components_register_node target)
     "${_RCLCPP_COMPONENTS_${resource_index}__NODES}${component};${_path}/$<TARGET_FILE_NAME:${target}>\n")
   list(APPEND _RCLCPP_COMPONENTS_PACKAGE_RESOURCE_INDICES ${resource_index})
 
+  if(ARGS_NO_UNDEFINED_SYMBOLS AND WIN32)
+    message(WARNING "NO_UNDEFINED_SYMBOLS is enabled for target \"${target}\", but this is unsupported on windows.")
+  elseif(ARGS_NO_UNDEFINED_SYMBOLS AND NOT WIN32)
+    check_cxx_compiler_flag("-Wl,--no-undefined" linker_supports_no_undefined)
+    if(linker_supports_no_undefined)
+      target_link_options("${target}" PRIVATE "-Wl,--no-undefined")
+    else()
+      message(WARNING "NO_UNDEFINED_SYMBOLS is enabled for target \"${target}\",\
+                but the linker does not support the \"--no-undefined\" flag.")
+    endif()
+
+    check_cxx_compiler_flag("-Wl,--no-allow-shlib-undefined" linker_supports_no_allow_shlib_undefined)
+    if(linker_supports_no_allow_shlib_undefined)
+      target_link_options("${target}" PRIVATE "-Wl,--no-allow-shlib-undefined")
+    else()
+      message(WARNING "NO_UNDEFINED_SYMBOLS is enabled for target \"${target}\",\
+                but the linker does not support the \"--no-allow-shlib-undefined\" flag.")
+    endif()
+  endif()
+
   configure_file(${rclcpp_components_NODE_TEMPLATE}
     ${PROJECT_BINARY_DIR}/rclcpp_components/node_main_configured_${node}.cpp.in)
   file(GENERATE OUTPUT ${PROJECT_BINARY_DIR}/rclcpp_components/node_main_${node}.cpp

rclcpp_components/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_components</name>
-  <version>28.1.0</version>
+  <version>28.1.10</version>
   <description>Package containing tools for dynamically loadable components</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_components/test/test_component_manager_api.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <memory>
 #include <string>
 

rclcpp_lifecycle/CHANGELOG.rst

@@ -3,6 +3,53 @@ Changelog for package rclcpp_lifecycle
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 
+28.1.10 (2025-06-23)
+--------------------
+* Added missing chrono includes (`#2854 <https://github.com/ros2/rclcpp/issues/2854>`_) (`#2856 <https://github.com/ros2/rclcpp/issues/2856>`_)
+* Contributors: mergify[bot]
+
+28.1.9 (2025-04-23)
+-------------------
+
+28.1.8 (2025-04-02)
+-------------------
+* should pull valid transition before trying to change the state. (`#2774 <https://github.com/ros2/rclcpp/issues/2774>`_) (`#2784 <https://github.com/ros2/rclcpp/issues/2784>`_)
+  (cherry picked from commit 7b6ee8a2e7a13d73f9f69368970390a9e0930448)
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Contributors: mergify[bot]
+
+28.1.7 (2025-03-26)
+-------------------
+
+28.1.6 (2024-12-18)
+-------------------
+
+28.1.5 (2024-09-19)
+-------------------
+
+28.1.4 (2024-09-06)
+-------------------
+
+28.1.3 (2024-06-27)
+-------------------
+* Revert "call shutdown in LifecycleNode dtor to avoid leaving the device in unknown state (2nd) (backport `#2528 <https://github.com/ros2/rclcpp/issues/2528>`_) (`#2542 <https://github.com/ros2/rclcpp/issues/2542>`_)" (`#2558 <https://github.com/ros2/rclcpp/issues/2558>`_)
+* call shutdown in LifecycleNode dtor to avoid leaving the device in unknown state (2nd) (backport `#2528 <https://github.com/ros2/rclcpp/issues/2528>`_) (`#2542 <https://github.com/ros2/rclcpp/issues/2542>`_)
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* rclcpp::shutdown should not be called before LifecycleNode dtor. (`#2527 <https://github.com/ros2/rclcpp/issues/2527>`_) (`#2540 <https://github.com/ros2/rclcpp/issues/2540>`_)
+  Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>
+* Contributors: Tomoya Fujita, mergify[bot]
+
+28.1.2 (2024-05-13)
+-------------------
+* Revert "call shutdown in LifecycleNode dtor to avoid leaving the device in un… (`#2450 <https://github.com/ros2/rclcpp/issues/2450>`_)" (`#2522 <https://github.com/ros2/rclcpp/issues/2522>`_) (`#2524 <https://github.com/ros2/rclcpp/issues/2524>`_)
+  This reverts commit 04ea0bb00293387791522590b7347a2282cda290.
+  (cherry picked from commit 42b0b5775b4e68718c5949308c9e1a059930ded7)
+  Co-authored-by: Chris Lalancette <clalancette@gmail.com>
+* Contributors: mergify[bot]
+
+28.1.1 (2024-04-24)
+-------------------
+
 28.1.0 (2024-04-16)
 -------------------
 * Remove references to index.ros.org. (`#2504 <https://github.com/ros2/rclcpp/issues/2504>`_)

rclcpp_lifecycle/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_lifecycle</name>
-  <version>28.1.0</version>
+  <version>28.1.10</version>
   <description>Package containing a prototype for lifecycle implementation</description>
 
   <maintainer email="ivanpauno@ekumenlabs.com">Ivan Paunovic</maintainer>

rclcpp_lifecycle/src/lifecycle_node.cpp

@@ -152,22 +152,6 @@ LifecycleNode::LifecycleNode(
 
 LifecycleNode::~LifecycleNode()
 {
-  // shutdown if necessary to avoid leaving the device in unknown state
-  if (LifecycleNode::get_current_state().id() !=
-    lifecycle_msgs::msg::State::PRIMARY_STATE_FINALIZED)
-  {
-    auto ret = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::ERROR;
-    auto finalized = LifecycleNode::shutdown(ret);
-    if (finalized.id() != lifecycle_msgs::msg::State::PRIMARY_STATE_FINALIZED ||
-      ret != rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS)
-    {
-      RCLCPP_WARN(
-        rclcpp::get_logger("rclcpp_lifecycle"),
-        "Shutdown error in destruction of LifecycleNode: final state(%s)",
-        finalized.label().c_str());
-    }
-  }
-
   // release sub-interfaces in an order that allows them to consult with node_base during tear-down
   node_waitables_.reset();
   node_time_source_.reset();

rclcpp_lifecycle/src/lifecycle_node_interface_impl.cpp

@@ -545,9 +545,7 @@ const State &
 LifecycleNode::LifecycleNodeInterfaceImpl::trigger_transition(uint8_t transition_id)
 {
   node_interfaces::LifecycleNodeInterface::CallbackReturn error;
-  change_state(transition_id, error);
-  (void) error;
-  return get_current_state();
+  return trigger_transition(transition_id, error);
 }
 
 const State &
@@ -555,7 +553,16 @@ LifecycleNode::LifecycleNodeInterfaceImpl::trigger_transition(
   uint8_t transition_id,
   node_interfaces::LifecycleNodeInterface::CallbackReturn & cb_return_code)
 {
-  change_state(transition_id, cb_return_code);
+  const rcl_lifecycle_transition_t * transition;
+  {
+    std::lock_guard<std::recursive_mutex> lock(state_machine_mutex_);
+
+    transition =
+      rcl_lifecycle_get_transition_by_id(state_machine_.current_state, transition_id);
+  }
+  if (transition) {
+    change_state(static_cast<uint8_t>(transition->id), cb_return_code);
+  }
   return get_current_state();
 }
 

rclcpp_lifecycle/test/benchmark/benchmark_lifecycle_client.cpp

@@ -14,6 +14,7 @@
 
 #include <benchmark/benchmark.h>
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <vector>

rclcpp_lifecycle/test/test_lifecycle_node.cpp

@@ -14,6 +14,7 @@
 
 
 #include <gtest/gtest.h>
+#include <chrono>
 #include <map>
 #include <memory>
 #include <set>
@@ -298,11 +299,146 @@ TEST_F(TestDefaultStateMachine, trigger_transition) {
   ASSERT_EQ(
     State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
       rclcpp_lifecycle::Transition(Transition::TRANSITION_CLEANUP)).id());
+  // supposed to fail because primary state is NOT active
+  ASSERT_EQ(
+    State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+      rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVE_SHUTDOWN)).id());
+  // supposed to fail because primary state is NOT inactive
+  ASSERT_EQ(
+    State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+      rclcpp_lifecycle::Transition(Transition::TRANSITION_INACTIVE_SHUTDOWN)).id());
   ASSERT_EQ(
     State::PRIMARY_STATE_FINALIZED, test_node->trigger_transition(
       rclcpp_lifecycle::Transition(Transition::TRANSITION_UNCONFIGURED_SHUTDOWN)).id());
 }
 
+TEST_F(TestDefaultStateMachine, trigger_transition_shutdown_id) {
+  // test Transition::TRANSITION_ACTIVE_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVE_SHUTDOWN)).id());
+  }
+
+  // test Transition::TRANSITION_INACTIVE_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_DEACTIVATE)).id());
+    // supposed to fail because primary state is NOT active
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVE_SHUTDOWN)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_INACTIVE_SHUTDOWN)).id());
+  }
+
+  // test Transition::TRANSITION_UNCONFIGURED_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_DEACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CLEANUP)).id());
+    // supposed to fail because primary state is NOT active
+    ASSERT_EQ(
+      State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVE_SHUTDOWN)).id());
+    // supposed to fail because primary state is NOT inactive
+    ASSERT_EQ(
+      State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_INACTIVE_SHUTDOWN)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_UNCONFIGURED_SHUTDOWN)).id());
+  }
+}
+
+TEST_F(TestDefaultStateMachine, trigger_transition_shutdown_label) {
+  // test Transition::TRANSITION_ACTIVE_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->shutdown().id());
+  }
+
+  // test Transition::TRANSITION_INACTIVE_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_DEACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->shutdown().id());
+  }
+
+  // test Transition::TRANSITION_UNCONFIGURED_SHUTDOWN
+  {
+    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
+
+    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_ACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_INACTIVE, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_DEACTIVATE)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_UNCONFIGURED, test_node->trigger_transition(
+        rclcpp_lifecycle::Transition(Transition::TRANSITION_CLEANUP)).id());
+    ASSERT_EQ(
+      State::PRIMARY_STATE_FINALIZED, test_node->shutdown().id());
+  }
+}
+
 TEST_F(TestDefaultStateMachine, trigger_transition_rcl_errors) {
   auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
 
@@ -447,146 +583,6 @@ TEST_F(TestDefaultStateMachine, bad_mood) {
   EXPECT_EQ(1u, test_node->number_of_callbacks);
 }
 
-
-TEST_F(TestDefaultStateMachine, shutdown_from_each_primary_state) {
-  auto success = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS;
-  auto reset_key = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::ERROR;
-
-  // PRIMARY_STATE_UNCONFIGURED to shutdown
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
-    auto finalized = test_node->shutdown(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(finalized.id(), State::PRIMARY_STATE_FINALIZED);
-  }
-
-  // PRIMARY_STATE_INACTIVE to shutdown
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    ret = reset_key;
-    auto finalized = test_node->shutdown(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(finalized.id(), State::PRIMARY_STATE_FINALIZED);
-  }
-
-  // PRIMARY_STATE_ACTIVE to shutdown
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    ret = reset_key;
-    auto activated = test_node->activate(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(activated.id(), State::PRIMARY_STATE_ACTIVE);
-    ret = reset_key;
-    auto finalized = test_node->shutdown(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(finalized.id(), State::PRIMARY_STATE_FINALIZED);
-  }
-
-  // PRIMARY_STATE_FINALIZED to shutdown
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<EmptyLifecycleNode>("testnode");
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    ret = reset_key;
-    auto activated = test_node->activate(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(activated.id(), State::PRIMARY_STATE_ACTIVE);
-    ret = reset_key;
-    auto finalized = test_node->shutdown(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(finalized.id(), State::PRIMARY_STATE_FINALIZED);
-    ret = reset_key;
-    auto finalized_again = test_node->shutdown(ret);
-    EXPECT_EQ(reset_key, ret);
-    EXPECT_EQ(finalized_again.id(), State::PRIMARY_STATE_FINALIZED);
-  }
-}
-
-TEST_F(TestDefaultStateMachine, test_shutdown_on_dtor) {
-  auto success = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS;
-  auto reset_key = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::ERROR;
-
-  bool shutdown_cb_called = false;
-  auto on_shutdown_callback =
-    [&shutdown_cb_called](const rclcpp_lifecycle::State &) ->
-    rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn {
-      shutdown_cb_called = true;
-      return rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS;
-    };
-
-  // PRIMARY_STATE_UNCONFIGURED to shutdown via dtor
-  shutdown_cb_called = false;
-  {
-    auto test_node = std::make_shared<rclcpp_lifecycle::LifecycleNode>("testnode");
-    test_node->register_on_shutdown(std::bind(on_shutdown_callback, std::placeholders::_1));
-    EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, test_node->get_current_state().id());
-    EXPECT_FALSE(shutdown_cb_called);
-  }
-  EXPECT_TRUE(shutdown_cb_called);
-
-  // PRIMARY_STATE_INACTIVE to shutdown via dtor
-  shutdown_cb_called = false;
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<rclcpp_lifecycle::LifecycleNode>("testnode");
-    test_node->register_on_shutdown(std::bind(on_shutdown_callback, std::placeholders::_1));
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    EXPECT_FALSE(shutdown_cb_called);
-  }
-  EXPECT_TRUE(shutdown_cb_called);
-
-  // PRIMARY_STATE_ACTIVE to shutdown via dtor
-  shutdown_cb_called = false;
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<rclcpp_lifecycle::LifecycleNode>("testnode");
-    test_node->register_on_shutdown(std::bind(on_shutdown_callback, std::placeholders::_1));
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    ret = reset_key;
-    auto activated = test_node->activate(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(activated.id(), State::PRIMARY_STATE_ACTIVE);
-    EXPECT_FALSE(shutdown_cb_called);
-  }
-  EXPECT_TRUE(shutdown_cb_called);
-
-  // PRIMARY_STATE_FINALIZED to shutdown via dtor
-  shutdown_cb_called = false;
-  {
-    auto ret = reset_key;
-    auto test_node = std::make_shared<rclcpp_lifecycle::LifecycleNode>("testnode");
-    test_node->register_on_shutdown(std::bind(on_shutdown_callback, std::placeholders::_1));
-    auto configured = test_node->configure(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(configured.id(), State::PRIMARY_STATE_INACTIVE);
-    ret = reset_key;
-    auto activated = test_node->activate(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(activated.id(), State::PRIMARY_STATE_ACTIVE);
-    ret = reset_key;
-    auto finalized = test_node->shutdown(ret);
-    EXPECT_EQ(success, ret);
-    EXPECT_EQ(finalized.id(), State::PRIMARY_STATE_FINALIZED);
-    EXPECT_TRUE(shutdown_cb_called);  // should be called already
-  }
-  EXPECT_TRUE(shutdown_cb_called);
-}
-
 TEST_F(TestDefaultStateMachine, lifecycle_subscriber) {
   auto test_node = std::make_shared<MoodyLifecycleNode<GoodMood>>("testnode");
 

rclcpp_lifecycle/test/test_lifecycle_publisher.cpp

@@ -55,12 +55,6 @@ public:
   explicit EmptyLifecycleNode(const std::string & node_name, const TimerType & timer_type)
   : rclcpp_lifecycle::LifecycleNode(node_name)
   {
-    rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> options;
-    publisher_ =
-      std::make_shared<rclcpp_lifecycle::LifecyclePublisher<test_msgs::msg::Empty>>(
-      get_node_base_interface().get(), std::string("topic"), rclcpp::QoS(10), options);
-    add_managed_entity(publisher_);
-
     // For coverage this is being added here
     switch (timer_type) {
       case TimerType::WALL_TIMER:
@@ -77,14 +71,6 @@ public:
         }
     }
   }
-
-  std::shared_ptr<rclcpp_lifecycle::LifecyclePublisher<test_msgs::msg::Empty>> publisher()
-  {
-    return publisher_;
-  }
-
-private:
-  std::shared_ptr<rclcpp_lifecycle::LifecyclePublisher<test_msgs::msg::Empty>> publisher_;
 };
 
 class TestLifecyclePublisher : public ::testing::TestWithParam<TimerType>
@@ -93,95 +79,103 @@ public:
   void SetUp()
   {
     rclcpp::init(0, nullptr);
-    node_ = std::make_shared<EmptyLifecycleNode>("node", GetParam());
   }
 
   void TearDown()
   {
     rclcpp::shutdown();
   }
-
-protected:
-  std::shared_ptr<EmptyLifecycleNode> node_;
 };
 
 TEST_P(TestLifecyclePublisher, publish_managed_by_node) {
+  auto node = std::make_shared<EmptyLifecycleNode>("node", GetParam());
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> options;
+  std::shared_ptr<rclcpp_lifecycle::LifecyclePublisher<test_msgs::msg::Empty>> publisher =
+    node->create_publisher<test_msgs::msg::Empty>(std::string("topic"), rclcpp::QoS(10), options);
+
   // transition via LifecycleNode
   auto success = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::SUCCESS;
   auto reset_key = rclcpp_lifecycle::node_interfaces::LifecycleNodeInterface::CallbackReturn::ERROR;
   auto ret = reset_key;
 
-  EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, node_->get_current_state().id());
-  node_->trigger_transition(
+  EXPECT_EQ(State::PRIMARY_STATE_UNCONFIGURED, node->get_current_state().id());
+  node->trigger_transition(
     rclcpp_lifecycle::Transition(Transition::TRANSITION_CONFIGURE), ret);
   ASSERT_EQ(success, ret);
   ret = reset_key;
-  node_->trigger_transition(
+  node->trigger_transition(
     rclcpp_lifecycle::Transition(Transition::TRANSITION_ACTIVATE), ret);
   ASSERT_EQ(success, ret);
   ret = reset_key;
-  EXPECT_TRUE(node_->publisher()->is_activated());
+  EXPECT_TRUE(publisher->is_activated());
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(*msg_ptr));
+    EXPECT_NO_THROW(publisher->publish(*msg_ptr));
   }
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(msg_ptr)));
+    EXPECT_NO_THROW(publisher->publish(std::move(msg_ptr)));
   }
   {
-    auto loaned_msg = node_->publisher()->borrow_loaned_message();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(loaned_msg)));
+    auto loaned_msg = publisher->borrow_loaned_message();
+    EXPECT_NO_THROW(publisher->publish(std::move(loaned_msg)));
   }
-  node_->trigger_transition(
+  node->trigger_transition(
     rclcpp_lifecycle::Transition(Transition::TRANSITION_DEACTIVATE), ret);
   ASSERT_EQ(success, ret);
   ret = reset_key;
   (void)ret;  // Just to make clang happy
-  EXPECT_FALSE(node_->publisher()->is_activated());
+  EXPECT_FALSE(publisher->is_activated());
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(*msg_ptr));
+    EXPECT_NO_THROW(publisher->publish(*msg_ptr));
   }
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(msg_ptr)));
+    EXPECT_NO_THROW(publisher->publish(std::move(msg_ptr)));
   }
   {
-    auto loaned_msg = node_->publisher()->borrow_loaned_message();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(loaned_msg)));
+    auto loaned_msg = publisher->borrow_loaned_message();
+    EXPECT_NO_THROW(publisher->publish(std::move(loaned_msg)));
   }
 }
 
 TEST_P(TestLifecyclePublisher, publish) {
+  auto node = std::make_shared<EmptyLifecycleNode>("node", GetParam());
+
+  rclcpp::PublisherOptionsWithAllocator<std::allocator<void>> options;
+  std::shared_ptr<rclcpp_lifecycle::LifecyclePublisher<test_msgs::msg::Empty>> publisher =
+    node->create_publisher<test_msgs::msg::Empty>(std::string("topic"), rclcpp::QoS(10), options);
+
   // transition via LifecyclePublisher
-  node_->publisher()->on_deactivate();
-  EXPECT_FALSE(node_->publisher()->is_activated());
+  publisher->on_deactivate();
+  EXPECT_FALSE(publisher->is_activated());
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(*msg_ptr));
+    EXPECT_NO_THROW(publisher->publish(*msg_ptr));
   }
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(msg_ptr)));
+    EXPECT_NO_THROW(publisher->publish(std::move(msg_ptr)));
   }
   {
-    auto loaned_msg = node_->publisher()->borrow_loaned_message();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(loaned_msg)));
+    auto loaned_msg = publisher->borrow_loaned_message();
+    EXPECT_NO_THROW(publisher->publish(std::move(loaned_msg)));
   }
-  node_->publisher()->on_activate();
-  EXPECT_TRUE(node_->publisher()->is_activated());
+  publisher->on_activate();
+  EXPECT_TRUE(publisher->is_activated());
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(*msg_ptr));
+    EXPECT_NO_THROW(publisher->publish(*msg_ptr));
   }
   {
     auto msg_ptr = std::make_unique<test_msgs::msg::Empty>();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(msg_ptr)));
+    EXPECT_NO_THROW(publisher->publish(std::move(msg_ptr)));
   }
   {
-    auto loaned_msg = node_->publisher()->borrow_loaned_message();
-    EXPECT_NO_THROW(node_->publisher()->publish(std::move(loaned_msg)));
+    auto loaned_msg = publisher->borrow_loaned_message();
+    EXPECT_NO_THROW(publisher->publish(std::move(loaned_msg)));
   }
 }
 

rclcpp_lifecycle/test/test_service.cpp

@@ -14,6 +14,7 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <string>
 #include <memory>
 #include <utility>