Writing publishers and subscribers in rclpy

In lesson 5 you explored a running ROS 2 system from the command line. Now you will write nodes of your own in Python with rclpy, the ROS 2 client library for Python. You will build a publisher that sends messages on a timer, a subscriber that reacts to them, and then a genuinely useful robot node that watches a laser scanner and refuses to let the robot drive forward into an obstacle. Along the way you will learn how rclpy runs your code through callbacks and executors, how to choose quality-of-service settings in code, how to log, how to measure timing and how to structure a node so its logic can be tested without a robot.

The listings in this lesson need a ROS 2 installation to run, so try them on a computer with ROS 2 Lyrical or Jazzy installed. Lesson 8 shows how to put them into a proper package; for now you can run a single file with python3 after sourcing ROS 2.

Anatomy of a node

An rclpy node is a Python class that inherits from rclpy.node.Node. Its constructor calls the parent constructor with the node's name, then creates the publishers, subscriptions and timers the node needs. The node does nothing until it is spun: rclpy.spin hands control to an executor, which waits for events, such as a timer expiring or a message arriving, and calls the matching function, called a callback. A main function initializes rclpy, creates the node, spins it and cleans up when the program is stopped. Here is a complete publisher that sends a greeting twice a second.

import rclpy
from rclpy.node import Node
from rclpy.executors import ExternalShutdownException
from std_msgs.msg import String


class Talker(Node):
    def __init__(self):
        super().__init__('talker')
        self.pub = self.create_publisher(String, 'chatter', 10)
        self.count = 0
        self.timer = self.create_timer(0.5, self.on_timer)  # 2 Hz

    def on_timer(self):
        msg = String()
        msg.data = f'hello {self.count}'
        self.pub.publish(msg)
        self.get_logger().info(f'sent: {msg.data}')
        self.count += 1


def main(args=None):
    rclpy.init(args=args)
    node = Talker()
    try:
        rclpy.spin(node)
    except (KeyboardInterrupt, ExternalShutdownException):
        pass
    finally:
        node.destroy_node()
        rclpy.try_shutdown()


if __name__ == '__main__':
    main()

Read it from the bottom up. main initializes the ROS 2 client library, creates a Talker and spins it until you press Ctrl+C or the system shuts down, at which point the try statement catches the resulting exception and the finally block destroys the node and shuts rclpy down. In the constructor, create_publisher takes the message type, the topic name and a queue depth of 10, which also selects the default QoS: reliable, volatile, keep last 10. create_timer asks the executor to call on_timer every 0.5 seconds. Each time it fires, the callback builds a String message, fills its data field, publishes it and logs what it sent.

Note what is absent: there is no loop and no sleep. You never write while True in a ROS 2 node. Instead you declare what should happen when events occur, and the executor runs your callbacks at the right times. This event-driven style is what lets one process handle several timers and subscriptions at once.

Subscribing

A subscriber is just as short. create_subscription takes the message type, the topic name, the callback to run for each message and a QoS setting. The callback receives the message as its only argument. The class below goes into a file with the same imports and main function as the talker, with Listener in place of Talker.

class Listener(Node):
    def __init__(self):
        super().__init__('listener')
        self.sub = self.create_subscription(
            String, 'chatter', self.on_msg, 10)

    def on_msg(self, msg):
        self.get_logger().info(f'heard: {msg.data}')

Run the talker in one terminal and the listener in another, both sourced, and the listener prints each greeting about two times a second. Start two listeners and both receive every message; stop the talker and the listeners simply wait. Run ros2 topic echo /chatter alongside them and it receives the same messages too. Because the default reliable, volatile QoS is used on both sides, they connect. If you start the listener after the talker has already sent some messages, it receives only the new ones: with volatile durability, nothing is stored for late joiners.

A useful node: obstacle stop

Now for a node that does real work. A person drives a robot with a joystick or keyboard, which publishes Twist commands on teleop_cmd. The obstacle_stop node passes those commands through to cmd_vel, where the base driver listens, but it also subscribes to the laser scan and, if anything is closer than 0.4 m in a cone straight ahead, it zeroes the forward speed while still allowing the robot to turn or reverse away. This is the heart of the capstone project in lesson 12.

Good practice is to put the decision logic in a plain Python function that knows nothing about ROS. The function below takes the list of ranges and the scan's starting angle and angle step, as found in every LaserScan message, and returns the nearest valid range within 20 degrees of straight ahead. Because it is plain Python, you can run it and test it anywhere.

import math


def front_min(ranges, angle_min, angle_increment,
              half_width=math.radians(20)):
    """Smallest valid range within +/- half_width of straight ahead."""
    first = round((-half_width - angle_min) / angle_increment)
    last = round((half_width - angle_min) / angle_increment)
    first, last = max(first, 0), min(last, len(ranges) - 1)
    window = [r for r in ranges[first:last + 1] if math.isfinite(r)]
    return min(window) if window else math.inf


