make it compile on osx

Signed-off-by: Karsten Knese <karsten@openrobotics.org>

rclcpp/CMakeLists.txt

@@ -4,7 +4,6 @@ project(rclcpp)
 
 find_package(ament_cmake_ros REQUIRED)
 find_package(builtin_interfaces REQUIRED)
-find_package(libstatistics_collector REQUIRED)
 find_package(rcl REQUIRED)
 find_package(rcl_interfaces REQUIRED)
 find_package(rcl_yaml_param_parser REQUIRED)
@@ -15,7 +14,6 @@ find_package(rosgraph_msgs REQUIRED)
 find_package(rosidl_runtime_cpp REQUIRED)
 find_package(rosidl_typesupport_c REQUIRED)
 find_package(rosidl_typesupport_cpp REQUIRED)
-find_package(statistics_msgs REQUIRED)
 find_package(tracetools REQUIRED)
 
 # Default to C++14
@@ -41,7 +39,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/detail/utilities.cpp
   src/rclcpp/duration.cpp
   src/rclcpp/event.cpp
-  src/rclcpp/exceptions/exceptions.cpp
+  src/rclcpp/exceptions.cpp
   src/rclcpp/executable_list.cpp
   src/rclcpp/executor.cpp
   src/rclcpp/executors.cpp
@@ -50,7 +48,6 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/executors/single_threaded_executor.cpp
   src/rclcpp/executors/static_executor_entities_collector.cpp
   src/rclcpp/executors/static_single_threaded_executor.cpp
-  src/rclcpp/future_return_code.cpp
   src/rclcpp/graph_listener.cpp
   src/rclcpp/guard_condition.cpp
   src/rclcpp/init_options.cpp
@@ -80,8 +77,6 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/publisher_base.cpp
   src/rclcpp/qos.cpp
   src/rclcpp/qos_event.cpp
-  src/rclcpp/serialization.cpp
-  src/rclcpp/serialized_message.cpp
   src/rclcpp/service.cpp
   src/rclcpp/signal_handler.cpp
   src/rclcpp/subscription_base.cpp
@@ -121,7 +116,6 @@ add_library(${PROJECT_NAME}
   ${${PROJECT_NAME}_SRCS})
 # specific order: dependents before dependencies
 ament_target_dependencies(${PROJECT_NAME}
-  "libstatistics_collector"
   "rcl"
   "rcl_yaml_param_parser"
   "rcpputils"
@@ -130,7 +124,6 @@ ament_target_dependencies(${PROJECT_NAME}
   "rosgraph_msgs"
   "rosidl_typesupport_cpp"
   "rosidl_runtime_cpp"
-  "statistics_msgs"
   "tracetools"
 )
 
@@ -150,7 +143,6 @@ install(
 ament_export_include_directories(include)
 ament_export_libraries(${PROJECT_NAME})
 
-ament_export_dependencies(libstatistics_collector)
 ament_export_dependencies(rcl)
 ament_export_dependencies(rcpputils)
 ament_export_dependencies(rcutils)
@@ -160,7 +152,6 @@ ament_export_dependencies(rosidl_typesupport_cpp)
 ament_export_dependencies(rosidl_typesupport_c)
 ament_export_dependencies(rosidl_runtime_cpp)
 ament_export_dependencies(rcl_yaml_param_parser)
-ament_export_dependencies(statistics_msgs)
 ament_export_dependencies(tracetools)
 
 if(BUILD_TESTING)
@@ -254,6 +245,18 @@ if(BUILD_TESTING)
     "test_msgs"
   )
   target_link_libraries(test_loaned_message ${PROJECT_NAME})
+  ament_add_gtest(test_intra_process_communication test/test_intra_process_communication.cpp
+    TIMEOUT 120)
+  if(TARGET test_intra_process_communication)
+    ament_target_dependencies(test_intra_process_communication
+      "rcl"
+      "rcl_interfaces"
+      "rmw"
+      "rosidl_typesupport_cpp"
+      "test_msgs"
+    )
+  endif()
+  target_link_libraries(test_intra_process_communication ${PROJECT_NAME})
 
   ament_add_gtest(test_node test/test_node.cpp TIMEOUT 240)
   if(TARGET test_node)
@@ -402,15 +405,6 @@ if(BUILD_TESTING)
       ${PROJECT_NAME}
     )
   endif()
-  ament_add_gtest(test_serialized_message test/test_serialized_message.cpp)
-  if(TARGET test_serialized_message)
-    ament_target_dependencies(test_serialized_message
-      test_msgs
-    )
-    target_link_libraries(test_serialized_message
-      ${PROJECT_NAME}
-    )
-  endif()
   ament_add_gtest(test_service test/test_service.cpp)
   if(TARGET test_service)
     ament_target_dependencies(test_service
@@ -449,7 +443,6 @@ if(BUILD_TESTING)
       "rcl"
       "test_msgs"
     )
-    target_link_libraries(test_subscription_traits ${PROJECT_NAME})
   endif()
   ament_add_gtest(test_find_weak_nodes test/test_find_weak_nodes.cpp)
   if(TARGET test_find_weak_nodes)
@@ -554,26 +547,6 @@ if(BUILD_TESTING)
     target_link_libraries(test_wait_set ${PROJECT_NAME})
   endif()
 
-  ament_add_gtest(test_subscription_topic_statistics test/topic_statistics/test_subscription_topic_statistics.cpp)
-  if(TARGET test_subscription_topic_statistics)
-    ament_target_dependencies(test_subscription_topic_statistics
-      "libstatistics_collector"
-      "rcl_interfaces"
-      "rcutils"
-      "rmw"
-      "rosidl_runtime_cpp"
-      "rosidl_typesupport_cpp"
-      "statistics_msgs"
-      "test_msgs")
-    target_link_libraries(test_subscription_topic_statistics ${PROJECT_NAME})
-  endif()
-
-  ament_add_gtest(test_subscription_options test/test_subscription_options.cpp)
-  if(TARGET test_subscription_options)
-    ament_target_dependencies(test_subscription_options "rcl")
-    target_link_libraries(test_subscription_options ${PROJECT_NAME})
-  endif()
-
   # Install test resources
   install(
     DIRECTORY test/resources

rclcpp/include/rclcpp/any_executable.hpp

@@ -29,6 +29,8 @@
 
 namespace rclcpp
 {
+namespace executor
+{
 
 struct AnyExecutable
 {
@@ -45,15 +47,10 @@ struct AnyExecutable
   rclcpp::ClientBase::SharedPtr client;
   rclcpp::Waitable::SharedPtr waitable;
   // These are used to keep the scope on the containing items
-  rclcpp::CallbackGroup::SharedPtr callback_group;
+  rclcpp::callback_group::CallbackGroup::SharedPtr callback_group;
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base;
 };
 
-namespace executor
-{
-
-using AnyExecutable [[deprecated("use rclcpp::AnyExecutable instead")]] = AnyExecutable;
-
 }  // namespace executor
 }  // namespace rclcpp
 

rclcpp/include/rclcpp/callback_group.hpp

@@ -40,6 +40,9 @@ class NodeTopics;
 class NodeWaitables;
 }  // namespace node_interfaces
 
+namespace callback_group
+{
+
 enum class CallbackGroupType
 {
   MutuallyExclusive,
@@ -159,12 +162,6 @@ private:
   }
 };
 
-namespace callback_group
-{
-
-using CallbackGroupType [[deprecated("use rclcpp::CallbackGroupType instead")]] = CallbackGroupType;
-using CallbackGroup [[deprecated("use rclcpp::CallbackGroup instead")]] = CallbackGroup;
-
 }  // namespace callback_group
 }  // namespace rclcpp
 

rclcpp/include/rclcpp/contexts/default_context.hpp

