From 87471079c18cdbc2652c07658d8351cb0bd8089b Mon Sep 17 00:00:00 2001
From: yukiendo-pfr <yukiendo@pfrobotics.jp>
Date: Fri, 9 Oct 2026 11:15:48 +0900
Subject: [PATCH] Release a dropped stream's subscription or timer

---
 .../include/rclcpp_async/co_context.hpp       | 13 ++++++----
 rclcpp_async/test/test_timer_stream.cpp       | 10 ++++++++
 rclcpp_async/test/test_topic_stream.cpp       | 12 ++++++++++
 rclcpp_async/test/test_wait_for_service.cpp   | 24 +++++++++++++++++++
 4 files changed, 55 insertions(+), 4 deletions(-)

diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp
index 6ca2d46..0baf2a8 100644
--- a/rclcpp_async/include/rclcpp_async/co_context.hpp
+++ b/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_) {
diff --git a/rclcpp_async/test/test_timer_stream.cpp b/rclcpp_async/test/test_timer_stream.cpp
index 74795bc..2b2b0c1 100644
--- a/rclcpp_async/test/test_timer_stream.cpp
+++ b/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());
+}
diff --git a/rclcpp_async/test/test_topic_stream.cpp b/rclcpp_async/test/test_topic_stream.cpp
index 3d8f22a..cf19e4f 100644
--- a/rclcpp_async/test/test_topic_stream.cpp
+++ b/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);
+}
diff --git a/rclcpp_async/test/test_wait_for_service.cpp b/rclcpp_async/test/test_wait_for_service.cpp
index c61d352..3821acd 100644
--- a/rclcpp_async/test/test_wait_for_service.cpp
+++ b/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);
+}