Skip to main content

Subscribers and Callbacks

Learning Objectives​

After completing this chapter, you will be able to:

  • Implement subscribers that handle messages efficiently and safely
  • Design callback functions that follow ROS 2 best practices
  • Handle different Quality of Service (QoS) settings in subscribers
  • Create robust message processing pipelines with proper error handling

Introduction​

Subscribers in ROS 2 are the counterpart to publishers and are responsible for receiving and processing messages published on topics. In humanoid robotics applications, subscribers are critical for processing sensor data, receiving commands, and reacting to system events. The effectiveness of a robot's behavior depends heavily on how well it processes incoming messages, making robust subscriber design essential for Physical AI applications.

Effective subscribers not only process messages correctly but also handle various edge cases like message loss, timing issues, and system failures. This chapter explores the implementation of subscribers that can operate reliably in dynamic robotic environments while maintaining real-time performance requirements.

Core Concepts​

Subscribers in ROS 2 are created with specific Quality of Service (QoS) settings that must be compatible with the publishers on the same topic. The callback functions associated with subscribers must be designed to handle messages efficiently without blocking other operations.

Message Callbacks​

Callback functions are executed when messages arrive. They must be designed to avoid blocking operations since ROS 2 uses a single-threaded executor by default for callbacks. Long-running operations should be handled in separate threads or through other mechanisms.

QoS Compatibility​

Subscribers must have compatible QoS settings with publishers on the same topic. While not all settings need to match exactly, they must be compatible to establish communication. Understanding these compatibility rules is important for debugging connectivity issues.

Message Processing Pipelines​

Complex robotic systems often require multiple processing steps for each message. Designing efficient processing pipelines that can handle high message rates is a key skill in robotics software development.

Hands-on Examples​

Let's implement subscribers with various processing strategies:

#!/usr/bin/env python3

"""
Basic subscriber example
"""

import rclpy
from rclpy.node import Node
from rclpy.qos import QoSProfile, ReliabilityPolicy, HistoryPolicy
from std_msgs.msg import String, Float32
from sensor_msgs.msg import JointState


class BasicSubscriber(Node):

    def __init__(self):
        super().__init__('basic_subscriber')
        
        # Create QoS profile for subscriptions
        qos_profile = QoSProfile(
            reliability=ReliabilityPolicy.BEST_EFFORT,
            history=HistoryPolicy.KEEP_LAST,
            depth=10
        )
        
        # Subscribe to simple string messages
        self.subscription = self.create_subscription(
            String,
            'robot_status',
            self.listener_callback,
            qos_profile
        )
        self.subscription  # prevent unused variable warning
        
        # Subscribe to floating point values
        self.float_subscription = self.create_subscription(
            Float32,
            'sensor_value',
            self.float_callback,
            qos_profile
        )
        
        # Subscribe to joint states
        self.joint_subscription = self.create_subscription(
            JointState,
            'joint_states',
            self.joint_state_callback,
            qos_profile
        )
        
        self.get_logger().info('Basic subscriber initialized')

    def listener_callback(self, msg):
        self.get_logger().info(f'Received: "{msg.data}"')

    def float_callback(self, msg):
        self.get_logger().info(f'Received sensor value: {msg.data:.2f}')

    def joint_state_callback(self, msg):
        if len(msg.name) > 0 and len(msg.position) > 0:
            self.get_logger().info(
                f'Received joint states for {len(msg.name)} joints, '
                f'first joint: {msg.name[0]} = {msg.position[0]:.3f}'
            )
        else:
            self.get_logger().info('Received joint state message with no data')


def main(args=None):
    rclpy.init(args=args)

    basic_subscriber = BasicSubscriber()

    try:
        rclpy.spin(basic_subscriber)
    except KeyboardInterrupt:
        basic_subscriber.get_logger().info('Subscriber stopped by user')
    finally:
        basic_subscriber.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

Expected Output:

[INFO] [1678882844.123456789] [basic_subscriber]: Basic subscriber initialized
[INFO] [1678882844.123456789] [basic_subscriber]: Received: "humanoid_robot status report: 0"
[INFO] [1678882844.123456789] [basic_subscriber]: Received sensor value: 1.23
[INFO] [1678882844.123456789] [basic_subscriber]: Received joint states for 12 joints, first joint: left_hip = 0.100
[INFO] [1678882845.123456789] [advanced_subscriber]: Advanced subscriber with buffering initialized
[INFO] [1678882846.123456789] [advanced_subscriber]: Buffer size: 1, Queue size: 0

Exercises​

Complete the following exercises to reinforce your understanding:

  1. Message Filtering: Create a subscriber that filters messages based on content

    • Implement a subscriber that only processes messages meeting certain criteria
    • Add filtering logic to ignore outdated messages
    • Create a statistics tracker for filtered messages
    • Test with messages that meet and don't meet criteria

  2. Threading Strategy: Implement a subscriber that uses multiple threads for processing

    • Create a subscriber with a dedicated thread for each message type
    • Implement proper synchronization between threads
    • Measure performance differences between single and multi-threaded approaches
    • Add thread safety mechanisms to prevent data races

Common Pitfalls and Solutions​

  • Pitfall 1: Blocking callbacks - Performing long operations in callback functions, blocking other callbacks
    • Solution: Move long operations to background threads or use async processing
  • Pitfall 2: Memory issues - Storing unlimited messages in buffers without bounds
    • Solution: Use bounded buffers like collections.deque with a maxlen parameter
  • Pitfall 3: Race conditions - Unsafe access to shared data in callbacks
    • Solution: Use locks, queues, or other thread-safe mechanisms for shared data
  • Pitfall 4: Unhandled exceptions - Exceptions in callbacks that crash the node
    • Solution: Wrap callback code in try-catch blocks with proper error handling

Summary​

  • Subscribers receive messages from topics using callback functions
  • Callbacks should be lightweight and avoid blocking operations
  • QoS settings must be compatible between publishers and subscribers
  • Message buffering and background processing can improve performance
  • Proper error handling is essential for robust subscriber design

Further Reading​