@@ -22,6 +22,8 @@ namespace rclcpp
 {
 namespace contexts
 {
+namespace default_context
+{
 
 class DefaultContext : public rclcpp::Context
 {
@@ -36,21 +38,6 @@ RCLCPP_PUBLIC
 DefaultContext::SharedPtr
 get_global_default_context();
 
-namespace default_context
-{
-
-using DefaultContext
-[[deprecated("use rclcpp::contexts::DefaultContext instead")]] = DefaultContext;
-
-[[deprecated("use rclcpp::contexts::get_global_default_context() instead")]]
-RCLCPP_PUBLIC
-inline
-DefaultContext::SharedPtr
-get_global_default_context()
-{
-  return rclcpp::contexts::get_global_default_context();
-}
-
 }  // namespace default_context
 }  // namespace contexts
 }  // namespace rclcpp

rclcpp/include/rclcpp/create_client.hpp

@@ -35,7 +35,7 @@ create_client(
   std::shared_ptr<node_interfaces::NodeServicesInterface> node_services,
   const std::string & service_name,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   rcl_client_options_t options = rcl_client_get_default_options();
   options.qos = qos_profile;

rclcpp/include/rclcpp/create_publisher.hpp

@@ -43,6 +43,7 @@ std::shared_ptr<PublisherT>
 create_publisher(
   NodeT & node,
   const std::string & topic_name,
+  const rosidl_message_type_support_t & type_support,
   const rclcpp::QoS & qos,
   const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options = (
     rclcpp::PublisherOptionsWithAllocator<AllocatorT>()
@@ -56,7 +57,7 @@ create_publisher(
   // Create the publisher.
   auto pub = node_topics->create_publisher(
     topic_name,
-    rclcpp::create_publisher_factory<MessageT, AllocatorT, PublisherT>(options),
+    rclcpp::create_publisher_factory<MessageT, AllocatorT, PublisherT>(options, type_support),
     qos
   );
 
@@ -66,6 +67,28 @@ create_publisher(
   return std::dynamic_pointer_cast<PublisherT>(pub);
 }
 
+template<
+  typename MessageT,
+  typename AllocatorT = std::allocator<void>,
+  typename PublisherT = rclcpp::Publisher<MessageT, AllocatorT>,
+  typename NodeT>
+std::shared_ptr<PublisherT>
+create_publisher(
+  NodeT & node,
+  const std::string & topic_name,
+  const rclcpp::QoS & qos,
+  const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options = (
+    rclcpp::PublisherOptionsWithAllocator<AllocatorT>()
+  )
+)
+{
+  const auto type_support = *rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>();
+
+  return create_publisher<MessageT, AllocatorT, PublisherT, NodeT>(
+    node, topic_name, type_support,
+    qos, options);
+}
+
 }  // namespace rclcpp
 
 #endif  // RCLCPP__CREATE_PUBLISHER_HPP_

rclcpp/include/rclcpp/create_service.hpp

@@ -37,7 +37,7 @@ create_service(
   const std::string & service_name,
   CallbackT && callback,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   rclcpp::AnyServiceCallback<ServiceT> any_service_callback;
   any_service_callback.set(std::forward<CallbackT>(callback));

rclcpp/include/rclcpp/create_subscription.hpp

@@ -51,14 +51,13 @@ typename std::shared_ptr<SubscriptionT>
 create_subscription(
   NodeT && node,
   const std::string & topic_name,
+  const rosidl_message_type_support_t & type_support,
   const rclcpp::QoS & qos,
   CallbackT && callback,
-  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options = (
-    rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>()
-  ),
-  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat = (
+  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options =
+    rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>(),
+  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat =
     MessageMemoryStrategyT::create_default()
-  )
 )
 {
   using rclcpp::node_interfaces::get_node_topics_interface;
@@ -67,7 +66,8 @@ create_subscription(
   auto factory = rclcpp::create_subscription_factory<MessageT>(
     std::forward<CallbackT>(callback),
     options,
-    msg_mem_strat
+    msg_mem_strat,
+    type_support
   );
 
   auto sub = node_topics->create_subscription(topic_name, factory, qos);
@@ -76,6 +76,39 @@ create_subscription(
   return std::dynamic_pointer_cast<SubscriptionT>(sub);
 }
 
+template<
+  typename MessageT,
+  typename CallbackT,
+  typename AllocatorT = std::allocator<void>,
+  typename CallbackMessageT =
+  typename rclcpp::subscription_traits::has_message_type<CallbackT>::type,
+  typename SubscriptionT = rclcpp::Subscription<CallbackMessageT, AllocatorT>,
+  typename MessageMemoryStrategyT = rclcpp::message_memory_strategy::MessageMemoryStrategy<
+    CallbackMessageT,
+    AllocatorT
+  >,
+  typename NodeT>
+typename std::shared_ptr<SubscriptionT>
+create_subscription(
+  NodeT && node,
+  const std::string & topic_name,
+  const rclcpp::QoS & qos,
+  CallbackT && callback,
+  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options =
+    rclcpp::SubscriptionOptionsWithAllocator<AllocatorT>(),
+  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat =
+    MessageMemoryStrategyT::create_default()
+)
+{
+  const auto type_support = *rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>();
+
+  return create_subscription<MessageT, CallbackT, AllocatorT, CallbackMessageT, SubscriptionT,
+           MessageMemoryStrategyT>(
+    std::forward<NodeT>(
+      node), topic_name, type_support, qos, std::forward<CallbackT>(
+      callback), options, msg_mem_strat);
+}
+
 }  // namespace rclcpp
 
 #endif  // RCLCPP__CREATE_SUBSCRIPTION_HPP_

rclcpp/include/rclcpp/create_timer.hpp

@@ -37,7 +37,7 @@ create_timer(
   rclcpp::Clock::SharedPtr clock,
   rclcpp::Duration period,
   CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr)
 {
   auto timer = rclcpp::GenericTimer<CallbackT>::make_shared(
     clock,
@@ -57,7 +57,7 @@ create_timer(
   rclcpp::Clock::SharedPtr clock,
   rclcpp::Duration period,
   CallbackT && callback,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr)
 {
   return create_timer(
     rclcpp::node_interfaces::get_node_base_interface(node),

rclcpp/include/rclcpp/detail/resolve_enable_topic_statistics.hpp

@@ -1,54 +0,0 @@
-// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
-//
-// 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__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_
-#define RCLCPP__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_
-
-#include <stdexcept>
-
-#include "rclcpp/topic_statistics_state.hpp"
-
-namespace rclcpp
-{
-namespace detail
-{
-
-/// Return whether or not topic statistics is enabled, resolving "NodeDefault" if needed.
-template<typename OptionsT, typename NodeBaseT>
-bool
-resolve_enable_topic_statistics(const OptionsT & options, const NodeBaseT & node_base)
-{
-  bool topic_stats_enabled;
-  switch (options.topic_stats_options.state) {
-    case TopicStatisticsState::Enable:
-      topic_stats_enabled = true;
-      break;
-    case TopicStatisticsState::Disable:
-      topic_stats_enabled = false;
-      break;
-    case TopicStatisticsState::NodeDefault:
-      topic_stats_enabled = node_base.get_enable_topic_statistics_default();
-      break;
-    default:
-      throw std::runtime_error("Unrecognized EnableTopicStatistics value");
-      break;
-  }
-
-  return topic_stats_enabled;
-}
-
-}  // namespace detail
-}  // namespace rclcpp
-
-#endif  // RCLCPP__DETAIL__RESOLVE_ENABLE_TOPIC_STATISTICS_HPP_

rclcpp/include/rclcpp/exceptions.hpp

@@ -15,6 +15,265 @@
 #ifndef RCLCPP__EXCEPTIONS_HPP_
 #define RCLCPP__EXCEPTIONS_HPP_
 
-#include "rclcpp/exceptions/exceptions.hpp"
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "rcl/error_handling.h"
+#include "rcl/types.h"
+#include "rclcpp/visibility_control.hpp"
+
+#include "rcpputils/join.hpp"
+
+namespace rclcpp
+{
+namespace exceptions
+{
+
+/// Thrown when a method is trying to use a node, but it is invalid.
+class InvalidNodeError : public std::runtime_error
+{
+public:
+  InvalidNodeError()
+  : std::runtime_error("node is invalid") {}
+};
+
+/// Thrown when a any kind of name (node, namespace, topic, etc.) is invalid.
+class NameValidationError : public std::invalid_argument
+{
+public:
+  NameValidationError(
+    const char * name_type_,
+    const char * name_,
+    const char * error_msg_,
+    size_t invalid_index_)
+  : std::invalid_argument(format_error(name_type_, name_, error_msg_, invalid_index_)),
+    name_type(name_type_), name(name_), error_msg(error_msg_), invalid_index(invalid_index_)
+  {}
+
+  static std::string
+  format_error(
+    const char * name_type,
+    const char * name,
+    const char * error_msg,
+    size_t invalid_index);
+
+  const std::string name_type;
+  const std::string name;
+  const std::string error_msg;
+  const size_t invalid_index;
+};
+
+/// Thrown when a node name is invalid.
+class InvalidNodeNameError : public NameValidationError
+{
+public:
+  InvalidNodeNameError(const char * node_name, const char * error_msg, size_t invalid_index)
+  : NameValidationError("node name", node_name, error_msg, invalid_index)
+  {}
+};
+
+/// Thrown when a node namespace is invalid.
+class InvalidNamespaceError : public NameValidationError
+{
+public:
+  InvalidNamespaceError(const char * namespace_, const char * error_msg, size_t invalid_index)
+  : NameValidationError("namespace", namespace_, error_msg, invalid_index)
+  {}
+};
+
+/// Thrown when a topic name is invalid.
+class InvalidTopicNameError : public NameValidationError
+{
+public:
+  InvalidTopicNameError(const char * namespace_, const char * error_msg, size_t invalid_index)
+  : NameValidationError("topic name", namespace_, error_msg, invalid_index)
+  {}
+};
+
+/// Thrown when a service name is invalid.
+class InvalidServiceNameError : public NameValidationError
+{
+public:
+  InvalidServiceNameError(const char * namespace_, const char * error_msg, size_t invalid_index)
+  : NameValidationError("service name", namespace_, error_msg, invalid_index)
+  {}
+};
+
+/// Throw a C++ std::exception which was created based on an rcl error.
+/**
+ * Passing nullptr for reset_error is safe and will avoid calling any function
+ * to reset the error.
+ *
+ * \param ret the return code for the current error state
+ * \param prefix string to prefix to the error if applicable (not all errors have custom messages)
+ * \param error_state error state to create exception from, if nullptr rcl_get_error_state is used
+ * \param reset_error function to be called before throwing which whill clear the error state
+ * \throws std::invalid_argument if ret is RCL_RET_OK
+ * \throws std::runtime_error if the rcl_get_error_state returns 0
+ * \throws RCLErrorBase some child class exception based on ret
+ */
+/* *INDENT-OFF* */  // Uncrustify cannot yet understand [[noreturn]] properly
+RCLCPP_PUBLIC
+void
+throw_from_rcl_error [[noreturn]] (
+  rcl_ret_t ret,
+  const std::string & prefix = "",
+  const rcl_error_state_t * error_state = nullptr,
+  void (* reset_error)() = rcl_reset_error);
+/* *INDENT-ON* */
+
+class RCLErrorBase
+{
+public:
+  RCLCPP_PUBLIC
+  RCLErrorBase(rcl_ret_t ret, const rcl_error_state_t * error_state);
+  virtual ~RCLErrorBase() {}
+
+  rcl_ret_t ret;
+  std::string message;
+  std::string file;
+  size_t line;
+  std::string formatted_message;
+};
+
+/// Created when the return code does not match one of the other specialized exceptions.
+class RCLError : public RCLErrorBase, public std::runtime_error
+{
+public:
+  RCLCPP_PUBLIC
+  RCLError(rcl_ret_t ret, const rcl_error_state_t * error_state, const std::string & prefix);
+  RCLCPP_PUBLIC
+  RCLError(const RCLErrorBase & base_exc, const std::string & prefix);
+};
+
+/// Created when the ret is RCL_RET_BAD_ALLOC.
+class RCLBadAlloc : public RCLErrorBase, public std::bad_alloc
+{
+public:
+  RCLCPP_PUBLIC
+  RCLBadAlloc(rcl_ret_t ret, const rcl_error_state_t * error_state);
+  RCLCPP_PUBLIC
+  explicit RCLBadAlloc(const RCLErrorBase & base_exc);
+};
+
+/// Created when the ret is RCL_RET_INVALID_ARGUMENT.
+class RCLInvalidArgument : public RCLErrorBase, public std::invalid_argument
+{
+public:
+  RCLCPP_PUBLIC
+  RCLInvalidArgument(
+    rcl_ret_t ret,
+    const rcl_error_state_t * error_state,
+    const std::string & prefix);
+  RCLCPP_PUBLIC
+  RCLInvalidArgument(const RCLErrorBase & base_exc, const std::string & prefix);
+};
+
+/// Created when the ret is RCL_RET_INVALID_ROS_ARGS.
+class RCLInvalidROSArgsError : public RCLErrorBase, public std::runtime_error
+{
+public:
+  RCLCPP_PUBLIC
+  RCLInvalidROSArgsError(
+    rcl_ret_t ret, const rcl_error_state_t * error_state, const std::string & prefix);
+  RCLCPP_PUBLIC
+  RCLInvalidROSArgsError(const RCLErrorBase & base_exc, const std::string & prefix);
+};
+
+/// Thrown when unparsed ROS specific arguments are found.
+class UnknownROSArgsError : public std::runtime_error
+{
+public:
+  explicit UnknownROSArgsError(std::vector<std::string> && unknown_ros_args_in)
+  : std::runtime_error(
+      "found unknown ROS arguments: '" + rcpputils::join(unknown_ros_args_in, "', '") + "'"),
+    unknown_ros_args(unknown_ros_args_in)
+  {
+  }
+
+  const std::vector<std::string> unknown_ros_args;
+};
+
+/// Thrown when an invalid rclcpp::Event object or SharedPtr is encountered.
+class InvalidEventError : public std::runtime_error
+{
+public:
+  InvalidEventError()
+  : std::runtime_error("event is invalid") {}
+};
+
+/// Thrown when an unregistered rclcpp::Event is encountered where a registered one was expected.
+class EventNotRegisteredError : public std::runtime_error
+{
+public:
+  EventNotRegisteredError()
+  : std::runtime_error("event already registered") {}
+};
+
+/// Thrown if passed parameters are inconsistent or invalid
+class InvalidParametersException : public std::runtime_error
+{
+public:
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+/// Thrown if passed parameter value is invalid.
+class InvalidParameterValueException : public std::runtime_error
+{
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+/// Thrown if requested parameter type is invalid.
+/**
+ * Essentially the same as rclcpp::ParameterTypeException, but with parameter
+ * name in the error message.
+ */
+class InvalidParameterTypeException : public std::runtime_error
+{
+public:
+  /// Construct an instance.
+  /**
+   * \param[in] name the name of the parameter.
+   * \param[in] message custom exception message.
+   */
+  RCLCPP_PUBLIC
+  InvalidParameterTypeException(const std::string & name, const std::string message)
+  : std::runtime_error("parameter '" + name + "' has invalid type: " + message)
+  {}
+};
+
+/// Thrown if parameter is already declared.
+class ParameterAlreadyDeclaredException : public std::runtime_error
+{
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+/// Thrown if parameter is not declared, e.g. either set or get was called without first declaring.
+class ParameterNotDeclaredException : public std::runtime_error
+{
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+/// Thrown if parameter is immutable and therefore cannot be undeclared.
+class ParameterImmutableException : public std::runtime_error
+{
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+/// Thrown if parameter is modified while in a set callback.
+class ParameterModifiedInCallbackException : public std::runtime_error
+{
+  // Inherit constructors from runtime_error.
+  using std::runtime_error::runtime_error;
+};
+
+}  // namespace exceptions
+}  // namespace rclcpp
 
 #endif  // RCLCPP__EXCEPTIONS_HPP_

rclcpp/include/rclcpp/exceptions/exceptions.hpp

@@ -1,279 +0,0 @@
-// Copyright 2016 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__EXCEPTIONS__EXCEPTIONS_HPP_
-#define RCLCPP__EXCEPTIONS__EXCEPTIONS_HPP_
-
-#include <stdexcept>
-#include <string>
-#include <vector>
-
-#include "rcl/error_handling.h"
-#include "rcl/types.h"
-#include "rclcpp/visibility_control.hpp"
-
-#include "rcpputils/join.hpp"
-
-namespace rclcpp
-{
-namespace exceptions
-{
-
-/// Thrown when a method is trying to use a node, but it is invalid.
-class InvalidNodeError : public std::runtime_error
-{
-public:
-  InvalidNodeError()
-  : std::runtime_error("node is invalid") {}
-};
-
-/// Thrown when a any kind of name (node, namespace, topic, etc.) is invalid.
-class NameValidationError : public std::invalid_argument
-{
-public:
-  NameValidationError(
-    const char * name_type_,
-    const char * name_,
-    const char * error_msg_,
-    size_t invalid_index_)
-  : std::invalid_argument(format_error(name_type_, name_, error_msg_, invalid_index_)),
-    name_type(name_type_), name(name_), error_msg(error_msg_), invalid_index(invalid_index_)
-  {}
-
-  static std::string
-  format_error(
-    const char * name_type,
-    const char * name,
-    const char * error_msg,
-    size_t invalid_index);
-
-  const std::string name_type;
-  const std::string name;
-  const std::string error_msg;
-  const size_t invalid_index;
-};
-
-/// Thrown when a node name is invalid.
-class InvalidNodeNameError : public NameValidationError
-{
-public:
-  InvalidNodeNameError(const char * node_name, const char * error_msg, size_t invalid_index)
-  : NameValidationError("node name", node_name, error_msg, invalid_index)
-  {}
-};
-
-/// Thrown when a node namespace is invalid.
-class InvalidNamespaceError : public NameValidationError
-{
-public:
-  InvalidNamespaceError(const char * namespace_, const char * error_msg, size_t invalid_index)
-  : NameValidationError("namespace", namespace_, error_msg, invalid_index)
-  {}
-};
-
-/// Thrown when a topic name is invalid.
-class InvalidTopicNameError : public NameValidationError
-{
-public:
-  InvalidTopicNameError(const char * namespace_, const char * error_msg, size_t invalid_index)
-  : NameValidationError("topic name", namespace_, error_msg, invalid_index)
-  {}
-};
-
-/// Thrown when a service name is invalid.
-class InvalidServiceNameError : public NameValidationError
-{
-public:
-  InvalidServiceNameError(const char * namespace_, const char * error_msg, size_t invalid_index)
-  : NameValidationError("service name", namespace_, error_msg, invalid_index)
-  {}
-};
-
-/// Throw a C++ std::exception which was created based on an rcl error.
-/**
- * Passing nullptr for reset_error is safe and will avoid calling any function
- * to reset the error.
- *
- * \param ret the return code for the current error state
- * \param prefix string to prefix to the error if applicable (not all errors have custom messages)
- * \param error_state error state to create exception from, if nullptr rcl_get_error_state is used
- * \param reset_error function to be called before throwing which whill clear the error state
- * \throws std::invalid_argument if ret is RCL_RET_OK
- * \throws std::runtime_error if the rcl_get_error_state returns 0
- * \throws RCLErrorBase some child class exception based on ret
- */
-/* *INDENT-OFF* */  // Uncrustify cannot yet understand [[noreturn]] properly
-RCLCPP_PUBLIC
-void
-throw_from_rcl_error [[noreturn]] (
-  rcl_ret_t ret,
-  const std::string & prefix = "",
-  const rcl_error_state_t * error_state = nullptr,
-  void (* reset_error)() = rcl_reset_error);
-/* *INDENT-ON* */
-
-class RCLErrorBase
-{
-public:
-  RCLCPP_PUBLIC
-  RCLErrorBase(rcl_ret_t ret, const rcl_error_state_t * error_state);
-  virtual ~RCLErrorBase() {}
-
-  rcl_ret_t ret;
-  std::string message;
-  std::string file;
-  size_t line;
-  std::string formatted_message;
-};
-
-/// Created when the return code does not match one of the other specialized exceptions.
-class RCLError : public RCLErrorBase, public std::runtime_error
-{
-public:
-  RCLCPP_PUBLIC
-  RCLError(rcl_ret_t ret, const rcl_error_state_t * error_state, const std::string & prefix);
-  RCLCPP_PUBLIC
-  RCLError(const RCLErrorBase & base_exc, const std::string & prefix);
-};
-
-/// Created when the ret is RCL_RET_BAD_ALLOC.
-class RCLBadAlloc : public RCLErrorBase, public std::bad_alloc
-{
-public:
-  RCLCPP_PUBLIC
-  RCLBadAlloc(rcl_ret_t ret, const rcl_error_state_t * error_state);
-  RCLCPP_PUBLIC
-  explicit RCLBadAlloc(const RCLErrorBase & base_exc);
-};
-
-/// Created when the ret is RCL_RET_INVALID_ARGUMENT.
-class RCLInvalidArgument : public RCLErrorBase, public std::invalid_argument
-{
-public:
-  RCLCPP_PUBLIC
-  RCLInvalidArgument(
-    rcl_ret_t ret,
-    const rcl_error_state_t * error_state,
-    const std::string & prefix);
-  RCLCPP_PUBLIC
-  RCLInvalidArgument(const RCLErrorBase & base_exc, const std::string & prefix);
-};
-
-/// Created when the ret is RCL_RET_INVALID_ROS_ARGS.
-class RCLInvalidROSArgsError : public RCLErrorBase, public std::runtime_error
-{
-public:
-  RCLCPP_PUBLIC
-  RCLInvalidROSArgsError(
-    rcl_ret_t ret, const rcl_error_state_t * error_state, const std::string & prefix);
-  RCLCPP_PUBLIC
-  RCLInvalidROSArgsError(const RCLErrorBase & base_exc, const std::string & prefix);
-};
-
-/// Thrown when unparsed ROS specific arguments are found.
-class UnknownROSArgsError : public std::runtime_error
-{
-public:
-  explicit UnknownROSArgsError(std::vector<std::string> && unknown_ros_args_in)
-  : std::runtime_error(
-      "found unknown ROS arguments: '" + rcpputils::join(unknown_ros_args_in, "', '") + "'"),
-    unknown_ros_args(unknown_ros_args_in)
-  {
-  }
-
-  const std::vector<std::string> unknown_ros_args;
-};
-
-/// Thrown when an invalid rclcpp::Event object or SharedPtr is encountered.
-class InvalidEventError : public std::runtime_error
-{
-public:
-  InvalidEventError()
-  : std::runtime_error("event is invalid") {}
-};
-
-/// Thrown when an unregistered rclcpp::Event is encountered where a registered one was expected.
-class EventNotRegisteredError : public std::runtime_error
-{
-public:
-  EventNotRegisteredError()
-  : std::runtime_error("event already registered") {}
-};
-
-/// Thrown if passed parameters are inconsistent or invalid
-class InvalidParametersException : public std::runtime_error
-{
-public:
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-/// Thrown if passed parameter value is invalid.
-class InvalidParameterValueException : public std::runtime_error
-{
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-/// Thrown if requested parameter type is invalid.
-/**
- * Essentially the same as rclcpp::ParameterTypeException, but with parameter
- * name in the error message.
- */
-class InvalidParameterTypeException : public std::runtime_error
-{
-public:
-  /// Construct an instance.
-  /**
-   * \param[in] name the name of the parameter.
-   * \param[in] message custom exception message.
-   */
-  RCLCPP_PUBLIC
-  InvalidParameterTypeException(const std::string & name, const std::string message)
-  : std::runtime_error("parameter '" + name + "' has invalid type: " + message)
-  {}
-};
-
-/// Thrown if parameter is already declared.
-class ParameterAlreadyDeclaredException : public std::runtime_error
-{
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-/// Thrown if parameter is not declared, e.g. either set or get was called without first declaring.
-class ParameterNotDeclaredException : public std::runtime_error
-{
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-/// Thrown if parameter is immutable and therefore cannot be undeclared.
-class ParameterImmutableException : public std::runtime_error
-{
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-/// Thrown if parameter is modified while in a set callback.
-class ParameterModifiedInCallbackException : public std::runtime_error
-{
-  // Inherit constructors from runtime_error.
-  using std::runtime_error::runtime_error;
-};
-
-}  // namespace exceptions
-}  // namespace rclcpp
-
-#endif  // RCLCPP__EXCEPTIONS__EXCEPTIONS_HPP_

rclcpp/include/rclcpp/executable_list.hpp

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#ifndef RCLCPP__EXPERIMENTAL__EXECUTABLE_LIST_HPP_
-#define RCLCPP__EXPERIMENTAL__EXECUTABLE_LIST_HPP_
+#ifndef RCLCPP__EXECUTABLE_LIST_HPP_
+#define RCLCPP__EXECUTABLE_LIST_HPP_
 
 #include <memory>
 #include <vector>
@@ -27,7 +27,7 @@
 
 namespace rclcpp
 {
-namespace experimental
+namespace executor
 {
 
 /// This class contains subscriptionbase, timerbase, etc. which can be used to run callbacks.
@@ -86,7 +86,7 @@ public:
   size_t number_of_waitables;
 };
 
-}  // namespace experimental
+}  // namespace executor
 }  // namespace rclcpp
 
-#endif  // RCLCPP__EXPERIMENTAL__EXECUTABLE_LIST_HPP_
+#endif  // RCLCPP__EXECUTABLE_LIST_HPP_

rclcpp/include/rclcpp/executor.hpp

@@ -29,241 +29,129 @@
 #include "rcl/guard_condition.h"
 #include "rcl/wait.h"
 
-#include "rclcpp/callback_group.hpp"
-#include "rclcpp/executor_options.hpp"
-#include "rclcpp/executor_policies/timer_favoring_priority_queue.hpp"
-#include "rclcpp/future_return_code.hpp"
-#include "rclcpp/guard_condition.hpp"
+#include "rclcpp/contexts/default_context.hpp"
 #include "rclcpp/memory_strategies.hpp"
 #include "rclcpp/memory_strategy.hpp"
 #include "rclcpp/node_interfaces/node_base_interface.hpp"
 #include "rclcpp/utilities.hpp"
 #include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_set.hpp"
 
 namespace rclcpp
 {
 
-/// Base class for Executor providing the common interface for adding items, spinning, etc.
-class ExecutorBase
+// Forward declaration is used in convenience method signature.
+class Node;
+
+namespace executor
+{
+
+/// Return codes to be used with spin_until_future_complete.
+/**
+ * SUCCESS: The future is complete and can be accessed with "get" without blocking.
+ *          This does not indicate that the operation succeeded; "get" may still throw an exception.
+ * INTERRUPTED: The future is not complete, spinning was interrupted by Ctrl-C or another error.
+ * TIMEOUT: Spinning timed out.
+ */
+enum class FutureReturnCode {SUCCESS, INTERRUPTED, TIMEOUT};
+
+RCLCPP_PUBLIC
+std::ostream &
+operator<<(std::ostream & os, const FutureReturnCode & future_return_code);
+
+RCLCPP_PUBLIC
+std::string
+to_string(const FutureReturnCode & future_return_code);
+
+///
+/**
+ * Options to be passed to the executor constructor.
+ */
+struct ExecutorArgs
+{
+  ExecutorArgs()
+  : memory_strategy(memory_strategies::create_default_strategy()),
+    context(rclcpp::contexts::default_context::get_global_default_context()),
+    max_conditions(0)
+  {}
+
+  memory_strategy::MemoryStrategy::SharedPtr memory_strategy;
+  std::shared_ptr<rclcpp::Context> context;
+  size_t max_conditions;
+};
+
+static inline ExecutorArgs create_default_executor_arguments()
+{
+  return ExecutorArgs();
+}
+
+/// Coordinate the order and timing of available communication tasks.
+/**
+ * Executor provides spin functions (including spin_node_once and spin_some).
+ * It coordinates the nodes and callback groups by looking for available work and completing it,
+ * based on the threading or concurrency scheme provided by the subclass implementation.
+ * An example of available work is executing a subscription callback, or a timer callback.
+ * The executor structure allows for a decoupling of the communication graph and the execution
+ * model.
+ * See SingleThreadedExecutor and MultiThreadedExecutor for examples of execution paradigms.
+ */
+class Executor
 {
 public:
-  RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(ExecutorBase)
+  RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(Executor)
 
   /// Default constructor.
-  /**
-   * \param[in] options Options for the executor.
-   */
+  // \param[in] ms The memory strategy to be used with this executor.
   RCLCPP_PUBLIC
-  explicit ExecutorBase(const ExecutorOptions & options = ExecutorOptions());
+  explicit Executor(const ExecutorArgs & args = ExecutorArgs());
 
   /// Default destructor.
   RCLCPP_PUBLIC
-  virtual ~ExecutorBase();
+  virtual ~Executor();
 
-  /// Execution loop which waits for work, executes work, and repeats until canceled.
-  /**
-   * This will block, continuing to wait for work and then execute it, until
-   * canceled, either by the cancel() method or by the associated context being
-   * shutdown, either explicitly or due to a SIGINT (perhaps due to ctrl-c).
-   */
-  virtual
-  void
+  /// Do work periodically as it becomes available to us. Blocking call, may block indefinitely.
+  // It is up to the implementation of Executor to implement spin.
+  virtual void
   spin() = 0;
 
-  /// Add all of a node's callback groups to the executor.
+  /// Add a node to the executor.
   /**
-   * Add all of the callback groups of a node to this executor.
-   *
-   * If any callback groups are associated with another executor, this method
-   * will throw a std::runtime_error.
-   *
-   * It will also trigger the interrupt guard condition which will cause the
-   * executor to wake up and consider the changes, then go back to waiting.
-   * Unless the notify parameter is passed false, in which case it will not
-   * interrupt the executor, and the changes may not be considered immediately.
-   *
-   * \param[in] node_ptr Shared pointer to the node which will have callback groups added.
-   * \param[in] notify If true, notfiy the executor of changes, otherwise do not.
-   * \throws std::runtime_error if any callback groups are associated with another executor.
-   */
-  template<class NodeT>
-  void
-  add_node(const std::shared_ptr<NodeT> & node_ptr, bool notify = true)
-  {
-    this->add_node(
-      node_ptr->get_node_base_interface(),
-      notify,
-      false  // raise on encountering already associated callback groups
-    );
-  }
-
-  /// Overload that takes the NodeBaseInterface directly.
-  template<class NodeT>
-  void
-  add_node(
-    const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr & node_ptr,
-    bool notify = true)
-  {
-    this->add_node(
-      node_ptr,
-      notify,
-      false  // raise on encountering already associated callback groups
-    );
-  }
-
-  /// Add all unassociated callback groups of the given node to this executor.
-  /**
-   * Same as add_node(), but instead of throwing if a callback group is already
-   * associated with another exector (already added to it) it will just ignore
-   * it rather than throwing.
-   *
-   * \param[in] node_ptr Shared pointer to the node which will have callback groups added.
-   * \param[in] notify If true, notfiy the executor of changes, otherwise do not.
-   */
-  template<class NodeT>
-  void
-  add_unassociated_callback_groups_from_node(
-    const std::shared_ptr<NodeT> & node_ptr,
-    bool notify = true)
-  {
-    this->add_node(
-      node_ptr->get_node_base_interface(),
-      notify,
-      true  // ignore already associated callback groups
-    );
-  }
-
-  /// Overload that takes the NodeBaseInterface directly.
-  template<class NodeT>
-  void
-  add_unassociated_callback_groups_from_node(
-    const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr & node_ptr,
-    bool notify = true)
-  {
-    this->add_node(
-      node_ptr,
-      notify,
-      true  // ignore already associated callback groups
-    );
-  }
-
-  /// Remove all of a node's callback groups from the executor.
-  /**
-   * Remove all of the callback groups of a node from this executor.
-   *
-   * It will also trigger the interrupt guard condition which will cause the
-   * executor to wake up and consider the changes, then go back to waiting.
-   * Unless the notify parameter is passed false, in which case it will not
-   * interrupt the executor, and the changes may not be considered immediately.
-   *
-   * \param[in] node Node which will have callback groups removed.
-   * \param[in] notify If true, notfiy the executor of changes, otherwise do not.
-   */
-  template<class NodeT>
-  void
-  remove_node(const NodeT & node, bool notify = true)
-  {
-    this->remove_node(*node.get_node_base_interface(), notify);
-  }
-
-  /// Overload that takes a shared pointer to the node.
-  /**
-   * This is kept for backwards compatibility from when executors shared
-   * ownership of Nodes.
-   */
-  template<class NodeT>
-  void
-  remove_node(const std::shared_ptr<NodeT> & node_ptr, bool notify = true)
-  {
-    this->remove_node(*node_ptr->get_node_base_interface(), notify);
-  }
-
-  /// Placeholder used to indicate that a method overload should not notify the executor.
-  struct DoNotNotify {};
-
-  /// Add a callback group to this executor.
-  /**
-   * If the given callback group is already associated with another executor,
-   * this method will throw a std::runtime_error.
-   *
-   * This overload of add_callback_group() will notify the executor so it will
-   * wake up if waiting and consider the changes.
-   *
-   * Weak ownership of the callback group is kept by the executor all of the
-   * time, but while waiting the weak ownership is periodically elevated to
-   * shared ownership.
-   * Therefore, if you let the callback group shared pointer go out of scope
-   * then it will stay in scope until this executor is done using it, at which
-   * point the callback group will be destructed and automatically removed from
-   * this executor in the next pass.
-   *
-   * \param[in] callback_group_ptr The callback group to be added.
-   * \throws std::runtime_error if the callback group is associated with another
-   *   executor already.
-   * \throws std::invalid_argument if the callback group pointer is nullptr.
+   * An executor can have zero or more nodes which provide work during `spin` functions.
+   * \param[in] node_ptr Shared pointer to the node to be added.
+   * \param[in] notify True to trigger the interrupt guard condition during this function. If
+   * the executor is blocked at the rmw layer while waiting for work and it is notified that a new
+   * node was added, it will wake up.
    */
   RCLCPP_PUBLIC
-  virtual
-  void
-  add_callback_group(rclcpp::CallbackGroup::SharedPtr callback_group_ptr) = 0;
+  virtual void
+  add_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify = true);
 
-  /// Add a callback group to this executor without notifying the executor.
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  RCLCPP_PUBLIC
+  virtual void
+  add_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify = true);
+
+  /// Remove a node from the executor.
   /**
-   * The same as the other overload of add_callback_group(), except it does not
-   * notify the executor, so it will not wake up and these changes may not be
-   * considered immediately.
-   *
-   * Note, a bool with a default value would be preferable for controlling the
-   * notify behavior, and we're using it in the add/remove node above, but
-   * in order to keep this function virtual, and to avoid using default values
-   * in conjunction with virtual methods, we use an overload instead, in the
-   * spirit of std::nothrow_t, e.g.:
-   *   https://en.cppreference.com/w/cpp/memory/new/nothrow
+   * \param[in] node_ptr Shared pointer to the node to remove.
+   * \param[in] notify True to trigger the interrupt guard condition and wake up the executor.
+   * This is useful if the last node was removed from the executor while the executor was blocked
+   * waiting for work in another thread, because otherwise the executor would never be notified.
    */
   RCLCPP_PUBLIC
-  virtual
-  void
-  add_callback_group(rclcpp::CallbackGroup::SharedPtr callback_group_ptr, DoNotNotify) = 0;
+  virtual void
+  remove_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify = true);
 
-  /// Remove a callback group from this executor.
-  /**
-   * If the given callback group is not associated with this executor, this
-   * method will throw a std::runtime_error.
-   *
-   * This overload of add_callback_group() will notify the executor so it will
-   * wake up if waiting and consider the changes.
-   *
-   * \param[in] callback_group The callback group to be removed.
-   * \throws std::runtime_error if the callback group is not associated with
-   *   this executor.
-   * \throws std::invalid_argument if the callback group pointer is nullptr.
-   */
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
   RCLCPP_PUBLIC
-  virtual
-  void
-  remove_callback_group(const rclcpp::CallbackGroup & callback_group) = 0;
-
-  /// Remove a callback group from this executor without notifying the executor.
-  /**
-   * The same as the other overload of remove_callback_group(), except it does not
-   * notify the executor, so it will not wake up and these changes may not be
-   * considered immediately.
-   *
-   * See add_callback_group() for a note about the use of DoNotNotify.
-   */
-  RCLCPP_PUBLIC
-  virtual
-  void
-  remove_callback_group(const rclcpp::CallbackGroup & callback_group, DoNotNotify) = 0;
+  virtual void
+  remove_node(std::shared_ptr<rclcpp::Node> node_ptr, bool notify = true);
 
   /// Add a node to executor, execute the next available unit of work, and remove the node.
   /**
    * \param[in] node Shared pointer to the node to add.
-   * \param[in] timeout How long to wait for work to become available.
-   *   Negative values cause spin_node_once to block indefinitely (the default
-   *   behavior).
-   *   A timeout of 0 causes this function to be non-blocking.
+   * \param[in] timeout How long to wait for work to become available. Negative values cause
+   * spin_node_once to block indefinitely (the default behavior). A timeout of 0 causes this
+   * function to be non-blocking.
    */
   template<typename RepT = int64_t, typename T = std::milli>
   void
@@ -278,7 +166,7 @@ public:
   }
 
   /// Convenience function which takes Node and forwards NodeBaseInterface.
-  template<typename NodeT, typename RepT = int64_t, typename T = std::milli>
+  template<typename NodeT = rclcpp::Node, typename RepT = int64_t, typename T = std::milli>
   void
   spin_node_once(
     std::shared_ptr<NodeT> node,
@@ -292,14 +180,16 @@ public:
 
   /// Add a node, complete all immediately available work, and remove the node.
   /**
-   * \param[in] node Shared pointer to the node to spin some.
+   * \param[in] node Shared pointer to the node to add.
    */
-  template<class NodeT>
+  RCLCPP_PUBLIC
   void
-  spin_node_some(const std::shared_ptr<NodeT> & node)
-  {
-    this->spin_node_some(node->get_node_base_interface());
-  }
+  spin_node_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node);
+
+  /// Convenience function which takes Node and forwards NodeBaseInterface.
+  RCLCPP_PUBLIC
+  void
+  spin_node_some(std::shared_ptr<rclcpp::Node> node);
 
   /// Complete all available queued work without blocking.
   /**
@@ -313,14 +203,12 @@ public:
    * been exceeded.
    */
   RCLCPP_PUBLIC
-  virtual
-  void
-  spin_some(std::chrono::nanoseconds max_duration = std::chrono::nanoseconds(0)) = 0;
+  virtual void
+  spin_some(std::chrono::nanoseconds max_duration = std::chrono::nanoseconds(0));
 
   RCLCPP_PUBLIC
-  virtual
-  void
-  spin_once(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1)) = 0;
+  virtual void
+  spin_once(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
 
   /// Spin (blocking) until the future is complete, it times out waiting, or rclcpp is interrupted.
   /**
@@ -382,79 +270,51 @@ public:
   }
 
   /// Cancel any running spin* function, causing it to return.
+  /* This function can be called asynchonously from any thread. */
+  RCLCPP_PUBLIC
+  void
+  cancel();
+
+  /// Support dynamic switching of the memory strategy.
   /**
-   * This function can be called asynchronously from any thread.
+   * Switching the memory strategy while the executor is spinning in another threading could have
+   * unintended consequences.
+   * \param[in] memory_strategy Shared pointer to the memory strategy to set.
    */
   RCLCPP_PUBLIC
-  virtual
   void
-  cancel() = 0;
+  set_memory_strategy(memory_strategy::MemoryStrategy::SharedPtr memory_strategy);
 
 protected:
-  /// Implementation of add_node().
-  /**
-   * \param[in] node_ptr The node which will have its callback groups added.
-   * \param[in] notify If true, the executor is interrupted to consider the
-   *   changes, otherwise it is not interrupted.
-   * \param[in] ignore_associated_callback_groups If true, then when a callback
-   *   group which is already been added to another executor is encountered
-   *   it will be ignored, if false then std::runtime_error is thrown instead.
-   * \throws std::runtime_error if ignore_associated_callback_groups is false
-   *   and a callback group which is already associated with another executor
-   *   is encountered.
-   * \throws std::invalid_argument if node_ptr is nullptr.
-   */
-  RCLCPP_PUBLIC
-  virtual
-  void
-  add_node(
-    rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
-    bool notify,
-    bool ignore_associated_callback_groups) = 0;
-
-  RCLCPP_PUBLIC
-  virtual
-  void
-  remove_node(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr, bool notify) = 0;
-
   RCLCPP_PUBLIC
   void
   spin_node_once_nanoseconds(
     rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node,
     std::chrono::nanoseconds timeout);
 
-  RCLCPP_PUBLIC
-  virtual
-  void
-  spin_node_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node);
-
   /// Find the next available executable and do the work associated with it.
   /** \param[in] any_exec Union structure that can hold any executable type (timer, subscription,
    * service, client).
    */
   RCLCPP_PUBLIC
   void
-  execute_any_executable(rclcpp::AnyExecutable & any_exec);
+  execute_any_executable(AnyExecutable & any_exec);
 
   RCLCPP_PUBLIC
-  static
-  void
+  static void
   execute_subscription(
     rclcpp::SubscriptionBase::SharedPtr subscription);
 
   RCLCPP_PUBLIC
-  static
-  void
+  static void
   execute_timer(rclcpp::TimerBase::SharedPtr timer);
 
   RCLCPP_PUBLIC
-  static
-  void
+  static void
   execute_service(rclcpp::ServiceBase::SharedPtr service);
 
   RCLCPP_PUBLIC
-  static
-  void
+  static void
   execute_client(rclcpp::ClientBase::SharedPtr client);
 
   RCLCPP_PUBLIC
@@ -462,103 +322,47 @@ protected:
   wait_for_work(std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
 
   RCLCPP_PUBLIC
-  rclcpp::CallbackGroup::SharedPtr
+  rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
+  get_node_by_group(rclcpp::callback_group::CallbackGroup::SharedPtr group);
+
+  RCLCPP_PUBLIC
+  rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_timer(rclcpp::TimerBase::SharedPtr timer);
 
   RCLCPP_PUBLIC
   bool
-  get_next_ready_executable(rclcpp::AnyExecutable & any_executable);
+  get_next_ready_executable(AnyExecutable & any_executable);
 
   RCLCPP_PUBLIC
   bool
   get_next_executable(
-    rclcpp::AnyExecutable & any_executable,
+    AnyExecutable & any_executable,
     std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));
 
-  RCLCPP_DISABLE_COPY(ExecutorBase)
-
   /// Spinning state, used to prevent multi threaded calls to spin and to cancel blocking spins.
   std::atomic_bool spinning;
 
   /// Guard condition for signaling the rmw layer to wake up for special events.
-  rclcpp::GuardCondition interrupt_guard_condition_;
-  // rcl_guard_condition_t interrupt_guard_condition_ = rcl_get_zero_initialized_guard_condition();
+  rcl_guard_condition_t interrupt_guard_condition_ = rcl_get_zero_initialized_guard_condition();
 
-  // /// Wait set for managing entities that the rmw layer waits on.
-  // rcl_wait_set_t wait_set_ = rcl_get_zero_initialized_wait_set();
+  /// Wait set for managing entities that the rmw layer waits on.
+  rcl_wait_set_t wait_set_ = rcl_get_zero_initialized_wait_set();
 
-  // // Mutex to protect the subsequent memory_strategy_.
-  // std::mutex memory_strategy_mutex_;
+  // Mutex to protect the subsequent memory_strategy_.
+  std::mutex memory_strategy_mutex_;
 
-  // /// The memory strategy: an interface for handling user-defined memory allocation strategies.
-  // memory_strategy::MemoryStrategy::SharedPtr memory_strategy_;
+  /// The memory strategy: an interface for handling user-defined memory allocation strategies.
+  memory_strategy::MemoryStrategy::SharedPtr memory_strategy_;
 
   /// The context associated with this executor.
-  rclcpp::Context::SharedPtr context_;
+  std::shared_ptr<rclcpp::Context> context_;
 
-  // std::list<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> weak_nodes_;
-  // std::list<const rcl_guard_condition_t *> guard_conditions_;
-  std::vector<rclcpp::CallbackGroup::WeakPtr> weak_guard_conditions_;
+  RCLCPP_DISABLE_COPY(Executor)
 
+  std::list<rclcpp::node_interfaces::NodeBaseInterface::WeakPtr> weak_nodes_;
+  std::list<const rcl_guard_condition_t *> guard_conditions_;
 };
 
-/// Template class which serves as the foundation of actual executors.
-/**
- * This class combines the wait set, scheduling policy, and the ExecutorBase
- * class, and implements all of the pure virtual functions of ExecutorBase
- * making it a concrete class.
- */
-template<class WaitSetT, class SchedulingPolicy>
-class ExecutorTemplate : public ExecutorBase, public SchedulingPolicy
-{
-public:
-  RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(ExecutorTemplate)
-
-  /// Default constructor.
-  /**
-   * \param[in] options Options for the executor.
-   */
-  explicit ExecutorTemplate(const ExecutorOptions & options = ExecutorOptions())
-  : ExecutorBase(options), SchedulingPolicy(options), WaitSetT(options.context)
-  {}
-
-  /// Default virtual destructor.
-  RCLCPP_PUBLIC
-  virtual
-  ~ExecutorTemplate() = default;
-
-protected:
-  WaitSetT wait_set_;
-};
-
-/// Executor concept which waits for work and coordinates execution of user callbacks.
-/**
- * Executor provides spin functions (including spin_node_once and spin_some).
- * It coordinates the nodes and callback groups by looking for available work
- * and completing it, based on the threading or concurrency scheme provided by
- * the subclass implementation.
- * An example of available work is executing a subscription callback, or a
- * timer callback.
- * The executor structure allows for a decoupling of the communication graph
- * and the execution model.
- * See SingleThreadedExecutor and MultiThreadedExecutor for examples of
- * different execution paradigms.
- *
- * By default this alias provides a foundation based on specific wait set type
- * and a scheduling policy.
- * The wait set is expected to be dynamic, i.e. items can be added or removed
- * after creation, and thread-safe, i.e. items can be added or removed while
- * also waiting concurrently.
- */
-using Executor = ExecutorTemplate<
-  rclcpp::ThreadSafeWaitSet,
-  rclcpp::executor_policies::TimerFavoringPriorityQueue>;
-
-namespace executor
-{
-
-using Executor [[deprecated("use rclcpp::Executor instead")]] = Executor;
-
 }  // namespace executor
 }  // namespace rclcpp
 

rclcpp/include/rclcpp/executor_options.hpp

@@ -1,49 +0,0 @@
-// Copyright 2014 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__EXECUTOR_OPTIONS_HPP_
-#define RCLCPP__EXECUTOR_OPTIONS_HPP_
-
-#include "rclcpp/context.hpp"
-#include "rclcpp/contexts/default_context.hpp"
-#include "rclcpp/memory_strategies.hpp"
-#include "rclcpp/memory_strategy.hpp"
-#include "rclcpp/visibility_control.hpp"
-
-namespace rclcpp
-{
-
-/// 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::memory_strategy::MemoryStrategy::SharedPtr memory_strategy;
-  rclcpp::Context::SharedPtr context;
-  size_t max_conditions;
-};
-
-namespace executor
-{
-
-using ExecutorArgs [[deprecated("use rclcpp::ExecutorOptions instead")]] = ExecutorOptions;
-
-}  // namespace executor
-}  // namespace rclcpp
-
-#endif  // RCLCPP__EXECUTOR_OPTIONS_HPP_

rclcpp/include/rclcpp/executor_policies/scheduling_result.hpp

@@ -1,35 +0,0 @@
-// Copyright 2020 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__EXECUTOR_POLICIES__SCHEDULING_RESULT_HPP_
-#define RCLCPP__EXECUTOR_POLICIES__SCHEDULING_RESULT_HPP_
-
-#include "rclcpp/visibility_control.hpp"
-
-namespace rclcpp
-{
-namespace executor_policies
-{
-
-/// Represents the directions a SchedulingPolicy can give to an Executor.
-enum RCLCPP_PUBLIC SchedulingResult
-{
-  ContinueExecuting,  //<! Indicates more work should be done before waiting.
-  WaitForWork,  //<! Indicates that the executor should wait on the wait set again.
-};
-
-}  // namespace executor_policies
-}  // namespace rclcpp
-
-#endif  // RCLCPP__EXECUTOR_POLICIES__SCHEDULING_RESULT_HPP_

rclcpp/include/rclcpp/executor_policies/timer_favoring_priority_queue.hpp

@@ -1,77 +0,0 @@
-// Copyright 2020 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__EXECUTOR_POLICIES__TIMER_FAVORING_PRIORITY_QUEUE_HPP_
-#define RCLCPP__EXECUTOR_POLICIES__TIMER_FAVORING_PRIORITY_QUEUE_HPP_
-
-#include "rclcpp/any_executable.hpp"
-#include "rclcpp/executor_options.hpp"
-#include "rclcpp/executor_policies/scheduling_result.hpp"
-#include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_result.hpp"
-
-namespace rclcpp
-{
-namespace executor_policies
-{
-
-/// A naive scheduling policy which selects Timers first, then Subscriptions and other items.
-/**
- * Items are executed in the order they were added, favoring Timers, then
- * Subscriptions, Service Servers, Service Clients, and finally Waitables.
- * All Timers are executed before any Subscriptions, and all Subscriptions
- * before any Service Servers, and so on.
- *
- * User guard conditions are not yet supported by the Executor and so all guard
- * condition are used by the executor itself and are handled before this policy
- * is consulted.
- * Therefore, guard conditions are ignored for the purposes of scheduling.
- *
- * This is a naive policy, but is the default until a better one is implemented.
- */
-class TimerFavoringPriorityQueue
-{
-public:
-  explicit TimerFavoringPriorityQueue(const rclcpp::ExecutorOptions &) {}
-
-  /// Select which item should next be executed, and indicate if waiting should resume.
-  /**
-   * Selects which item to be executed next, assign it to the any_executable, or
-   * assigning nullptr if no work should be done right now.
-   *
-   * This method is called by the executor after waiting on a wait set, in
-   * order to determine what to execute next based on the result.
-   *
-   * Additionally, if returning SchedulingResult::ContinueExecuting then the
-   * executor will call this function again without waiting on the wait set, or
-   * if returning SchedulingResult::WaitForWork then the executor will wait on
-   * the wait set again after executing the selected any_executable, or
-   * immediately if any_executable was assigned nullptr.
-   */
-  template<class WaitSetT>
-  rclcpp::executor_policies::SchedulingResult
-  schedule_next_any_executable(
-    const WaitResult<WaitSetT> & wait_result,
-    rclcpp::AnyExecutable & any_executable)
-  {
-    // Explicitly ignore guard conditions.
-    // Check Timers for being ready.
-
-  }
-};
-
-}  // namespace executor_policies
-}  // namespace rclcpp
-
-#endif  // RCLCPP__EXECUTOR_POLICIES__TIMER_FAVORING_PRIORITY_QUEUE_HPP_

rclcpp/include/rclcpp/executors.hpp

@@ -67,9 +67,9 @@ using rclcpp::executors::SingleThreadedExecutor;
  * \return The return code, one of `SUCCESS`, `INTERRUPTED`, or `TIMEOUT`.
  */
 template<typename ResponseT, typename TimeRepT = int64_t, typename TimeT = std::milli>
-rclcpp::FutureReturnCode
+rclcpp::executor::FutureReturnCode
 spin_node_until_future_complete(
-  rclcpp::Executor & executor,
+  rclcpp::executor::Executor & executor,
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
   const std::shared_future<ResponseT> & future,
   std::chrono::duration<TimeRepT, TimeT> timeout = std::chrono::duration<TimeRepT, TimeT>(-1))
@@ -84,9 +84,9 @@ spin_node_until_future_complete(
 
 template<typename NodeT = rclcpp::Node, typename ResponseT, typename TimeRepT = int64_t,
   typename TimeT = std::milli>
-rclcpp::FutureReturnCode
+rclcpp::executor::FutureReturnCode
 spin_node_until_future_complete(
-  rclcpp::Executor & executor,
+  rclcpp::executor::Executor & executor,
   std::shared_ptr<NodeT> node_ptr,
   const std::shared_future<ResponseT> & future,
   std::chrono::duration<TimeRepT, TimeT> timeout = std::chrono::duration<TimeRepT, TimeT>(-1))
@@ -101,7 +101,7 @@ spin_node_until_future_complete(
 }  // namespace executors
 
 template<typename FutureT, typename TimeRepT = int64_t, typename TimeT = std::milli>
-rclcpp::FutureReturnCode
+rclcpp::executor::FutureReturnCode
 spin_until_future_complete(
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_ptr,
   const std::shared_future<FutureT> & future,
@@ -113,7 +113,7 @@ spin_until_future_complete(
 
 template<typename NodeT = rclcpp::Node, typename FutureT, typename TimeRepT = int64_t,
   typename TimeT = std::milli>
-rclcpp::FutureReturnCode
+rclcpp::executor::FutureReturnCode
 spin_until_future_complete(
   std::shared_ptr<NodeT> node_ptr,
   const std::shared_future<FutureT> & future,

rclcpp/include/rclcpp/executors/multi_threaded_executor.hpp

@@ -32,7 +32,7 @@ namespace rclcpp
 namespace executors
 {
 
-class MultiThreadedExecutor : public rclcpp::Executor
+class MultiThreadedExecutor : public executor::Executor
 {
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(MultiThreadedExecutor)
@@ -45,14 +45,14 @@ public:
    * This is useful for reproducing some bugs related to taking work more than
    * once.
    *
-   * \param options common options for all executors
+   * \param args common arguments for all executors
    * \param number_of_threads number of threads to have in the thread pool,
    *   the default 0 will use the number of cpu cores found instead
    * \param yield_before_execute if true std::this_thread::yield() is called
    */
   RCLCPP_PUBLIC
   MultiThreadedExecutor(
-    const rclcpp::ExecutorOptions & options = rclcpp::ExecutorOptions(),
+    const executor::ExecutorArgs & args = executor::ExecutorArgs(),
     size_t number_of_threads = 0,
     bool yield_before_execute = false,
     std::chrono::nanoseconds timeout = std::chrono::nanoseconds(-1));

rclcpp/include/rclcpp/executors/single_threaded_executor.hpp

@@ -35,31 +35,25 @@ namespace rclcpp
 namespace executors
 {
 
-/// Single-threaded executor implementation.
-/**
- * This is the default executor created by rclcpp::spin.
- */
-class SingleThreadedExecutor : public rclcpp::Executor
+/// Single-threaded executor implementation
+// This is the default executor created by rclcpp::spin.
+class SingleThreadedExecutor : public executor::Executor
 {
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(SingleThreadedExecutor)
 
   /// Default constructor. See the default constructor for Executor.
   RCLCPP_PUBLIC
-  explicit SingleThreadedExecutor(
-    const rclcpp::ExecutorOptions & options = rclcpp::ExecutorOptions());
+  SingleThreadedExecutor(
+    const executor::ExecutorArgs & args = executor::ExecutorArgs());
 
-  /// Default destructor.
+  /// Default destrcutor.
   RCLCPP_PUBLIC
   virtual ~SingleThreadedExecutor();
 
   /// Single-threaded implementation of spin.
-  /**
-   * This function will block until work comes in, execute it, and then repeat
-   * the process until canceled.
-   * It may be interrupt by a call to rclcpp::Executor::cancel() or by ctrl-c
-   * if the associated context is configured to shutdown on SIGINT.
-   */
+  // This function will block until work comes in, execute it, and keep blocking.
+  // It will only be interrupt by a CTRL-C (managed by the global signal handler).
   RCLCPP_PUBLIC
   void
   spin() override;

rclcpp/include/rclcpp/executors/static_executor_entities_collector.hpp

@@ -22,7 +22,7 @@
 #include "rcl/guard_condition.h"
 #include "rcl/wait.h"
 
-#include "rclcpp/experimental/executable_list.hpp"
+#include "rclcpp/executable_list.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/memory_strategy.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -159,7 +159,7 @@ private:
   rcl_wait_set_t * p_wait_set_ = nullptr;
 
   /// Executable list: timers, subscribers, clients, services and waitables
-  rclcpp::experimental::ExecutableList exec_list_;
+  rclcpp::executor::ExecutableList exec_list_;
 };
 
 }  // namespace executors

rclcpp/include/rclcpp/executors/static_single_threaded_executor.hpp

@@ -23,9 +23,9 @@
 
 #include "rmw/rmw.h"
 
+#include "rclcpp/executable_list.hpp"
 #include "rclcpp/executor.hpp"
 #include "rclcpp/executors/static_executor_entities_collector.hpp"
-#include "rclcpp/experimental/executable_list.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/memory_strategies.hpp"
 #include "rclcpp/node.hpp"
@@ -54,7 +54,7 @@ namespace executors
  * exec.spin();
  * exec.remove_node(node);
  */
-class StaticSingleThreadedExecutor : public rclcpp::Executor
+class StaticSingleThreadedExecutor : public executor::Executor
 {
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(StaticSingleThreadedExecutor)
@@ -62,7 +62,7 @@ public:
   /// Default constructor. See the default constructor for Executor.
   RCLCPP_PUBLIC
   explicit StaticSingleThreadedExecutor(
-    const rclcpp::ExecutorOptions & options = rclcpp::ExecutorOptions());
+    const executor::ExecutorArgs & args = executor::ExecutorArgs());
 
   /// Default destrcutor.
   RCLCPP_PUBLIC
@@ -131,14 +131,14 @@ public:
    *  exec.spin_until_future_complete(future);
    */
   template<typename ResponseT, typename TimeRepT = int64_t, typename TimeT = std::milli>
-  rclcpp::FutureReturnCode
+  rclcpp::executor::FutureReturnCode
   spin_until_future_complete(
     std::shared_future<ResponseT> & future,
     std::chrono::duration<TimeRepT, TimeT> timeout = std::chrono::duration<TimeRepT, TimeT>(-1))
   {
     std::future_status status = future.wait_for(std::chrono::seconds(0));
     if (status == std::future_status::ready) {
-      return rclcpp::FutureReturnCode::SUCCESS;
+      return rclcpp::executor::FutureReturnCode::SUCCESS;
     }
 
     auto end_time = std::chrono::steady_clock::now();
@@ -159,7 +159,7 @@ public:
       // Check if the future is set, return SUCCESS if it is.
       status = future.wait_for(std::chrono::seconds(0));
       if (status == std::future_status::ready) {
-        return rclcpp::FutureReturnCode::SUCCESS;
+        return rclcpp::executor::FutureReturnCode::SUCCESS;
       }
       // If the original timeout is < 0, then this is blocking, never TIMEOUT.
       if (timeout_ns < std::chrono::nanoseconds::zero()) {
@@ -168,14 +168,14 @@ public:
       // Otherwise check if we still have time to wait, return TIMEOUT if not.
       auto now = std::chrono::steady_clock::now();
       if (now >= end_time) {
-        return rclcpp::FutureReturnCode::TIMEOUT;
+        return rclcpp::executor::FutureReturnCode::TIMEOUT;
       }
       // Subtract the elapsed time from the original timeout.
       timeout_left = std::chrono::duration_cast<std::chrono::nanoseconds>(end_time - now);
     }
 
     // The future did not complete before ok() returned false, return INTERRUPTED.
-    return rclcpp::FutureReturnCode::INTERRUPTED;
+    return rclcpp::executor::FutureReturnCode::INTERRUPTED;
   }
 
 protected:

rclcpp/include/rclcpp/experimental/intra_process_manager.hpp

@@ -110,11 +110,14 @@ public:
    * In addition this generates a unique intra process id for the subscription.
    *
    * \param subscription the SubscriptionIntraProcess to register.
+   * \param is_serialized true if the buffer expects serialized messages
    * \return an unsigned 64-bit integer which is the subscription's unique id.
    */
   RCLCPP_PUBLIC
   uint64_t
-  add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription);
+  add_subscription(
+    rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription,
+    const bool is_serialized = false);
 
   /// Unregister a subscription using the subscription's unique id.
   /**
@@ -134,11 +137,12 @@ public:
    * In addition this generates a unique intra process id for the publisher.
    *
    * \param publisher publisher to be registered with the manager.
+   * \param is_serialized true if the buffer expects serialized messages
    * \return an unsigned 64-bit integer which is the publisher's unique id.
    */
   RCLCPP_PUBLIC
   uint64_t
-  add_publisher(rclcpp::PublisherBase::SharedPtr publisher);
+  add_publisher(rclcpp::PublisherBase::SharedPtr publisher, const bool is_serialized = false);
 
   /// Unregister a publisher using the publisher's unique id.
   /**
@@ -311,6 +315,7 @@ private:
     rmw_qos_profile_t qos;
     const char * topic_name;
     bool use_take_shared_method;
+    bool is_serialized;
   };
 
   struct PublisherInfo
@@ -320,6 +325,7 @@ private:
     rclcpp::PublisherBase::WeakPtr publisher;
     rmw_qos_profile_t qos;
     const char * topic_name;
+    bool is_serialized;
   };
 
   struct SplittedSubscriptions

rclcpp/include/rclcpp/experimental/subscription_intra_process.hpp

@@ -28,6 +28,8 @@
 #include "rclcpp/experimental/buffers/intra_process_buffer.hpp"
 #include "rclcpp/experimental/create_intra_process_buffer.hpp"
 #include "rclcpp/experimental/subscription_intra_process_base.hpp"
+#include "rclcpp/serialization.hpp"
+#include "rclcpp/serialized_message.hpp"
 #include "rclcpp/type_support_decl.hpp"
 #include "rclcpp/waitable.hpp"
 #include "tracetools/tracetools.h"
@@ -37,6 +39,8 @@ namespace rclcpp
 namespace experimental
 {
 
+class SerializedMessage;
+
 template<
   typename MessageT,
   typename Alloc = std::allocator<void>,
@@ -51,6 +55,10 @@ public:
   using MessageAlloc = typename MessageAllocTraits::allocator_type;
   using ConstMessageSharedPtr = std::shared_ptr<const MessageT>;
   using MessageUniquePtr = std::unique_ptr<MessageT, Deleter>;
+  using CallbackMessageAllocTraits = allocator::AllocRebind<CallbackMessageT, Alloc>;
+  using CallbackMessageAlloc = typename CallbackMessageAllocTraits::allocator_type;
+  using CallbackMessageUniquePtr = std::unique_ptr<CallbackMessageT>;
+  using CallbackMessageSharedPtr = std::shared_ptr<CallbackMessageT>;
 
   using BufferUniquePtr = typename rclcpp::experimental::buffers::IntraProcessBuffer<
     MessageT,
@@ -64,11 +72,15 @@ public:
     rclcpp::Context::SharedPtr context,
     const std::string & topic_name,
     rmw_qos_profile_t qos_profile,
-    rclcpp::IntraProcessBufferType buffer_type)
+    rclcpp::IntraProcessBufferType buffer_type,
+    std::shared_ptr<SerializationBase> serializer)
   : SubscriptionIntraProcessBase(topic_name, qos_profile),
-    any_callback_(callback)
+    any_callback_(callback), serializer_(serializer)
   {
-    if (!std::is_same<MessageT, CallbackMessageT>::value) {
+    if (!std::is_same<MessageT, CallbackMessageT>::value &&
+      !std::is_same<MessageT, rclcpp::SerializedMessage>::value &&
+      !std::is_same<CallbackMessageT, rcl_serialized_message_t>::value)
+    {
       throw std::runtime_error("SubscriptionIntraProcess wrong callback type");
     }
 
@@ -142,18 +154,47 @@ private:
     (void)ret;
   }
 
+  // convert from ROS2 message to rcl_serialized_message_t (serilizatino needed)
   template<typename T>
-  typename std::enable_if<std::is_same<T, rcl_serialized_message_t>::value, void>::type
+  typename std::enable_if<
+    std::is_same<T, rcl_serialized_message_t>::value &&
+    !std::is_same<MessageT, rclcpp::SerializedMessage>::value,
+    void>::type
   execute_impl()
   {
-    throw std::runtime_error("Subscription intra-process can't handle serialized messages");
+    if (nullptr == serializer_) {
+      throw std::runtime_error("Subscription intra-process can't handle serialized messages");
+    }
+
+    rmw_message_info_t msg_info = {};
+    msg_info.from_intra_process = true;
+
+    ConstMessageSharedPtr msg = buffer_->consume_shared();
+    auto serialized_msg =
+      serializer_->serialize_message(reinterpret_cast<const void *>(msg.get()));
+
+    if (nullptr == serialized_msg) {
+      throw std::runtime_error("Subscription intra-process could not serialize message");
+    }
+
+    if (any_callback_.use_take_shared_method()) {
+      any_callback_.dispatch_intra_process(serialized_msg, msg_info);
+    } else {
+      throw std::runtime_error(
+              "Subscription intra-process for serialized "
+              "messages does not support unique pointers.");
+    }
   }
 
+  // forward from ROS2 message to ROS2 message (same type)
   template<class T>
-  typename std::enable_if<!std::is_same<T, rcl_serialized_message_t>::value, void>::type
+  typename std::enable_if<
+    !std::is_same<T, rcl_serialized_message_t>::value &&
+    !std::is_same<MessageT, rclcpp::SerializedMessage>::value,
+    void>::type
   execute_impl()
   {
-    rmw_message_info_t msg_info;
+    rmw_message_info_t msg_info = {};
     msg_info.publisher_gid = {0, {0}};
     msg_info.from_intra_process = true;
 
@@ -166,8 +207,80 @@ private:
     }
   }
 
+  // forward from rcl_serialized_message_t to SerializationMessage (no conversion needed)
+  template<typename T>
+  typename std::enable_if<
+    std::is_same<T, rcl_serialized_message_t>::value &&
+    std::is_same<MessageT, rclcpp::SerializedMessage>::value,
+    void>::type
+  execute_impl()
+  {
+    rmw_message_info_t msg_info = {};
+    msg_info.from_intra_process = true;
+
+    if (any_callback_.use_take_shared_method()) {
+      ConstMessageSharedPtr msg = buffer_->consume_shared();
+      if (msg == nullptr) {
+        throw std::runtime_error("Subscription intra-process could not get serialized message");
+      }
+      any_callback_.dispatch_intra_process(msg, msg_info);
+    } else {
+      throw std::runtime_error(
+              "Subscription intra-process for serialized "
+              "messages does not support unique pointers.");
+    }
+  }
+
+  // convert from rcl_serialized_message_t to ROS2 message (deserialization needed)
+  template<class T>
+  typename std::enable_if<
+    !std::is_same<T, rcl_serialized_message_t>::value &&
+    std::is_same<MessageT, rclcpp::SerializedMessage>::value,
+    void>::type
+  execute_impl()
+  {
+    if (nullptr == serializer_) {
+      throw std::runtime_error("Subscription intra-process can't handle unserialized messages");
+    }
+
+    ConstMessageSharedPtr serialized_container = buffer_->consume_shared();
+    if (nullptr == serialized_container) {
+      throw std::runtime_error("Subscription intra-process could not get serialized message");
+    }
+
+    rmw_message_info_t msg_info = {};
+    msg_info.from_intra_process = true;
+
+    if (any_callback_.use_take_shared_method()) {
+      CallbackMessageSharedPtr msg = construct_unique();
+      serializer_->deserialize_message(
+        serialized_container.get(),
+        reinterpret_cast<void *>(msg.get()));
+      any_callback_.dispatch_intra_process(msg, msg_info);
+    } else {
+      CallbackMessageUniquePtr msg = construct_unique();
+      serializer_->deserialize_message(
+        serialized_container.get(),
+        reinterpret_cast<void *>(msg.get()));
+      any_callback_.dispatch_intra_process(std::move(msg), msg_info);
+    }
+  }
+
+  CallbackMessageUniquePtr construct_unique()
+  {
+    CallbackMessageUniquePtr unique_msg;
+    auto ptr = CallbackMessageAllocTraits::allocate(*message_allocator_.get(), 1);
+    CallbackMessageAllocTraits::construct(*message_allocator_.get(), ptr);
+    unique_msg = CallbackMessageUniquePtr(ptr);
+
+    return unique_msg;
+  }
+
   AnySubscriptionCallback<CallbackMessageT, Alloc> any_callback_;
   BufferUniquePtr buffer_;
+  std::shared_ptr<SerializationBase> serializer_;
+  std::shared_ptr<CallbackMessageAlloc> message_allocator_ =
+    std::make_shared<CallbackMessageAlloc>();
 };
 
 }  // namespace experimental

rclcpp/include/rclcpp/future_return_code.hpp

@@ -1,53 +0,0 @@
-// Copyright 2014 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__FUTURE_RETURN_CODE_HPP_
-#define RCLCPP__FUTURE_RETURN_CODE_HPP_
-
-#include <iostream>
-#include <string>
-
-#include "rclcpp/visibility_control.hpp"
-
-namespace rclcpp
-{
-
-/// Return codes to be used with spin_until_future_complete.
-/**
- * SUCCESS: The future is complete and can be accessed with "get" without blocking.
- *          This does not indicate that the operation succeeded; "get" may still throw an exception.
- * INTERRUPTED: The future is not complete, spinning was interrupted by Ctrl-C or another error.
- * TIMEOUT: Spinning timed out.
- */
-enum class FutureReturnCode {SUCCESS, INTERRUPTED, TIMEOUT};
-
-/// Stream operator for FutureReturnCode.
-RCLCPP_PUBLIC
-std::ostream &
-operator<<(std::ostream & os, const FutureReturnCode & future_return_code);
-
-/// String conversion function for FutureReturnCode.
-RCLCPP_PUBLIC
-std::string
-to_string(const FutureReturnCode & future_return_code);
-
-namespace executor
-{
-
-using FutureReturnCode [[deprecated("use rclcpp::FutureReturnCode instead")]] = FutureReturnCode;
-
-}  // namespace executor
-}  // namespace rclcpp
-
-#endif  // RCLCPP__FUTURE_RETURN_CODE_HPP_

rclcpp/include/rclcpp/guard_condition.hpp

@@ -47,7 +47,7 @@ public:
   RCLCPP_PUBLIC
   explicit GuardCondition(
     rclcpp::Context::SharedPtr context =
-    rclcpp::contexts::get_global_default_context());
+    rclcpp::contexts::default_context::get_global_default_context());
 
   RCLCPP_PUBLIC
   virtual

rclcpp/include/rclcpp/memory_strategy.hpp

@@ -67,27 +67,27 @@ public:
 
   virtual void
   get_next_subscription(
-    rclcpp::AnyExecutable & any_exec,
+    rclcpp::executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) = 0;
 
   virtual void
   get_next_service(
-    rclcpp::AnyExecutable & any_exec,
+    rclcpp::executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) = 0;
 
   virtual void
   get_next_client(
-    rclcpp::AnyExecutable & any_exec,
+    rclcpp::executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) = 0;
 
   virtual void
   get_next_timer(
-    rclcpp::AnyExecutable & any_exec,
+    rclcpp::executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) = 0;
 
   virtual void
   get_next_waitable(
-    rclcpp::AnyExecutable & any_exec,
+    rclcpp::executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) = 0;
 
   virtual rcl_allocator_t
@@ -115,30 +115,30 @@ public:
 
   static rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
   get_node_by_group(
-    rclcpp::CallbackGroup::SharedPtr group,
+    rclcpp::callback_group::CallbackGroup::SharedPtr group,
     const WeakNodeList & weak_nodes);
 
-  static rclcpp::CallbackGroup::SharedPtr
+  static rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_subscription(
     rclcpp::SubscriptionBase::SharedPtr subscription,
     const WeakNodeList & weak_nodes);
 
-  static rclcpp::CallbackGroup::SharedPtr
+  static rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_service(
     rclcpp::ServiceBase::SharedPtr service,
     const WeakNodeList & weak_nodes);
 
-  static rclcpp::CallbackGroup::SharedPtr
+  static rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_client(
     rclcpp::ClientBase::SharedPtr client,
     const WeakNodeList & weak_nodes);
 
-  static rclcpp::CallbackGroup::SharedPtr
+  static rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_timer(
     rclcpp::TimerBase::SharedPtr timer,
     const WeakNodeList & weak_nodes);
 
-  static rclcpp::CallbackGroup::SharedPtr
+  static rclcpp::callback_group::CallbackGroup::SharedPtr
   get_group_by_waitable(
     rclcpp::Waitable::SharedPtr waitable,
     const WeakNodeList & weak_nodes);

rclcpp/include/rclcpp/node.hpp

@@ -135,12 +135,12 @@ public:
 
   /// Create and return a callback group.
   RCLCPP_PUBLIC
-  rclcpp::CallbackGroup::SharedPtr
-  create_callback_group(rclcpp::CallbackGroupType group_type);
+  rclcpp::callback_group::CallbackGroup::SharedPtr
+  create_callback_group(rclcpp::callback_group::CallbackGroupType group_type);
 
   /// Return the list of callback groups in the node.
   RCLCPP_PUBLIC
-  const std::vector<rclcpp::CallbackGroup::WeakPtr> &
+  const std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> &
   get_callback_groups() const;
 
   /// Create and return a Publisher.
@@ -227,7 +227,7 @@ public:
   create_wall_timer(
     std::chrono::duration<DurationRepT, DurationT> period,
     CallbackT callback,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   /* Create and return a Client. */
   template<typename ServiceT>
@@ -235,7 +235,7 @@ public:
   create_client(
     const std::string & service_name,
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_services_default,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   /* Create and return a Service. */
   template<typename ServiceT, typename CallbackT>
@@ -244,7 +244,7 @@ public:
     const std::string & service_name,
     CallbackT && callback,
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_services_default,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   /// Declare and initialize a parameter, return the effective value.
   /**
@@ -1142,7 +1142,7 @@ private:
 
   RCLCPP_PUBLIC
   bool
-  group_in_node(CallbackGroup::SharedPtr group);
+  group_in_node(callback_group::CallbackGroup::SharedPtr group);
 
   rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_;
   rclcpp::node_interfaces::NodeGraphInterface::SharedPtr node_graph_;

rclcpp/include/rclcpp/node_impl.hpp

@@ -38,7 +38,6 @@
 #include "rclcpp/create_publisher.hpp"
 #include "rclcpp/create_service.hpp"
 #include "rclcpp/create_subscription.hpp"
-#include "rclcpp/detail/resolve_enable_topic_statistics.hpp"
 #include "rclcpp/parameter.hpp"
 #include "rclcpp/qos.hpp"
 #include "rclcpp/type_support_decl.hpp"
@@ -106,7 +105,7 @@ typename rclcpp::WallTimer<CallbackT>::SharedPtr
 Node::create_wall_timer(
   std::chrono::duration<DurationRepT, DurationT> period,
   CallbackT callback,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   auto timer = rclcpp::WallTimer<CallbackT>::make_shared(
     std::chrono::duration_cast<std::chrono::nanoseconds>(period),
@@ -121,7 +120,7 @@ typename Client<ServiceT>::SharedPtr
 Node::create_client(
   const std::string & service_name,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   return rclcpp::create_client<ServiceT>(
     node_base_,
@@ -138,7 +137,7 @@ Node::create_service(
   const std::string & service_name,
   CallbackT && callback,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   return rclcpp::create_service<ServiceT, CallbackT>(
     node_base_,

rclcpp/include/rclcpp/node_interfaces/node_base.hpp

@@ -42,8 +42,7 @@ public:
     const std::string & namespace_,
     rclcpp::Context::SharedPtr context,
     const rcl_node_options_t & rcl_node_options,
-    bool use_intra_process_default,
-    bool enable_topic_statistics_default);
+    bool use_intra_process_default);
 
   RCLCPP_PUBLIC
   virtual
@@ -96,22 +95,22 @@ public:
 
   RCLCPP_PUBLIC
 
-  rclcpp::CallbackGroup::SharedPtr
-  create_callback_group(rclcpp::CallbackGroupType group_type) override;
+  rclcpp::callback_group::CallbackGroup::SharedPtr
+  create_callback_group(rclcpp::callback_group::CallbackGroupType group_type) override;
 
   RCLCPP_PUBLIC
 
-  rclcpp::CallbackGroup::SharedPtr
+  rclcpp::callback_group::CallbackGroup::SharedPtr
   get_default_callback_group() override;
 
   RCLCPP_PUBLIC
 
   bool
-  callback_group_in_node(rclcpp::CallbackGroup::SharedPtr group) override;
+  callback_group_in_node(rclcpp::callback_group::CallbackGroup::SharedPtr group) override;
 
   RCLCPP_PUBLIC
 
-  const std::vector<rclcpp::CallbackGroup::WeakPtr> &
+  const std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> &
   get_callback_groups() const override;
 
   RCLCPP_PUBLIC
@@ -134,20 +133,16 @@ public:
   bool
   get_use_intra_process_default() const override;
 
-  bool
-  get_enable_topic_statistics_default() const override;
-
 private:
   RCLCPP_DISABLE_COPY(NodeBase)
 
   rclcpp::Context::SharedPtr context_;
   bool use_intra_process_default_;
-  bool enable_topic_statistics_default_;
 
   std::shared_ptr<rcl_node_t> node_handle_;
 
-  rclcpp::CallbackGroup::SharedPtr default_callback_group_;
-  std::vector<rclcpp::CallbackGroup::WeakPtr> callback_groups_;
+  rclcpp::callback_group::CallbackGroup::SharedPtr default_callback_group_;
+  std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> callback_groups_;
 
   std::atomic_bool associated_with_executor_;
 

rclcpp/include/rclcpp/node_interfaces/node_base_interface.hpp

@@ -111,25 +111,25 @@ public:
   /// Create and return a callback group.
   RCLCPP_PUBLIC
   virtual
-  rclcpp::CallbackGroup::SharedPtr
-  create_callback_group(rclcpp::CallbackGroupType group_type) = 0;
+  rclcpp::callback_group::CallbackGroup::SharedPtr
+  create_callback_group(rclcpp::callback_group::CallbackGroupType group_type) = 0;
 
   /// Return the default callback group.
   RCLCPP_PUBLIC
   virtual
-  rclcpp::CallbackGroup::SharedPtr
+  rclcpp::callback_group::CallbackGroup::SharedPtr
   get_default_callback_group() = 0;
 
   /// Return true if the given callback group is associated with this node.
   RCLCPP_PUBLIC
   virtual
   bool
-  callback_group_in_node(rclcpp::CallbackGroup::SharedPtr group) = 0;
+  callback_group_in_node(rclcpp::callback_group::CallbackGroup::SharedPtr group) = 0;
 
   /// Return list of callback groups associated with this node.
   RCLCPP_PUBLIC
   virtual
-  const std::vector<rclcpp::CallbackGroup::WeakPtr> &
+  const std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> &
   get_callback_groups() const = 0;
 
   /// Return the atomic bool which is used to ensure only one executor is used.
@@ -161,12 +161,6 @@ public:
   virtual
   bool
   get_use_intra_process_default() const = 0;
-
-  /// Return the default preference for enabling topic statistics collection.
-  RCLCPP_PUBLIC
-  virtual
-  bool
-  get_enable_topic_statistics_default() const = 0;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_interfaces/node_services.hpp

@@ -46,14 +46,14 @@ public:
   void
   add_client(
     rclcpp::ClientBase::SharedPtr client_base_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) override;
 
   RCLCPP_PUBLIC
 
   void
   add_service(
     rclcpp::ServiceBase::SharedPtr service_base_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) override;
 
 private:
   RCLCPP_DISABLE_COPY(NodeServices)

rclcpp/include/rclcpp/node_interfaces/node_services_interface.hpp

@@ -41,14 +41,14 @@ public:
   void
   add_client(
     rclcpp::ClientBase::SharedPtr client_base_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) = 0;
 
   RCLCPP_PUBLIC
   virtual
   void
   add_service(
     rclcpp::ServiceBase::SharedPtr service_base_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) = 0;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_interfaces/node_timers.hpp

@@ -46,7 +46,7 @@ public:
   void
   add_timer(
     rclcpp::TimerBase::SharedPtr timer,
-    rclcpp::CallbackGroup::SharedPtr callback_group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) override;
 
 private:
   RCLCPP_DISABLE_COPY(NodeTimers)

rclcpp/include/rclcpp/node_interfaces/node_timers_interface.hpp

@@ -41,7 +41,7 @@ public:
   void
   add_timer(
     rclcpp::TimerBase::SharedPtr timer,
-    rclcpp::CallbackGroup::SharedPtr callback_group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) = 0;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_interfaces/node_topics.hpp

@@ -55,7 +55,7 @@ public:
   void
   add_publisher(
     rclcpp::PublisherBase::SharedPtr publisher,
-    rclcpp::CallbackGroup::SharedPtr callback_group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) override;
 
   RCLCPP_PUBLIC
   rclcpp::SubscriptionBase::SharedPtr
@@ -68,7 +68,7 @@ public:
   void
   add_subscription(
     rclcpp::SubscriptionBase::SharedPtr subscription,
-    rclcpp::CallbackGroup::SharedPtr callback_group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) override;
 
   RCLCPP_PUBLIC
   rclcpp::node_interfaces::NodeBaseInterface *

rclcpp/include/rclcpp/node_interfaces/node_topics_interface.hpp

@@ -59,7 +59,7 @@ public:
   void
   add_publisher(
     rclcpp::PublisherBase::SharedPtr publisher,
-    rclcpp::CallbackGroup::SharedPtr callback_group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) = 0;
 
   RCLCPP_PUBLIC
   virtual
@@ -74,7 +74,7 @@ public:
   void
   add_subscription(
     rclcpp::SubscriptionBase::SharedPtr subscription,
-    rclcpp::CallbackGroup::SharedPtr callback_group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr callback_group) = 0;
 
   RCLCPP_PUBLIC
   virtual

rclcpp/include/rclcpp/node_interfaces/node_waitables.hpp

@@ -45,14 +45,14 @@ public:
   void
   add_waitable(
     rclcpp::Waitable::SharedPtr waitable_base_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) override;
 
   RCLCPP_PUBLIC
 
   void
   remove_waitable(
     rclcpp::Waitable::SharedPtr waitable_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) noexcept override;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) noexcept override;
 
 private:
   RCLCPP_DISABLE_COPY(NodeWaitables)

rclcpp/include/rclcpp/node_interfaces/node_waitables_interface.hpp

@@ -40,7 +40,7 @@ public:
   void
   add_waitable(
     rclcpp::Waitable::SharedPtr waitable_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) = 0;
 
   /// \note this function should not throw because it may be called in destructors
   RCLCPP_PUBLIC
@@ -48,7 +48,7 @@ public:
   void
   remove_waitable(
     rclcpp::Waitable::SharedPtr waitable_ptr,
-    rclcpp::CallbackGroup::SharedPtr group) noexcept = 0;
+    rclcpp::callback_group::CallbackGroup::SharedPtr group) noexcept = 0;
 };
 
 }  // namespace node_interfaces

rclcpp/include/rclcpp/node_options.hpp

@@ -38,12 +38,11 @@ public:
   /**
    * Default values for the node options:
    *
-   *   - context = rclcpp::contexts::get_global_default_context()
+   *   - context = rclcpp::contexts::default_context::get_global_default_context()
    *   - arguments = {}
    *   - parameter_overrides = {}
    *   - use_global_arguments = true
    *   - use_intra_process_comms = false
-   *   - enable_topic_statistics = false
    *   - start_parameter_services = true
    *   - start_parameter_event_publisher = true
    *   - parameter_event_qos = rclcpp::ParameterEventQoS
@@ -188,23 +187,6 @@ public:
   NodeOptions &
   use_intra_process_comms(bool use_intra_process_comms);
 
-  /// Return the enable_topic_statistics flag.
-  RCLCPP_PUBLIC
-  bool
-  enable_topic_statistics() const;
-
-  /// Set the enable_topic_statistics flag, return this for parameter idiom.
-  /**
-   * If true, topic statistics collection and publication will be enabled
-   * for all subscriptions.
-   * This can be used to override the global topic statistics setting.
-   *
-   * Defaults to false.
-   */
-  RCLCPP_PUBLIC
-  NodeOptions &
-  enable_topic_statistics(bool enable_topic_statistics);
-
   /// Return the start_parameter_services flag.
   RCLCPP_PUBLIC
   bool
@@ -338,7 +320,7 @@ private:
   // documentation in this class.
 
   rclcpp::Context::SharedPtr context_ {
-    rclcpp::contexts::get_global_default_context()};
+    rclcpp::contexts::default_context::get_global_default_context()};
 
   std::vector<std::string> arguments_ {};
 
@@ -350,8 +332,6 @@ private:
 
   bool use_intra_process_comms_ {false};
 
-  bool enable_topic_statistics_ {false};
-
   bool start_parameter_services_ {true};
 
   bool start_parameter_event_publisher_ {true};

rclcpp/include/rclcpp/parameter_client.hpp

@@ -54,21 +54,21 @@ public:
     const rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services_interface,
     const std::string & remote_node_name = "",
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_parameters,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   RCLCPP_PUBLIC
   AsyncParametersClient(
     const rclcpp::Node::SharedPtr node,
     const std::string & remote_node_name = "",
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_parameters,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   RCLCPP_PUBLIC
   AsyncParametersClient(
     rclcpp::Node * node,
     const std::string & remote_node_name = "",
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_parameters,
-    rclcpp::CallbackGroup::SharedPtr group = nullptr);
+    rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr);
 
   RCLCPP_PUBLIC
   std::shared_future<std::vector<rclcpp::Parameter>>
@@ -205,7 +205,7 @@ public:
 
   RCLCPP_PUBLIC
   SyncParametersClient(
-    rclcpp::Executor::SharedPtr executor,
+    rclcpp::executor::Executor::SharedPtr executor,
     rclcpp::Node::SharedPtr node,
     const std::string & remote_node_name = "",
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_parameters);
@@ -218,14 +218,14 @@ public:
 
   RCLCPP_PUBLIC
   SyncParametersClient(
-    rclcpp::Executor::SharedPtr executor,
+    rclcpp::executor::Executor::SharedPtr executor,
     rclcpp::Node * node,
     const std::string & remote_node_name = "",
     const rmw_qos_profile_t & qos_profile = rmw_qos_profile_parameters);
 
   RCLCPP_PUBLIC
   SyncParametersClient(
-    rclcpp::Executor::SharedPtr executor,
+    rclcpp::executor::Executor::SharedPtr executor,
     const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_interface,
     const rclcpp::node_interfaces::NodeTopicsInterface::SharedPtr node_topics_interface,
     const rclcpp::node_interfaces::NodeGraphInterface::SharedPtr node_graph_interface,
@@ -339,7 +339,7 @@ public:
   }
 
 private:
-  rclcpp::Executor::SharedPtr executor_;
+  rclcpp::executor::Executor::SharedPtr executor_;
   const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_interface_;
   AsyncParametersClient::SharedPtr async_parameters_client_;
 };

rclcpp/include/rclcpp/publisher.hpp

@@ -56,6 +56,12 @@ public:
   using MessageDeleter = allocator::Deleter<MessageAllocator, MessageT>;
   using MessageUniquePtr = std::unique_ptr<MessageT, MessageDeleter>;
   using MessageSharedPtr = std::shared_ptr<const MessageT>;
+  using SerializedMessageAllocatorTraits =
+    allocator::AllocRebind<rclcpp::SerializedMessage,
+      AllocatorT>;
+  using SerializedMessageAllocator = typename SerializedMessageAllocatorTraits::allocator_type;
+  using SerializedMessageDeleter = allocator::Deleter<SerializedMessageAllocator,
+      rclcpp::SerializedMessage>;
 
   RCLCPP_SMART_PTR_DEFINITIONS(Publisher<MessageT, AllocatorT>)
 
@@ -70,37 +76,28 @@ public:
       *rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>(),
       options.template to_rcl_publisher_options<MessageT>(qos)),
     options_(options),
-    message_allocator_(new MessageAllocator(*options.get_allocator().get()))
+    message_allocator_(new MessageAllocator(*options.get_allocator().get())),
+    message_allocator_serialized_(new SerializedMessageAllocator(*options.get_allocator().get()))
   {
-    allocator::set_allocator_for_deleter(&message_deleter_, message_allocator_.get());
+    init_setup();
+  }
 
-    if (options_.event_callbacks.deadline_callback) {
-      this->add_event_handler(
-        options_.event_callbacks.deadline_callback,
-        RCL_PUBLISHER_OFFERED_DEADLINE_MISSED);
-    }
-    if (options_.event_callbacks.liveliness_callback) {
-      this->add_event_handler(
-        options_.event_callbacks.liveliness_callback,
-        RCL_PUBLISHER_LIVELINESS_LOST);
-    }
-    if (options_.event_callbacks.incompatible_qos_callback) {
-      this->add_event_handler(
-        options_.event_callbacks.incompatible_qos_callback,
-        RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS);
-    } else if (options_.use_default_callbacks) {
-      // Register default callback when not specified
-      try {
-        this->add_event_handler(
-          [this](QOSOfferedIncompatibleQoSInfo & info) {
-            this->default_incompatible_qos_callback(info);
-          },
-          RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS);
-      } catch (UnsupportedEventTypeException & /*exc*/) {
-        // pass
-      }
-    }
-    // Setup continues in the post construction method, post_init_setup().
+  Publisher(
+    rclcpp::node_interfaces::NodeBaseInterface * node_base,
+    const std::string & topic,
+    const rclcpp::QoS & qos,
+    const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options,
+    const rosidl_message_type_support_t & type_support)
+  : PublisherBase(
+      node_base,
+      topic,
+      type_support,
+      options.template to_rcl_publisher_options<MessageT>(qos)),
+    options_(options),
+    message_allocator_(new MessageAllocator(*options.get_allocator().get())),
+    message_allocator_serialized_(new SerializedMessageAllocator(*options.get_allocator().get()))
+  {
+    init_setup();
   }
 
   /// Called post construction, so that construction may continue after shared_from_this() works.
@@ -135,8 +132,11 @@ public:
                 "intraprocess communication allowed only with volatile durability");
       }
       uint64_t intra_process_publisher_id = ipm->add_publisher(this->shared_from_this());
+      uint64_t intra_process_publisher_id_serialized = ipm->add_publisher(
+        this->shared_from_this(), true);
       this->setup_intra_process(
         intra_process_publisher_id,
+        intra_process_publisher_id_serialized,
         ipm);
     }
   }
@@ -173,6 +173,11 @@ public:
   virtual void
   publish(std::unique_ptr<MessageT, MessageDeleter> msg)
   {
+    if (std::is_same<MessageT, rcl_serialized_message_t>::value) {
+      this->template publish<MessageDeleter>(std::move(msg));
+      return;
+    }
+
     if (!intra_process_is_enabled_) {
       this->do_inter_process_publish(*msg);
       return;
@@ -187,34 +192,41 @@ public:
       get_subscription_count() > get_intra_process_subscription_count();
 
     if (inter_process_publish_needed) {
-      auto shared_msg = this->do_intra_process_publish_and_return_shared(std::move(msg));
+      auto shared_msg = this->do_intra_process_publish_and_return_shared(
+        std::move(msg), message_allocator_);
       this->do_inter_process_publish(*shared_msg);
     } else {
-      this->do_intra_process_publish(std::move(msg));
+      this->do_intra_process_publish(std::move(msg), message_allocator_);
     }
   }
 
   virtual void
   publish(const MessageT & msg)
   {
-    // Avoid allocating when not using intra process.
-    if (!intra_process_is_enabled_) {
-      // In this case we're not using intra process.
-      return this->do_inter_process_publish(msg);
-    }
-    // Otherwise we have to allocate memory in a unique_ptr and pass it along.
-    // As the message is not const, a copy should be made.
-    // A shared_ptr<const MessageT> could also be constructed here.
-    auto ptr = MessageAllocatorTraits::allocate(*message_allocator_.get(), 1);
-    MessageAllocatorTraits::construct(*message_allocator_.get(), ptr, msg);
-    MessageUniquePtr unique_msg(ptr, message_deleter_);
-    this->publish(std::move(unique_msg));
+    this->do_publish_message(msg);
   }
 
-  void
+  template<class T = MessageT>
+  typename std::enable_if<!std::is_same<T, rcl_serialized_message_t>::value>::type
   publish(const rcl_serialized_message_t & serialized_msg)
   {
-    return this->do_serialized_publish(&serialized_msg);
+    this->do_publish_message<rcl_serialized_message_t>(serialized_msg);
+  }
+
+  /// Publish a serialized message. Non specialized version to prevent compiling errors.
+  template<typename TDeleter, typename T>
+  void publish(std::unique_ptr<T, TDeleter> serialized_msg)
+  {
+    (void)serialized_msg;
+    throw std::runtime_error(
+            "publishing unique_ptr with custom deleter only supported for serialized messages");
+  }
+
+  /// Publish a serialized message.
+  template<typename TDeleter>
+  void publish(std::unique_ptr<rcl_serialized_message_t, TDeleter> serialized_msg)
+  {
+    this->do_serialized_publish(*serialized_msg);
   }
 
   /// Publish an instance of a LoanedMessage.
@@ -259,6 +271,69 @@ public:
   }
 
 protected:
+  void init_setup()
+  {
+    allocator::set_allocator_for_deleter(&message_deleter_, message_allocator_.get());
+
+    if (options_.event_callbacks.deadline_callback) {
+      this->add_event_handler(
+        options_.event_callbacks.deadline_callback,
+        RCL_PUBLISHER_OFFERED_DEADLINE_MISSED);
+    }
+    if (options_.event_callbacks.liveliness_callback) {
+      this->add_event_handler(
+        options_.event_callbacks.liveliness_callback,
+        RCL_PUBLISHER_LIVELINESS_LOST);
+    }
+    if (options_.event_callbacks.incompatible_qos_callback) {
+      this->add_event_handler(
+        options_.event_callbacks.incompatible_qos_callback,
+        RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS);
+    } else if (options_.use_default_callbacks) {
+      // Register default callback when not specified
+      try {
+        this->add_event_handler(
+          [this](QOSOfferedIncompatibleQoSInfo & info) {
+            this->default_incompatible_qos_callback(info);
+          },
+          RCL_PUBLISHER_OFFERED_INCOMPATIBLE_QOS);
+      } catch (UnsupportedEventTypeException & /*exc*/) {
+        RCLCPP_WARN_ONCE(
+          rclcpp::get_logger(rcl_node_get_logger_name(rcl_node_handle_.get())),
+          "This rmw implementation does not support ON_OFFERED_INCOMPATIBLE_QOS "
+          "events, you will not be notified when Publishers offer an incompatible "
+          "QoS profile to Subscriptions on the same topic.");
+      }
+    }
+    // Setup continues in the post construction method, post_init_setup().
+  }
+
+  template<class T = MessageT>
+  typename std::enable_if<!std::is_same<T, rcl_serialized_message_t>::value>::type
+  do_publish_message(const T & msg)
+  {
+    // Avoid allocating when not using intra process.
+    if (!intra_process_is_enabled_) {
+      // In this case we're not using intra process.
+      return this->do_inter_process_publish(msg);
+    }
+    // Otherwise we have to allocate memory in a unique_ptr and pass it along.
+    // As the message is not const, a copy should be made.
+    // A shared_ptr<const MessageT> could also be constructed here.
+    auto ptr = MessageAllocatorTraits::allocate(*message_allocator_.get(), 1);
+    MessageAllocatorTraits::construct(*message_allocator_.get(), ptr, msg);
+    MessageUniquePtr unique_msg(ptr, message_deleter_);
+    this->publish(std::move(unique_msg));
+  }
+
+  template<class T = MessageT>
+  typename std::enable_if<std::is_same<T, rcl_serialized_message_t>::value>::type
+  do_publish_message(const T & msg)
+  {
+    // Kept for backwards compatibility. Copies compelete memory!
+    this->publish(std::make_unique<rclcpp::SerializedMessage>(msg));
+  }
+
   void
   do_inter_process_publish(const MessageT & msg)
   {
@@ -280,15 +355,24 @@ protected:
   }
 
   void
-  do_serialized_publish(const rcl_serialized_message_t * serialized_msg)
+  do_serialized_publish(rcl_serialized_message_t serialized_msg)
   {
-    if (intra_process_is_enabled_) {
-      // TODO(Karsten1987): support serialized message passed by intraprocess
-      throw std::runtime_error("storing serialized messages in intra process is not supported yet");
+    bool inter_process_publish_needed =
+      get_subscription_count() > get_intra_process_subscription_count();
+
+    if (inter_process_publish_needed) {
+      // declare here to avoid deletion before returning method
+      auto status = rcl_publish_serialized_message(&publisher_handle_, &serialized_msg, nullptr);
+      if (RCL_RET_OK != status) {
+        rclcpp::exceptions::throw_from_rcl_error(status, "failed to publish serialized message");
+      }
     }
-    auto status = rcl_publish_serialized_message(&publisher_handle_, serialized_msg, nullptr);
-    if (RCL_RET_OK != status) {
-      rclcpp::exceptions::throw_from_rcl_error(status, "failed to publish serialized message");
+
+    auto msg = std::make_unique<rclcpp::SerializedMessage>(
+      std::move(serialized_msg));
+
+    if (intra_process_is_enabled_) {
+      do_intra_process_publish(std::move(msg), message_allocator_serialized_);
     }
   }
 
@@ -312,8 +396,11 @@ protected:
     }
   }
 
+  template<typename T, typename Deleter, typename Allocator>
   void
-  do_intra_process_publish(std::unique_ptr<MessageT, MessageDeleter> msg)
+  do_intra_process_publish(
+    std::unique_ptr<T, Deleter> msg,
+    std::shared_ptr<Allocator> & message_allocator)
   {
     auto ipm = weak_ipm_.lock();
     if (!ipm) {
@@ -324,14 +411,21 @@ protected:
       throw std::runtime_error("cannot publish msg which is a null pointer");
     }
 
-    ipm->template do_intra_process_publish<MessageT, AllocatorT>(
-      intra_process_publisher_id_,
+    const uint64_t intra_process_publisher_id = std::is_same<T,
+        rclcpp::SerializedMessage>::value ?
+      intra_process_publisher_id_serialized_ : intra_process_publisher_id_;
+
+    ipm->template do_intra_process_publish<T, AllocatorT>(
+      intra_process_publisher_id,
       std::move(msg),
-      message_allocator_);
+      message_allocator);
   }
 
-  std::shared_ptr<const MessageT>
-  do_intra_process_publish_and_return_shared(std::unique_ptr<MessageT, MessageDeleter> msg)
+  template<typename T, typename Deleter, typename Allocator>
+  std::shared_ptr<const T>
+  do_intra_process_publish_and_return_shared(
+    std::unique_ptr<T, Deleter> msg,
+    std::shared_ptr<Allocator> & message_allocator)
   {
     auto ipm = weak_ipm_.lock();
     if (!ipm) {
@@ -342,10 +436,14 @@ protected:
       throw std::runtime_error("cannot publish msg which is a null pointer");
     }
 
-    return ipm->template do_intra_process_publish_and_return_shared<MessageT, AllocatorT>(
-      intra_process_publisher_id_,
+    const uint64_t intra_process_publisher_id = std::is_same<T,
+        rclcpp::SerializedMessage>::value ?
+      intra_process_publisher_id_serialized_ : intra_process_publisher_id_;
+
+    return ipm->template do_intra_process_publish_and_return_shared<T, AllocatorT>(
+      intra_process_publisher_id,
       std::move(msg),
-      message_allocator_);
+      message_allocator);
   }
 
   /// Copy of original options passed during construction.
@@ -356,6 +454,7 @@ protected:
   const rclcpp::PublisherOptionsWithAllocator<AllocatorT> options_;
 
   std::shared_ptr<MessageAllocator> message_allocator_;
+  std::shared_ptr<SerializedMessageAllocator> message_allocator_serialized_;
 
   MessageDeleter message_deleter_;
 };

rclcpp/include/rclcpp/publisher_base.hpp

@@ -191,6 +191,7 @@ public:
   void
   setup_intra_process(
     uint64_t intra_process_publisher_id,
+    uint64_t intra_process_publisher_id_serialized,
     IntraProcessManagerSharedPtr ipm);
 
 protected:
@@ -222,6 +223,7 @@ protected:
   bool intra_process_is_enabled_;
   IntraProcessManagerWeakPtr weak_ipm_;
   uint64_t intra_process_publisher_id_;
+  uint64_t intra_process_publisher_id_serialized_;
 
   rmw_gid_t rmw_gid_;
 };

rclcpp/include/rclcpp/publisher_factory.hpp

@@ -63,17 +63,21 @@ struct PublisherFactory
 /// Return a PublisherFactory with functions setup for creating a PublisherT<MessageT, AllocatorT>.
 template<typename MessageT, typename AllocatorT, typename PublisherT>
 PublisherFactory
-create_publisher_factory(const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options)
+create_publisher_factory(
+  const rclcpp::PublisherOptionsWithAllocator<AllocatorT> & options,
+  const rosidl_message_type_support_t & type_support)
 {
   PublisherFactory factory {
     // factory function that creates a MessageT specific PublisherT
-    [options](
+    [options, type_support](
       rclcpp::node_interfaces::NodeBaseInterface * node_base,
       const std::string & topic_name,
       const rclcpp::QoS & qos
     ) -> std::shared_ptr<PublisherT>
     {
-      auto publisher = std::make_shared<PublisherT>(node_base, topic_name, qos, options);
+      auto publisher = std::make_shared<PublisherT>(
+        node_base, topic_name, qos, options,
+        type_support);
       // This is used for setting up things like intra process comms which
       // require this->shared_from_this() which cannot be called from
       // the constructor.

rclcpp/include/rclcpp/publisher_options.hpp

@@ -30,7 +30,10 @@
 namespace rclcpp
 {
 
+namespace callback_group
+{
 class CallbackGroup;
+}  // namespace callback_group
 
 /// Non-templated part of PublisherOptionsWithAllocator<Allocator>.
 struct PublisherOptionsBase
@@ -45,7 +48,7 @@ struct PublisherOptionsBase
   bool use_default_callbacks = true;
 
   /// Callback group in which the waitable items from the publisher should be placed.
-  std::shared_ptr<rclcpp::CallbackGroup> callback_group;
+  std::shared_ptr<rclcpp::callback_group::CallbackGroup> callback_group;
 
   /// Optional RMW implementation specific payload to be used during creation of the publisher.
   std::shared_ptr<rclcpp::detail::RMWImplementationSpecificPublisherPayload>

rclcpp/include/rclcpp/serialization.hpp

@@ -15,86 +15,107 @@
 #ifndef RCLCPP__SERIALIZATION_HPP_
 #define RCLCPP__SERIALIZATION_HPP_
 
+#include <rmw/rmw.h>
+
 #include <memory>
-#include <stdexcept>
 #include <string>
 
-#include "rclcpp/visibility_control.hpp"
-
-#include "rcl/types.h"
-
-#include "rosidl_runtime_c/message_type_support_struct.h"
-
-#include "rosidl_typesupport_cpp/message_type_support.hpp"
+#include "rcl/error_handling.h"
 
 namespace rclcpp
 {
 
-class SerializedMessage;
-
-namespace serialization_traits
-{
-// trait to check if type is the object oriented serialized message
-template<typename T>
-struct is_serialized_message_class : std::false_type
-{};
-
-template<>
-struct is_serialized_message_class<rcl_serialized_message_t>: std::true_type
-{};
-
-template<>
-struct is_serialized_message_class<SerializedMessage>: std::true_type
-{};
-}  // namespace serialization_traits
-
 /// Interface to (de)serialize a message
-class RCLCPP_PUBLIC_TYPE SerializationBase
+class SerializationBase
 {
 public:
-  /// Constructor of SerializationBase
-  /**
-   * \param[in] type_support handle for the message type support
-   * to be used for serialization and deserialization.
-   */
-  explicit SerializationBase(const rosidl_message_type_support_t * type_support);
-
-  /// Destructor of SerializationBase
   virtual ~SerializationBase() = default;
 
   /// Serialize a ROS2 message to a serialized stream
   /**
    * \param[in] message The ROS2 message which is read and serialized by rmw.
-   * \param[out] serialized_message The serialized message.
    */
-  void serialize_message(
-    const void * ros_message, SerializedMessage * serialized_message) const;
+  virtual std::shared_ptr<rcl_serialized_message_t> serialize_message(const void * message) = 0;
 
   /// Deserialize a serialized stream to a ROS message
   /**
    * \param[in] serialized_message The serialized message to be converted to ROS2 by rmw.
    * \param[out] message The deserialized ROS2 message.
    */
-  void deserialize_message(
-    const SerializedMessage * serialized_message, void * ros_message) const;
-
-private:
-  const rosidl_message_type_support_t * type_support_;
+  virtual void deserialize_message(
+    const rcl_serialized_message_t * serialized_message,
+    void * msg) = 0;
 };
 
 /// Default implementation to (de)serialize a message by using rmw_(de)serialize
-template<typename MessageT>
 class Serialization : public SerializationBase
 {
 public:
-  /// Constructor of Serialization
-  Serialization()
-  : SerializationBase(rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>())
+  Serialization(
+    const rosidl_message_type_support_t & type_support,
+    const rcutils_allocator_t allocator = rcutils_get_default_allocator())
+  : type_support_(type_support), rcutils_allocator_(allocator)
+  {}
+
+  std::shared_ptr<rcl_serialized_message_t> serialize_message(const void * message) override
   {
-    static_assert(
-      !serialization_traits::is_serialized_message_class<MessageT>::value,
-      "Serialization of serialized message to serialized message is not possible.");
+    auto serialized_message = new rcl_serialized_message_t;
+    *serialized_message = rmw_get_zero_initialized_serialized_message();
+    const auto ret = rmw_serialized_message_init(serialized_message, 0, &rcutils_allocator_);
+    if (ret != RCUTILS_RET_OK) {
+      throw std::runtime_error(
+              "Error allocating resources for serialized message: " +
+              std::string(rcutils_get_error_string().str));
+    }
+
+    if (nullptr == message) {
+      delete serialized_message;
+      throw std::runtime_error("Message is nullpointer while serialization.");
+    }
+
+    const auto error = rmw_serialize(
+      message,
+      &type_support_,
+      serialized_message);
+    if (error != RCL_RET_OK) {
+      delete serialized_message;
+      throw std::runtime_error("Failed to serialize.");
+    }
+
+    auto shared_serialized_msg = std::shared_ptr<rcl_serialized_message_t>(
+      serialized_message,
+      [](rcl_serialized_message_t * msg) {
+        auto fini_ret = rmw_serialized_message_fini(msg);
+        delete msg;
+        if (fini_ret != RCL_RET_OK) {
+          RCUTILS_LOG_ERROR_NAMED(
+            "rclcpp",
+            "Failed to destroy serialized message: %s", rcl_get_error_string().str);
+        }
+      });
+
+    return shared_serialized_msg;
   }
+
+  void deserialize_message(const rcl_serialized_message_t * serialized_message, void * msg) override
+  {
+    if (nullptr == serialized_message ||
+      serialized_message->buffer_capacity == 0 ||
+      serialized_message->buffer_length == 0 ||
+      !serialized_message->buffer)
+    {
+      throw std::runtime_error("Failed to deserialize nullptr serialized message.");
+    }
+
+    const auto ret = rmw_deserialize(serialized_message, &type_support_, msg);
+    if (ret != RMW_RET_OK) {
+      throw std::runtime_error("Failed to deserialize serialized message.");
+    }
+  }
+
+private:
+  rosidl_message_type_support_t type_support_;
+  rcutils_allocator_t rcutils_allocator_;
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/serialized_message.hpp

@@ -15,91 +15,64 @@
 #ifndef RCLCPP__SERIALIZED_MESSAGE_HPP_
 #define RCLCPP__SERIALIZED_MESSAGE_HPP_
 
-#include "rcl/allocator.h"
-#include "rcl/types.h"
+#include <rclcpp/exceptions.hpp>
 
-#include "rclcpp/visibility_control.hpp"
+#include <cstring>
+
+#include "rcutils/logging_macros.h"
+
+#include "rmw/serialized_message.h"
 
 namespace rclcpp
 {
 
 /// Object oriented version of rcl_serialized_message_t with destructor to avoid memory leaks
-class RCLCPP_PUBLIC_TYPE SerializedMessage
+class SerializedMessage : public rcl_serialized_message_t
 {
 public:
-  /// Default constructor for a SerializedMessage
-  /**
-   * Default constructs a serialized message and initalizes it
-   * with initial capacity of 0.
-   * The allocator defaults to `rcl_get_default_allocator()`.
-   *
-   * \param[in] allocator The allocator to be used for the initialization.
-   */
-  explicit SerializedMessage(
-    const rcl_allocator_t & allocator = rcl_get_default_allocator());
+  SerializedMessage()
+  : rcl_serialized_message_t(rmw_get_zero_initialized_serialized_message())
+  {}
 
-  /// Default constructor for a SerializedMessage
-  /**
-   * Default constructs a serialized message and initalizes it
-   * with the provided capacity.
-   * The allocator defaults to `rcl_get_default_allocator()`.
-   *
-   * \param[in] initial_capacity The amount of memory to be allocated.
-   * \param[in] allocator The allocator to be used for the initialization.
-   */
-  SerializedMessage(
-    size_t initial_capacity,
-    const rcl_allocator_t & allocator = rcl_get_default_allocator());
+  explicit SerializedMessage(const SerializedMessage & serialized_message)
+  : SerializedMessage(static_cast<const rcl_serialized_message_t>(serialized_message))
+  {}
 
-  /// Copy Constructor for a SerializedMessage
-  SerializedMessage(const SerializedMessage & other);
+  explicit SerializedMessage(const rcl_serialized_message_t & serialized_message)
+  : rcl_serialized_message_t(rmw_get_zero_initialized_serialized_message())
+  {
+    const auto ret = rmw_serialized_message_init(
+      this, serialized_message.buffer_length,
+      &serialized_message.allocator);
+    if (ret != RCL_RET_OK) {
+      rclcpp::exceptions::throw_from_rcl_error(ret);
+    }
 
-  /// Constructor for a SerializedMessage from a rcl_serialized_message_t
-  explicit SerializedMessage(const rcl_serialized_message_t & other);
+    // do not call memcpy if the pointer is "static"
+    if (buffer != serialized_message.buffer) {
+      std::memcpy(buffer, serialized_message.buffer, serialized_message.buffer_length);
+    }
+    buffer_length = serialized_message.buffer_length;
+  }
 
-  /// Move Constructor for a SerializedMessage
-  SerializedMessage(SerializedMessage && other);
+  explicit SerializedMessage(rcl_serialized_message_t && msg)
+  : rcl_serialized_message_t(msg)
+  {
+    // reset buffer to prevent double free
+    msg = rmw_get_zero_initialized_serialized_message();
+  }
 
-  /// Constructor for a SerializedMessage from a moved rcl_serialized_message_t
-  explicit SerializedMessage(rcl_serialized_message_t && other);
-
-  /// Copy assignment operator
-  SerializedMessage & operator=(const SerializedMessage & other);
-
-  /// Copy assignment operator from a rcl_serialized_message_t
-  SerializedMessage & operator=(const rcl_serialized_message_t & other);
-
-  /// Move assignment operator
-  SerializedMessage & operator=(SerializedMessage && other);
-
-  /// Move assignment operator from a rcl_serialized_message_t
-  SerializedMessage & operator=(rcl_serialized_message_t && other);
-
-  /// Destructor for a SerializedMessage
-  virtual ~SerializedMessage();
-
-  /// Get the underlying rcl_serialized_t handle
-  rcl_serialized_message_t & get_rcl_serialized_message();
-
-  // Get a const handle to the underlying rcl_serialized_message_t
-  const rcl_serialized_message_t & get_rcl_serialized_message() const;
-
-  /// Get the size of the serialized data buffer
-  /**
-   * Note, this is different from the actual amount of allocated memory.
-   * This can be obtained via a call to `capacity`.
-   */
-  size_t size() const;
-
-  /// Get the size of allocated memory for the data buffer
-  /**
-   * Note, this is different from the amount of content in the buffer.
-   * This can be obtained via a call to `size`.
-   */
-  size_t capacity() const;
-
-private:
-  rcl_serialized_message_t serialized_message_;
+  ~SerializedMessage()
+  {
+    if (nullptr != buffer) {
+      const auto fini_ret = rmw_serialized_message_fini(this);
+      if (fini_ret != RCL_RET_OK) {
+        RCUTILS_LOG_ERROR_NAMED(
+          "rclcpp",
+          "Failed to destroy serialized message: %s", rcl_get_error_string().str);
+      }
+    }
+  }
 };
 
 }  // namespace rclcpp

rclcpp/include/rclcpp/strategies/allocator_memory_strategy.hpp

@@ -263,7 +263,7 @@ public:
 
   void
   get_next_subscription(
-    rclcpp::AnyExecutable & any_exec,
+    executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) override
   {
     auto it = subscription_handles_.begin();
@@ -298,7 +298,7 @@ public:
 
   void
   get_next_service(
-    rclcpp::AnyExecutable & any_exec,
+    executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) override
   {
     auto it = service_handles_.begin();
@@ -332,7 +332,7 @@ public:
   }
 
   void
-  get_next_client(rclcpp::AnyExecutable & any_exec, const WeakNodeList & weak_nodes) override
+  get_next_client(executor::AnyExecutable & any_exec, const WeakNodeList & weak_nodes) override
   {
     auto it = client_handles_.begin();
     while (it != client_handles_.end()) {
@@ -366,7 +366,7 @@ public:
 
   void
   get_next_timer(
-    rclcpp::AnyExecutable & any_exec,
+    executor::AnyExecutable & any_exec,
     const WeakNodeList & weak_nodes) override
   {
     auto it = timer_handles_.begin();
@@ -400,7 +400,7 @@ public:
   }
 
   void
-  get_next_waitable(rclcpp::AnyExecutable & any_exec, const WeakNodeList & weak_nodes) override
+  get_next_waitable(executor::AnyExecutable & any_exec, const WeakNodeList & weak_nodes) override
   {
     auto it = waitable_handles_.begin();
     while (it != waitable_handles_.end()) {

rclcpp/include/rclcpp/strategies/message_pool_memory_strategy.hpp

@@ -17,8 +17,6 @@
 
 #include <memory>
 
-#include "rosidl_runtime_cpp/traits.hpp"
-
 #include "rclcpp/macros.hpp"
 #include "rclcpp/message_memory_strategy.hpp"
 #include "rclcpp/visibility_control.hpp"

rclcpp/include/rclcpp/subscription.hpp

@@ -41,6 +41,7 @@
 #include "rclcpp/message_info.hpp"
 #include "rclcpp/message_memory_strategy.hpp"
 #include "rclcpp/node_interfaces/node_base_interface.hpp"
+#include "rclcpp/serialized_message.hpp"
 #include "rclcpp/subscription_base.hpp"
 #include "rclcpp/subscription_options.hpp"
 #include "rclcpp/subscription_traits.hpp"
@@ -155,29 +156,84 @@ public:
                 "intraprocess communication allowed only with volatile durability");
       }
 
-      // First create a SubscriptionIntraProcess which will be given to the intra-process manager.
-      auto context = node_base->get_context();
-      using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<
-        CallbackMessageT,
-        AllocatorT,
-        typename MessageUniquePtr::deleter_type>;
-      auto subscription_intra_process = std::make_shared<SubscriptionIntraProcessT>(
-        callback,
-        options.get_allocator(),
-        context,
-        this->get_topic_name(),  // important to get like this, as it has the fully-qualified name
-        qos_profile,
-        resolve_intra_process_buffer_type(options.intra_process_buffer_type, callback));
-      TRACEPOINT(
-        rclcpp_subscription_init,
-        (const void *)get_subscription_handle().get(),
-        (const void *)subscription_intra_process.get());
-
-      // Add it to the intra process manager.
       using rclcpp::experimental::IntraProcessManager;
+      uint64_t intra_process_subscription_id;
+      uint64_t intra_process_subscription_id_serialized;
+
+      auto context = node_base->get_context();
       auto ipm = context->get_sub_context<IntraProcessManager>();
-      uint64_t intra_process_subscription_id = ipm->add_subscription(subscription_intra_process);
-      this->setup_intra_process(intra_process_subscription_id, ipm);
+
+      {
+        // First create a SubscriptionIntraProcess which will be given to the intra-process manager.
+        auto subscription_intra_process = std::make_shared<
+          rclcpp::experimental::SubscriptionIntraProcess<
+            CallbackMessageT,
+            AllocatorT,
+            typename MessageUniquePtr::deleter_type
+          >>(
+          callback,
+          options.get_allocator(),
+          context,
+          this->get_topic_name(),  // important to get it by the fully-qualified name
+          qos.get_rmw_qos_profile(),
+          resolve_intra_process_buffer_type(options.intra_process_buffer_type, callback),
+          std::make_shared<rclcpp::Serialization>(
+            type_support_handle,
+            options.template to_rcl_subscription_options<CallbackMessageT>(qos).allocator)
+          );
+        TRACEPOINT(
+          rclcpp_subscription_init,
+          (const void *)get_subscription_handle().get(),
+          (const void *)subscription_intra_process.get());
+
+        // Add it to the intra process manager.
+        intra_process_subscription_id = ipm->add_subscription(subscription_intra_process);
+      }
+
+      {
+        using SerializedMessageAllocatorTraits =
+          allocator::AllocRebind<rclcpp::SerializedMessage,
+            AllocatorT>;
+        using SerializedMessageAllocator =
+          typename SerializedMessageAllocatorTraits::allocator_type;
+        using SerializedMessageDeleter = allocator::Deleter<SerializedMessageAllocator,
+            rclcpp::SerializedMessage>;
+        using SerializedMessageUniquePtr =
+          std::unique_ptr<rclcpp::SerializedMessage,
+            SerializedMessageDeleter>;
+
+        // First create a SubscriptionIntraProcess which will be given to the intra-process manager.
+        auto subscription_intra_process = std::make_shared<
+          rclcpp::experimental::SubscriptionIntraProcess<
+            rclcpp::SerializedMessage,
+            AllocatorT,
+            typename SerializedMessageUniquePtr::deleter_type,
+            CallbackMessageT
+          >>(
+          callback,
+          options.get_allocator(),
+          context,
+          this->get_topic_name(),    // important to get it by the fully-qualified name
+          qos.get_rmw_qos_profile(),
+          resolve_intra_process_buffer_type(options.intra_process_buffer_type, callback),
+          std::make_shared<rclcpp::Serialization>(
+            type_support_handle,
+            options.template to_rcl_subscription_options<CallbackMessageT>(qos).allocator)
+          );
+        TRACEPOINT(
+          rclcpp_subscription_init,
+          (const void *)get_subscription_handle().get(),
+          (const void *)subscription_intra_process.get());
+
+        // Add it to the intra process manager.
+        intra_process_subscription_id_serialized = ipm->add_subscription(
+          subscription_intra_process,
+          true);
+      }
+
+      this->setup_intra_process(
+        {intra_process_subscription_id,
+          intra_process_subscription_id_serialized}, ipm);
     }
 
     TRACEPOINT(

rclcpp/include/rclcpp/subscription_base.hpp

@@ -228,13 +228,13 @@ public:
   RCLCPP_PUBLIC
   void
   setup_intra_process(
-    uint64_t intra_process_subscription_id,
+    const std::vector<uint64_t> & intra_process_subscription_ids,
     IntraProcessManagerWeakPtr weak_ipm);
 
   /// Return the waitable for intra-process, or nullptr if intra-process is not setup.
   RCLCPP_PUBLIC
-  rclcpp::Waitable::SharedPtr
-  get_intra_process_waitable() const;
+  std::vector<rclcpp::Waitable::SharedPtr>
+  get_intra_process_waitables() const;
 
   /// Exchange state of whether or not a part of the subscription is used by a wait set.
   /**
@@ -286,7 +286,7 @@ protected:
 
   bool use_intra_process_;
   IntraProcessManagerWeakPtr weak_ipm_;
-  uint64_t intra_process_subscription_id_;
+  std::vector<uint64_t> intra_process_subscription_ids_;
 
 private:
   RCLCPP_DISABLE_COPY(SubscriptionBase)

rclcpp/include/rclcpp/subscription_factory.hpp

@@ -78,7 +78,8 @@ SubscriptionFactory
 create_subscription_factory(
   CallbackT && callback,
   const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options,
-  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat)
+  typename MessageMemoryStrategyT::SharedPtr msg_mem_strat,
+  const rosidl_message_type_support_t & type_support)
 {
   auto allocator = options.get_allocator();
 
@@ -88,7 +89,7 @@ create_subscription_factory(
 
   SubscriptionFactory factory {
     // factory function that creates a MessageT specific SubscriptionT
-    [options, msg_mem_strat, any_subscription_callback](
+    [options, msg_mem_strat, any_subscription_callback, type_support](
       rclcpp::node_interfaces::NodeBaseInterface * node_base,
       const std::string & topic_name,
       const rclcpp::QoS & qos
@@ -99,7 +100,7 @@ create_subscription_factory(
 
       auto sub = Subscription<CallbackMessageT, AllocatorT>::make_shared(
         node_base,
-        *rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>(),
+        type_support,
         topic_name,
         qos,
         any_subscription_callback,

rclcpp/include/rclcpp/subscription_options.hpp

@@ -15,7 +15,6 @@
 #ifndef RCLCPP__SUBSCRIPTION_OPTIONS_HPP_
 #define RCLCPP__SUBSCRIPTION_OPTIONS_HPP_
 
-#include <chrono>
 #include <memory>
 #include <string>
 #include <vector>
@@ -26,7 +25,6 @@
 #include "rclcpp/intra_process_setting.hpp"
 #include "rclcpp/qos.hpp"
 #include "rclcpp/qos_event.hpp"
-#include "rclcpp/topic_statistics_state.hpp"
 #include "rclcpp/visibility_control.hpp"
 
 namespace rclcpp
@@ -45,7 +43,7 @@ struct SubscriptionOptionsBase
   bool ignore_local_publications = false;
 
   /// The callback group for this subscription. NULL to use the default callback group.
-  rclcpp::CallbackGroup::SharedPtr callback_group = nullptr;
+  rclcpp::callback_group::CallbackGroup::SharedPtr callback_group = nullptr;
 
   /// Setting to explicitly set intraprocess communications.
   IntraProcessSetting use_intra_process_comm = IntraProcessSetting::NodeDefault;
@@ -56,21 +54,6 @@ struct SubscriptionOptionsBase
   /// Optional RMW implementation specific payload to be used during creation of the subscription.
   std::shared_ptr<rclcpp::detail::RMWImplementationSpecificSubscriptionPayload>
   rmw_implementation_payload = nullptr;
-
-  // Options to configure topic statistics collector in the subscription.
-  struct TopicStatisticsOptions
-  {
-    // Enable and disable topic statistics calculation and publication. Defaults to disabled.
-    TopicStatisticsState state = TopicStatisticsState::NodeDefault;
-
-    // Topic to which topic statistics get published when enabled. Defaults to /statistics.
-    std::string publish_topic = "/statistics";
-
-    // Topic statistics publication period in ms. Defaults to one minute.
-    std::chrono::milliseconds publish_period{std::chrono::seconds(1)};
-  };
-
-  TopicStatisticsOptions topic_stats_options;
 };
 
 /// Structure containing optional configuration for Subscriptions.
@@ -121,6 +104,7 @@ struct SubscriptionOptionsWithAllocator : public SubscriptionOptionsBase
 };
 
 using SubscriptionOptions = SubscriptionOptionsWithAllocator<std::allocator<void>>;
+
 }  // namespace rclcpp
 
 #endif  // RCLCPP__SUBSCRIPTION_OPTIONS_HPP_

rclcpp/include/rclcpp/subscription_traits.hpp

@@ -18,7 +18,6 @@
 #include <memory>
 
 #include "rclcpp/function_traits.hpp"
-#include "rclcpp/serialized_message.hpp"
 #include "rclcpp/subscription_options.hpp"
 #include "rcl/types.h"
 
@@ -51,15 +50,6 @@ struct is_serialized_subscription_argument<std::shared_ptr<rcl_serialized_messag
   : std::true_type
 {};
 
-template<>
-struct is_serialized_subscription_argument<SerializedMessage>: std::true_type
-{};
-
-template<>
-struct is_serialized_subscription_argument<std::shared_ptr<SerializedMessage>>
-  : std::true_type
-{};
-
 template<typename T>
 struct is_serialized_subscription : is_serialized_subscription_argument<T>
 {};

rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp

@@ -1,204 +0,0 @@
-// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
-//
-// 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__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_
-#define RCLCPP__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_
-
-#include <memory>
-#include <string>
-#include <utility>
-#include <vector>
-
-#include "libstatistics_collector/collector/generate_statistics_message.hpp"
-#include "libstatistics_collector/moving_average_statistics/types.hpp"
-#include "libstatistics_collector/topic_statistics_collector/constants.hpp"
-#include "libstatistics_collector/topic_statistics_collector/received_message_period.hpp"
-
-#include "rcl/time.h"
-#include "rclcpp/time.hpp"
-#include "rclcpp/publisher.hpp"
-#include "rclcpp/timer.hpp"
-
-#include "statistics_msgs/msg/metrics_message.hpp"
-
-namespace rclcpp
-{
-namespace topic_statistics
-{
-
-constexpr const char kDefaultPublishTopicName[]{"/statistics"};
-constexpr const std::chrono::milliseconds kDefaultPublishingPeriod{std::chrono::seconds(1)};
-
-using libstatistics_collector::collector::GenerateStatisticMessage;
-using statistics_msgs::msg::MetricsMessage;
-using libstatistics_collector::moving_average_statistics::StatisticData;
-
-/**
- * Class used to collect, measure, and publish topic statistics data. Current statistics
- * supported for subscribers are received message age and received message period.
- *
- * \tparam CallbackMessageT the subscribed message type
- */
-template<typename CallbackMessageT>
-class SubscriptionTopicStatistics
-{
-  using TopicStatsCollector =
-    libstatistics_collector::topic_statistics_collector::TopicStatisticsCollector<
-    CallbackMessageT>;
-  using ReceivedMessagePeriod =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessagePeriodCollector<
-    CallbackMessageT>;
-
-public:
-  /// Construct a SubscriptionTopicStatistics object.
-  /**
-   * This object wraps utilities, defined in libstatistics_collector, to collect,
-   * measure, and publish topic statistics data. This throws an invalid_argument
-   * if the input publisher is null.
-   *
-   * \param node_name the name of the node, which created this instance, in order to denote
-   * topic source
-   * \param publisher instance constructed by the node in order to publish statistics data.
-   * This class owns the publisher.
-   */
-  SubscriptionTopicStatistics(
-    const std::string & node_name,
-    rclcpp::Publisher<statistics_msgs::msg::MetricsMessage>::SharedPtr publisher)
-  : node_name_(node_name),
-    publisher_(std::move(publisher))
-  {
-    // TODO(dbbonnie): ros-tooling/aws-roadmap/issues/226, received message age
-
-    if (nullptr == publisher_) {
-      throw std::invalid_argument("publisher pointer is nullptr");
-    }
-
-    bring_up();
-  }
-
-  virtual ~SubscriptionTopicStatistics()
-  {
-    tear_down();
-  }
-
-  /// Handle a message received by the subscription to collect statistics.
-  /**
-   * \param received_message the message received by the subscription
-   * \param now_nanoseconds current time in nanoseconds
-   */
-  virtual void handle_message(
-    const CallbackMessageT & received_message,
-    const rclcpp::Time now_nanoseconds) const
-  {
-    for (const auto & collector : subscriber_statistics_collectors_) {
-      collector->OnMessageReceived(received_message, now_nanoseconds.nanoseconds());
-    }
-  }
-
-  /// Set the timer used to publish statistics messages.
-  /**
-   * \param measurement_timer the timer to fire the publisher, created by the node
-   */
-  void set_publisher_timer(rclcpp::TimerBase::SharedPtr publisher_timer)
-  {
-    publisher_timer_ = publisher_timer;
-  }
-
-  /// Publish a populated MetricsStatisticsMessage.
-  virtual void publish_message()
-  {
-    rclcpp::Time window_end{get_current_nanoseconds_since_epoch()};
-
-    for (auto & collector : subscriber_statistics_collectors_) {
-      const auto collected_stats = collector->GetStatisticsResults();
-
-      auto message = libstatistics_collector::collector::GenerateStatisticMessage(
-        node_name_,
-        collector->GetMetricName(),
-        collector->GetMetricUnit(),
-        window_start_,
-        window_end,
-        collected_stats);
-      publisher_->publish(message);
-    }
-    window_start_ = window_end;
-  }
-
-protected:
-  /// Return a vector of all the currently collected data.
-  /**
-   * \return a vector of all the collected data
-   */
-  std::vector<StatisticData> get_current_collector_data() const
-  {
-    std::vector<StatisticData> data;
-    for (const auto & collector : subscriber_statistics_collectors_) {
-      data.push_back(collector->GetStatisticsResults());
-    }
-    return data;
-  }
-
-private:
-  /// Construct and start all collectors and set window_start_.
-  void bring_up()
-  {
-    auto received_message_period = std::make_unique<ReceivedMessagePeriod>();
-    received_message_period->Start();
-    subscriber_statistics_collectors_.emplace_back(std::move(received_message_period));
-
-    window_start_ = rclcpp::Time(get_current_nanoseconds_since_epoch());
-  }
-
-  /// Stop all collectors, clear measurements, stop publishing timer, and reset publisher.
-  void tear_down()
-  {
-    for (auto & collector : subscriber_statistics_collectors_) {
-      collector->Stop();
-    }
-
-    subscriber_statistics_collectors_.clear();
-
-    if (publisher_timer_) {
-      publisher_timer_->cancel();
-      publisher_timer_.reset();
-    }
-
-    publisher_.reset();
-  }
-
-  /// Return the current nanoseconds (count) since epoch.
-  /**
-   * \return the current nanoseconds (count) since epoch
-   */
-  int64_t get_current_nanoseconds_since_epoch() const
-  {
-    const auto now = std::chrono::system_clock::now();
-    return std::chrono::duration_cast<std::chrono::nanoseconds>(now.time_since_epoch()).count();
-  }
-
-  /// Collection of statistics collectors
-  std::vector<std::unique_ptr<TopicStatsCollector>> subscriber_statistics_collectors_{};
-  /// Node name used to generate topic statistics messages to be published
-  const std::string node_name_;
-  /// Publisher, created by the node, used to publish topic statistics messages
-  rclcpp::Publisher<statistics_msgs::msg::MetricsMessage>::SharedPtr publisher_;
-  /// Timer which fires the publisher
-  rclcpp::TimerBase::SharedPtr publisher_timer_;
-  /// The start of the collection window, used in the published topic statistics message
-  rclcpp::Time window_start_;
-};
-}  // namespace topic_statistics
-}  // namespace rclcpp
-
-#endif  // RCLCPP__TOPIC_STATISTICS__SUBSCRIPTION_TOPIC_STATISTICS_HPP_

rclcpp/include/rclcpp/topic_statistics_state.hpp

@@ -1,35 +0,0 @@
-// Copyright 2019 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__TOPIC_STATISTICS_STATE_HPP_
-#define RCLCPP__TOPIC_STATISTICS_STATE_HPP_
-
-namespace rclcpp
-{
-
-/// Represent the state of topic statistics collector.
-/// Used as argument in create_subscriber.
-enum class TopicStatisticsState
-{
-  /// Explicitly enable topic statistics at subscription level.
-  Enable,
-  /// Explicitly disable topic statistics at subscription level.
-  Disable,
-  /// Take topic statistics state from the node.
-  NodeDefault
-};
-
-}  // namespace rclcpp
-
-#endif  // RCLCPP__TOPIC_STATISTICS_STATE_HPP_

rclcpp/include/rclcpp/utilities.hpp

@@ -45,7 +45,7 @@ namespace rclcpp
 /// Initialize communications via the rmw implementation and set up a global signal handler.
 /**
  * Initializes the global context which is accessible via the function
- * rclcpp::contexts::get_global_default_context().
+ * rclcpp::contexts::default_context::get_global_default_context().
  * Also, installs the global signal handlers with the function
  * rclcpp::install_signal_handlers().
  *

rclcpp/include/rclcpp/wait_set_algorithms.hpp

@@ -1,48 +0,0 @@
-// Copyright 2020 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__WAIT_SET_ALGORITHMS_HPP_
-#define RCLCPP__WAIT_SET_ALGORITHMS_HPP_
-
-#include <algorithm>
-
-#include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_result.hpp"
-
-namespace rclcpp
-{
-
-/// Find next ready entity out of the given sequence from start to last.
-template<class InputIt>
-InputIt
-find_next_ready_wait_set_entity(InputIt start, InputIt last)
-{
-  return std::find_if(start, last, [](const auto & entity) {return nullptr != entity;});
-}
-
-/// Find next ready guard condition given a WaitResult.
-template<class WaitSetT>
-typename WaitSetT::StoragePolicy::GuardConditionsIterable::const_iterator
-find_next_ready_guard_condition(
-  const typename rclcpp::WaitResult<WaitSetT> & wait_result,
-  typename WaitSetT::StoragePolicy::GuardConditionsIterable::const_iterator start = (
-    wait_result.get_wait_set()))
-{
-  return
-}
-
-
-}  // namespace rclcpp
-
-#endif  // RCLCPP__WAIT_SET_ALGORITHMS_HPP_

rclcpp/include/rclcpp/wait_set_policies/detail/entry_types.hpp

@@ -1,123 +0,0 @@
-// Copyright 2020 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__WAIT_SET_POLICIES__DETAIL__ENTRY_TYPES_HPP_
-#define RCLCPP__WAIT_SET_POLICIES__DETAIL__ENTRY_TYPES_HPP_
-
-#include <limits>
-#include <memory>
-
-#include "rclcpp/client.hpp"
-#include "rclcpp/guard_condition.hpp"
-#include "rclcpp/service.hpp"
-#include "rclcpp/subscription_base.hpp"
-#include "rclcpp/subscription_wait_set_mask.hpp"
-#include "rclcpp/timer.hpp"
-#include "rclcpp/visibility_control.hpp"
-#include "rclcpp/waitable.hpp"
-
-namespace rclcpp
-{
-namespace wait_set_policies
-{
-namespace detail
-{
-
-constexpr const size_t not_in_wait_set = (std::numeric_limits<std::size_t>::max)();
-
-/// Templated ownership entries used for storage.
-template<class PointerT>
-class EntryTemplate : public PointerT
-{
-public:
-  size_t rcl_wait_set_index;
-  constexpr static const size_t not_in_wait_set = not_in_wait_set;
-
-  /// Conversion constructor, which is intentionally not marked explicit.
-  EntryTemplate(
-    const PointerT & entity_in = nullptr,
-    size_t rcl_wait_set_index_in = not_in_wait_set) noexcept
-  : PointerT(entity_in),
-    rcl_wait_set_index(rcl_wait_set_index_in)
-  {}
-};
-
-/// Templated ownership subscription entries used for storage.
-template<class PointerT>
-class SubscriptionEntryTemplate : public EntryTemplate<PointerT>
-{
-public:
-  rclcpp::SubscriptionWaitSetMask mask;
-
-  /// Conversion constructor, which is intentionally not marked explicit.
-  SubscriptionEntryTemplate(
-    const PointerT & entity_in = nullptr,
-    const rclcpp::SubscriptionWaitSetMask & mask_in = {},
-    size_t rcl_wait_set_index_in = not_in_wait_set) noexcept
-  : EntryTemplate<PointerT>(entity_in, rcl_wait_set_index_in),
-    mask(mask_in)
-  {}
-};
-
-/// Templated ownership entries with associated entity used for storage.
-template<class PointerT>
-class EntryWithAssociatedEntityTemplate : public PointerT
-{
-public:
-  std::shared_ptr<void> associated_entity;
-
-  /// Conversion constructor, which is intentionally not marked explicit.
-  EntryWithAssociatedEntityTemplate(
-    const PointerT & entity_in = nullptr,
-    const std::shared_ptr<void> & associated_entity_in = nullptr) noexcept
-  : PointerT(entity_in),
-    associated_entity(associated_entity_in)
-  {}
-};
-
-/// Shared ownership SubscriptionBase entry for storage.
-using SubscriptionEntry = SubscriptionEntryTemplate<std::shared_ptr<rclcpp::SubscriptionBase>>;
-/// Weak ownership SubscriptionBase entry for storage.
-using WeakSubscriptionEntry = SubscriptionEntryTemplate<std::weak_ptr<rclcpp::SubscriptionBase>>;
-
-/// Shared ownership GuardCondition entry for storage.
-using GuardConditionEntry = EntryTemplate<std::shared_ptr<rclcpp::GuardCondition>>;
-/// Weak ownership GuardCondition entry for storage.
-using WeakGuardConditionEntry = EntryTemplate<std::weak_ptr<rclcpp::GuardCondition>>;
-
-/// Shared ownership Timer entry for storage.
-using TimerEntry = EntryTemplate<std::shared_ptr<rclcpp::TimerBase>>;
-/// Weak ownership Timer entry for storage.
-using WeakTimerEntry = EntryTemplate<std::weak_ptr<rclcpp::TimerBase>>;
-
-/// Shared ownership Client entry for storage.
-using ClientEntry = EntryTemplate<std::shared_ptr<rclcpp::ClientBase>>;
-/// Weak ownership Client entry for storage.
-using WeakClientEntry = EntryTemplate<std::weak_ptr<rclcpp::ClientBase>>;
-
-/// Shared ownership Service entry for storage.
-using ServiceEntry = EntryTemplate<std::shared_ptr<rclcpp::ServiceBase>>;
-/// Weak ownership Service entry for storage.
-using WeakServiceEntry = EntryTemplate<std::weak_ptr<rclcpp::ServiceBase>>;
-
-/// Shared ownership Waitable entry for storage.
-using WaitableEntry = EntryWithAssociatedEntityTemplate<std::shared_ptr<rclcpp::Waitable>>;
-/// Weak ownership Waitable entry for storage.
-using WeakWaitableEntry = EntryWithAssociatedEntityTemplate<std::weak_ptr<rclcpp::Waitable>>;
-
-}  // namespace detail
-}  // namespace wait_set_policies
-}  // namespace rclcpp
-
-#endif  // RCLCPP__WAIT_SET_POLICIES__DETAIL__ENTRY_TYPES_HPP_

rclcpp/include/rclcpp/wait_set_policies/detail/storage_policy_common.hpp

@@ -59,9 +59,7 @@ protected:
     const WaitablesIterable & waitables,
     rclcpp::Context::SharedPtr context
   )
-  : rcl_wait_set_(rcl_get_zero_initialized_wait_set()),
-    context_(context),
-    previous_size_of_extra_guard_conditions_(extra_guard_conditions.size())
+  : rcl_wait_set_(rcl_get_zero_initialized_wait_set()), context_(context)
   {
     // Check context is not nullptr.
     if (nullptr == context) {
@@ -168,11 +166,6 @@ protected:
   )
   {
     bool was_resized = false;
-    // Check if the extra_guard_conditions has changed.
-    if (previous_size_of_extra_guard_conditions_ != extra_guard_conditions.size()) {
-      previous_size_of_extra_guard_conditions_ = extra_guard_conditions.size();
-      needs_resize_ = true;
-    }
     // Resize the wait set, but only if it needs to be.
     if (needs_resize_) {
       // Resizing with rcl_wait_set_resize() is a no-op if nothing has changed,
@@ -242,7 +235,7 @@ protected:
     // Add subscriptions.
     for (const auto & subscription_entry : subscriptions) {
       auto subscription_ptr_pair =
-        get_raw_pointer_from_smart_pointer(subscription_entry);
+        get_raw_pointer_from_smart_pointer(subscription_entry.subscription);
       if (nullptr == subscription_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
@@ -257,7 +250,7 @@ protected:
       rcl_ret_t ret = rcl_wait_set_add_subscription(
         &rcl_wait_set_,
         subscription_ptr_pair.second->get_subscription_handle().get(),
-        &subscription_entry.rcl_wait_set_index);
+        nullptr);
       if (RCL_RET_OK != ret) {
         rclcpp::exceptions::throw_from_rcl_error(ret);
       }
@@ -283,7 +276,7 @@ protected:
           rcl_ret_t ret = rcl_wait_set_add_guard_condition(
             &rcl_wait_set_,
             &guard_condition_ptr_pair.second->get_rcl_guard_condition(),
-            &guard_condition.rcl_wait_set_index);
+            nullptr);
           if (RCL_RET_OK != ret) {
             rclcpp::exceptions::throw_from_rcl_error(ret);
           }
@@ -313,7 +306,7 @@ protected:
       rcl_ret_t ret = rcl_wait_set_add_timer(
         &rcl_wait_set_,
         timer_ptr_pair.second->get_timer_handle().get(),
-        &timer.rcl_wait_set_index);
+        nullptr);
       if (RCL_RET_OK != ret) {
         rclcpp::exceptions::throw_from_rcl_error(ret);
       }
@@ -336,7 +329,7 @@ protected:
       rcl_ret_t ret = rcl_wait_set_add_client(
         &rcl_wait_set_,
         client_ptr_pair.second->get_client_handle().get(),
-        &clients.rcl_wait_set_index);
+        nullptr);
       if (RCL_RET_OK != ret) {
         rclcpp::exceptions::throw_from_rcl_error(ret);
       }
@@ -359,7 +352,7 @@ protected:
       rcl_ret_t ret = rcl_wait_set_add_service(
         &rcl_wait_set_,
         service_ptr_pair.second->get_service_handle().get(),
-        &service.rcl_wait_set_index);
+        nullptr);
       if (RCL_RET_OK != ret) {
         rclcpp::exceptions::throw_from_rcl_error(ret);
       }
@@ -367,7 +360,7 @@ protected:
 
     // Add waitables.
     for (auto & waitable_entry : waitables) {
-      auto waitable_ptr_pair = get_raw_pointer_from_smart_pointer(waitable_entry);
+      auto waitable_ptr_pair = get_raw_pointer_from_smart_pointer(waitable_entry.waitable);
       if (nullptr == waitable_ptr_pair.second) {
         // In this case it was probably stored as a weak_ptr, but is now locking to nullptr.
         if (HasStrongOwnership) {
@@ -410,7 +403,6 @@ protected:
 
   bool needs_pruning_ = false;
   bool needs_resize_ = false;
-  size_t previous_size_of_extra_guard_conditions_ = 0;
 };
 
 }  // namespace detail

rclcpp/include/rclcpp/wait_set_policies/dynamic_storage.hpp

@@ -20,9 +20,14 @@
 #include <utility>
 #include <vector>
 
+#include "rclcpp/client.hpp"
+#include "rclcpp/guard_condition.hpp"
 #include "rclcpp/macros.hpp"
+#include "rclcpp/service.hpp"
+#include "rclcpp/subscription_base.hpp"
+#include "rclcpp/subscription_wait_set_mask.hpp"
+#include "rclcpp/timer.hpp"
 #include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_set_policies/detail/entry_types.hpp"
 #include "rclcpp/wait_set_policies/detail/storage_policy_common.hpp"
 #include "rclcpp/waitable.hpp"
 
@@ -37,25 +42,126 @@ class DynamicStorage : public rclcpp::wait_set_policies::detail::StoragePolicyCo
 protected:
   using is_mutable = std::true_type;
 
-  using SubscriptionEntry = detail::SubscriptionEntry;
-  using SequenceOfWeakSubscriptions = std::vector<detail::WeakSubscriptionEntry>;
-  using SubscriptionsIterable = std::vector<detail::SubscriptionEntry>;
+  class SubscriptionEntry
+  {
+    // (wjwwood): indent of 'public:' is weird, I know. uncrustify is dumb.
 
-  using SequenceOfWeakGuardConditions = std::vector<detail::WeakGuardConditionEntry>;
-  using GuardConditionsIterable = std::vector<detail::GuardConditionEntry>;
+public:
+    std::shared_ptr<rclcpp::SubscriptionBase> subscription;
+    rclcpp::SubscriptionWaitSetMask mask;
 
-  using SequenceOfWeakTimers = std::vector<detail::WeakTimerEntry>;
-  using TimersIterable = std::vector<detail::TimerEntry>;
+    /// Conversion constructor, which is intentionally not marked explicit.
+    SubscriptionEntry(
+      const std::shared_ptr<rclcpp::SubscriptionBase> & subscription_in = nullptr,
+      const rclcpp::SubscriptionWaitSetMask & mask_in = {})
+    : subscription(subscription_in),
+      mask(mask_in)
+    {}
 
-  using SequenceOfWeakClients = std::vector<detail::WeakClientEntry>;
-  using ClientsIterable = std::vector<detail::ClientEntry>;
+    void
+    reset() noexcept
+    {
+      subscription.reset();
+    }
+  };
+  class WeakSubscriptionEntry
+  {
+public:
+    std::weak_ptr<rclcpp::SubscriptionBase> subscription;
+    rclcpp::SubscriptionWaitSetMask mask;
 
-  using SequenceOfWeakServices = std::vector<detail::WeakServiceEntry>;
-  using ServicesIterable = std::vector<detail::ServiceEntry>;
+    explicit WeakSubscriptionEntry(
+      const std::shared_ptr<rclcpp::SubscriptionBase> & subscription_in,
+      const rclcpp::SubscriptionWaitSetMask & mask_in) noexcept
+    : subscription(subscription_in),
+      mask(mask_in)
+    {}
 
-  using WaitableEntry = detail::WaitableEntry;
-  using SequenceOfWeakWaitables = std::vector<detail::WeakWaitableEntry>;
-  using WaitablesIterable = std::vector<detail::WaitableEntry>;
+    explicit WeakSubscriptionEntry(const SubscriptionEntry & other)
+    : subscription(other.subscription),
+      mask(other.mask)
+    {}
+
+    std::shared_ptr<rclcpp::SubscriptionBase>
+    lock() const
+    {
+      return subscription.lock();
+    }
+
+    bool
+    expired() const noexcept
+    {
+      return subscription.expired();
+    }
+  };
+  using SequenceOfWeakSubscriptions = std::vector<WeakSubscriptionEntry>;
+  using SubscriptionsIterable = std::vector<SubscriptionEntry>;
+
+  using SequenceOfWeakGuardConditions = std::vector<std::weak_ptr<rclcpp::GuardCondition>>;
+  using GuardConditionsIterable = std::vector<std::shared_ptr<rclcpp::GuardCondition>>;
+
+  using SequenceOfWeakTimers = std::vector<std::weak_ptr<rclcpp::TimerBase>>;
+  using TimersIterable = std::vector<std::shared_ptr<rclcpp::TimerBase>>;
+
+  using SequenceOfWeakClients = std::vector<std::weak_ptr<rclcpp::ClientBase>>;
+  using ClientsIterable = std::vector<std::shared_ptr<rclcpp::ClientBase>>;
+
+  using SequenceOfWeakServices = std::vector<std::weak_ptr<rclcpp::ServiceBase>>;
+  using ServicesIterable = std::vector<std::shared_ptr<rclcpp::ServiceBase>>;
+
+  class WaitableEntry
+  {
+public:
+    std::shared_ptr<rclcpp::Waitable> waitable;
+    std::shared_ptr<void> associated_entity;
+
+    /// Conversion constructor, which is intentionally not marked explicit.
+    WaitableEntry(
+      const std::shared_ptr<rclcpp::Waitable> & waitable_in = nullptr,
+      const std::shared_ptr<void> & associated_entity_in = nullptr) noexcept
+    : waitable(waitable_in),
+      associated_entity(associated_entity_in)
+    {}
+
+    void
+    reset() noexcept
+    {
+      waitable.reset();
+      associated_entity.reset();
+    }
+  };
+  class WeakWaitableEntry
+  {
+public:
+    std::weak_ptr<rclcpp::Waitable> waitable;
+    std::weak_ptr<void> associated_entity;
+
+    explicit WeakWaitableEntry(
+      const std::shared_ptr<rclcpp::Waitable> & waitable_in,
+      const std::shared_ptr<void> & associated_entity_in) noexcept
+    : waitable(waitable_in),
+      associated_entity(associated_entity_in)
+    {}
+
+    explicit WeakWaitableEntry(const WaitableEntry & other)
+    : waitable(other.waitable),
+      associated_entity(other.associated_entity)
+    {}
+
+    std::shared_ptr<rclcpp::Waitable>
+    lock() const
+    {
+      return waitable.lock();
+    }
+
+    bool
+    expired() const noexcept
+    {
+      return waitable.expired();
+    }
+  };
+  using SequenceOfWeakWaitables = std::vector<WeakWaitableEntry>;
+  using WaitablesIterable = std::vector<WaitableEntry>;
 
   template<class ArrayOfExtraGuardConditions>
   explicit
@@ -99,13 +205,13 @@ protected:
   storage_rebuild_rcl_wait_set(const ArrayOfExtraGuardConditions & extra_guard_conditions)
   {
     this->storage_rebuild_rcl_wait_set_with_sets(
-      shared_subscriptions_,
-      shared_guard_conditions_,
+      subscriptions_,
+      guard_conditions_,
       extra_guard_conditions,
-      shared_timers_,
-      shared_clients_,
-      shared_services_,
-      shared_waitables_
+      timers_,
+      clients_,
+      services_,
+      waitables_
     );
   }
 
@@ -137,7 +243,7 @@ protected:
     if (this->storage_has_entity(*subscription, subscriptions_)) {
       throw std::runtime_error("subscription already in wait set");
     }
-    detail::WeakSubscriptionEntry weak_entry{std::move(subscription), {}, detail::not_in_wait_set};
+    WeakSubscriptionEntry weak_entry{std::move(subscription), {}};
     subscriptions_.push_back(std::move(weak_entry));
     this->storage_flag_for_resize();
   }
@@ -159,7 +265,7 @@ protected:
     if (this->storage_has_entity(*guard_condition, guard_conditions_)) {
       throw std::runtime_error("guard_condition already in wait set");
     }
-    guard_conditions_.push_back({std::move(guard_condition), detail::not_in_wait_set});
+    guard_conditions_.push_back(std::move(guard_condition));
     this->storage_flag_for_resize();
   }
 
@@ -180,7 +286,7 @@ protected:
     if (this->storage_has_entity(*timer, timers_)) {
       throw std::runtime_error("timer already in wait set");
     }
-    timers_.push_back({std::move(timer), detail::not_in_wait_set});
+    timers_.push_back(std::move(timer));
     this->storage_flag_for_resize();
   }
 
@@ -201,7 +307,7 @@ protected:
     if (this->storage_has_entity(*client, clients_)) {
       throw std::runtime_error("client already in wait set");
     }
-    clients_.push_back({std::move(client), detail::not_in_wait_set});
+    clients_.push_back(std::move(client));
     this->storage_flag_for_resize();
   }
 
@@ -222,7 +328,7 @@ protected:
     if (this->storage_has_entity(*service, services_)) {
       throw std::runtime_error("service already in wait set");
     }
-    services_.push_back({std::move(service), detail::not_in_wait_set});
+    services_.push_back(std::move(service));
     this->storage_flag_for_resize();
   }
 
@@ -245,10 +351,7 @@ protected:
     if (this->storage_has_entity(*waitable, waitables_)) {
       throw std::runtime_error("waitable already in wait set");
     }
-    detail::WeakWaitableEntry weak_entry(
-      std::move(waitable),
-      std::move(associated_entity),
-      detail::not_in_wait_set);
+    WeakWaitableEntry weak_entry(std::move(waitable), std::move(associated_entity));
     waitables_.push_back(std::move(weak_entry));
     this->storage_flag_for_resize();
   }
@@ -279,7 +382,6 @@ protected:
         return weak_ptr.expired();
       };
     // remove guard conditions which have been deleted
-    subscriptions_.erase(std::remove_if(subscriptions_.begin(), subscriptions_.end(), p));
     guard_conditions_.erase(std::remove_if(guard_conditions_.begin(), guard_conditions_.end(), p));
     timers_.erase(std::remove_if(timers_.begin(), timers_.end(), p));
     clients_.erase(std::remove_if(clients_.begin(), clients_.end(), p));
@@ -303,7 +405,6 @@ protected:
         }
       };
     // Lock all the weak pointers and hold them until released.
-    lock_all(subscriptions_, shared_subscriptions_);
     lock_all(guard_conditions_, shared_guard_conditions_);
     lock_all(timers_, shared_timers_);
     lock_all(clients_, shared_clients_);
@@ -314,9 +415,9 @@ protected:
         shared_ptrs.resize(weak_ptrs.size());
         size_t index = 0;
         for (const auto & weak_ptr : weak_ptrs) {
-          shared_ptrs[index++] = detail::WaitableEntry{
-            weak_ptr.lock(),
-            weak_ptr.associated_entity};
+          shared_ptrs[index++] = WaitableEntry{
+            weak_ptr.waitable.lock(),
+            weak_ptr.associated_entity.lock()};
         }
       };
     lock_all_waitables(waitables_, shared_waitables_);
@@ -335,7 +436,6 @@ protected:
           shared_ptr.reset();
         }
       };
-    reset_all(shared_subscriptions_);
     reset_all(shared_guard_conditions_);
     reset_all(shared_timers_);
     reset_all(shared_clients_);

rclcpp/include/rclcpp/wait_set_policies/static_storage.hpp

@@ -26,7 +26,6 @@
 #include "rclcpp/subscription_wait_set_mask.hpp"
 #include "rclcpp/timer.hpp"
 #include "rclcpp/visibility_control.hpp"
-#include "rclcpp/wait_set_policies/detail/entry_types.hpp"
 #include "rclcpp/wait_set_policies/detail/storage_policy_common.hpp"
 #include "rclcpp/waitable.hpp"
 
@@ -53,38 +52,65 @@ class StaticStorage : public rclcpp::wait_set_policies::detail::StoragePolicyCom
 protected:
   using is_mutable = std::false_type;
 
+  class SubscriptionEntry
+  {
+public:
+    std::shared_ptr<rclcpp::SubscriptionBase> subscription;
+    rclcpp::SubscriptionWaitSetMask mask;
+
+    /// Conversion constructor, which is intentionally not marked explicit.
+    SubscriptionEntry(
+      const std::shared_ptr<rclcpp::SubscriptionBase> & subscription_in = nullptr,
+      const rclcpp::SubscriptionWaitSetMask & mask_in = {})
+    : subscription(subscription_in),
+      mask(mask_in)
+    {}
+  };
   using ArrayOfSubscriptions = std::array<
-    detail::SubscriptionEntry,
+    SubscriptionEntry,
     NumberOfSubscriptions
   >;
   using SubscriptionsIterable = ArrayOfSubscriptions;
 
   using ArrayOfGuardConditions = std::array<
-    detail::GuardConditionEntry,
+    std::shared_ptr<rclcpp::GuardCondition>,
     NumberOfGuardCondtions
   >;
   using GuardConditionsIterable = ArrayOfGuardConditions;
 
   using ArrayOfTimers = std::array<
-    detail::TimerEntry,
+    std::shared_ptr<rclcpp::TimerBase>,
     NumberOfTimers
   >;
   using TimersIterable = ArrayOfTimers;
 
   using ArrayOfClients = std::array<
-    detail::ClientEntry,
+    std::shared_ptr<rclcpp::ClientBase>,
     NumberOfClients
   >;
   using ClientsIterable = ArrayOfClients;
 
   using ArrayOfServices = std::array<
-    detail::ServiceEntry,
+    std::shared_ptr<rclcpp::ServiceBase>,
     NumberOfServices
   >;
   using ServicesIterable = ArrayOfServices;
 
+  struct WaitableEntry
+  {
+    /// Conversion constructor, which is intentionally not marked explicit.
+    WaitableEntry(
+      const std::shared_ptr<rclcpp::Waitable> & waitable_in = nullptr,
+      const std::shared_ptr<void> & associated_entity_in = nullptr) noexcept
+    : waitable(waitable_in),
+      associated_entity(associated_entity_in)
+    {}
+
+    std::shared_ptr<rclcpp::Waitable> waitable;
+    std::shared_ptr<void> associated_entity;
+  };
   using ArrayOfWaitables = std::array<
-    detail::WaitableEntry,
+    WaitableEntry,
     NumberOfWaitables
   >;
   using WaitablesIterable = ArrayOfWaitables;

rclcpp/include/rclcpp/wait_set_template.hpp

@@ -73,8 +73,8 @@ public:
     const typename StoragePolicy::ClientsIterable & clients = {},
     const typename StoragePolicy::ServicesIterable & services = {},
     const typename StoragePolicy::WaitablesIterable & waitables = {},
-    rclcpp::Context::SharedPtr context = (
-      rclcpp::contexts::get_global_default_context()))
+    rclcpp::Context::SharedPtr context =
+    rclcpp::contexts::default_context::get_global_default_context())
   : SynchronizationPolicy(context),
     StoragePolicy(
       subscriptions,
@@ -165,18 +165,19 @@ public:
         }
         if (mask.include_intra_process_waitable) {
           auto local_subscription = inner_subscription;
-          auto waitable = inner_subscription->get_intra_process_waitable();
-          if (nullptr != waitable) {
-            bool already_in_use = local_subscription->exchange_in_use_by_wait_set_state(
-              waitable.get(),
-              true);
-            if (already_in_use) {
-              throw std::runtime_error(
-                "subscription intra-process waitable already associated with a wait set");
+          for (auto waitable : inner_subscription->get_intra_process_waitables()) {
+            if (nullptr != waitable) {
+              bool already_in_use = local_subscription->exchange_in_use_by_wait_set_state(
+                waitable.get(),
+                true);
+              if (already_in_use) {
+                throw std::runtime_error(
+                  "subscription intra-process waitable already associated with a wait set");
+              }
+              this->storage_add_waitable(
+                std::move(waitable),
+                std::move(local_subscription));
             }
-            this->storage_add_waitable(
-              std::move(inner_subscription->get_intra_process_waitable()),
-              std::move(local_subscription));
           }
         }
       });
@@ -230,11 +231,12 @@ public:
           }
         }
         if (mask.include_intra_process_waitable) {
-          auto local_waitable = inner_subscription->get_intra_process_waitable();
-          inner_subscription->exchange_in_use_by_wait_set_state(local_waitable.get(), false);
-          if (nullptr != local_waitable) {
-            // This is the case when intra process is disabled for the subscription.
-            this->storage_remove_waitable(std::move(local_waitable));
+          for (auto local_waitable : inner_subscription->get_intra_process_waitables()) {
+            inner_subscription->exchange_in_use_by_wait_set_state(local_waitable.get(), false);
+            if (nullptr != local_waitable) {
+              // This is the case when intra process is disabled for the subscription.
+              this->storage_remove_waitable(std::move(local_waitable));
+            }
           }
         }
       });
@@ -249,7 +251,7 @@ public:
    *
    * Except in the case of a fixed sized storage, where changes to the wait set
    * cannot occur after construction, in which case it holds shared ownership
-   * at all times until the wait set is destroyed, but this method also does not
+   * at all times until the wait set is destroy, but this method also does not
    * exist on a fixed sized wait set.
    *
    * This function may be thread-safe depending on the SynchronizationPolicy

rclcpp/package.xml

@@ -22,13 +22,11 @@
   <build_export_depend>rosidl_typesupport_c</build_export_depend>
   <build_export_depend>rosidl_typesupport_cpp</build_export_depend>
 
-  <depend>libstatistics_collector</depend>
   <depend>rcl</depend>
   <depend>rcl_yaml_param_parser</depend>
   <depend>rcpputils</depend>
   <depend>rcutils</depend>
   <depend>rmw</depend>
-  <depend>statistics_msgs</depend>
   <depend>tracetools</depend>
 
   <test_depend>ament_cmake_gmock</test_depend>

rclcpp/src/rclcpp/any_executable.cpp

@@ -14,7 +14,7 @@
 
 #include "rclcpp/any_executable.hpp"
 
-using rclcpp::AnyExecutable;
+using rclcpp::executor::AnyExecutable;
 
 AnyExecutable::AnyExecutable()
 : subscription(nullptr),

rclcpp/src/rclcpp/callback_group.cpp

@@ -16,8 +16,8 @@
 
 #include <vector>
 
-using rclcpp::CallbackGroup;
-using rclcpp::CallbackGroupType;
+using rclcpp::callback_group::CallbackGroup;
+using rclcpp::callback_group::CallbackGroupType;
 
 CallbackGroup::CallbackGroup(CallbackGroupType group_type)
 : type_(group_type), can_be_taken_from_(true)

rclcpp/src/rclcpp/contexts/default_context.cpp

@@ -14,13 +14,13 @@
 
 #include "rclcpp/contexts/default_context.hpp"
 
-using rclcpp::contexts::DefaultContext;
+using rclcpp::contexts::default_context::DefaultContext;
 
 DefaultContext::DefaultContext()
 {}
 
 DefaultContext::SharedPtr
-rclcpp::contexts::get_global_default_context()
+rclcpp::contexts::default_context::get_global_default_context()
 {
   static DefaultContext::SharedPtr default_context = DefaultContext::make_shared();
   return default_context;

rclcpp/src/rclcpp/executable_list.cpp

@@ -14,9 +14,9 @@
 
 #include <utility>
 
-#include "rclcpp/experimental/executable_list.hpp"
+#include "rclcpp/executable_list.hpp"
 
-using rclcpp::experimental::ExecutableList;
+using rclcpp::executor::ExecutableList;
 
 ExecutableList::ExecutableList()
 : number_of_subscriptions(0),

rclcpp/src/rclcpp/executor.cpp

@@ -29,14 +29,53 @@
 #include "rcutils/logging_macros.h"
 
 using rclcpp::exceptions::throw_from_rcl_error;
-using rclcpp::AnyExecutable;
-using rclcpp::Executor;
-using rclcpp::ExecutorOptions;
-using rclcpp::FutureReturnCode;
+using rclcpp::executor::AnyExecutable;
+using rclcpp::executor::Executor;
+using rclcpp::executor::ExecutorArgs;
+using rclcpp::executor::FutureReturnCode;
 
-Executor::Executor(const ExecutorOptions & options)
-: spinning(false), interrupt_guard_condition_(options.context)
+Executor::Executor(const ExecutorArgs & args)
+: spinning(false),
+  memory_strategy_(args.memory_strategy)
 {
+  rcl_guard_condition_options_t guard_condition_options = rcl_guard_condition_get_default_options();
+  rcl_ret_t ret = rcl_guard_condition_init(
+    &interrupt_guard_condition_, args.context->get_rcl_context().get(), guard_condition_options);
+  if (RCL_RET_OK != ret) {
+    throw_from_rcl_error(ret, "Failed to create interrupt guard condition in Executor constructor");
+  }
+
+  // The number of guard conditions is always at least 2: 1 for the ctrl-c guard cond,
+  // and one for the executor's guard cond (interrupt_guard_condition_)
+
+  // Put the global ctrl-c guard condition in
+  memory_strategy_->add_guard_condition(args.context->get_interrupt_guard_condition(&wait_set_));
+
+  // Put the executor's guard condition in
+  memory_strategy_->add_guard_condition(&interrupt_guard_condition_);
+  rcl_allocator_t allocator = memory_strategy_->get_allocator();
+
+  // Store the context for later use.
+  context_ = args.context;
+
+  ret = rcl_wait_set_init(
+    &wait_set_,
+    0, 2, 0, 0, 0, 0,
+    context_->get_rcl_context().get(),
+    allocator);
+  if (RCL_RET_OK != ret) {
+    RCUTILS_LOG_ERROR_NAMED(
+      "rclcpp",
+      "failed to create wait set: %s", rcl_get_error_string().str);
+    rcl_reset_error();
+    if (rcl_guard_condition_fini(&interrupt_guard_condition_) != RCL_RET_OK) {
+      RCUTILS_LOG_ERROR_NAMED(
+        "rclcpp",
+        "failed to destroy guard condition: %s", rcl_get_error_string().str);
+      rcl_reset_error();
+    }
+    throw std::runtime_error("Failed to create wait set in Executor constructor");
+  }
 }
 
 Executor::~Executor()
@@ -167,6 +206,12 @@ Executor::spin_node_some(rclcpp::node_interfaces::NodeBaseInterface::SharedPtr n
   this->remove_node(node, false);
 }
 
+void
+Executor::spin_node_some(std::shared_ptr<rclcpp::Node> node)
+{
+  this->spin_node_some(node->get_node_base_interface());
+}
+
 void
 Executor::spin_some(std::chrono::nanoseconds max_duration)
 {
@@ -221,6 +266,15 @@ Executor::cancel()
   }
 }
 
+void
+Executor::set_memory_strategy(rclcpp::memory_strategy::MemoryStrategy::SharedPtr memory_strategy)
+{
+  if (memory_strategy == nullptr) {
+    throw std::runtime_error("Received NULL memory strategy in executor.");
+  }
+  memory_strategy_ = memory_strategy;
+}
+
 void
 Executor::execute_any_executable(AnyExecutable & any_exec)
 {
@@ -449,7 +503,7 @@ Executor::wait_for_work(std::chrono::nanoseconds timeout)
 }
 
 rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
-Executor::get_node_by_group(rclcpp::CallbackGroup::SharedPtr group)
+Executor::get_node_by_group(rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   if (!group) {
     return nullptr;
@@ -469,7 +523,7 @@ Executor::get_node_by_group(rclcpp::CallbackGroup::SharedPtr group)
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 Executor::get_group_by_timer(rclcpp::TimerBase::SharedPtr timer)
 {
   for (auto & weak_node : weak_nodes_) {
@@ -491,7 +545,7 @@ Executor::get_group_by_timer(rclcpp::TimerBase::SharedPtr timer)
       }
     }
   }
-  return rclcpp::CallbackGroup::SharedPtr();
+  return rclcpp::callback_group::CallbackGroup::SharedPtr();
 }
 
 bool
@@ -572,3 +626,29 @@ Executor::get_next_executable(AnyExecutable & any_executable, std::chrono::nanos
   }
   return success;
 }
+
+std::ostream &
+rclcpp::executor::operator<<(std::ostream & os, const FutureReturnCode & future_return_code)
+{
+  return os << to_string(future_return_code);
+}
+
+std::string
+rclcpp::executor::to_string(const FutureReturnCode & future_return_code)
+{
+  using enum_type = std::underlying_type<FutureReturnCode>::type;
+  std::string prefix = "Unknown enum value (";
+  std::string ret_as_string = std::to_string(static_cast<enum_type>(future_return_code));
+  switch (future_return_code) {
+    case FutureReturnCode::SUCCESS:
+      prefix = "SUCCESS (";
+      break;
+    case FutureReturnCode::INTERRUPTED:
+      prefix = "INTERRUPTED (";
+      break;
+    case FutureReturnCode::TIMEOUT:
+      prefix = "TIMEOUT (";
+      break;
+  }
+  return prefix + ret_as_string + ")";
+}

rclcpp/src/rclcpp/executors/multi_threaded_executor.cpp

@@ -25,11 +25,11 @@
 using rclcpp::executors::MultiThreadedExecutor;
 
 MultiThreadedExecutor::MultiThreadedExecutor(
-  const rclcpp::ExecutorOptions & options,
+  const rclcpp::executor::ExecutorArgs & args,
   size_t number_of_threads,
   bool yield_before_execute,
   std::chrono::nanoseconds next_exec_timeout)
-: rclcpp::Executor(options),
+: executor::Executor(args),
   yield_before_execute_(yield_before_execute),
   next_exec_timeout_(next_exec_timeout)
 {
@@ -74,7 +74,7 @@ void
 MultiThreadedExecutor::run(size_t)
 {
   while (rclcpp::ok(this->context_) && spinning.load()) {
-    rclcpp::AnyExecutable any_exec;
+    executor::AnyExecutable any_exec;
     {
       std::lock_guard<std::mutex> wait_lock(wait_mutex_);
       if (!rclcpp::ok(this->context_) || !spinning.load()) {

rclcpp/src/rclcpp/executors/single_threaded_executor.cpp

@@ -18,8 +18,8 @@
 
 using rclcpp::executors::SingleThreadedExecutor;
 
-SingleThreadedExecutor::SingleThreadedExecutor(const rclcpp::ExecutorOptions & options)
-: rclcpp::Executor(options) {}
+SingleThreadedExecutor::SingleThreadedExecutor(const rclcpp::executor::ExecutorArgs & args)
+: executor::Executor(args) {}
 
 SingleThreadedExecutor::~SingleThreadedExecutor() {}
 
@@ -31,7 +31,7 @@ SingleThreadedExecutor::spin()
   }
   RCLCPP_SCOPE_EXIT(this->spinning.store(false); );
   while (rclcpp::ok(this->context_) && spinning.load()) {
-    rclcpp::AnyExecutable any_executable;
+    rclcpp::executor::AnyExecutable any_executable;
     if (get_next_executable(any_executable)) {
       execute_any_executable(any_executable);
     }

rclcpp/src/rclcpp/executors/static_executor_entities_collector.cpp

@@ -46,7 +46,7 @@ StaticExecutorEntitiesCollector::init(
   rcl_guard_condition_t * executor_guard_condition)
 {
   // Empty initialize executable list
-  exec_list_ = rclcpp::experimental::ExecutableList();
+  exec_list_ = executor::ExecutableList();
   // Get executor's wait_set_ pointer
   p_wait_set_ = p_wait_set;
   // Get executor's memory strategy ptr

rclcpp/src/rclcpp/executors/static_single_threaded_executor.cpp

@@ -19,11 +19,11 @@
 #include "rclcpp/scope_exit.hpp"
 
 using rclcpp::executors::StaticSingleThreadedExecutor;
-using rclcpp::experimental::ExecutableList;
+using rclcpp::executor::ExecutableList;
 
 StaticSingleThreadedExecutor::StaticSingleThreadedExecutor(
-  const rclcpp::ExecutorOptions & options)
-: rclcpp::Executor(options)
+  const rclcpp::executor::ExecutorArgs & args)
+: executor::Executor(args)
 {
   entities_collector_ = std::make_shared<StaticExecutorEntitiesCollector>();
 }

rclcpp/src/rclcpp/future_return_code.cpp

@@ -1,50 +0,0 @@
-// Copyright 2015 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 <iostream>
-#include <string>
-#include <type_traits>
-
-#include "rclcpp/future_return_code.hpp"
-
-namespace rclcpp
-{
-
-std::ostream &
-operator<<(std::ostream & os, const rclcpp::FutureReturnCode & future_return_code)
-{
-  return os << to_string(future_return_code);
-}
-
-std::string
-to_string(const rclcpp::FutureReturnCode & future_return_code)
-{
-  using enum_type = std::underlying_type<FutureReturnCode>::type;
-  std::string prefix = "Unknown enum value (";
-  std::string ret_as_string = std::to_string(static_cast<enum_type>(future_return_code));
-  switch (future_return_code) {
-    case FutureReturnCode::SUCCESS:
-      prefix = "SUCCESS (";
-      break;
-    case FutureReturnCode::INTERRUPTED:
-      prefix = "INTERRUPTED (";
-      break;
-    case FutureReturnCode::TIMEOUT:
-      prefix = "TIMEOUT (";
-      break;
-  }
-  return prefix + ret_as_string + ")";
-}
-
-}  // namespace rclcpp

rclcpp/src/rclcpp/intra_process_manager.cpp

@@ -32,7 +32,9 @@ IntraProcessManager::~IntraProcessManager()
 {}
 
 uint64_t
-IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
+IntraProcessManager::add_publisher(
+  rclcpp::PublisherBase::SharedPtr publisher,
+  const bool is_serialized)
 {
   std::unique_lock<std::shared_timed_mutex> lock(mutex_);
 
@@ -41,6 +43,7 @@ IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
   publishers_[id].publisher = publisher;
   publishers_[id].topic_name = publisher->get_topic_name();
   publishers_[id].qos = publisher->get_actual_qos().get_rmw_qos_profile();
+  publishers_[id].is_serialized = is_serialized;
 
   // Initialize the subscriptions storage for this publisher.
   pub_to_subs_[id] = SplittedSubscriptions();
@@ -56,7 +59,9 @@ IntraProcessManager::add_publisher(rclcpp::PublisherBase::SharedPtr publisher)
 }
 
 uint64_t
-IntraProcessManager::add_subscription(SubscriptionIntraProcessBase::SharedPtr subscription)
+IntraProcessManager::add_subscription(
+  SubscriptionIntraProcessBase::SharedPtr subscription,
+  const bool is_serialized)
 {
   std::unique_lock<std::shared_timed_mutex> lock(mutex_);
 
@@ -66,6 +71,7 @@ IntraProcessManager::add_subscription(SubscriptionIntraProcessBase::SharedPtr su
   subscriptions_[id].topic_name = subscription->get_topic_name();
   subscriptions_[id].qos = subscription->get_actual_qos();
   subscriptions_[id].use_take_shared_method = subscription->use_take_shared_method();
+  subscriptions_[id].is_serialized = is_serialized;
 
   // adds the subscription id to all the matchable publishers
   for (auto & pair : publishers_) {
@@ -220,6 +226,13 @@ IntraProcessManager::can_communicate(
     return false;
   }
 
+  // a publisher and a subscription with different content type can't communicate
+  // if is_serialized is true, the expected message typ is rcl_serialized_message_t
+  // otherwise the templated ROS2 message type
+  if (sub_info.is_serialized != pub_info.is_serialized) {
+    return false;
+  }
+
   return true;
 }
 

rclcpp/src/rclcpp/memory_strategy.cpp

@@ -127,7 +127,7 @@ MemoryStrategy::get_timer_by_handle(
 
 rclcpp::node_interfaces::NodeBaseInterface::SharedPtr
 MemoryStrategy::get_node_by_group(
-  rclcpp::CallbackGroup::SharedPtr group,
+  rclcpp::callback_group::CallbackGroup::SharedPtr group,
   const WeakNodeList & weak_nodes)
 {
   if (!group) {
@@ -148,7 +148,7 @@ MemoryStrategy::get_node_by_group(
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 MemoryStrategy::get_group_by_subscription(
   rclcpp::SubscriptionBase::SharedPtr subscription,
   const WeakNodeList & weak_nodes)
@@ -175,7 +175,7 @@ MemoryStrategy::get_group_by_subscription(
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 MemoryStrategy::get_group_by_service(
   rclcpp::ServiceBase::SharedPtr service,
   const WeakNodeList & weak_nodes)
@@ -202,7 +202,7 @@ MemoryStrategy::get_group_by_service(
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 MemoryStrategy::get_group_by_client(
   rclcpp::ClientBase::SharedPtr client,
   const WeakNodeList & weak_nodes)
@@ -229,7 +229,7 @@ MemoryStrategy::get_group_by_client(
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 MemoryStrategy::get_group_by_timer(
   rclcpp::TimerBase::SharedPtr timer,
   const WeakNodeList & weak_nodes)
@@ -256,7 +256,7 @@ MemoryStrategy::get_group_by_timer(
   return nullptr;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 MemoryStrategy::get_group_by_waitable(
   rclcpp::Waitable::SharedPtr waitable,
   const WeakNodeList & weak_nodes)

rclcpp/src/rclcpp/node.cpp

@@ -103,8 +103,7 @@ Node::Node(
       namespace_,
       options.context(),
       *(options.get_rcl_node_options()),
-      options.use_intra_process_comms(),
-      options.enable_topic_statistics())),
+      options.use_intra_process_comms())),
   node_graph_(new rclcpp::node_interfaces::NodeGraph(node_base_.get())),
   node_logging_(new rclcpp::node_interfaces::NodeLogging(node_base_.get())),
   node_timers_(new rclcpp::node_interfaces::NodeTimers(node_base_.get())),
@@ -210,14 +209,15 @@ Node::get_logger() const
   return node_logging_->get_logger();
 }
 
-rclcpp::CallbackGroup::SharedPtr
-Node::create_callback_group(rclcpp::CallbackGroupType group_type)
+rclcpp::callback_group::CallbackGroup::SharedPtr
+Node::create_callback_group(
+  rclcpp::callback_group::CallbackGroupType group_type)
 {
   return node_base_->create_callback_group(group_type);
 }
 
 bool
-Node::group_in_node(rclcpp::CallbackGroup::SharedPtr group)
+Node::group_in_node(rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   return node_base_->callback_group_in_node(group);
 }
@@ -376,7 +376,7 @@ Node::get_subscriptions_info_by_topic(const std::string & topic_name, bool no_ma
   return node_graph_->get_subscriptions_info_by_topic(topic_name, no_mangle);
 }
 
-const std::vector<rclcpp::CallbackGroup::WeakPtr> &
+const std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> &
 Node::get_callback_groups() const
 {
   return node_base_->get_callback_groups();

rclcpp/src/rclcpp/node_interfaces/node_base.cpp

@@ -34,11 +34,9 @@ NodeBase::NodeBase(
   const std::string & namespace_,
   rclcpp::Context::SharedPtr context,
   const rcl_node_options_t & rcl_node_options,
-  bool use_intra_process_default,
-  bool enable_topic_statistics_default)
+  bool use_intra_process_default)
 : context_(context),
   use_intra_process_default_(use_intra_process_default),
-  enable_topic_statistics_default_(enable_topic_statistics_default),
   node_handle_(nullptr),
   default_callback_group_(nullptr),
   associated_with_executor_(false),
@@ -133,7 +131,7 @@ NodeBase::NodeBase(
     });
 
   // Create the default callback group.
-  using rclcpp::CallbackGroupType;
+  using rclcpp::callback_group::CallbackGroupType;
   default_callback_group_ = create_callback_group(CallbackGroupType::MutuallyExclusive);
 
   // Indicate the notify_guard_condition is now valid.
@@ -208,24 +206,24 @@ NodeBase::assert_liveliness() const
   return RCL_RET_OK == rcl_node_assert_liveliness(get_rcl_node_handle());
 }
 
-rclcpp::CallbackGroup::SharedPtr
-NodeBase::create_callback_group(rclcpp::CallbackGroupType group_type)
+rclcpp::callback_group::CallbackGroup::SharedPtr
+NodeBase::create_callback_group(rclcpp::callback_group::CallbackGroupType group_type)
 {
-  using rclcpp::CallbackGroup;
-  using rclcpp::CallbackGroupType;
+  using rclcpp::callback_group::CallbackGroup;
+  using rclcpp::callback_group::CallbackGroupType;
   auto group = CallbackGroup::SharedPtr(new CallbackGroup(group_type));
   callback_groups_.push_back(group);
   return group;
 }
 
-rclcpp::CallbackGroup::SharedPtr
+rclcpp::callback_group::CallbackGroup::SharedPtr
 NodeBase::get_default_callback_group()
 {
   return default_callback_group_;
 }
 
 bool
-NodeBase::callback_group_in_node(rclcpp::CallbackGroup::SharedPtr group)
+NodeBase::callback_group_in_node(rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   bool group_belongs_to_this_node = false;
   for (auto & weak_group : this->callback_groups_) {
@@ -237,7 +235,7 @@ NodeBase::callback_group_in_node(rclcpp::CallbackGroup::SharedPtr group)
   return group_belongs_to_this_node;
 }
 
-const std::vector<rclcpp::CallbackGroup::WeakPtr> &
+const std::vector<rclcpp::callback_group::CallbackGroup::WeakPtr> &
 NodeBase::get_callback_groups() const
 {
   return callback_groups_;
@@ -270,9 +268,3 @@ NodeBase::get_use_intra_process_default() const
 {
   return use_intra_process_default_;
 }
-
-bool
-NodeBase::get_enable_topic_statistics_default() const
-{
-  return enable_topic_statistics_default_;
-}

rclcpp/src/rclcpp/node_interfaces/node_services.cpp

@@ -28,7 +28,7 @@ NodeServices::~NodeServices()
 void
 NodeServices::add_service(
   rclcpp::ServiceBase::SharedPtr service_base_ptr,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {
@@ -55,7 +55,7 @@ NodeServices::add_service(
 void
 NodeServices::add_client(
   rclcpp::ClientBase::SharedPtr client_base_ptr,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {

rclcpp/src/rclcpp/node_interfaces/node_timers.cpp

@@ -28,7 +28,7 @@ NodeTimers::~NodeTimers()
 void
 NodeTimers::add_timer(
   rclcpp::TimerBase::SharedPtr timer,
-  rclcpp::CallbackGroup::SharedPtr callback_group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr callback_group)
 {
   if (callback_group) {
     if (!node_base_->callback_group_in_node(callback_group)) {

rclcpp/src/rclcpp/node_interfaces/node_topics.cpp

@@ -42,7 +42,7 @@ NodeTopics::create_publisher(
 void
 NodeTopics::add_publisher(
   rclcpp::PublisherBase::SharedPtr publisher,
-  rclcpp::CallbackGroup::SharedPtr callback_group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr callback_group)
 {
   // Assign to a group.
   if (callback_group) {
@@ -81,7 +81,7 @@ NodeTopics::create_subscription(
 void
 NodeTopics::add_subscription(
   rclcpp::SubscriptionBase::SharedPtr subscription,
-  rclcpp::CallbackGroup::SharedPtr callback_group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr callback_group)
 {
   // Assign to a group.
   if (callback_group) {
@@ -99,10 +99,12 @@ NodeTopics::add_subscription(
     callback_group->add_waitable(subscription_event);
   }
 
-  auto intra_process_waitable = subscription->get_intra_process_waitable();
-  if (nullptr != intra_process_waitable) {
-    // Add to the callback group to be notified about intra-process msgs.
-    callback_group->add_waitable(intra_process_waitable);
+  const auto intra_process_waitables = subscription->get_intra_process_waitables();
+  for (auto & intra_process_waitable : intra_process_waitables) {
+    if (nullptr != intra_process_waitable) {
+      // Add to the callback group to be notified about intra-process msgs.
+      callback_group->add_waitable(intra_process_waitable);
+    }
   }
 
   // Notify the executor that a new subscription was created using the parent Node.

rclcpp/src/rclcpp/node_interfaces/node_waitables.cpp

@@ -28,7 +28,7 @@ NodeWaitables::~NodeWaitables()
 void
 NodeWaitables::add_waitable(
   rclcpp::Waitable::SharedPtr waitable_ptr,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {
@@ -55,7 +55,7 @@ NodeWaitables::add_waitable(
 void
 NodeWaitables::remove_waitable(
   rclcpp::Waitable::SharedPtr waitable_ptr,
-  rclcpp::CallbackGroup::SharedPtr group) noexcept
+  rclcpp::callback_group::CallbackGroup::SharedPtr group) noexcept
 {
   if (group) {
     if (!node_base_->callback_group_in_node(group)) {

rclcpp/src/rclcpp/node_options.cpp

@@ -73,7 +73,6 @@ NodeOptions::operator=(const NodeOptions & other)
     this->use_global_arguments_ = other.use_global_arguments_;
     this->enable_rosout_ = other.enable_rosout_;
     this->use_intra_process_comms_ = other.use_intra_process_comms_;
-    this->enable_topic_statistics_ = other.enable_topic_statistics_;
     this->start_parameter_services_ = other.start_parameter_services_;
     this->allocator_ = other.allocator_;
     this->allow_undeclared_parameters_ = other.allow_undeclared_parameters_;
@@ -214,19 +213,6 @@ NodeOptions::use_intra_process_comms(bool use_intra_process_comms)
   return *this;
 }
 
-bool
-NodeOptions::enable_topic_statistics() const
-{
-  return this->enable_topic_statistics_;
-}
-
-NodeOptions &
-NodeOptions::enable_topic_statistics(bool enable_topic_statistics)
-{
-  this->enable_topic_statistics_ = enable_topic_statistics;
-  return *this;
-}
-
 bool
 NodeOptions::start_parameter_services() const
 {

rclcpp/src/rclcpp/parameter_client.cpp

@@ -31,7 +31,7 @@ AsyncParametersClient::AsyncParametersClient(
   const rclcpp::node_interfaces::NodeServicesInterface::SharedPtr node_services_interface,
   const std::string & remote_node_name,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 : node_topics_interface_(node_topics_interface)
 {
   if (remote_node_name != "") {
@@ -103,7 +103,7 @@ AsyncParametersClient::AsyncParametersClient(
   const rclcpp::Node::SharedPtr node,
   const std::string & remote_node_name,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 : AsyncParametersClient(
     node->get_node_base_interface(),
     node->get_node_topics_interface(),
@@ -118,7 +118,7 @@ AsyncParametersClient::AsyncParametersClient(
   rclcpp::Node * node,
   const std::string & remote_node_name,
   const rmw_qos_profile_t & qos_profile,
-  rclcpp::CallbackGroup::SharedPtr group)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group)
 : AsyncParametersClient(
     node->get_node_base_interface(),
     node->get_node_topics_interface(),
@@ -349,7 +349,7 @@ SyncParametersClient::SyncParametersClient(
 {}
 
 SyncParametersClient::SyncParametersClient(
-  rclcpp::Executor::SharedPtr executor,
+  rclcpp::executor::Executor::SharedPtr executor,
   rclcpp::Node::SharedPtr node,
   const std::string & remote_node_name,
   const rmw_qos_profile_t & qos_profile)
@@ -375,7 +375,7 @@ SyncParametersClient::SyncParametersClient(
 {}
 
 SyncParametersClient::SyncParametersClient(
-  rclcpp::Executor::SharedPtr executor,
+  rclcpp::executor::Executor::SharedPtr executor,
   rclcpp::Node * node,
   const std::string & remote_node_name,
   const rmw_qos_profile_t & qos_profile)
@@ -390,7 +390,7 @@ SyncParametersClient::SyncParametersClient(
 {}
 
 SyncParametersClient::SyncParametersClient(
-  rclcpp::Executor::SharedPtr executor,
+  rclcpp::executor::Executor::SharedPtr executor,
   const rclcpp::node_interfaces::NodeBaseInterface::SharedPtr node_base_interface,
   const rclcpp::node_interfaces::NodeTopicsInterface::SharedPtr node_topics_interface,
   const rclcpp::node_interfaces::NodeGraphInterface::SharedPtr node_graph_interface,
@@ -417,7 +417,7 @@ SyncParametersClient::get_parameters(const std::vector<std::string> & parameter_
   if (
     spin_node_until_future_complete(
       *executor_, node_base_interface_,
-      f) == rclcpp::FutureReturnCode::SUCCESS)
+      f) == rclcpp::executor::FutureReturnCode::SUCCESS)
   {
     return f.get();
   }
@@ -443,7 +443,7 @@ SyncParametersClient::get_parameter_types(const std::vector<std::string> & param
   if (
     spin_node_until_future_complete(
       *executor_, node_base_interface_,
-      f) == rclcpp::FutureReturnCode::SUCCESS)
+      f) == rclcpp::executor::FutureReturnCode::SUCCESS)
   {
     return f.get();
   }
@@ -460,7 +460,7 @@ SyncParametersClient::set_parameters(
   if (
     spin_node_until_future_complete(
       *executor_, node_base_interface_,
-      f) == rclcpp::FutureReturnCode::SUCCESS)
+      f) == rclcpp::executor::FutureReturnCode::SUCCESS)
   {
     return f.get();
   }
@@ -477,7 +477,7 @@ SyncParametersClient::set_parameters_atomically(
   if (
     spin_node_until_future_complete(
       *executor_, node_base_interface_,
-      f) == rclcpp::FutureReturnCode::SUCCESS)
+      f) == rclcpp::executor::FutureReturnCode::SUCCESS)
   {
     return f.get();
   }
@@ -496,7 +496,7 @@ SyncParametersClient::list_parameters(
   if (
     spin_node_until_future_complete(
       *executor_, node_base_interface_,
-      f) == rclcpp::FutureReturnCode::SUCCESS)
+      f) == rclcpp::executor::FutureReturnCode::SUCCESS)
   {
     return f.get();
   }

rclcpp/src/rclcpp/publisher_base.cpp

@@ -44,7 +44,8 @@ PublisherBase::PublisherBase(
   const rosidl_message_type_support_t & type_support,
   const rcl_publisher_options_t & publisher_options)
 : rcl_node_handle_(node_base->get_shared_rcl_node_handle()),
-  intra_process_is_enabled_(false), intra_process_publisher_id_(0)
+  intra_process_is_enabled_(false), intra_process_publisher_id_(0),
+  intra_process_publisher_id_serialized_(0)
 {
   rcl_ret_t ret = rcl_publisher_init(
     &publisher_handle_,
@@ -105,6 +106,7 @@ PublisherBase::~PublisherBase()
     return;
   }
   ipm->remove_publisher(intra_process_publisher_id_);
+  ipm->remove_publisher(intra_process_publisher_id_serialized_);
 }
 
 const char *
@@ -238,9 +240,11 @@ PublisherBase::operator==(const rmw_gid_t * gid) const
 void
 PublisherBase::setup_intra_process(
   uint64_t intra_process_publisher_id,
+  uint64_t intra_process_publisher_id_serialized,
   IntraProcessManagerSharedPtr ipm)
 {
   intra_process_publisher_id_ = intra_process_publisher_id;
+  intra_process_publisher_id_serialized_ = intra_process_publisher_id_serialized;
   weak_ipm_ = ipm;
   intra_process_is_enabled_ = true;
 }

rclcpp/src/rclcpp/serialization.cpp

@@ -1,69 +0,0 @@
-// Copyright 2020 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/serialization.hpp"
-
-#include <memory>
-#include <string>
-
-#include "rclcpp/exceptions.hpp"
-#include "rclcpp/serialized_message.hpp"
-
-#include "rcpputils/asserts.hpp"
-
-#include "rmw/rmw.h"
-
-namespace rclcpp
-{
-
-SerializationBase::SerializationBase(const rosidl_message_type_support_t * type_support)
-: type_support_(type_support)
-{
-  rcpputils::check_true(nullptr != type_support, "Typesupport is nullpointer.");
-}
-
-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.");
-
-  const auto ret = rmw_serialize(
-    ros_message,
-    type_support_,
-    &serialized_message->get_rcl_serialized_message());
-  if (ret != RMW_RET_OK) {
-    rclcpp::exceptions::throw_from_rcl_error(ret, "Failed to serialize ROS 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(
-    0 != serialized_message->capacity() && 0 != serialized_message->size(),
-    "Serialized message is wrongly initialized.");
-  rcpputils::check_true(nullptr != ros_message, "ROS message is a nullpointer.");
-
-  const auto ret = rmw_deserialize(
-    &serialized_message->get_rcl_serialized_message(), type_support_, ros_message);
-  if (ret != RMW_RET_OK) {
-    rclcpp::exceptions::throw_from_rcl_error(ret, "Failed to deserialize ROS message.");
-  }
-}
-
-}  // namespace rclcpp

rclcpp/src/rclcpp/serialized_message.cpp

@@ -1,147 +0,0 @@
-// Copyright 2020 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/serialized_message.hpp"
-
-#include <cstring>
-
-#include "rclcpp/exceptions.hpp"
-#include "rclcpp/logging.hpp"
-
-#include "rmw/types.h"
-
-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);
-  if (RCL_RET_OK != ret) {
-    rclcpp::exceptions::throw_from_rcl_error(ret);
-  }
-
-  // do not call memcpy if the pointer is "static"
-  if (to.buffer != from.buffer) {
-    std::memcpy(to.buffer, from.buffer, from.buffer_length);
-  }
-  to.buffer_length = from.buffer_length;
-}
-
-/// Object oriented version of rcl_serialized_message_t with destructor to avoid memory leaks
-SerializedMessage::SerializedMessage(const rcl_allocator_t & allocator)
-: SerializedMessage(0u, allocator)
-{}
-
-SerializedMessage::SerializedMessage(
-  size_t initial_capacity, const rcl_allocator_t & allocator)
-: serialized_message_(rmw_get_zero_initialized_serialized_message())
-{
-  const auto ret = rmw_serialized_message_init(
-    &serialized_message_, initial_capacity, &allocator);
-  if (RCL_RET_OK != ret) {
-    rclcpp::exceptions::throw_from_rcl_error(ret);
-  }
-}
-
-SerializedMessage::SerializedMessage(const SerializedMessage & other)
-: SerializedMessage(other.serialized_message_)
-{}
-
-SerializedMessage::SerializedMessage(const rcl_serialized_message_t & other)
-: serialized_message_(rmw_get_zero_initialized_serialized_message())
-{
-  copy_rcl_message(other, serialized_message_);
-}
-
-SerializedMessage::SerializedMessage(SerializedMessage && other)
-: SerializedMessage(other.serialized_message_)
-{
-  other.serialized_message_ = rmw_get_zero_initialized_serialized_message();
-}
-
-SerializedMessage::SerializedMessage(rcl_serialized_message_t && other)
-: serialized_message_(other)
-{
-  // reset buffer to prevent double free
-  other = rmw_get_zero_initialized_serialized_message();
-}
-
-SerializedMessage & SerializedMessage::operator=(const SerializedMessage & other)
-{
-  serialized_message_ = rmw_get_zero_initialized_serialized_message();
-  copy_rcl_message(other.serialized_message_, serialized_message_);
-
-  return *this;
-}
-
-SerializedMessage & SerializedMessage::operator=(const rcl_serialized_message_t & other)
-{
-  serialized_message_ = rmw_get_zero_initialized_serialized_message();
-  copy_rcl_message(other, serialized_message_);
-
-  return *this;
-}
-
-SerializedMessage & SerializedMessage::operator=(SerializedMessage && other)
-{
-  *this = other.serialized_message_;
-  other.serialized_message_ = rmw_get_zero_initialized_serialized_message();
-
-  return *this;
-}
-
-SerializedMessage & SerializedMessage::operator=(rcl_serialized_message_t && other)
-{
-  serialized_message_ = rmw_get_zero_initialized_serialized_message();
-  serialized_message_ = other;
-
-  // reset original to prevent double free
-  other = rmw_get_zero_initialized_serialized_message();
-
-  return *this;
-}
-
-SerializedMessage::~SerializedMessage()
-{
-  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);
-    }
-  }
-}
-
-rcl_serialized_message_t & SerializedMessage::get_rcl_serialized_message()
-{
-  return serialized_message_;
-}
-
-const rcl_serialized_message_t & SerializedMessage::get_rcl_serialized_message() const
-{
-  return serialized_message_;
-}
-
-size_t SerializedMessage::size() const
-{
-  return serialized_message_.buffer_length;
-}
-
-size_t SerializedMessage::capacity() const
-{
-  return serialized_message_.buffer_capacity;
-}
-}  // namespace rclcpp

rclcpp/src/rclcpp/subscription_base.cpp

@@ -39,7 +39,6 @@ SubscriptionBase::SubscriptionBase(
 : node_base_(node_base),
   node_handle_(node_base_->get_shared_rcl_node_handle()),
   use_intra_process_(false),
-  intra_process_subscription_id_(0),
   type_support_(type_support_handle),
   is_serialized_(is_serialized)
 {
@@ -93,7 +92,9 @@ SubscriptionBase::~SubscriptionBase()
       "Intra process manager died before than a subscription.");
     return;
   }
-  ipm->remove_subscription(intra_process_subscription_id_);
+  for (auto intra_process_subscription_id : intra_process_subscription_ids_) {
+    ipm->remove_subscription(intra_process_subscription_id);
+  }
 }
 
 const char *
@@ -204,10 +205,10 @@ SubscriptionBase::get_publisher_count() const
 
 void
 SubscriptionBase::setup_intra_process(
-  uint64_t intra_process_subscription_id,
+  const std::vector<uint64_t> & intra_process_subscription_ids,
   IntraProcessManagerWeakPtr weak_ipm)
 {
-  intra_process_subscription_id_ = intra_process_subscription_id;
+  intra_process_subscription_ids_ = intra_process_subscription_ids;
   weak_ipm_ = weak_ipm;
   use_intra_process_ = true;
 }
@@ -218,23 +219,29 @@ SubscriptionBase::can_loan_messages() const
   return rcl_subscription_can_loan_messages(subscription_handle_.get());
 }
 
-rclcpp::Waitable::SharedPtr
-SubscriptionBase::get_intra_process_waitable() const
+std::vector<rclcpp::Waitable::SharedPtr>
+SubscriptionBase::get_intra_process_waitables() const
 {
   // If not using intra process, shortcut to nullptr.
   if (!use_intra_process_) {
-    return nullptr;
+    return std::vector<rclcpp::Waitable::SharedPtr>();
   }
   // Get the intra process manager.
   auto ipm = weak_ipm_.lock();
   if (!ipm) {
     throw std::runtime_error(
-            "SubscriptionBase::get_intra_process_waitable() called "
+            "SubscriptionBase::get_intra_process_waitables() called "
             "after destruction of intra process manager");
   }
 
-  // Use the id to retrieve the subscription intra-process from the intra-process manager.
-  return ipm->get_subscription_intra_process(intra_process_subscription_id_);
+  std::vector<rclcpp::Waitable::SharedPtr> waitables(intra_process_subscription_ids_.size());
+
+  for (auto i = 0u; i < intra_process_subscription_ids_.size(); ++i) {
+    // Use the id to retrieve the subscription intra-process from the intra-process manager.
+    waitables[i] = ipm->get_subscription_intra_process(intra_process_subscription_ids_[i]);
+  }
+
+  return waitables;
 }
 
 void
@@ -274,8 +281,10 @@ SubscriptionBase::exchange_in_use_by_wait_set_state(
   if (this == pointer_to_subscription_part) {
     return subscription_in_use_by_wait_set_.exchange(in_use_state);
   }
-  if (get_intra_process_waitable().get() == pointer_to_subscription_part) {
-    return intra_process_subscription_waitable_in_use_by_wait_set_.exchange(in_use_state);
+  for (auto & waitable : get_intra_process_waitables()) {
+    if (waitable.get() == pointer_to_subscription_part) {
+      return intra_process_subscription_waitable_in_use_by_wait_set_.exchange(in_use_state);
+    }
   }
   for (const auto & qos_event : event_handlers_) {
     if (qos_event.get() == pointer_to_subscription_part) {

rclcpp/src/rclcpp/timer.cpp

@@ -33,7 +33,7 @@ TimerBase::TimerBase(
 : clock_(clock), timer_handle_(nullptr)
 {
   if (nullptr == context) {
-    context = rclcpp::contexts::get_global_default_context();
+    context = rclcpp::contexts::default_context::get_global_default_context();
   }
 
   auto rcl_context = context->get_rcl_context();

rclcpp/src/rclcpp/utilities.cpp

@@ -30,7 +30,7 @@ namespace rclcpp
 void
 init(int argc, char const * const argv[], const InitOptions & init_options)
 {
-  using rclcpp::contexts::get_global_default_context;
+  using contexts::default_context::get_global_default_context;
   get_global_default_context()->init(argc, argv, init_options);
   // Install the signal handlers.
   install_signal_handlers();
@@ -125,7 +125,7 @@ remove_ros_arguments(int argc, char const * const argv[])
 bool
 ok(Context::SharedPtr context)
 {
-  using rclcpp::contexts::get_global_default_context;
+  using contexts::default_context::get_global_default_context;
   if (nullptr == context) {
     context = get_global_default_context();
   }
@@ -141,7 +141,7 @@ is_initialized(Context::SharedPtr context)
 bool
 shutdown(Context::SharedPtr context, const std::string & reason)
 {
-  using rclcpp::contexts::get_global_default_context;
+  using contexts::default_context::get_global_default_context;
   auto default_context = get_global_default_context();
   if (nullptr == context) {
     context = default_context;
@@ -156,7 +156,7 @@ shutdown(Context::SharedPtr context, const std::string & reason)
 void
 on_shutdown(std::function<void()> callback, Context::SharedPtr context)
 {
-  using rclcpp::contexts::get_global_default_context;
+  using contexts::default_context::get_global_default_context;
   if (nullptr == context) {
     context = get_global_default_context();
   }
@@ -166,7 +166,7 @@ on_shutdown(std::function<void()> callback, Context::SharedPtr context)
 bool
 sleep_for(const std::chrono::nanoseconds & nanoseconds, Context::SharedPtr context)
 {
-  using rclcpp::contexts::get_global_default_context;
+  using contexts::default_context::get_global_default_context;
   if (nullptr == context) {
     context = get_global_default_context();
   }

rclcpp/test/executors/test_multi_threaded_executor.cpp

@@ -51,14 +51,14 @@ TEST_F(TestMultiThreadedExecutor, timer_over_take) {
   bool yield_before_execute = true;
 
   rclcpp::executors::MultiThreadedExecutor executor(
-    rclcpp::ExecutorOptions(), 2u, yield_before_execute);
+    rclcpp::executor::create_default_executor_arguments(), 2u, yield_before_execute);
 
   ASSERT_GT(executor.get_number_of_threads(), 1u);
 
   std::shared_ptr<rclcpp::Node> node =
     std::make_shared<rclcpp::Node>("test_multi_threaded_executor_timer_over_take");
 
-  auto cbg = node->create_callback_group(rclcpp::CallbackGroupType::Reentrant);
+  auto cbg = node->create_callback_group(rclcpp::callback_group::CallbackGroupType::Reentrant);
 
   rclcpp::Clock system_clock(RCL_STEADY_TIME);
   std::mutex last_mutex;

rclcpp/test/test_intra_process_communication.cpp

@@ -0,0 +1,367 @@
+// Copyright 2020 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 <test_msgs/message_fixtures.hpp>
+
+#include <iostream>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "rclcpp/exceptions.hpp"
+#include "rclcpp/publisher.hpp"
+#include "rclcpp/rclcpp.hpp"
+
+#include "rclcpp/serialized_message.hpp"
+
+int32_t & get_test_allocation_counter()
+{
+  static int32_t counter = 0;
+  return counter;
+}
+
+void * custom_allocate(size_t size, void * state)
+{
+  static auto m_allocator = rcutils_get_default_allocator();
+
+  ++get_test_allocation_counter();
+  auto r = m_allocator.allocate(size, state);
+  return r;
+}
+
+void * custom_zero_allocate(size_t number_of_elements, size_t size_of_element, void * state)
+{
+  static auto m_allocator = rcutils_get_default_allocator();
+
+  ++get_test_allocation_counter();
+  auto r = m_allocator.zero_allocate(number_of_elements, size_of_element, state);
+  return r;
+}
+
+void * custom_reallocate(void * pointer, size_t size, void * state)
+{
+  static auto m_allocator = rcutils_get_default_allocator();
+
+  if (pointer == nullptr) {
+    ++get_test_allocation_counter();
+  }
+
+  auto r = m_allocator.reallocate(pointer, size, state);
+  return r;
+}
+
+void custom_deallocate(void * pointer, void * state)
+{
+  static auto m_allocator = rcutils_get_default_allocator();
+
+  --get_test_allocation_counter();
+  m_allocator.deallocate(pointer, state);
+}
+
+rcl_serialized_message_t make_serialized_string_msg(
+  const std::shared_ptr<test_msgs::msg::Strings> & stringMsg)
+{
+  auto m_allocator = rcutils_get_default_allocator();
+
+  // add custom (de)allocator to count the references to the object
+  m_allocator.allocate = &custom_allocate;
+  m_allocator.deallocate = &custom_deallocate;
+  m_allocator.reallocate = &custom_reallocate;
+  m_allocator.zero_allocate = &custom_zero_allocate;
+
+  rcl_serialized_message_t msg = rmw_get_zero_initialized_serialized_message();
+  auto ret = rmw_serialized_message_init(&msg, 0, &m_allocator);
+  if (ret != RCL_RET_OK) {
+    rclcpp::exceptions::throw_from_rcl_error(ret);
+  }
+
+  static auto type =
+    rosidl_typesupport_cpp::get_message_type_support_handle
+    <test_msgs::msg::Strings>();
+  auto error = rmw_serialize(stringMsg.get(), type, &msg);
+  if (error != RMW_RET_OK) {
+    RCUTILS_LOG_ERROR_NAMED(
+      "test_intra_process_communication",
+      "Something went wrong preparing the serialized message");
+  }
+
+  return msg;
+}
+
+/**
+ * Parameterized test.
+ * The first param are the NodeOptions used to create the nodes.
+ * The second param are the expected intraprocess count results.
+ */
+struct TestParameters
+{
+  rclcpp::NodeOptions node_options[2];
+  uint64_t intraprocess_count_results[2];
+  size_t runs;
+  std::string description;
+};
+
+std::ostream & operator<<(std::ostream & out, const TestParameters & params)
+{
+  out << params.description;
+  return out;
+}
+
+class TestPublisherSubscriptionSerialized : public ::testing::TestWithParam<TestParameters>
+{
+public:
+  static void SetUpTestCase()
+  {
+    if (!rclcpp::ok()) {
+      rclcpp::init(0, nullptr);
+    }
+  }
+
+protected:
+  static std::chrono::milliseconds offset;
+};
+
+std::chrono::milliseconds TestPublisherSubscriptionSerialized::offset = std::chrono::milliseconds(
+  250);
+std::array<uint32_t, 2> counts;
+
+void OnMessageSerialized(const std::shared_ptr<const rcl_serialized_message_t> msg)
+{
+  EXPECT_NE(msg->buffer, nullptr);
+  EXPECT_GT(msg->buffer_capacity, 0u);
+
+  ++counts[0];
+}
+
+void OnMessageConst(std::shared_ptr<const test_msgs::msg::Strings> msg)
+{
+  EXPECT_EQ(msg->string_value.back(), '9');
+
+  ++counts[1];
+}
+
+void OnMessageUniquePtr(std::unique_ptr<test_msgs::msg::Strings> msg)
+{
+  EXPECT_EQ(msg->string_value.back(), '9');
+
+  ++counts[1];
+}
+
+void OnMessage(std::shared_ptr<test_msgs::msg::Strings> msg)
+{
+  EXPECT_EQ(msg->string_value.back(), '9');
+
+  ++counts[1];
+}
+
+TEST_P(TestPublisherSubscriptionSerialized, publish_serialized)
+{
+  get_test_allocation_counter() = 0;
+
+  TestParameters parameters = GetParam();
+  {
+    rclcpp::Node::SharedPtr node = std::make_shared<rclcpp::Node>(
+      "my_node",
+      "/ns",
+      parameters.node_options[0]);
+    auto publisher = node->create_publisher<test_msgs::msg::Strings>("/topic", 10);
+
+    auto sub_shared = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessage);
+    auto sub_unique = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageUniquePtr);
+    auto sub_const_shared = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageConst);
+    auto sub_serialized = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageSerialized);
+
+    rclcpp::sleep_for(offset);
+
+    counts.fill(0);
+    auto stringMsg = get_messages_strings()[3];
+
+    for (size_t i = 0; i < parameters.runs; i++) {
+      auto msg0 = make_serialized_string_msg(stringMsg);
+
+      std::unique_ptr<test_msgs::msg::Strings> stringMsgU(
+        new test_msgs::msg::Strings(
+          *stringMsg));
+
+      publisher->publish(std::make_unique<rcl_serialized_message_t>(msg0));
+      publisher->publish(*stringMsg);
+      publisher->publish(std::move(stringMsgU));
+    }
+    for (uint32_t i = 0; i < 3; ++i) {
+      rclcpp::spin_some(node);
+      rclcpp::sleep_for(offset);
+    }
+
+    rclcpp::spin_some(node);
+  }
+
+  if (parameters.runs == 1) {
+    EXPECT_EQ(counts[0], 3u);
+    EXPECT_EQ(counts[1], 9u);
+  }
+
+  EXPECT_EQ(get_test_allocation_counter(), 0);
+}
+
+TEST_P(TestPublisherSubscriptionSerialized, publish_serialized_generic)
+{
+  get_test_allocation_counter() = 0;
+
+  TestParameters parameters = GetParam();
+  {
+    rclcpp::Node::SharedPtr node = std::make_shared<rclcpp::Node>(
+      "my_node",
+      "/ns",
+      parameters.node_options[0]);
+    auto publisher = rclcpp::create_publisher<rcl_serialized_message_t>(
+      node,
+      "/topic",
+      *rosidl_typesupport_cpp::get_message_type_support_handle<test_msgs::msg::Strings>(),
+      rclcpp::QoS(10));
+
+    auto sub_gen_serialized = rclcpp::create_subscription<rcl_serialized_message_t>(
+      node,
+      "/topic",
+      *rosidl_typesupport_cpp::get_message_type_support_handle<test_msgs::msg::Strings>(),
+      rclcpp::QoS(10),
+      &OnMessageSerialized);
+
+    auto sub_shared = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessage);
+    auto sub_unique = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageUniquePtr);
+    auto sub_const_shared = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageConst);
+    auto sub_serialized = node->create_subscription<test_msgs::msg::Strings>(
+      "/topic", 10,
+      &OnMessageSerialized);
+
+    rclcpp::sleep_for(offset);
+
+    counts.fill(0);
+    auto stringMsg = get_messages_strings()[3];
+
+    for (size_t i = 0; i < parameters.runs; i++) {
+      auto msg0 = make_serialized_string_msg(stringMsg);
+
+      publisher->publish(std::make_unique<rcl_serialized_message_t>(msg0));
+    }
+    rclcpp::spin_some(node);
+    rclcpp::sleep_for(offset);
+
+    rclcpp::spin_some(node);
+  }
+
+  if (parameters.runs == 1) {
+    EXPECT_EQ(counts[0], 2u);
+    EXPECT_EQ(counts[1], 3u);
+  }
+
+  EXPECT_EQ(get_test_allocation_counter(), 0);
+}
+
+auto get_new_context()
+{
+  auto context = rclcpp::Context::make_shared();
+  context->init(0, nullptr);
+  return context;
+}
+
+std::vector<TestParameters> parameters = {
+  /*
+     Testing publisher subscription count api and internal process subscription count.
+     Two subscriptions in the same topic, both using intraprocess comm.
+   */
+  {
+    {
+      rclcpp::NodeOptions().use_intra_process_comms(true),
+      rclcpp::NodeOptions().use_intra_process_comms(true)
+    },
+    {1u, 2u},
+    1,
+    "two_subscriptions_intraprocess_comm"
+  },
+  /*
+     Testing publisher subscription count api and internal process subscription count.
+     Two subscriptions, one using intra-process comm and the other not using it.
+   */
+  {
+    {
+      rclcpp::NodeOptions().use_intra_process_comms(true),
+      rclcpp::NodeOptions().use_intra_process_comms(false)
+    },
+    {1u, 1u},
+    1,
+    "two_subscriptions_one_intraprocess_one_not"
+  },
+  /*
+     Testing publisher subscription count api and internal process subscription count.
+     Two contexts, both using intra-process.
+   */
+  {
+    {
+      rclcpp::NodeOptions().use_intra_process_comms(true),
+      rclcpp::NodeOptions().context(get_new_context()).use_intra_process_comms(true)
+    },
+    {1u, 1u},
+    1,
+    "two_subscriptions_in_two_contexts_with_intraprocess_comm"
+  },
+  /*
+     Testing publisher subscription count api and internal process subscription count.
+     Two contexts, both of them not using intra-process comm.
+   */
+  {
+    {
+      rclcpp::NodeOptions().use_intra_process_comms(false),
+      rclcpp::NodeOptions().context(get_new_context()).use_intra_process_comms(false)
+    },
+    {0u, 0u},
+    1,
+    "two_subscriptions_in_two_contexts_without_intraprocess_comm"
+  }
+};
+
+std::vector<TestParameters> setRuns(const std::vector<TestParameters> & in, const size_t runs)
+{
+  std::vector<TestParameters> out = in;
+  for (auto & p : out) {
+    p.runs = runs;
+  }
+  return out;
+}
+
+INSTANTIATE_TEST_CASE_P(
+  TestWithDifferentNodeOptions, TestPublisherSubscriptionSerialized,
+  ::testing::ValuesIn(parameters),
+  ::testing::PrintToStringParamName());
+
+INSTANTIATE_TEST_CASE_P(
+  TestWithDifferentNodeOptions1000Runs, TestPublisherSubscriptionSerialized,
+  ::testing::ValuesIn(setRuns(parameters, 1000)),
+  ::testing::PrintToStringParamName());

rclcpp/test/test_serialized_message.cpp

@@ -1,154 +0,0 @@
-// Copyright 2020 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 <cstring>
-#include <memory>
-#include <string>
-#include <utility>
-
-#include "rclcpp/serialization.hpp"
-#include "rclcpp/serialized_message.hpp"
-#include "rclcpp/rclcpp.hpp"
-
-#include "rcpputils/asserts.hpp"
-
-#include "test_msgs/message_fixtures.hpp"
-#include "test_msgs/msg/basic_types.hpp"
-
-TEST(TestSerializedMessage, empty_initialize) {
-  rclcpp::SerializedMessage serialized_message;
-  EXPECT_EQ(0u, serialized_message.size());
-  EXPECT_EQ(0u, serialized_message.capacity());
-}
-
-TEST(TestSerializedMessage, initialize_with_capacity) {
-  rclcpp::SerializedMessage serialized_message(13);
-  EXPECT_EQ(0u, serialized_message.size());
-  EXPECT_EQ(13u, serialized_message.capacity());
-}
-
-TEST(TestSerializedMessage, various_constructors) {
-  const std::string content = "Hello World";
-  const auto content_size = content.size() + 1;  // accounting for null terminator
-
-  rclcpp::SerializedMessage serialized_message(content_size);
-  // manually copy some content
-  auto & rcl_handle = serialized_message.get_rcl_serialized_message();
-  std::memcpy(rcl_handle.buffer, content.c_str(), content.size());
-  rcl_handle.buffer[content.size()] = '\0';
-  rcl_handle.buffer_length = content_size;
-  EXPECT_STREQ(content.c_str(), reinterpret_cast<char *>(rcl_handle.buffer));
-  EXPECT_EQ(content_size, serialized_message.capacity());
-
-  // Copy Constructor
-  rclcpp::SerializedMessage other_serialized_message(serialized_message);
-  EXPECT_EQ(content_size, other_serialized_message.capacity());
-  EXPECT_EQ(content_size, other_serialized_message.size());
-  auto & other_rcl_handle = other_serialized_message.get_rcl_serialized_message();
-  EXPECT_STREQ(
-    reinterpret_cast<char *>(rcl_handle.buffer),
-    reinterpret_cast<char *>(other_rcl_handle.buffer));
-
-  // Move Constructor
-  rclcpp::SerializedMessage yet_another_serialized_message(std::move(other_serialized_message));
-  auto & yet_another_rcl_handle = yet_another_serialized_message.get_rcl_serialized_message();
-  EXPECT_TRUE(nullptr == other_rcl_handle.buffer);
-  EXPECT_EQ(0u, other_serialized_message.capacity());
-  EXPECT_EQ(0u, other_serialized_message.size());
-  EXPECT_TRUE(nullptr != yet_another_rcl_handle.buffer);
-  EXPECT_EQ(content_size, yet_another_serialized_message.size());
-  EXPECT_EQ(content_size, yet_another_serialized_message.capacity());
-}
-
-TEST(TestSerializedMessage, various_constructors_from_rcl) {
-  const std::string content = "Hello World";
-  const auto content_size = content.size() + 1;  // accounting for null terminator
-
-  auto default_allocator = rcl_get_default_allocator();
-
-  auto rcl_serialized_msg = rmw_get_zero_initialized_serialized_message();
-  auto ret = rmw_serialized_message_init(&rcl_serialized_msg, 13, &default_allocator);
-  ASSERT_EQ(RCL_RET_OK, ret);
-  // manually copy some content
-  std::memcpy(rcl_serialized_msg.buffer, content.c_str(), content.size());
-  rcl_serialized_msg.buffer[content.size()] = '\0';
-  rcl_serialized_msg.buffer_length = content_size;
-  EXPECT_EQ(13u, rcl_serialized_msg.buffer_capacity);
-
-  // Copy Constructor from rcl_serialized_message_t
-  rclcpp::SerializedMessage serialized_message(rcl_serialized_msg);
-  EXPECT_EQ(13u, serialized_message.capacity());
-  EXPECT_EQ(content_size, serialized_message.size());
-
-  // Move Constructor from rcl_serialized_message_t
-  rclcpp::SerializedMessage another_serialized_message(std::move(rcl_serialized_msg));
-  EXPECT_TRUE(nullptr == rcl_serialized_msg.buffer);
-  EXPECT_EQ(0u, rcl_serialized_msg.buffer_capacity);
-  EXPECT_EQ(0u, rcl_serialized_msg.buffer_length);
-  EXPECT_EQ(13u, another_serialized_message.capacity());
-  EXPECT_EQ(content_size, another_serialized_message.size());
-
-  // Verify that despite being fini'd, the copy is real
-  ret = rmw_serialized_message_fini(&rcl_serialized_msg);
-  ASSERT_EQ(RCUTILS_RET_INVALID_ARGUMENT, ret);  // Buffer is null, because it was moved
-  EXPECT_EQ(nullptr, rcl_serialized_msg.buffer);
-  EXPECT_EQ(0u, rcl_serialized_msg.buffer_capacity);
-  EXPECT_EQ(0u, rcl_serialized_msg.buffer_length);
-  EXPECT_EQ(13u, serialized_message.capacity());
-  EXPECT_EQ(content_size, serialized_message.size());
-
-  auto rcl_handle = serialized_message.get_rcl_serialized_message();
-  EXPECT_TRUE(nullptr != rcl_handle.buffer);
-}
-
-TEST(TestSerializedMessage, serialization) {
-  using MessageT = test_msgs::msg::BasicTypes;
-
-  rclcpp::Serialization<MessageT> serializer;
-
-  auto basic_type_ros_msgs = get_messages_basic_types();
-  for (const auto & ros_msg : basic_type_ros_msgs) {
-    // convert ros msg to serialized msg
-    rclcpp::SerializedMessage serialized_msg;
-    serializer.serialize_message(ros_msg.get(), &serialized_msg);
-
-    // convert serialized msg back to ros msg
-    MessageT deserialized_ros_msg;
-    serializer.deserialize_message(&serialized_msg, &deserialized_ros_msg);
-
-    EXPECT_EQ(*ros_msg, deserialized_ros_msg);
-  }
-}
-
-template<typename MessageT>
-void test_empty_msg_serialize()
-{
-  rclcpp::Serialization<MessageT> serializer;
-  MessageT ros_msg;
-  rclcpp::SerializedMessage serialized_msg;
-
-  serializer.serialize_message(&ros_msg, &serialized_msg);
-}
-
-template<typename MessageT>
-void test_empty_msg_deserialize()
-{
-  rclcpp::Serialization<MessageT> serializer;
-  MessageT ros_msg;
-  rclcpp::SerializedMessage serialized_msg;
-
-  serializer.deserialize_message(&serialized_msg, &ros_msg);
-}

rclcpp/test/test_subscription_options.cpp

@@ -1,88 +0,0 @@
-// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
-//
-// 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 <memory>
-#include <string>
-#include <vector>
-
-#include "rclcpp/node.hpp"
-#include "rclcpp/node_options.hpp"
-#include "rclcpp/subscription_options.hpp"
-
-using namespace std::chrono_literals;
-
-namespace
-{
-constexpr const char defaultPublishTopic[] = "/statistics";
-}
-
-class TestSubscriptionOptions : public ::testing::Test
-{
-public:
-  static void SetUpTestCase()
-  {
-    rclcpp::init(0, nullptr);
-  }
-
-protected:
-  void initialize(const rclcpp::NodeOptions & node_options = rclcpp::NodeOptions())
-  {
-    node = std::make_shared<rclcpp::Node>("test_subscription_options", node_options);
-  }
-
-  void TearDown()
-  {
-    node.reset();
-  }
-
-  rclcpp::Node::SharedPtr node;
-};
-
-TEST_F(TestSubscriptionOptions, topic_statistics_options_default_and_set) {
-  auto options = rclcpp::SubscriptionOptions();
-
-  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);
-
-  options.topic_stats_options.state = rclcpp::TopicStatisticsState::Enable;
-  options.topic_stats_options.publish_topic = "topic_statistics";
-  options.topic_stats_options.publish_period = 5min;
-
-  EXPECT_EQ(options.topic_stats_options.state, rclcpp::TopicStatisticsState::Enable);
-  EXPECT_EQ(options.topic_stats_options.publish_topic, "topic_statistics");
-  EXPECT_EQ(options.topic_stats_options.publish_period, 5min);
-}
-
-TEST_F(TestSubscriptionOptions, topic_statistics_options_node_default_mode) {
-  initialize();
-  auto subscription_options = rclcpp::SubscriptionOptions();
-
-  EXPECT_EQ(
-    subscription_options.topic_stats_options.state,
-    rclcpp::TopicStatisticsState::NodeDefault);
-  EXPECT_FALSE(
-    rclcpp::detail::resolve_enable_topic_statistics(
-      subscription_options,
-      *(node->get_node_base_interface())));
-
-  initialize(rclcpp::NodeOptions().enable_topic_statistics(true));
-  EXPECT_TRUE(
-    rclcpp::detail::resolve_enable_topic_statistics(
-      subscription_options,
-      *(node->get_node_base_interface())));
-}

rclcpp/test/test_subscription_traits.cpp

@@ -21,7 +21,6 @@
 #include "rcl/types.h"
 
 #include "rclcpp/allocator/allocator_common.hpp"
-#include "rclcpp/serialized_message.hpp"
 #include "rclcpp/subscription_traits.hpp"
 
 #include "test_msgs/msg/empty.hpp"
@@ -53,16 +52,6 @@ void not_serialized_unique_ptr_callback(
   (void) unused;
 }
 
-void rclcpp_serialized_callback_copy(rclcpp::SerializedMessage unused)
-{
-  (void) unused;
-}
-
-void rclcpp_serialized_callback_shared_ptr(std::shared_ptr<rclcpp::SerializedMessage> unused)
-{
-  (void) unused;
-}
-
 TEST(TestSubscriptionTraits, is_serialized_callback) {
   // Test regular functions
   auto cb1 = &serialized_callback_copy;
@@ -108,16 +97,6 @@ TEST(TestSubscriptionTraits, is_serialized_callback) {
   static_assert(
     rclcpp::subscription_traits::is_serialized_callback<decltype(cb7)>::value == false,
     "passing a fancy unique_ptr of test_msgs::msg::Empty is not a serialized callback");
-
-  auto cb8 = &rclcpp_serialized_callback_copy;
-  static_assert(
-    rclcpp::subscription_traits::is_serialized_callback<decltype(cb8)>::value == true,
-    "rclcpp::SerializedMessage in a first argument callback makes it a serialized callback");
-
-  auto cb9 = &rclcpp_serialized_callback_shared_ptr;
-  static_assert(
-    rclcpp::subscription_traits::is_serialized_callback<decltype(cb9)>::value == true,
-    "std::shared_ptr<rclcpp::SerializedMessage> in a callback makes it a serialized callback");
 }
 
 TEST(TestSubscriptionTraits, callback_messages) {

rclcpp/test/test_utilities.cpp

@@ -52,8 +52,8 @@ TEST(TestUtilities, init_with_args) {
 }
 
 TEST(TestUtilities, multi_init) {
-  auto context1 = std::make_shared<rclcpp::contexts::DefaultContext>();
-  auto context2 = std::make_shared<rclcpp::contexts::DefaultContext>();
+  auto context1 = std::make_shared<rclcpp::contexts::default_context::DefaultContext>();
+  auto context2 = std::make_shared<rclcpp::contexts::default_context::DefaultContext>();
 
   EXPECT_FALSE(rclcpp::ok(context1));
   EXPECT_FALSE(rclcpp::ok(context2));

rclcpp/test/topic_statistics/test_subscription_topic_statistics.cpp

@@ -1,140 +0,0 @@
-// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
-//
-// 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 <memory>
-#include <string>
-#include <vector>
-
-#include "libstatistics_collector/moving_average_statistics/types.hpp"
-
-#include "rclcpp/create_publisher.hpp"
-#include "rclcpp/node.hpp"
-#include "rclcpp/qos.hpp"
-#include "rclcpp/rclcpp.hpp"
-
-#include "rclcpp/topic_statistics/subscription_topic_statistics.hpp"
-
-#include "statistics_msgs/msg/metrics_message.hpp"
-#include "test_msgs/msg/empty.hpp"
-
-namespace
-{
-constexpr const char kTestNodeName[]{"test_sub_stats_node"};
-constexpr const char kTestSubStatsTopic[]{"/test_sub_stats_topic"};
-constexpr const char kTestTopicStatisticsTopic[]{"/test_topic_statistics_topic"};
-constexpr const uint64_t kNoSamples{0};
-}  // namespace
-
-using test_msgs::msg::Empty;
-using statistics_msgs::msg::MetricsMessage;
-using rclcpp::topic_statistics::SubscriptionTopicStatistics;
-using libstatistics_collector::moving_average_statistics::StatisticData;
-
-template<typename CallbackMessageT>
-class TestSubscriptionTopicStatistics : public SubscriptionTopicStatistics<CallbackMessageT>
-{
-public:
-  TestSubscriptionTopicStatistics(
-    const std::string & node_name,
-    rclcpp::Publisher<statistics_msgs::msg::MetricsMessage>::SharedPtr publisher)
-  : SubscriptionTopicStatistics<CallbackMessageT>(node_name, publisher)
-  {
-  }
-
-  virtual ~TestSubscriptionTopicStatistics() = default;
-
-  /// Exposed for testing
-  std::vector<StatisticData> get_current_collector_data() const
-  {
-    return SubscriptionTopicStatistics<CallbackMessageT>::get_current_collector_data();
-  }
-};
-
-/**
- * Empty subscriber node: used to create subscriber topic statistics requirements
- */
-class EmptySubscriber : public rclcpp::Node
-{
-public:
-  EmptySubscriber(const std::string & name, const std::string & topic)
-  : Node(name)
-  {
-    auto callback = [this](Empty::UniquePtr msg) {
-        this->receive_message(*msg);
-      };
-    subscription_ = create_subscription<Empty,
-        std::function<void(Empty::UniquePtr)>>(
-      topic,
-      rclcpp::QoS(rclcpp::KeepAll()),
-      callback);
-  }
-
-  virtual ~EmptySubscriber() = default;
-
-private:
-  void receive_message(const Empty &) const
-  {
-  }
-
-  rclcpp::Subscription<Empty>::SharedPtr subscription_;
-};
-
-/**
- * Test fixture to bring up and teardown rclcpp
- */
-class TestSubscriptionTopicStatisticsFixture : public ::testing::Test
-{
-protected:
-  void SetUp()
-  {
-    rclcpp::init(0 /* argc */, nullptr /* argv */);
-    empty_subscriber = std::make_shared<EmptySubscriber>(
-      kTestNodeName,
-      kTestSubStatsTopic);
-  }
-
-  void TearDown()
-  {
-    rclcpp::shutdown();
-    empty_subscriber.reset();
-  }
-  std::shared_ptr<EmptySubscriber> empty_subscriber;
-};
-
-TEST_F(TestSubscriptionTopicStatisticsFixture, test_manual_construction)
-{
-  // manually create publisher tied to the node
-  auto topic_stats_publisher =
-    empty_subscriber->create_publisher<MetricsMessage>(
-    kTestTopicStatisticsTopic,
-    10);
-
-  // construct the instance
-  auto sub_topic_stats = std::make_unique<TestSubscriptionTopicStatistics<Empty>>(
-    empty_subscriber->get_name(),
-    topic_stats_publisher);
-
-  using libstatistics_collector::moving_average_statistics::StatisticData;
-
-  // expect no data has been collected / no samples received
-  for (const auto & data : sub_topic_stats->get_current_collector_data()) {
-    EXPECT_TRUE(std::isnan(data.average));
-    EXPECT_TRUE(std::isnan(data.min));
-    EXPECT_TRUE(std::isnan(data.max));
-    EXPECT_TRUE(std::isnan(data.standard_deviation));
-    EXPECT_EQ(kNoSamples, data.sample_count);
-  }
-}

rclcpp_action/include/rclcpp_action/create_client.hpp

@@ -46,11 +46,11 @@ create_client(
   rclcpp::node_interfaces::NodeLoggingInterface::SharedPtr node_logging_interface,
   rclcpp::node_interfaces::NodeWaitablesInterface::SharedPtr node_waitables_interface,
   const std::string & name,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr)
 {
   std::weak_ptr<rclcpp::node_interfaces::NodeWaitablesInterface> weak_node =
     node_waitables_interface;
-  std::weak_ptr<rclcpp::CallbackGroup> weak_group = group;
+  std::weak_ptr<rclcpp::callback_group::CallbackGroup> weak_group = group;
   bool group_is_null = (nullptr == group.get());
 
   auto deleter = [weak_node, weak_group, group_is_null](Client<ActionT> * ptr)
@@ -102,7 +102,7 @@ typename Client<ActionT>::SharedPtr
 create_client(
   NodeT node,
   const std::string & name,
-  rclcpp::CallbackGroup::SharedPtr group = nullptr)
+  rclcpp::callback_group::CallbackGroup::SharedPtr group = nullptr)
 {
   return create_client<ActionT>(
     node->get_node_base_interface(),

rclcpp_components/include/rclcpp_components/component_manager.hpp

@@ -62,7 +62,7 @@ public:
 
   RCLCPP_COMPONENTS_PUBLIC
   ComponentManager(
-    std::weak_ptr<rclcpp::Executor> executor,
+    std::weak_ptr<rclcpp::executor::Executor> executor,
     std::string node_name = "ComponentManager",
     const rclcpp::NodeOptions & node_options = rclcpp::NodeOptions());
 
@@ -103,7 +103,7 @@ protected:
     std::shared_ptr<ListNodes::Response> response);
 
 private:
-  std::weak_ptr<rclcpp::Executor> executor_;
+  std::weak_ptr<rclcpp::executor::Executor> executor_;
 
   uint64_t unique_id_ {1};
   std::map<std::string, std::unique_ptr<class_loader::ClassLoader>> loaders_;