Support required nodes (ros2#179)

* Support required nodes

Currently, the only way to specify that a given node is required for a
given `LaunchDescription` is to register an `OnProcessExit` handler for
the exit event of the required node that then emits a `Shutdown` event.
Instead of requiring this boilerplate for a common use-case, add an
`on_exit` parameter to `ExecuteProcess` that can take actions, and add a
new action called `Shutdown` that immediately causes the launched system
to shut down. This allows for the concept of a required node to be
expressed simply with:

    launch_ros.actions.Node(
        # ...
        on_exit=Shutdown()
    )

Resolve ros2#174

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Inherit from EmitEvent

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Properly align type

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Expose reason

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

launch/launch/actions/__init__.py

@@ -25,6 +25,7 @@
 from .push_launch_configurations import PushLaunchConfigurations
 from .register_event_handler import RegisterEventHandler
 from .set_launch_configuration import SetLaunchConfiguration
+from .shutdown_action import Shutdown
 from .timer_action import TimerAction
 from .unregister_event_handler import UnregisterEventHandler
 from .unset_launch_configuration import UnsetLaunchConfiguration
@@ -41,6 +42,7 @@
     'PushLaunchConfigurations',
     'RegisterEventHandler',
     'SetLaunchConfiguration',
+    'Shutdown',
     'TimerAction',
     'UnregisterEventHandler',
     'UnsetLaunchConfiguration',

launch/launch/actions/execute_process.py

@@ -23,13 +23,15 @@
 import threading
 import traceback
 from typing import Any  # noqa: F401
+from typing import Callable
 from typing import cast
 from typing import Dict
 from typing import Iterable
 from typing import List
 from typing import Optional
 from typing import Text
 from typing import Tuple  # noqa: F401
+from typing import Union
 
 from osrf_pycommon.process_utils import async_execute_process
 from osrf_pycommon.process_utils import AsyncSubprocessProtocol
@@ -40,6 +42,7 @@
 from ..action import Action
 from ..event import Event
 from ..event_handler import EventHandler
+from ..event_handlers import OnProcessExit
 from ..event_handlers import OnShutdown
 from ..events import Shutdown
 from ..events.process import matches_action
@@ -86,6 +89,10 @@ def __init__(
         prefix: Optional[SomeSubstitutionsType] = None,
         output: Optional[Text] = None,
         log_cmd: bool = False,
+        on_exit: Optional[Union[
+            SomeActionsType,
+            Callable[[ProcessExited, LaunchContext], Optional[SomeActionsType]]
+        ]] = None,
         **kwargs
     ) -> None:
         """
@@ -162,6 +169,7 @@ def __init__(
         :param: log_cmd if True, prints the final cmd before executing the
             process, which is useful for debugging when substitutions are
             involved.
+        :param: on_exit list of actions to execute upon process exit.
         """
         super().__init__(**kwargs)
         self.__cmd = [normalize_to_list_of_substitutions(x) for x in cmd]
@@ -190,6 +198,7 @@ def __init__(
                 )
             )
         self.__log_cmd = log_cmd
+        self.__on_exit = on_exit
 
         self.__process_event_args = None  # type: Optional[Dict[Text, Any]]
         self._subprocess_protocol = None  # type: Optional[Any]
@@ -436,7 +445,7 @@ async def __execute_process(self, context: LaunchContext) -> None:
 
         returncode = await self._subprocess_protocol.complete
         if returncode == 0:
