Support component nodes by changing CoContext to accept Node& instead of SharedPtr (#23)

Author: otamachan

README.md

@@ -51,14 +51,14 @@ colcon build --packages-select rclcpp_async
 rclcpp_async::Task<void> run(rclcpp_async::CoContext & ctx)
 {
   co_await ctx.sleep(std::chrono::seconds(1));
-  RCLCPP_INFO(ctx.node()->get_logger(), "Hello from coroutine!");
+  RCLCPP_INFO(ctx.node().get_logger(), "Hello from coroutine!");
 }
 
 int main(int argc, char * argv[])
 {
   rclcpp::init(argc, argv);
   auto node = std::make_shared<rclcpp::Node>("hello");
-  rclcpp_async::CoContext ctx(node);
+  rclcpp_async::CoContext ctx(*node);
 
   auto task = ctx.create_task(run(ctx));
 
@@ -110,7 +110,7 @@ Task<void> listen(CoContext & ctx, std::string topic)
     if (!msg.has_value()) {
       break;
     }
-    RCLCPP_INFO(ctx.node()->get_logger(), "Heard: %s", (*msg)->data.c_str());
+    RCLCPP_INFO(ctx.node().get_logger(), "Heard: %s", (*msg)->data.c_str());
   }
 }
 ```
@@ -133,12 +133,12 @@ using SetBool = std_srvs::srv::SetBool;
 
 Task<void> call_service(CoContext & ctx)
 {
-  auto client = ctx.node()->create_client<SetBool>("set_bool");
+  auto client = ctx.node().create_client<SetBool>("set_bool");
 
   // Wait for the service to become available
   auto wait_result = co_await ctx.wait_for_service(client, std::chrono::seconds(10));
   if (!wait_result.ok()) {
-    RCLCPP_ERROR(ctx.node()->get_logger(), "Service not available");
+    RCLCPP_ERROR(ctx.node().get_logger(), "Service not available");
     co_return;
   }
 
@@ -147,7 +147,7 @@ Task<void> call_service(CoContext & ctx)
   req->data = true;
 
   auto resp = co_await ctx.send_request<SetBool>(client, req);
-  RCLCPP_INFO(ctx.node()->get_logger(), "Response: %s", resp->message.c_str());
+  RCLCPP_INFO(ctx.node().get_logger(), "Response: %s", resp->message.c_str());
 }
 ```
 
@@ -163,7 +163,7 @@ Task<void> tick(CoContext & ctx)
 
   while (true) {
     co_await timer->next();
-    RCLCPP_INFO(ctx.node()->get_logger(), "Tick %d", count++);
+    RCLCPP_INFO(ctx.node().get_logger(), "Tick %d", count++);
   }
 }
 ```
@@ -273,7 +273,7 @@ Send a goal and iterate over feedback with `GoalStream`:
 ```cpp
 Task<void> send_action(CoContext & ctx)
 {
-  auto client = rclcpp_action::create_client<Fibonacci>(ctx.node(), "fibonacci");
+  auto client = rclcpp_action::create_client<Fibonacci>(&ctx.node(), "fibonacci");
 
   auto wait_result = co_await ctx.wait_for_action(client, std::chrono::seconds(10));
   if (!wait_result.ok()) { co_return; }
@@ -293,13 +293,13 @@ Task<void> send_action(CoContext & ctx)
       break;
     }
     auto & seq = (*feedback)->sequence;
-    RCLCPP_INFO(ctx.node()->get_logger(), "Feedback: last=%d", seq.back());
+    RCLCPP_INFO(ctx.node().get_logger(), "Feedback: last=%d", seq.back());
   }
 
   // Get final result
   auto result = stream->result();
   auto & seq = result.result->sequence;
-  RCLCPP_INFO(ctx.node()->get_logger(), "Result: last=%d", seq.back());
+  RCLCPP_INFO(ctx.node().get_logger(), "Result: last=%d", seq.back());
 }
 ```
 
@@ -313,7 +313,7 @@ rclcpp_async::Event event(ctx);
 // Task 1: wait for the event
 auto waiter = ctx.create_task([&]() -> Task<void> {
   co_await event.wait();
-  RCLCPP_INFO(ctx.node()->get_logger(), "Event received!");
+  RCLCPP_INFO(ctx.node().get_logger(), "Event received!");
 });
 
 // Task 2: signal the event
@@ -340,7 +340,7 @@ rclcpp_async::Mutex mutex(ctx);
 Task<void> critical_section(CoContext & ctx, Mutex & mutex, const std::string & name)
 {
   co_await mutex.lock();
-  RCLCPP_INFO(ctx.node()->get_logger(), "%s: acquired lock", name.c_str());
+  RCLCPP_INFO(ctx.node().get_logger(), "%s: acquired lock", name.c_str());
   co_await ctx.sleep(std::chrono::seconds(1));
   mutex.unlock();
 }
@@ -364,7 +364,7 @@ Task<void> run(CoContext & ctx)
   // Sync lookup (latest) -- returns nullopt if not yet available
   auto opt = tf.lookup_transform("map", "base_link");
   if (opt) {
-    RCLCPP_INFO(ctx.node()->get_logger(), "x=%.2f", opt->transform.translation.x);
+    RCLCPP_INFO(ctx.node().get_logger(), "x=%.2f", opt->transform.translation.x);
   }
 
   // Async lookup -- suspends until the transform becomes available
@@ -403,7 +403,7 @@ auto task = ctx.create_task([&]() -> Task<void> {
     if (!val.has_value()) {
       break;                       // Channel closed
     }
-    RCLCPP_INFO(ctx.node()->get_logger(), "Received: %d", *val);
+    RCLCPP_INFO(ctx.node().get_logger(), "Received: %d", *val);
   }
 });
 ```
@@ -483,14 +483,14 @@ Task<void> run(CoContext & ctx)
 ```cpp
 Task<void> run(CoContext & ctx)
 {
-  auto client = ctx.node()->create_client<SetBool>("set_bool");
+  auto client = ctx.node().create_client<SetBool>("set_bool");
 
   // Race an awaitable against a timeout
   auto req = std::make_shared<SetBool::Request>();
   req->data = true;
   auto resp = co_await ctx.wait_for(ctx.send_request<SetBool>(client, req), 5s);
   if (resp.ok()) {
-    RCLCPP_INFO(ctx.node()->get_logger(), "Response: %s", resp.value->message.c_str());
+    RCLCPP_INFO(ctx.node().get_logger(), "Response: %s", resp.value->message.c_str());
   }
 }
 ```
@@ -546,7 +546,7 @@ Task<void> run(CoContext & ctx)
   try {
     co_await ctx.sleep(std::chrono::seconds(10));
   } catch (const CancelledException &) {
-    RCLCPP_INFO(ctx.node()->get_logger(), "Cancelled!");
+    RCLCPP_INFO(ctx.node().get_logger(), "Cancelled!");
   }
 }
 ```
@@ -579,7 +579,7 @@ See [`nested_demo.cpp`](rclcpp_async_example/src/nested_demo.cpp) for a full dem
 | `when_all(awaitables...)` | *awaitable* `tuple<Ts...>` | Await all tasks/awaitables concurrently |
 | `when_any(awaitables...)` | *awaitable* `variant<Ts...>` | Race tasks/awaitables, return first result |
 | `post(fn)` | `void` | Post a callback to the executor thread (thread-safe) |
-| `node()` | `Node::SharedPtr` | Access the underlying node |
+| `node()` | `Node&` | Access the underlying node |
 
 ### TfBuffer
 

rclcpp_async/include/rclcpp_async/co_context.hpp

@@ -116,22 +116,25 @@ class DrainWaitable : public rclcpp::Waitable
 class CoContext
 {
   std::shared_ptr<DrainWaitable> drain_;
-  rclcpp::Node::SharedPtr node_;
+  rclcpp::Node & node_;
 
 public:
-  explicit CoContext(rclcpp::Node::SharedPtr node) : node_(std::move(node))
+  explicit CoContext(rclcpp::Node & node) : node_(node)
   {
     drain_ = std::make_shared<DrainWaitable>();
-    node_->get_node_waitables_interface()->add_waitable(drain_, nullptr);
+    node_.get_node_waitables_interface()->add_waitable(drain_, nullptr);
   }
 
+  CoContext(const CoContext &) = delete;
+  CoContext & operator=(const CoContext &) = delete;
+
   // Post a callback to be executed on the executor thread (thread-safe).
   void post(std::function<void()> fn) { drain_->post(std::move(fn)); }
 
   // Resume a coroutine directly. Call only from the executor thread.
   void resume(std::coroutine_handle<> h) { h.resume(); }
 
-  rclcpp::Node::SharedPtr node() { return node_; }
+  rclcpp::Node & node() { return node_; }
 
   template <typename T>
   [[nodiscard]] Task<T> create_task(Task<T> task)
@@ -182,7 +185,7 @@ class CoContext
   std::shared_ptr<TopicStream<MsgT>> subscribe(const std::string & topic, const rclcpp::QoS & qos)
   {
     auto stream = std::make_shared<TopicStream<MsgT>>(*this);
-    stream->sub_ = node_->template create_subscription<MsgT>(
+    stream->sub_ = node_.template create_subscription<MsgT>(
       topic, qos, [s = stream](typename MsgT::SharedPtr msg) {
         if (s->closed_) {
           return;
@@ -224,13 +227,13 @@ class CoContext
           srv->send_response(*req_id, response);
         }(cb, std::move(service_handle), std::move(request_id), std::move(request));
       };
-    return node_->template create_service<ServiceT>(name, std::move(handler));
+    return node_.template create_service<ServiceT>(name, std::move(handler));
   }
 
   std::shared_ptr<TimerStream> create_timer(std::chrono::nanoseconds period)
   {
     auto stream = std::make_shared<TimerStream>(*this);
-    stream->timer_ = node_->create_wall_timer(period, [s = stream, this]() {
+    stream->timer_ = node_.create_wall_timer(period, [s = stream, this]() {
       if (s->waiter_) {
         auto w = s->waiter_;
         s->waiter_ = nullptr;
@@ -286,7 +289,9 @@ class CoContext
     using GoalHandleT = rclcpp_action::ServerGoalHandle<ActionT>;
 
     return rclcpp_action::create_server<ActionT>(
-      node_, name, std::forward<GoalCbT>(goal_callback), std::forward<CancelCbT>(cancel_callback),
+      node_.get_node_base_interface(), node_.get_node_clock_interface(),
+      node_.get_node_logging_interface(), node_.get_node_waitables_interface(), name,
+      std::forward<GoalCbT>(goal_callback), std::forward<CancelCbT>(cancel_callback),
       [cb = std::forward<CallbackT>(callback)](
         const std::shared_ptr<GoalHandleT> goal_handle) mutable {
         [](CallbackT & cb, std::shared_ptr<GoalHandleT> gh) -> SpawnedTask {
@@ -377,7 +382,7 @@ inline void SleepAwaiter::await_suspend(std::coroutine_handle<> h)
     ctx.resume(h);
   };
 
-  timer = ctx.node()->create_wall_timer(duration, [finish]() { finish(); });
+  timer = ctx.node().create_wall_timer(duration, [finish]() { finish(); });
 
   register_cancel(
     cancel_cb_, token, ctx, h, [this]() { return done; },

rclcpp_async/include/rclcpp_async/task.hpp

@@ -129,8 +129,21 @@ struct Task
   {
     o.handle = nullptr;
   }
-  Task & operator=(Task &&) = delete;
+  Task & operator=(Task && o) noexcept
+  {
+    if (this != &o) {
+      if (handle) {
+        handle.destroy();
+      }
+      handle = o.handle;
+      started_ = o.started_;
+      parent_cancel_cb_ = std::move(o.parent_cancel_cb_);
+      o.handle = nullptr;
+    }
+    return *this;
+  }
   Task(const Task &) = delete;
+  Task() noexcept : handle(nullptr) {}
 
 private:
   explicit Task(std::coroutine_handle<promise_type> h) : handle(h) {}
@@ -231,8 +244,21 @@ struct Task<void>
   {
     o.handle = nullptr;
   }
-  Task & operator=(Task &&) = delete;
+  Task & operator=(Task && o) noexcept
+  {
+    if (this != &o) {
+      if (handle) {
+        handle.destroy();
+      }
+      handle = o.handle;
+      started_ = o.started_;
+      parent_cancel_cb_ = std::move(o.parent_cancel_cb_);
+      o.handle = nullptr;
+    }
+    return *this;
+  }
   Task(const Task &) = delete;
+  Task() noexcept : handle(nullptr) {}
 
 private:
   explicit Task(std::coroutine_handle<promise_type> h) : handle(h) {}

rclcpp_async/include/rclcpp_async/tf_buffer.hpp

@@ -59,10 +59,10 @@ class TfBuffer
   explicit TfBuffer(CoContext & ctx)
   : ctx_(ctx),
     tf_node_(std::make_shared<rclcpp::Node>(
-      "_tf_listener", ctx.node()->get_namespace(),
+      "_tf_listener", ctx.node().get_namespace(),
       rclcpp::NodeOptions().start_parameter_services(false).start_parameter_event_publisher(
         false))),
-    buffer_(std::make_shared<tf2_ros::Buffer>(ctx.node()->get_clock()))
+    buffer_(std::make_shared<tf2_ros::Buffer>(ctx.node().get_clock()))
   {
     auto cb = [this](tf2_msgs::msg::TFMessage::ConstSharedPtr msg) {
       on_tf_message(std::move(msg), false);

rclcpp_async/test/benchmark/bench_concurrent.cpp

@@ -130,7 +130,7 @@ int main(int argc, char ** argv)
   rclcpp::init(argc, argv);
 
   auto node = std::make_shared<rclcpp::Node>("bench_concurrent_node");
-  auto ctx = std::make_unique<CoContext>(node);
+  auto ctx = std::make_unique<CoContext>(*node);
   rclcpp::executors::SingleThreadedExecutor executor;
   executor.add_node(node);
 

rclcpp_async/test/benchmark/bench_pubsub.cpp

@@ -50,7 +50,7 @@ int main(int argc, char ** argv)
   rclcpp::init(argc, argv);
 
   auto node = std::make_shared<rclcpp::Node>("bench_pubsub_node");
-  auto ctx = std::make_unique<CoContext>(node);
+  auto ctx = std::make_unique<CoContext>(*node);
   rclcpp::executors::SingleThreadedExecutor executor;
   executor.add_node(node);
 

rclcpp_async/test/benchmark/bench_service.cpp

@@ -46,7 +46,7 @@ int main(int argc, char ** argv)
   rclcpp::init(argc, argv);
 
   auto node = std::make_shared<rclcpp::Node>("bench_service_node");
-  auto ctx = std::make_unique<CoContext>(node);
+  auto ctx = std::make_unique<CoContext>(*node);
   rclcpp::executors::SingleThreadedExecutor executor;
   executor.add_node(node);
 

rclcpp_async/test/benchmark/bench_sleep.cpp

@@ -92,7 +92,7 @@ int main(int argc, char ** argv)
   rclcpp::init(argc, argv);
 
   auto node = std::make_shared<rclcpp::Node>("bench_sleep_node");
-  auto ctx = std::make_unique<CoContext>(node);
+  auto ctx = std::make_unique<CoContext>(*node);
   rclcpp::executors::SingleThreadedExecutor executor;
   executor.add_node(node);
 

rclcpp_async/test/test_chained_goals.cpp

@@ -45,7 +45,7 @@ class ChainedGoalsTest : public ::testing::Test
   void SetUp() override
   {
     node_ = std::make_shared<rclcpp::Node>("test_chained_goals_node");
-    ctx_ = std::make_unique<CoContext>(node_);
+    ctx_ = std::make_unique<CoContext>(*node_);
     executor_.add_node(node_);
     action_client_ = rclcpp_action::create_client<Fibonacci>(node_, "test_chained_action");
   }

rclcpp_async/test/test_channel.cpp

@@ -32,7 +32,7 @@ class ChannelTest : public ::testing::Test
   void SetUp() override
   {
     node_ = std::make_shared<rclcpp::Node>("test_channel_node");
-    ctx_ = std::make_unique<CoContext>(node_);
+    ctx_ = std::make_unique<CoContext>(*node_);
     executor_.add_node(node_);
   }