ROS 2 Message Design: Custom Messages, Interfaces & The Schema Hell

The Problem: Standard Messages Aren’t Enough

ROS 2 comes with standard messages:

geometry_msgs/Twist - velocity commands
sensor_msgs/LaserScan - lidar data
nav_msgs/Odometry - odometry

But your custom sensor publishes:

Proprietary temperature readings
Custom calibration format
Weird button states

Do you squeeze them into std_msgs/Float64? Into a topic name? Into a message field you hack?

No. You create a custom message. But then you hit the gotchas.

Part 1: Creating Custom Messages

1.1 Message File Structure

Create a .msg file in your package:

my_robot_interfaces/
├── msg/
│   ├── Temperature.msg
│   ├── RobotState.msg
│   └── SensorReading.msg
├── srv/
│   ├── Calibrate.srv
│   └── Reset.srv
├── action/
│   └── MoveArm.action
├── CMakeLists.txt
├── package.xml
└── src/

1.2 Simple Message

# msg/Temperature.msg
float64 celsius
float64 fahrenheit
string sensor_name

Usage:

from my_robot_interfaces.msg import Temperature
 
msg = Temperature()
msg.celsius = 25.0
msg.fahrenheit = 77.0
msg.sensor_name = "thermal_camera"
 
self.publisher.publish(msg)

1.3 Nested Messages (Composition)

# msg/SensorReading.msg
std_msgs/Header header      # Standard header with timestamp
Temperature temperature     # Nested custom message
float32[] raw_data         # Array of floats

In your code:

from my_robot_interfaces.msg import SensorReading, Temperature
from std_msgs.msg import Header
 
reading = SensorReading()
reading.header.stamp = self.get_clock().now().to_msg()
reading.header.frame_id = "sensor_frame"
 
reading.temperature = Temperature()
reading.temperature.celsius = 23.5
reading.temperature.fahrenheit = 74.3
reading.temperature.sensor_name = "main"
 
reading.raw_data = [1.2, 3.4, 5.6, 7.8]
 
self.publisher.publish(reading)

1.4 Complex Message with Various Types

# msg/RobotState.msg
# State information
string state_name                     # String
int32 state_id                        # Integer
float64 confidence                    # Float

# Position
geometry_msgs/Point position          # Nested standard message
geometry_msgs/Quaternion orientation  # Nested standard message

# Timestamps
builtin_interfaces/Time timestamp     # Time message
duration time_elapsed                 # Duration type

# Arrays
float32[] sensor_values               # Variable-length array
int8[5] calibration_data              # Fixed-size array

# Boolean
bool is_error

Part 2: Building and Using Custom Messages

2.1 CMakeLists.txt Configuration

cmake_minimum_required(VERSION 3.8)
project(my_robot_interfaces)
 
find_package(ament_cmake REQUIRED)
find_package(std_msgs REQUIRED)
find_package(geometry_msgs REQUIRED)
find_package(rosidl_default_generators REQUIRED)
 
rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Temperature.msg"
  "msg/SensorReading.msg"
  "msg/RobotState.msg"
  "srv/Calibrate.srv"
  "action/MoveArm.action"
  DEPENDENCIES std_msgs geometry_msgs
)
 
ament_package()

2.2 In Python Node

from my_robot_interfaces.msg import RobotState, Temperature
from geometry_msgs.msg import Point, Quaternion
from std_msgs.msg import Header
 
class StatePublisher(Node):
    def __init__(self):
        super().__init__('state_publisher')
        
        self.publisher = self.create_publisher(
            RobotState,
            'robot_state',
            10
        )
        
        self.create_timer(0.1, self.publish_state)
    
    def publish_state(self):
        msg = RobotState()
        msg.state_name = "moving"
        msg.state_id = 1
        msg.confidence = 0.95
        
        msg.position = Point(x=1.0, y=2.0, z=0.0)
        msg.orientation = Quaternion(x=0.0, y=0.0, z=0.0, w=1.0)
        
        msg.timestamp = self.get_clock().now().to_msg()
        
        msg.sensor_values = [1.2, 3.4, 5.6]
        msg.calibration_data = [10, 20, 30, 40, 50]
        
        msg.is_error = False
        
        self.publisher.publish(msg)

Part 3: Services and Actions with Custom Interfaces

3.1 Custom Service

# srv/Calibrate.srv
# Request
int32 calibration_mode
float64 target_value
---
# Response
bool success
string error_message
float64 actual_value

Server:

from my_robot_interfaces.srv import Calibrate
 
def calibrate_callback(self, request, response):
    """Handle calibration request"""
    self.get_logger().info(
        f'Calibrating with mode {request.calibration_mode}'
    )
    
    # Perform calibration
    try:
        actual = self.perform_calibration(request.calibration_mode)
        response.success = True
        response.actual_value = actual
    except Exception as e:
        response.success = False
        response.error_message = str(e)
    
    return response
 
def perform_calibration(self, mode):
    # Simulated calibration
    return 23.5

Client:

def request_calibration(self):
    request = Calibrate.Request()
    request.calibration_mode = 1
    request.target_value = 25.0
    
    future = self.client.call_async(request)
    rclpy.spin_until_future_complete(self, future)
    
    if future.result():
        response = future.result()
        if response.success:
            self.get_logger().info(f'Calibration succeeded: {response.actual_value}')
        else:
            self.get_logger().error(f'Calibration failed: {response.error_message}')

3.2 Custom Action

# action/MoveArm.action
# Goal
float64 target_x
float64 target_y
float64 target_z
---
# Result
bool success
string error
---
# Feedback
float32 percent_complete
geometry_msgs/Point current_position

Server:

from my_robot_interfaces.action import MoveArm
from geometry_msgs.msg import Point
 
def execute_callback(self, goal_handle):
    goal = goal_handle.request
    
    for i in range(0, 101, 10):
        if goal_handle.is_cancel_requested:
            goal_handle.canceled()
            return MoveArm.Result()
        
        # Simulate movement
        current_x = goal.target_x * (i / 100)
        current_y = goal.target_y * (i / 100)
        current_z = goal.target_z * (i / 100)
        
        feedback = MoveArm.Feedback()
        feedback.percent_complete = i
        feedback.current_position = Point(x=current_x, y=current_y, z=current_z)
        goal_handle.publish_feedback(feedback)
        
        time.sleep(0.1)
    
    goal_handle.succeed()
    result = MoveArm.Result()
    result.success = True
    result.error = ""
    return result

Part 4: The Gotchas

Gotcha 1: Field Name Reserved Words

# ❌ WRONG: Using Python reserved words
int32 import
int32 class
int32 return

# ✅ CORRECT: Use suffixes or different names
int32 import_flag
int32 class_type
int32 return_value

Gotcha 2: Inconsistent Message Versions

# Version 1: msg/Data.msg
float64 value

# You publish with version 1
publisher.publish(Data(value=1.0))

# Later, someone adds a field:
# Version 2: msg/Data.msg
float64 value
string metadata  # NEW FIELD

# Old code breaks if metadata is required

Solution: Always provide defaults or handle gracefully.

Gotcha 3: Array Memory Management

# Large array in message
float32[1000000] data  # 4MB per message!

# Publishing at 100 Hz = 400 MB/s
# Your network and memory explode

Solution: Use reasonable array sizes or split across multiple messages.

Gotcha 4: Missing std_msgs/Header

# ❌ WRONG: No timestamp or frame info
float64 value

# ✅ CORRECT: Include header for time synchronization
std_msgs/Header header
float64 value

This matters for:

Time synchronization across nodes
Debugging (which message is this?)
Frame transforms

Gotcha 5: Backward Compatibility Hell

# Your old message
string robot_name
float64 velocity

# You add a field for new feature
string robot_name
float64 velocity
string new_feature  # BREAKS OLD NODES!

# Old nodes expect 2 fields, get 3
# Deserialization fails

Solution: Plan your message structure carefully. Add with defaults.

Part 5: Message Design Best Practices

5.1 Always Include Header

std_msgs/Header header
# Other fields...

Why:

Timestamp for synchronization
Frame ID for spatial data
Sequence number for debugging

5.2 Use Standard Messages When Possible

# ✅ GOOD: Uses standard messages
geometry_msgs/Point position
geometry_msgs/Vector3 velocity
sensor_msgs/Imu imu_data

# ❌ BAD: Reinventing the wheel
float64 x
float64 y
float64 z
float64 vx
float64 vy
float64 vz

5.3 Meaningful Field Names

# ❌ BAD: Unclear
float64 val1
float64 val2
int32 flag

# ✅ GOOD: Clear intent
float64 temperature_celsius
float64 pressure_pascals
int32 error_code

5.4 Add Comments to Complex Fields

# msg/RobotState.msg
std_msgs/Header header

# State machine: 0=idle, 1=moving, 2=error
int32 state_id

# Confidence in state identification (0-1)
float32 state_confidence

# Current position in world frame (meters)
geometry_msgs/Point position

5.5 Use Enums for State

# ❌ WRONG: Magic numbers
int32 state  # What do 0, 1, 2 mean?

# ✅ BETTER: Define constants
# msg/RobotState.msg
int32 STATE_IDLE = 0
int32 STATE_MOVING = 1
int32 STATE_ERROR = 2

int32 state

Part 6: Real-World Pattern - Complete Robot State Message

# msg/RobotState.msg
std_msgs/Header header

# State identification
int8 STATE_IDLE = 0
int8 STATE_MOVING = 1
int8 STATE_CHARGING = 2
int8 STATE_ERROR = 3
int8 current_state

# Position and orientation
geometry_msgs/Point position
geometry_msgs/Quaternion orientation

# Velocity (base frame)
geometry_msgs/Twist velocity

# Sensor data
sensor_msgs/BatteryState battery

# Errors
string[] error_messages

# System diagnostics
int32 uptime_seconds
float32 cpu_load
float32 memory_percent

Publisher:

from my_robot_interfaces.msg import RobotState
from geometry_msgs.msg import Point, Quaternion, Twist
from sensor_msgs.msg import BatteryState
 
class RobotStatePublisher(Node):
    def __init__(self):
        super().__init__('robot_state_pub')
        self.pub = self.create_publisher(RobotState, 'robot_state', 10)
        self.create_timer(0.1, self.publish_state)
    
    def publish_state(self):
        msg = RobotState()
        msg.header.stamp = self.get_clock().now().to_msg()
        msg.header.frame_id = "world"
        
        msg.current_state = RobotState.STATE_MOVING
        
        msg.position = Point(x=1.0, y=2.0, z=0.0)
        msg.orientation = Quaternion(x=0, y=0, z=0, w=1)
        
        msg.velocity = Twist()
        msg.velocity.linear.x = 0.5
        msg.velocity.angular.z = 0.1
        
        msg.battery = BatteryState()
        msg.battery.percentage = 0.85
        
        msg.error_messages = []
        
        msg.uptime_seconds = 3600
        msg.cpu_load = 0.45
        msg.memory_percent = 62.3
        
        self.pub.publish(msg)

Debugging Custom Messages

# See what fields a message has
ros2 msg show my_robot_interfaces/msg/RobotState
 
# Publish a message from command line
ros2 topic pub /robot_state my_robot_interfaces/msg/RobotState \
  "{header: {stamp: now, frame_id: world}, current_state: 1, position: {x: 1.0, y: 2.0, z: 0.0}}"
 
# Echo incoming messages
ros2 topic echo /robot_state

Quick Checklist

Message has std_msgs/Header for timestamps
Field names are clear and descriptive
Using standard messages where applicable
No reserved Python keywords as fields
Array sizes are reasonable
Backward compatibility considered
Comments on complex/unclear fields
Enums used for state values
CMakeLists.txt configured correctly

Key Takeaways

Custom messages organize complex data elegantly
Always include Header for time/frame info
Use standard messages when they exist
Plan for versioning - backward compatibility matters
Array sizes matter - watch your bandwidth
Comments help - future you will appreciate it

Remember: A good message design is invisible. A bad one haunts you for years.

Vaibhav Shende

M.Eng. in Robotics @ University of Maryland

ROS 2 Message Design: Custom Messages, Interfaces & The Schema Hell