# a 360-beam scan, one beam per degree, starting at -180 degrees
ranges = [3.0] * 360
ranges[175] = 0.35          # obstacle 5 degrees to the right
ranges[260] = 0.20          # close object far to the left, ignored
ranges[181] = math.inf      # a beam with no return
d = front_min(ranges, math.radians(-180), math.radians(1))
print(f"nearest in front: {d:.2f} m")
print("STOP" if d < 0.4 else "GO")

nearest in front: 0.35 m
STOP

With the logic tested, the node itself is mostly wiring. It uses the sensor-data QoS profile for the scan, because most laser drivers publish with best-effort reliability and a reliable subscriber would receive nothing, as lesson 5 explained.

import math
import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from geometry_msgs.msg import Twist
from sensor_msgs.msg import LaserScan

# front_min() from the previous listing goes here


class ObstacleStop(Node):
    def __init__(self):
        super().__init__('obstacle_stop')
        self.stop_distance = 0.4
        self.blocked = False
        self.create_subscription(LaserScan, 'scan', self.on_scan,
                                 qos_profile_sensor_data)
        self.create_subscription(Twist, 'teleop_cmd', self.on_cmd, 10)
        self.pub = self.create_publisher(Twist, 'cmd_vel', 10)

    def on_scan(self, scan):
        d = front_min(scan.ranges, scan.angle_min,
                      scan.angle_increment)
        self.blocked = d < self.stop_distance

    def on_cmd(self, cmd):
        if self.blocked and cmd.linear.x > 0.0:
            cmd.linear.x = 0.0          # allow turning and reversing
            self.get_logger().warning('obstacle ahead',
                                      throttle_duration_sec=1.0)
        self.pub.publish(cmd)

The node keeps one piece of state, whether the way ahead is blocked, updated by the scan callback and read by the command callback. The warning uses throttle_duration_sec so that it is logged at most once a second instead of flooding the console at the command rate. Hard-coding the 0.4 m distance is a weakness that lesson 7 fixes with a parameter.

Callbacks, executors and threads

By default rclpy.spin uses a single-threaded executor, which runs one callback at a time, in the order events arrive. That is simple and safe: two callbacks in the same node never run at once, so they can share state such as self.blocked without locks. But it means a slow callback delays everything else. If on_scan took half a second, commands would pile up and the robot would react to obstacles half a second late. The rule is to keep callbacks short: do the work, publish the result, return. Never call time.sleep or wait for user input inside a callback.

When a node genuinely needs concurrency, for example a callback that waits for a service response, rclpy offers a multi-threaded executor and callback groups, which control which callbacks may run in parallel. A mutually exclusive group never runs two of its callbacks at once; a reentrant group allows it. Lyrical adds an AsyncNode class that lets callbacks use Python's asyncio and await operations such as service calls. Start with the single-threaded executor and reach for these tools only when you have measured a need.

Timers, clocks and timing

Timers in ROS 2 use the node's clock, which normally follows the computer's clock but can follow simulated time instead when the use_sim_time parameter is true, as lesson 11 explains. Get the current time with self.get_clock().now(), not time.time(), so that your node works in both real and simulated time. A timer period is the reciprocal of the rate you want: 0.05 s for 20 Hz, 0.02 s for 50 Hz. ROS 2 timers schedule against deadlines, so they do not drift the way the naive loop in lesson 3 did, but each callback can still start a little early or late. The variation is called jitter, and it is worth measuring for any node that controls motion.

# Arrival times (s) of ten messages from a nominal 20 Hz timer
t = [0.000, 0.050, 0.101, 0.149, 0.200, 0.262,
     0.300, 0.350, 0.399, 0.450]

periods = [b - a for a, b in zip(t, t[1:])]
mean = sum(periods) / len(periods)
worst = max(abs(p - 0.050) for p in periods)
print(f"mean period {mean * 1000:.1f} ms "
      f"({1 / mean:.1f} Hz)")
print(f"worst deviation {worst * 1000:.1f} ms")
print("periods (ms):", [round(p * 1000) for p in periods])

mean period 50.0 ms (20.0 Hz)
worst deviation 12.0 ms
periods (ms): [50, 51, 48, 51, 62, 38, 50, 49, 51]

The average rate is exactly 20 Hz, yet one interval was 62 ms and the next 38 ms: the timer caught up after a late callback. A controller tuned for a 50 ms step that suddenly sees 62 ms behaves slightly differently, which rarely matters at these rates but becomes important for fast control. Common causes of jitter are other callbacks in the same executor taking too long, the computer being overloaded and Python's memory management pausing briefly. Measuring is easy: record timestamps with the node's clock in the callback, or run ros2 topic hz on the published topic, which reports the minimum, maximum and standard deviation of the period.

Stopping safely

When you press Ctrl+C, rclpy's default signal handling shuts down ROS 2 communication and spin returns or raises an exception. By then the node can no longer reliably publish, so a final zero-velocity command sent in the finally block may never reach the base. This is why well-designed base drivers do not depend on a last stop message: they apply a command timeout, stopping the motors if no new command arrives within, say, half a second. When you write a node that commands motion, publish commands continuously at a steady rate rather than once, and make sure the driver you send them to has a timeout. If the publisher crashes, the robot stops.