-            _logger.info('process[{}]: process has finished cleanly'.format(name, pid))
+            _logger.info('process[{}]: process has finished cleanly'.format(name))
         else:
             _logger.error("process[{}] process has died [pid {}, exit code {}, cmd '{}'].".format(
                 name, pid, returncode, ' '.join(cmd)
@@ -474,6 +483,10 @@ def execute(self, context: LaunchContext) -> Optional[List['Action']]:
             OnShutdown(
                 on_shutdown=self.__on_shutdown,
             ),
+            OnProcessExit(
+                target_action=self,
+                on_exit=self.__on_exit,
+            ),
         ]
         for event_handler in event_handlers:
             context.register_event_handler(event_handler)

launch/launch/actions/shutdown_action.py

@@ -0,0 +1,45 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Module for the Shutdown action."""
+
+import logging
+from typing import Text
+
+from .emit_event import EmitEvent
+from ..events import Shutdown as ShutdownEvent
+from ..events.process import ProcessExited
+from ..launch_context import LaunchContext
+
+_logger = logging.getLogger(name='launch')
+
+
+class Shutdown(EmitEvent):
+    """Action that shuts down a launched system by emitting Shutdown when executed."""
+
+    def __init__(self, *, reason: Text = 'reason not given', **kwargs):
+        super().__init__(event=ShutdownEvent(reason=reason), **kwargs)
+
+    def execute(self, context: LaunchContext):
+        """Execute the action."""
+        try:
+            event = context.locals.event
+        except AttributeError:
+            event = None
+
+        if isinstance(event, ProcessExited):
+            _logger.info('process[{}] was required: shutting down launched system'.format(
+                event.process_name))
+
+        super().execute(context)

launch/test/launch/actions/test_shutdown_action.py

@@ -0,0 +1,57 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the Shutdown action class."""
+
+from launch import LaunchContext
+from launch.actions import Shutdown
+from launch.conditions import IfCondition
+from launch.events import Shutdown as ShutdownEvent
+
+
+def test_shutdown_execute():
+    """Test the execute (or visit) of the Shutdown class."""
+    action = Shutdown()
+    context = LaunchContext()
+    assert context._event_queue.qsize() == 0
+    assert action.visit(context) is None
+    assert context._event_queue.qsize() == 1
+    event = context._event_queue.get_nowait()
+    assert isinstance(event, ShutdownEvent)
+
+
+def test_shutdown_execute_conditional():
+    """Test the conditional execution (or visit) of the Shutdown class."""
+    true_action = Shutdown(condition=IfCondition('True'))
+    false_action = Shutdown(condition=IfCondition('False'))
+    context = LaunchContext()
+
+    assert context._event_queue.qsize() == 0
+    assert false_action.visit(context) is None
+    assert context._event_queue.qsize() == 0
+    assert true_action.visit(context) is None
+    assert context._event_queue.qsize() == 1
+    event = context._event_queue.get_nowait()
+    assert isinstance(event, ShutdownEvent)
+
+
+def test_shutdown_reason():
+    """Test the execute (or visit) of a Shutdown class that has a reason."""
+    action = Shutdown(reason='test reason')
+    context = LaunchContext()
+    assert action.visit(context) is None
+    assert context._event_queue.qsize() == 1
+    event = context._event_queue.get_nowait()
+    assert isinstance(event, ShutdownEvent)
+    assert event.reason == 'test reason'

test_launch_ros/test/test_launch_ros/actions/test_node.py

@@ -20,6 +20,7 @@
 
 from launch import LaunchDescription
 from launch import LaunchService
+from launch.actions import Shutdown
 from launch.substitutions import EnvironmentVariable
 import launch_ros.actions.node
 import yaml
@@ -85,6 +86,26 @@ def test_launch_node_with_remappings(self):
         for i in range(2):
             assert expanded_remappings[i] == ('chatter', 'new_chatter')
 
+    def test_launch_required_node(self):
+        # This node will never exit on its own, it'll keep publishing forever.
+        long_running_node = launch_ros.actions.Node(
+            package='demo_nodes_py', node_executable='talker_qos', output='screen',
+            node_namespace='my_ns',
+        )
+
+        # This node will exit after publishing a single message. It is required, so we
+        # tie on_exit to the Shutdown action which means that, once it exits, it should
+        # bring down the whole launched system, including the above node that will never
+        # exit on its own.
+        required_node = launch_ros.actions.Node(
+            package='demo_nodes_py', node_executable='talker_qos', output='screen',
+            node_namespace='my_ns2', arguments=['--number_of_cycles', '1'],
+            on_exit=Shutdown()
+        )
+
+        # If the on_exit functionality or Shutdown action breaks, this will never return.
+        self._assert_launch_no_errors([required_node, long_running_node])
+
     def test_create_node_with_invalid_remappings(self):
         """Test creating a node with invalid remappings."""
         self._assert_type_error_creating_node(