Replace individual handler functions with single generic one.

Allows arbitrary signal handlers to easily be added in future without
writing new functions, DRY.

Also allows at least one signal of each type, printing a warning when
more than one is received. This is to allow in future each signal type
to be passed to the children of launch, so a user can manually escalate
the kill signals.

Signed-off-by: Cian Donovan <[email protected]>

Author: ciandonovan

launch/launch/launch_service.py

@@ -186,34 +186,25 @@ def _prepare_run_loop(self):
 
             self.__this_task = this_task
             # Setup custom signal handlers for SIGINT & SIGTERM
-            sigint_received = False
+            signals_received: dict = {}
 
-            def _on_sigint(signum):
-                nonlocal sigint_received
-                base_msg = 'user interrupted with ctrl-c (SIGINT)'
-                if not sigint_received:
+            def _on_signal(signum):
+                nonlocal signals_received
+                base_msg = 'caught ' + signal.Signals(signum).name
+                if signum not in signals_received or signals_received[signum] is not True:
+                    signals_received[signum] = True
                     self.__logger.warning(base_msg)
                     ret = self._shutdown(
-                        reason='ctrl-c (SIGINT)', due_to_sigint=True, force_sync=True
+                        reason=base_msg, due_to_sigint=True, force_sync=True
                     )
                     assert ret is None, ret
-                    sigint_received = True
                 else:
                     self.__logger.warning('{} again, ignoring...'.format(base_msg))
 
-            def _on_sigterm(signum):
-                signame = signal.Signals(signum).name
-                self.__logger.error(
-                    'user interrupted with ctrl-\\ ({}), terminating...'.format(signame))
-                # TODO(wjwwood): try to terminate running subprocesses before exiting.
-                self.__logger.error('using {} can result in orphaned processes'.format(signame))
-                self.__logger.error('make sure no processes launched are still running')
-                this_loop.call_soon(this_task.cancel)
-
             with AsyncSafeSignalManager(this_loop) as manager:
                 # Setup signal handlers
-                manager.handle(signal.SIGINT, _on_sigint)
-                manager.handle(signal.SIGTERM, _on_sigterm)
+                manager.handle(signal.SIGINT, _on_signal)
+                manager.handle(signal.SIGTERM, _on_signal)
                 # Yield asyncio loop and current task.
                 yield this_loop, this_task
         finally: