feat: Add an option to use EventsExecutor (backport #1157)

Author: bjsowa

rosbridge_server/CMakeLists.txt

@@ -30,16 +30,36 @@ if(BUILD_TESTING)
     TIMEOUT 60
   )
 
-  add_launch_test(test/websocket/advertise_action.test.py)
-  add_launch_test(test/websocket/advertise_action_feedback.test.py)
-  add_launch_test(test/websocket/advertise_publish.test.py)
-  add_launch_test(test/websocket/advertise_service.test.py)
-  add_launch_test(test/websocket/call_service.test.py)
-  add_launch_test(test/websocket/send_action_goal.test.py)
-  add_launch_test(test/websocket/smoke.test.py)
-  add_launch_test(test/websocket/transient_local_publisher.test.py)
-  add_launch_test(test/websocket/best_effort_publisher.test.py)
-  add_launch_test(test/websocket/multiple_subscribers_raw.test.py)
+  # Run each test with both SingleThreadedExecutor and EventsExecutor
+  set(TEST_FILES
+    test/websocket/advertise_action.test.py
+    test/websocket/advertise_action_feedback.test.py
+    test/websocket/advertise_publish.test.py
+    test/websocket/advertise_service.test.py
+    test/websocket/call_service.test.py
+    test/websocket/send_action_goal.test.py
+    test/websocket/smoke.test.py
+    test/websocket/transient_local_publisher.test.py
+    test/websocket/best_effort_publisher.test.py
+    test/websocket/multiple_subscribers_raw.test.py
+  )
+
+  foreach(TEST_FILE ${TEST_FILES})
+    # Extract base name for test naming
+    get_filename_component(TEST_NAME ${TEST_FILE} NAME_WE)
+
+    # Run with SingleThreadedExecutor (default)
+    add_launch_test(${TEST_FILE}
+      TARGET ${TEST_NAME}_singlethreaded
+      ARGS "use_events_executor:=false"
+    )
+
+    # Run with EventsExecutor
+    add_launch_test(${TEST_FILE}
+      TARGET ${TEST_NAME}_events
+      ARGS "use_events_executor:=true"
+    )
+  endforeach()
 
   find_package(ament_cmake_mypy REQUIRED)
   ament_mypy()

rosbridge_server/scripts/rosbridge_websocket.py

@@ -44,6 +44,7 @@
 import rclpy
 from rcl_interfaces.msg import ParameterDescriptor
 from rclpy.executors import SingleThreadedExecutor
+from rclpy.experimental import EventsExecutor
 from rclpy.node import Node
 from rclpy.utilities import remove_ros_args
 from tornado.httpserver import HTTPServer
@@ -69,6 +70,8 @@
     ("websocket_ping_timeout", float, 30.0, "Timeout in seconds for WebSocket ping responses."),
     # Websocket handler parameters
     ("use_compression", bool, False, "Enable compression for WebSocket messages."),
+    # Executor parameters
+    ("use_events_executor", bool, False, "Use EventsExecutor instead of SingleThreadedExecutor."),
 )
 
 PROTOCOL_PARAMETERS = (
@@ -195,6 +198,11 @@ def _handle_parameters(self) -> None:
             self.get_parameter("use_compression").get_parameter_value().bool_value
         )
 
+        # Executor parameters
+        self.use_events_executor = (
+            self.get_parameter("use_events_executor").get_parameter_value().bool_value
+        )
+
     def _start_server(self) -> None:
         handlers = [(r"/", RosbridgeWebSocket), (r"", RosbridgeWebSocket)]
         if self.url_path != "/":
@@ -231,7 +239,11 @@ async def async_main() -> None:
 
     node = RosbridgeWebsocketNode()
 
-    executor = SingleThreadedExecutor()
+    if node.use_events_executor:
+        executor = EventsExecutor()
+    else:
+        executor = SingleThreadedExecutor()
+
     executor.add_node(node)
 
     spin_thread = threading.Thread(target=executor.spin)

rosbridge_server/test/websocket/common.py

@@ -7,7 +7,9 @@
 import launch_ros
 import rclpy
 from autobahn.twisted.websocket import WebSocketClientFactory, WebSocketClientProtocol
+from launch.actions import DeclareLaunchArgument
 from launch.launch_description import LaunchDescription
+from launch.substitutions import LaunchConfiguration
 from launch_testing.actions import ReadyToTest
 from rcl_interfaces.srv import GetParameters
 from rclpy.executors import SingleThreadedExecutor
@@ -49,21 +51,30 @@ def onMessage(self, payload: str, binary: bool) -> None:  # noqa: N802
         self.message_handler(payload if binary else json.loads(payload))
 
 
-def _generate_node() -> launch_ros.actions.Node:
-    return launch_ros.actions.Node(
-        executable="rosbridge_websocket",
-        package="rosbridge_server",
-        parameters=[{"port": 0}],
-    )
-
-
 def generate_test_description() -> LaunchDescription:
     """
     Generate a launch description that runs the websocket server.
 
     Re-export this from a test file and use add_launch_test() to run the test.
+    This supports parameterization via the 'use_events_executor' launch argument.
     """
-    return LaunchDescription([_generate_node(), ReadyToTest()])
+    return LaunchDescription(
+        [
+            DeclareLaunchArgument(
+                "use_events_executor",
+                default_value="false",
+                description="Use EventsExecutor instead of SingleThreadedExecutor",
+            ),
+            launch_ros.actions.Node(
+                executable="rosbridge_websocket",
+                package="rosbridge_server",
+                parameters=[
+                    {"port": 0, "use_events_executor": LaunchConfiguration("use_events_executor")}
+                ],
+            ),
+            ReadyToTest(),
+        ]
+    )
 
 
 async def get_server_port(node: Node) -> int: