Thread-safe promise and nodiscard

CHANGELOG.md

@@ -5,20 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
-
-### Removed 
-
-- Callback-based ServiceClient::call overload
-- ServiceClient::cancel because it does not cancel the timeout timer correctly
+## 0.4.0
 
 ### Added
 
 - Support for actions
 - Multi-threaded executor support for the async/await API
 
+### Removed 
+
+- Callback-based ServiceClient::call overload
+- ServiceClient::cancel because it does not cancel the timeout timer correctly
+
 ### Fixed
 
+- allowing coroutines without suspension points (i.e. co_await) but only a single co_return
 - Correct result type implementation
 - Missing request cleanup in ServiceClientImpl::our_to_real_req_id_
 - Added missing cancellation of timeout timer in Promise destruction

icey/include/icey/icey_image_transport.hpp

@@ -23,7 +23,7 @@ namespace icey {
 
 /// Image_transport does not yet support lifecycle_nodes:
 void assert_is_not_lifecycle_node(Context &context) {
-  if (context.node_base().is_lifecycle_node())
+  if (context.node_base()->is_lifecycle_node())
     throw std::runtime_error(
         "You tried to use image_transport with a lifecycle node, unfortunately ROS Humble does not "
         "currently support image_transport with lifecycle nodes.");
@@ -51,7 +51,7 @@ struct ImageTransportSubscription
     };
     try {
       this->impl()->subscription =
-          image_transport::create_subscription(&context.node_base().as_node(), base_topic_name, cb,
+          image_transport::create_subscription(&context.node_base()->as_node(), base_topic_name, cb,
                                                transport, qos.get_rmw_qos_profile(), options);
     } catch (const image_transport::TransportLoadException &exception) {
       this->impl()->put_error(exception);
@@ -68,7 +68,7 @@ struct ImageTransportPublisher : public Stream<sensor_msgs::msg::Image::SharedPt
         context);  /// NodeBookkeeping acts a type-erasing common interface between regular Nodes
                    /// and lifecycle nodes, so we can only assert this at runtime
     image_transport::Publisher publisher = image_transport::create_publisher(
-        &context.node_base().as_node(), base_topic_name, qos.get_rmw_qos_profile());
+        &context.node_base()->as_node(), base_topic_name, qos.get_rmw_qos_profile());
     this->impl()->register_handler([publisher](const auto &new_state) {
       publisher.publish(new_state.value());  /// There can be no error
     });
@@ -101,7 +101,7 @@ struct CameraSubscription
                    /// and lifecycle nodes, so we can only assert this at runtime
     try {
       this->impl()->subscription = image_transport::create_camera_subscription(
-          &context.node_base().as_node(), base_topic_name, cb, transport,
+          &context.node_base()->as_node(), base_topic_name, cb, transport,
           qos.get_rmw_qos_profile());
     } catch (const image_transport::TransportLoadException &exception) {
       this->impl()->put_error(exception);
@@ -120,7 +120,7 @@ struct CameraPublisher
         context);  /// NodeBookkeeping acts a type-erasing common interface between regular Nodes
                    /// and lifecycle nodes, so we can only assert this at runtime
     image_transport::CameraPublisher publisher = image_transport::create_camera_publisher(
-        &context.node_base().as_node(), base_topic_name, qos.get_rmw_qos_profile());
+        &context.node_base()->as_node(), base_topic_name, qos.get_rmw_qos_profile());
     this->impl()->register_handler([publisher](const auto &new_state) {
       const auto [image_msg, camera_info_msg] = new_state.value();  /// There can be no error;
       publisher.publish(image_msg, camera_info_msg);

icey/include/icey/icey_rx.hpp

@@ -85,9 +85,6 @@ struct Weak {
 /// `AnyStream` concept. \sa AnyStream
 struct StreamTag {};
 
-template <class T>
-constexpr bool is_stream = std::is_base_of_v<StreamTag, T>;
-
 /// A stream type with any error or value type. \sa StreamTag
 template <class T>
 concept AnyStream = std::is_base_of_v<StreamTag, T>;
@@ -264,10 +261,9 @@ class Context : public ContextAsyncAwait,
                      FValidate f_validate = {}, bool ignore_override = false) {
     parameter_validators_.emplace(name, f_validate);
     add_parameter_validator_if_needed();
-    auto param = node_parameters_->declare_parameter(name, rclcpp::ParameterValue(default_value),
-                                                     parameter_descriptor, ignore_override);
-    auto param_subscription =
-        std::make_shared<rclcpp::ParameterEventHandler>(static_cast<NodeBase &>(*this));
+    auto param = this->node_base()->get_node_parameters_interface()->declare_parameter(
+        name, rclcpp::ParameterValue(default_value), parameter_descriptor, ignore_override);
+    auto param_subscription = std::make_shared<rclcpp::ParameterEventHandler>(this->node_base());
     auto cb_handle =
         param_subscription->add_parameter_callback(name, std::forward<CallbackT>(update_callback));
     parameters_.emplace(name, std::make_pair(param_subscription, cb_handle));
@@ -307,26 +303,25 @@ class Context : public ContextAsyncAwait,
     return impl;
   }
 
-  /// Get the NodeBase, i.e. the ROS node using which this Context was created.
-  NodeBase &node_base() { return static_cast<NodeBase &>(*this); }
-
 protected:
   /// Installs the validator callback
   void add_parameter_validator_if_needed() {
     if (validate_param_cb_) return;
-    this->validate_param_cb_ = node_parameters_->add_on_set_parameters_callback(
-        [this](const std::vector<rclcpp::Parameter> &parameters) {
-          rcl_interfaces::msg::SetParametersResult result;
-          for (const auto &parameter : parameters) {
-            /// We want to skip validating parameters that we didn't declare, for example here we
-            /// are getting called for parameters like "qos_overrides./tf.publisher.durability"
-            if (parameter_validators_.contains(parameter.get_name())) {
-              result.reason = parameter_validators_.at(parameter.get_name())(parameter);
-            }
-          }
-          result.successful = result.reason == "";
-          return result;
-        });
+    this->validate_param_cb_ =
+        this->node_base()->get_node_parameters_interface()->add_on_set_parameters_callback(
+            [this](const std::vector<rclcpp::Parameter> &parameters) {
+              rcl_interfaces::msg::SetParametersResult result;
+              for (const auto &parameter : parameters) {
+                /// We want to skip validating parameters that we didn't declare, for example here
+                /// we are getting called for parameters like
+                /// "qos_overrides./tf.publisher.durability"
+                if (parameter_validators_.contains(parameter.get_name())) {
+                  result.reason = parameter_validators_.at(parameter.get_name())(parameter);
+                }
+              }
+              result.successful = result.reason == "";
+              return result;
+            });
   }
 
   /// A map that stores for each parameter name some ROS entities that we need to hold to be able to
@@ -951,7 +946,7 @@ struct ParameterStream : public Stream<_Value> {
         },
         this->create_descriptor(), this->validator.validate, this->ignore_override);
     /// Set the default value
-    this->impl()->put_value(context.node_base().get_parameter<Value>(this->parameter_name));
+    this->impl()->put_value(context.node_base()->get_parameter<Value>(this->parameter_name));
   }
 
   /// Get the value. Parameters are initialized always at the beginning, so they always have a
@@ -1029,7 +1024,7 @@ struct SubscriptionStream
   SubscriptionStream(Context &context, const std::string &topic_name, const rclcpp::QoS &qos,
                      const rclcpp::SubscriptionOptions &options)
       : Base(context) {
-    this->impl()->subscription = context.node_base().create_subscription<_Message>(
+    this->impl()->subscription = context.node_base()->create_subscription<_Message>(
         topic_name,
         [impl = this->impl()](typename _Message::SharedPtr msg) { impl->put_value(msg); }, qos,
         options);
@@ -1076,7 +1071,7 @@ struct TimerStream : public Stream<size_t, Nothing, TimerImpl> {
   TimerStream() = default;
   TimerStream(Context &context, const Duration &interval, bool is_one_off_timer) : Base(context) {
     this->impl()->timer =
-        context.node_base().create_wall_timer(interval, [impl = this->impl(), is_one_off_timer]() {
+        context.node_base()->create_wall_timer(interval, [impl = this->impl(), is_one_off_timer]() {
           /// Needed as separate state as it might be reset in async/await mode
           auto cnt = impl->ticks_counter;
           impl->ticks_counter++;
@@ -1128,7 +1123,7 @@ struct PublisherStream : public Stream<_Value, Nothing, PublisherImpl<_Value>> {
                   Input *maybe_input = nullptr)
       : Base(context) {
     this->impl()->publisher =
-        context.node_base().create_publisher<Message>(topic_name, qos, publisher_options);
+        context.node_base()->create_publisher<Message>(topic_name, qos, publisher_options);
     this->impl()->register_handler(
         [impl = this->impl()](const auto &new_state) { impl->publish(new_state.value()); });
     if (maybe_input) {
@@ -1182,7 +1177,7 @@ struct TimeoutFilter
   TimeoutFilter(Context &context, Input input, const Duration &max_age,
                 bool create_extra_timer = true)
       : Base(context) {
-    auto node_clock = context.node_base().get_node_clock_interface();
+    auto node_clock = context.node_base()->get_node_clock_interface();
     rclcpp::Duration max_age_ros(max_age);
     auto check_state = [impl = this->impl(), node_clock, max_age_ros](const auto &new_state) {
       if (!new_state.has_value()) return true;
@@ -1339,8 +1334,8 @@ struct TransformSynchronizerImpl {
     /// because we must buffer every message for as long as we are waiting for a transform.
     synchronizer = std::make_shared<tf2_ros::MessageFilter<Message>>(
         *input_filter->impl(), *tf_listener->buffer_, target_frame, 0,
-        context.node_base().get_node_logging_interface(),
-        context.node_base().get_node_clock_interface(), lookup_timeout);
+        context.node_base()->get_node_logging_interface(),
+        context.node_base()->get_node_clock_interface(), lookup_timeout);
 
     synchronizer->registerCallback(&Self::on_message, this);
   }
@@ -1437,7 +1432,7 @@ void Context::declare_parameter_struct(
                                                std::string_view field_name, auto &field_value) {
     using Field = std::remove_reference_t<decltype(field_value)>;
     std::string field_name_r = name_prefix + std::string(field_name);
-    if constexpr (is_stream<Field>) {
+    if constexpr (AnyStream<Field>) {
       field_value.impl_ = this->create_stream_impl<typename Field::Impl>();
       field_value.impl()->context =
           std::enable_shared_from_this<Context>::shared_from_this();  /// First, give it the missing