Release a dropped stream's subscription or timer

rclcpp_async/include/rclcpp_async/co_context.hpp

@@ -220,10 +220,13 @@ class CoContext : public Executor
   std::shared_ptr<TopicStream<MsgT>> subscribe(const std::string & topic, const rclcpp::QoS & qos)
   {
     auto stream = std::make_shared<TopicStream<MsgT>>(*this, qos.depth());
+    // The subscription is owned by the stream, so the callback holds it weakly:
+    // dropping the stream releases the subscription with it.
     stream->sub_ = node_.template create_subscription<MsgT>(
       topic, qos,
-      [s = stream](std::shared_ptr<const MsgT> msg) {
-        if (s->closed_) {
+      [weak = std::weak_ptr(stream)](std::shared_ptr<const MsgT> msg) {
+        auto s = weak.lock();
+        if (!s || s->closed_) {
           return;
         }
         s->queue_.push(std::move(msg));
@@ -281,10 +284,12 @@ class CoContext : public Executor
   std::shared_ptr<TimerStream> create_timer(std::chrono::nanoseconds period)
   {
     auto stream = std::make_shared<TimerStream>(*this);
+    // As with subscribe(): dropping the stream stops and releases the timer.
     stream->timer_ = node_.create_wall_timer(
       period,
-      [s = stream, this]() {
-        if (s->closed_) {
+      [weak = std::weak_ptr(stream), this]() {
+        auto s = weak.lock();
+        if (!s || s->closed_) {
           return;
         }
         if (s->waiter_) {

rclcpp_async/test/test_timer_stream.cpp

@@ -169,3 +169,13 @@ TEST_F(TimerStreamTest, CancelDuringNext)
   ASSERT_TRUE(running.handle.done());
   EXPECT_TRUE(was_cancelled);
 }
+
+TEST_F(TimerStreamTest, DroppingTheStreamReleasesTheTimer)
+{
+  std::weak_ptr<TimerStream> weak;
+  {
+    auto timer = ctx_->create_timer(10ms);
+    weak = timer;
+  }
+  EXPECT_TRUE(weak.expired());
+}

rclcpp_async/test/test_topic_stream.cpp

@@ -213,3 +213,15 @@ TEST_F(TopicStreamTest, DestroyingASuspendedWaiterDetachesIt)
   spin_until_done(next);
   EXPECT_TRUE(next.handle.done());
 }
+
+TEST_F(TopicStreamTest, DroppingTheStreamUnsubscribes)
+{
+  std::weak_ptr<TopicStream<StringMsg>> weak;
+  {
+    auto stream = ctx_->subscribe<StringMsg>("dropped_stream_topic", 10);
+    weak = stream;
+    EXPECT_EQ(node_->count_subscribers("dropped_stream_topic"), 1u);
+  }
+  EXPECT_TRUE(weak.expired());
+  EXPECT_EQ(node_->count_subscribers("dropped_stream_topic"), 0u);
+}

rclcpp_async/test/test_wait_for_service.cpp

@@ -143,3 +143,27 @@ TEST_F(WaitForServiceTest, Cancel)
   ASSERT_TRUE(task.handle.done());
   EXPECT_TRUE(was_cancelled);
 }
+
+static int count_timers(rclcpp::Node & node)
+{
+  int timers = 0;
+  node.get_node_base_interface()->for_each_callback_group(
+    [&timers](const rclcpp::CallbackGroup::SharedPtr & group) {
+      group->collect_all_ptrs(
+        [](const rclcpp::SubscriptionBase::SharedPtr &) {},
+        [](const rclcpp::ServiceBase::SharedPtr &) {}, [](const rclcpp::ClientBase::SharedPtr &) {},
+        [&timers](const rclcpp::TimerBase::SharedPtr &) { ++timers; },
+        [](const rclcpp::Waitable::SharedPtr &) {});
+    });
+  return timers;
+}
+
+TEST_F(WaitForServiceTest, TimeoutLeavesNoTimerBehind)
+{
+  auto coro = [&]() -> Task<void> { co_await ctx_->wait_for_service(client_, 200ms); };
+  auto task = ctx_->create_task(coro());
+  spin_until_done(task);
+
+  ASSERT_TRUE(task.handle.done());
+  EXPECT_EQ(count_timers(*node_), 0);
+}