Feature request: Launch action to make a node wait for its predecessor to publish a topic before starting dependent nodes

[RobotBramhana]

Background

This reference to the issue opened here

Solution

Since the solution is simpler, I did not create any design docs with diagram, instead directly came up with a working PoC.
I have implemented it as in the code below. I will raise a PR once the draft is approved in terms of the design. Renaming of class name, variables and so on can be handled as part of the code review.

import asyncio
import rclpy
from rclpy.node import Node
from launch import Action
import logging
import signal

class WaitForTopic(Action):
    """Custom Action that waits for a topic to become available."""

    def __init__(self, topic_name: str):
        super().__init__()
        self.topic_name = topic_name
        self.logger = logging.getLogger('WaitForTopic')
        self.logger.setLevel(logging.INFO)

    def execute(self, context):
        """Wait for the topic to be available before continuing."""
        self.logger.info(f"Starting to wait for topic '{self.topic_name}'...")

        rclpy.init(args=None)
        self.node = rclpy.create_node("topic_waiter")

        # Create a completion future
        self.future = asyncio.get_event_loop().create_future()

        # Set up signal handler for graceful shutdown
        signal.signal(signal.SIGINT, self._shutdown_handler)

        def check_topic():
            topics = [t[0] for t in self.node.get_topic_names_and_types()]
            if self.topic_name in topics:
                self.logger.info(f"Topic '{self.topic_name}' is available.")
                if not self.future.done():
                    self.future.set_result(True)
            else:
                self.logger.info(f"Waiting for topic '{self.topic_name}'...")

        # Periodically check for the topic
        self.node.create_timer(0.5, check_topic)

        # Run the asyncio event loop until future completes
        asyncio.ensure_future(self._wait_for_future())

        context.add_completion_future(self.future)
        return []

    async def _wait_for_future(self):
        """Wait until the topic is available and clean up resources."""
        try:
            await self.future
            self.logger.info(f"Topic '{self.topic_name}' found, proceeding...")
        finally:
            self._cleanup()

    def _shutdown_handler(self, signum, frame):
        """Handle SIGINT for graceful shutdown."""
        self.logger.info("Received SIGINT, shutting down...")
        if not self.future.done():
            self.future.set_exception(asyncio.CancelledError())

    def _cleanup(self):
        """Shutdown rclpy properly and destroy the node."""
        self.node.destroy_node()
        rclpy.shutdown()

Tests

I have tested it by developing the below launch file and running demo nodes. The same will provided in the examples directory for reference. If there is any other tests to be done, kindly feel free to let me know.

import os
from launch import LaunchDescription
from launch_ros.actions import Node
from launch_ros.actions import WaitForTopic

def generate_launch_description():
    # Nodes for the camera and processing
    talker = Node(
        package='demo_nodes_cpp',
        executable='talker',
        name='talker',
    )

    listener = Node(
        package='demo_nodes_cpp',
        executable='listener',
        name='listener',
    )
    # Wait for the topic '/chatter' before launching the processing node
    wait_for_a_topic = WaitForTopic('/chatter')

    return LaunchDescription([
        talker,
        wait_for_a_topic ,  # Action that waits for the topic
        listener         # Node that will be launched after topic availability
    ])

Issue Assignment
This issue can be assigned to me.