Nodes, Topics, and Services

Learning Objectives

By the end of this lesson, you will be able to:

1. Create and manage ROS 2 nodes in Python and C++
2. Implement publishers and subscribers for topic-based communication
3. Build service servers and clients for request/response patterns
4. Use ROS 2 command-line tools for debugging and introspection
5. Apply best practices for node design and error handling

---

ROS 2 Nodes: The Building Blocks

A node is a process that performs computation. ROS 2 applications are composed of many nodes working together.

Node Design Principles

1. Single Responsibility: Each node does one thing well
2. Loose Coupling: Nodes communicate via topics/services, not function calls
3. Configurability: Use parameters for runtime configuration
4. Error Handling: Handle failures gracefully, log errors
5. Resource Management: Clean up on shutdown

Creating a Minimal Node (Python)

import rclpy
from rclpy.node import Node

class MinimalNode(Node):
    def __init__(self):
        # Initialize with node name
        super().__init__('minimal_node')

        # Log startup
        self.get_logger().info('MinimalNode has been started')

def main(args=None):
    # Initialize ROS 2
    rclpy.init(args=args)

    # Create node
    node = MinimalNode()

    # Keep node running
    rclpy.spin(node)

    # Cleanup
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Run the node:

python3 minimal_node.py

# Or via ros2 run (after packaging)
ros2 run my_package minimal_node

---

Publishers and Subscribers

Publisher: Sending Data

A publisher sends messages on a topic.

Example: Robot Pose Publisher

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import PoseStamped
import math

class PosePublisher(Node):
    def __init__(self):
        super().__init__('pose_publisher')

        # Create publisher
        self.publisher_ = self.create_publisher(
            PoseStamped,           # Message type
            '/robot/pose',         # Topic name
            10                     # Queue size (QoS depth)
        )

        # Publish at 10 Hz
        self.timer = self.create_timer(0.1, self.publish_pose)
        self.counter = 0

        self.get_logger().info('PosePublisher started')

    def publish_pose(self):
        msg = PoseStamped()

        # Header
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.header.frame_id = 'map'

        # Pose (circular motion)
        t = self.counter * 0.1
        msg.pose.position.x = 5.0 * math.cos(t)
        msg.pose.position.y = 5.0 * math.sin(t)
        msg.pose.position.z = 0.0

        # Publish
        self.publisher_.publish(msg)
        self.get_logger().info(f'Published pose: x={msg.pose.position.x:.2f}, y={msg.pose.position.y:.2f}')

        self.counter += 1

def main():
    rclpy.init()
    node = PosePublisher()
    rclpy.spin(node)
    rclpy.shutdown()

Subscriber: Receiving Data

A subscriber listens to messages on a topic.

Example: Pose Subscriber

import rclpy
from rclpy.node import Node
from geometry_msgs.msg import PoseStamped

class PoseSubscriber(Node):
    def __init__(self):
        super().__init__('pose_subscriber')

        # Create subscription
        self.subscription = self.create_subscription(
            PoseStamped,
            '/robot/pose',
            self.pose_callback,  # Callback function
            10
        )

        self.get_logger().info('PoseSubscriber started, waiting for messages...')

    def pose_callback(self, msg):
        x = msg.pose.position.x
        y = msg.pose.position.y

        # Calculate distance from origin
        distance = (x**2 + y**2)**0.5

        self.get_logger().info(f'Received pose at ({x:.2f}, {y:.2f}), distance: {distance:.2f}m')

def main():
    rclpy.init()
    node = PoseSubscriber()
    rclpy.spin(node)
    rclpy.shutdown()

Test the pub/sub pair:

# Terminal 1: Run publisher
ros2 run my_package pose_publisher

# Terminal 2: Run subscriber
ros2 run my_package pose_subscriber

# Terminal 3: Inspect topic
ros2 topic list
ros2 topic info /robot/pose
ros2 topic echo /robot/pose
ros2 topic hz /robot/pose

---

Combining Publisher and Subscriber

Many nodes both subscribe AND publish (e.g., sensor fusion, filters).

Example: LiDAR Filter Node

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import LaserScan

class LaserFilter(Node):
    def __init__(self):
        super().__init__('laser_filter')

        # Subscribe to raw laser scan
        self.subscription = self.create_subscription(
            LaserScan,
            '/scan_raw',
            self.scan_callback,
            10
        )

        # Publish filtered laser scan
        self.publisher_ = self.create_publisher(
            LaserScan,
            '/scan_filtered',
            10
        )

        # Parameters
        self.declare_parameter('min_range', 0.5)  # Ignore readings < 0.5m
        self.declare_parameter('max_range', 10.0) # Ignore readings > 10m

        self.get_logger().info('LaserFilter started')

    def scan_callback(self, msg_in):
        # Get parameters
        min_range = self.get_parameter('min_range').value
        max_range = self.get_parameter('max_range').value

        # Create output message (copy input)
        msg_out = LaserScan()
        msg_out.header = msg_in.header
        msg_out.angle_min = msg_in.angle_min
        msg_out.angle_max = msg_in.angle_max
        msg_out.angle_increment = msg_in.angle_increment
        msg_out.time_increment = msg_in.time_increment
        msg_out.scan_time = msg_in.scan_time
        msg_out.range_min = min_range
        msg_out.range_max = max_range

        # Filter ranges
        msg_out.ranges = [
            r if (min_range <= r <= max_range) else float('inf')
            for r in msg_in.ranges
        ]

        # Publish filtered scan
        self.publisher_.publish(msg_out)

        valid_readings = sum(1 for r in msg_out.ranges if r != float('inf'))
        self.get_logger().info(f'Filtered scan: {valid_readings}/{len(msg_out.ranges)} valid readings')

def main():
    rclpy.init()
    node = LaserFilter()
    rclpy.spin(node)
    rclpy.shutdown()

Run with custom parameters:

ros2 run my_package laser_filter --ros-args \
  -p min_range:=1.0 \
  -p max_range:=8.0

---

Services: Request/Response Communication

Services are for one-time interactions where a client requests something and waits for a response.

Service Server

Example: String Reversal Service

import rclpy
from rclpy.node import Node
from example_interfaces.srv import SetBool  # Using standard service
# For custom service, you'd import your own .srv definition

class StringReverseService(Node):
    def __init__(self):
        super().__init__('string_reverse_server')

        # Create service
        self.srv = self.create_service(
            SetBool,                     # Service type
            'reverse_string',            # Service name
            self.reverse_callback        # Callback function
        )

        self.get_logger().info('StringReverseService ready')

    def reverse_callback(self, request, response):
        # Process request
        input_str = "True" if request.data else "False"
        reversed_str = input_str[::-1]

        # Build response
        response.success = True
        response.message = f'Reversed: {reversed_str}'

        self.get_logger().info(f'Request: {input_str} -> Response: {reversed_str}')

        return response

def main():
    rclpy.init()
    node = StringReverseService()
    rclpy.spin(node)
    rclpy.shutdown()

Service Client

Example: Service Caller

import rclpy
from rclpy.node import Node
from example_interfaces.srv import SetBool
import sys

class StringReverseClient(Node):
    def __init__(self):
        super().__init__('string_reverse_client')

        # Create client
        self.client = self.create_client(SetBool, 'reverse_string')

        # Wait for service to be available
        while not self.client.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('Waiting for reverse_string service...')

    def send_request(self, value):
        # Create request
        request = SetBool.Request()
        request.data = value

        # Call service asynchronously
        self.future = self.client.call_async(request)

        self.get_logger().info(f'Sent request: {value}')

def main():
    rclpy.init()
    node = StringReverseClient()

    # Send request
    node.send_request(True)

    # Wait for response
    rclpy.spin_until_future_complete(node, node.future)

    # Get result
    response = node.future.result()
    node.get_logger().info(f'Received response: {response.message}')

    rclpy.shutdown()

Test the service:

# Terminal 1: Start server
ros2 run my_package string_reverse_server

# Terminal 2: Call service via CLI
ros2 service list
ros2 service type /reverse_string
ros2 service call /reverse_string example_interfaces/srv/SetBool "{data: true}"

# Or run client node
ros2 run my_package string_reverse_client

---

Real-World Example: Robot Arm Controller

Let's combine topics and services in a practical robot arm controller.

Architecture:

• Topic: /joint_states (publishes current joint positions)
• Service: /set_joint_target (sets target joint positions)

import rclpy
from rclpy.node import Node
from sensor_msgs.msg import JointState
from example_interfaces.srv import Trigger
import math

class RobotArmController(Node):
    def __init__(self):
        super().__init__('robot_arm_controller')

        # Joint state publisher
        self.joint_pub = self.create_publisher(JointState, '/joint_states', 10)

        # Calibration service
        self.calibrate_srv = self.create_service(
            Trigger,
            '/calibrate_arm',
            self.calibrate_callback
        )

        # Internal state
        self.joint_positions = [0.0, 0.0, 0.0, 0.0, 0.0]  # 5 joints
        self.is_calibrated = False

        # Publish joint states at 50 Hz
        self.timer = self.create_timer(0.02, self.publish_joint_states)

        self.get_logger().info('RobotArmController started (uncalibrated)')

    def publish_joint_states(self):
        msg = JointState()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.name = ['joint_1', 'joint_2', 'joint_3', 'joint_4', 'joint_5']
        msg.position = self.joint_positions

        self.joint_pub.publish(msg)

    def calibrate_callback(self, request, response):
        self.get_logger().info('Calibrating arm...')

        # Simulate calibration (move to home position)
        self.joint_positions = [0.0] * 5
        self.is_calibrated = True

        response.success = True
        response.message = 'Arm calibrated successfully'

        self.get_logger().info('Calibration complete')

        return response

def main():
    rclpy.init()
    node = RobotArmController()
    rclpy.spin(node)
    rclpy.shutdown()

Test the arm controller:

# Start controller
ros2 run my_package robot_arm_controller

# Monitor joint states
ros2 topic echo /joint_states

# Call calibration service
ros2 service call /calibrate_arm example_interfaces/srv/Trigger

---

ROS 2 Command-Line Tools

Topic Introspection

# List all topics
ros2 topic list

# Show topic type
ros2 topic info /scan

# Echo messages
ros2 topic echo /scan

# Measure frequency
ros2 topic hz /scan

# Publish from CLI
ros2 topic pub /cmd_vel geometry_msgs/msg/Twist "{linear: {x: 0.5}, angular: {z: 0.1}}"

Service Introspection

# List services
ros2 service list

# Show service type
ros2 service type /set_motor_speed

# Call service
ros2 service call /calibrate_arm example_interfaces/srv/Trigger

Node Introspection

# List running nodes
ros2 node list

# Show node info
ros2 node info /laser_filter

# See computation graph
rqt_graph

---

Best Practices

1. Use Namespaces

Organize topics/services under namespaces:

# Good: Namespaced
self.publisher_ = self.create_publisher(Image, '/camera/left/image', 10)

# Bad: Flat namespace
self.publisher_ = self.create_publisher(Image, '/left_camera_image', 10)

2. Handle Uninitialized Data

class SafeSubscriber(Node):
    def __init__(self):
        super().__init__('safe_subscriber')
        self.latest_pose = None
        self.subscription = self.create_subscription(
            PoseStamped, '/pose', self.pose_callback, 10
        )

    def pose_callback(self, msg):
        self.latest_pose = msg

    def get_current_pose(self):
        if self.latest_pose is None:
            self.get_logger().warn('No pose received yet!')
            return None
        return self.latest_pose

3. Use Timers, Not Loops

# Good: Timer-based publishing
self.timer = self.create_timer(0.1, self.publish_data)

# Bad: While loop
while rclpy.ok():
    self.publish_data()
    time.sleep(0.1)

4. Clean Shutdown

def main():
    rclpy.init()
    node = MyNode()

    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        node.get_logger().info('Shutting down gracefully...')
    finally:
        node.destroy_node()
        rclpy.shutdown()

---

Hands-On Exercise: Teleoperation System

Goal: Build a keyboard teleoperation system for a mobile robot.

Requirements:

1. Keyboard Node: Read arrow keys, publish Twist to /cmd_vel
2. Motor Controller Node: Subscribe to /cmd_vel, control motors
3. Emergency Stop Service: /emergency_stop service to halt robot

Starter Code:

# keyboard_node.py
import rclpy
from rclpy.node import Node
from geometry_msgs.msg import Twist
import sys, tty, termios

class KeyboardTeleop(Node):
    def __init__(self):
        super().__init__('keyboard_teleop')
        self.publisher_ = self.create_publisher(Twist, '/cmd_vel', 10)
        self.get_logger().info('Use arrow keys to move (q to quit)')

    def get_key(self):
        fd = sys.stdin.fileno()
        old_settings = termios.tcgetattr(fd)
        try:
            tty.setraw(fd)
            key = sys.stdin.read(1)
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
        return key

    def run(self):
        while rclpy.ok():
            key = self.get_key()
            msg = Twist()

            if key == '\x1b[A':  # Up arrow
                msg.linear.x = 0.5
            elif key == '\x1b[B':  # Down arrow
                msg.linear.x = -0.5
            elif key == '\x1b[C':  # Right arrow
                msg.angular.z = -0.5
            elif key == '\x1b[D':  # Left arrow
                msg.angular.z = 0.5
            elif key == 'q':
                break

            self.publisher_.publish(msg)
            self.get_logger().info(f'Published: linear={msg.linear.x}, angular={msg.angular.z}')

# Implement motor_controller.py yourself!

Test:

ros2 run my_package keyboard_teleop
ros2 topic echo /cmd_vel

---

Key Takeaways

✅ Nodes are processes: Each does one thing well ✅ Topics for streaming: Sensor data, robot state ✅ Services for requests: One-shot commands, queries ✅ Use CLI tools: Debug with ros2 topic, ros2 service, ros2 node ✅ Follow best practices: Namespaces, error handling, timers

---

What's Next?

Now that you can build nodes, let's package and distribute them:

👉 Building ROS 2 Packages — Create reusable packages

---

Additional Resources

• ROS 2 rclpy API
• ROS 2 Standard Messages
• Writing ROS 2 Nodes Best Practices